更新 scada 数据清洗方法
This commit is contained in:
JIANG
@@ -2,7 +2,8 @@ from typing import List, Optional, Any
|
||||
from datetime import datetime
|
||||
from psycopg import AsyncConnection
|
||||
import pandas as pd
|
||||
import api_ex
|
||||
from api_ex.Fdataclean import clean_flow_data_df_kf
|
||||
from api_ex.Pdataclean import clean_pressure_data_df_km
|
||||
|
||||
from postgresql.scada_info import ScadaRepository as PostgreScadaRepository
|
||||
from timescaledb.schemas.realtime import RealtimeRepository
|
||||
@@ -236,87 +237,94 @@ class CompositeQueries:
|
||||
# 将列表转换为字典，以 device_id 为键
|
||||
scada_device_info_dict = {info["id"]: info for info in scada_infos}
|
||||
|
||||
# 按设备类型分组设备
|
||||
type_groups = {}
|
||||
for device_id in device_ids:
|
||||
device_info = scada_device_info_dict.get(device_id, {})
|
||||
device_type = device_info.get("type", "unknown")
|
||||
if device_type not in type_groups:
|
||||
type_groups[device_type] = []
|
||||
type_groups[device_type].append(device_id)
|
||||
# 如果 device_ids 为空，则处理所有 SCADA 设备
|
||||
if not device_ids:
|
||||
device_ids = list(scada_device_info_dict.keys())
|
||||
|
||||
# 批量处理每种类型的设备
|
||||
for device_type, ids in type_groups.items():
|
||||
if device_type not in ["pressure", "pipe_flow"]:
|
||||
continue # 跳过未知类型
|
||||
# 批量查询所有设备的数据
|
||||
data = await ScadaRepository.get_scada_field_by_id_time_range(
|
||||
timescale_conn, device_ids, start_time, end_time, "monitored_value"
|
||||
)
|
||||
|
||||
# 查询 monitored_value 数据
|
||||
data = await ScadaRepository.get_scada_field_by_id_time_range(
|
||||
timescale_conn, ids, start_time, end_time, "monitored_value"
|
||||
)
|
||||
if not data:
|
||||
return "error: fetch none scada data" # 没有数据，直接返回
|
||||
|
||||
if not data:
|
||||
continue
|
||||
# 将嵌套字典转换为 DataFrame，使用 time 作为索引
|
||||
# data 格式: {device_id: [{"time": "...", "value": ...}, ...]}
|
||||
all_records = []
|
||||
for device_id, records in data.items():
|
||||
for record in records:
|
||||
all_records.append(
|
||||
{
|
||||
"time": record["time"],
|
||||
"device_id": device_id,
|
||||
"value": record["value"],
|
||||
}
|
||||
)
|
||||
|
||||
# 将嵌套字典转换为 DataFrame，使用 time 作为索引
|
||||
# data 格式: {device_id: [{"time": "...", "value": ...}, ...]}
|
||||
all_records = []
|
||||
for device_id, records in data.items():
|
||||
for record in records:
|
||||
all_records.append(
|
||||
{
|
||||
"time": record["time"],
|
||||
"device_id": device_id,
|
||||
"value": record["value"],
|
||||
}
|
||||
)
|
||||
if not all_records:
|
||||
return "error: fetch none scada data" # 没有数据，直接返回
|
||||
|
||||
if not all_records:
|
||||
continue
|
||||
# 创建 DataFrame 并透视，使 device_id 成为列
|
||||
df_long = pd.DataFrame(all_records)
|
||||
df = df_long.pivot(index="time", columns="device_id", values="value")
|
||||
|
||||
# 创建 DataFrame 并透视，使 device_id 成为列
|
||||
df_long = pd.DataFrame(all_records)
|
||||
df = df_long.pivot(index="time", columns="device_id", values="value")
|
||||
# 根据type分类设备
|
||||
pressure_ids = [
|
||||
id
|
||||
for id in df.columns
|
||||
if scada_device_info_dict.get(id, {}).get("type") == "pressure"
|
||||
]
|
||||
flow_ids = [
|
||||
id
|
||||
for id in df.columns
|
||||
if scada_device_info_dict.get(id, {}).get("type") == "pipe_flow"
|
||||
]
|
||||
|
||||
# 确保所有请求的设备都在列中（即使没有数据）
|
||||
for device_id in ids:
|
||||
if device_id not in df.columns:
|
||||
df[device_id] = None
|
||||
|
||||
# 只保留请求的设备列
|
||||
df = df[ids]
|
||||
# 处理pressure数据
|
||||
# if pressure_ids:
|
||||
# pressure_df = df[pressure_ids]
|
||||
# # 重置索引，将 time 变为普通列
|
||||
# pressure_df = pressure_df.reset_index()
|
||||
# # 移除 time 列，准备输入给清洗方法
|
||||
# value_df = pressure_df.drop(columns=["time"])
|
||||
# # 调用清洗方法
|
||||
# cleaned_value_df = clean_pressure_data_df_km(value_df)
|
||||
# # 添加 time 列到首列
|
||||
# cleaned_df = pd.concat([pressure_df["time"], cleaned_value_df], axis=1)
|
||||
# # 将清洗后的数据写回数据库
|
||||
# for device_id in pressure_ids:
|
||||
# if device_id in cleaned_df.columns:
|
||||
# cleaned_values = cleaned_df[device_id].tolist()
|
||||
# time_values = cleaned_df["time"].tolist()
|
||||
# for i, time_str in enumerate(time_values):
|
||||
# time_dt = datetime.fromisoformat(time_str)
|
||||
# value = cleaned_values[i]
|
||||
# await ScadaRepository.update_scada_field(
|
||||
# timescale_conn,
|
||||
# time_dt,
|
||||
# device_id,
|
||||
# "cleaned_value",
|
||||
# value,
|
||||
# )
|
||||
|
||||
# 处理flow数据
|
||||
if flow_ids:
|
||||
flow_df = df[flow_ids]
|
||||
# 重置索引，将 time 变为普通列
|
||||
df = df.reset_index()
|
||||
|
||||
flow_df = flow_df.reset_index()
|
||||
# 移除 time 列，准备输入给清洗方法
|
||||
value_df = df.drop(columns=["time"])
|
||||
|
||||
value_df = flow_df.drop(columns=["time"])
|
||||
# 调用清洗方法
|
||||
if device_type == "pressure":
|
||||
cleaned_dict = api_ex.Pdataclean.clean_pressure_data_dict_km(
|
||||
value_df.to_dict(orient="list")
|
||||
)
|
||||
elif device_type == "pipe_flow":
|
||||
cleaned_dict = api_ex.Fdataclean.clean_flow_data_dict(
|
||||
value_df.to_dict(orient="list")
|
||||
)
|
||||
else:
|
||||
continue
|
||||
|
||||
# 将字典转换为 DataFrame（字典键为设备ID，值为值列表）
|
||||
cleaned_value_df = pd.DataFrame(cleaned_dict)
|
||||
|
||||
cleaned_value_df = clean_flow_data_df_kf(value_df)
|
||||
# 添加 time 列到首列
|
||||
cleaned_df = pd.concat([df["time"], cleaned_value_df], axis=1)
|
||||
|
||||
cleaned_df = pd.concat([flow_df["time"], cleaned_value_df], axis=1)
|
||||
# 将清洗后的数据写回数据库
|
||||
for device_id in ids:
|
||||
for device_id in flow_ids:
|
||||
if device_id in cleaned_df.columns:
|
||||
cleaned_values = cleaned_df[device_id].tolist()
|
||||
time_values = cleaned_df["time"].tolist()
|
||||
for i, time_str in enumerate(time_values):
|
||||
# time_str 已经是 ISO 格式字符串
|
||||
time_dt = datetime.fromisoformat(time_str)
|
||||
value = cleaned_values[i]
|
||||
await ScadaRepository.update_scada_field(
|
||||
|
||||
@@ -521,11 +521,14 @@ async def clean_scada_data(
|
||||
根据 device_ids 查询 monitored_value，清洗后更新 cleaned_value
|
||||
"""
|
||||
try:
|
||||
device_ids_list = (
|
||||
[id.strip() for id in device_ids.split(",") if id.strip()]
|
||||
if device_ids
|
||||
else []
|
||||
)
|
||||
if device_ids == "all":
|
||||
device_ids_list = []
|
||||
else:
|
||||
device_ids_list = (
|
||||
[id.strip() for id in device_ids.split(",") if id.strip()]
|
||||
if device_ids
|
||||
else []
|
||||
)
|
||||
return await CompositeQueries.clean_scada_data(
|
||||
timescale_conn, postgres_conn, device_ids_list, start_time, end_time
|
||||
)
|
||||
|
||||
@@ -589,7 +589,11 @@ class RealtimeRepository:
|
||||
raise ValueError(f"Invalid type: {type}. Must be 'node' or 'link'")
|
||||
|
||||
# Format the results
|
||||
return [{"ID": item["id"], "value": item["value"]} for item in data]
|
||||
result = []
|
||||
for id, items in data.items():
|
||||
for item in items:
|
||||
result.append({"ID": id, "value": item["value"]})
|
||||
return result
|
||||
|
||||
@staticmethod
|
||||
async def query_simulation_result_by_id_time(
|
||||
|
||||
Reference in New Issue
Block a user