diff --git a/app/algorithms/__init__.py b/app/algorithms/__init__.py index 57dc324..a31f4b3 100644 --- a/app/algorithms/__init__.py +++ b/app/algorithms/__init__.py @@ -1,36 +1,45 @@ -from app.algorithms.cleaning import flow_data_clean, pressure_data_clean -from app.algorithms.sensor import ( - pressure_sensor_placement_sensitivity, - pressure_sensor_placement_kmeans, -) -from app.algorithms.isolation.valve import valve_isolation_analysis -from app.algorithms.leakage import LeakageIdentifier -from app.algorithms.health import PipelineHealthAnalyzer -from app.algorithms.burst_location import run_burst_location -from app.algorithms.simulation.scenarios import ( - convert_to_local_unit, - burst_analysis, - valve_close_analysis, - flushing_analysis, - contaminant_simulation, - age_analysis, - pressure_regulation, -) +"""Algorithm package with side-effect-free, lazy compatibility exports.""" -__all__ = [ - "flow_data_clean", - "pressure_data_clean", - "pressure_sensor_placement_sensitivity", - "pressure_sensor_placement_kmeans", - "convert_to_local_unit", - "burst_analysis", - "valve_close_analysis", - "flushing_analysis", - "contaminant_simulation", - "age_analysis", - "pressure_regulation", - "valve_isolation_analysis", - "LeakageIdentifier", - "PipelineHealthAnalyzer", - "run_burst_location", -] +from importlib import import_module +from typing import Any + + +_EXPORT_MODULES = { + "flow_data_clean": "app.algorithms.cleaning", + "pressure_data_clean": "app.algorithms.cleaning", + "pressure_sensor_placement_sensitivity": "app.algorithms.sensor", + "pressure_sensor_placement_kmeans": "app.algorithms.sensor", + "valve_isolation_analysis": "app.algorithms.isolation.valve", + "LeakageIdentifier": "app.algorithms.leakage", + "PipelineHealthAnalyzer": "app.algorithms.health", + "run_burst_location": "app.algorithms.burst_location", + **{ + name: "app.algorithms.simulation.scenarios" + for name in ( + "convert_to_local_unit", + "burst_analysis", + "valve_close_analysis", + "flushing_analysis", + "contaminant_simulation", + "age_analysis", + "pressure_regulation", + ) + }, +} + +__all__ = list(_EXPORT_MODULES) + + +def __getattr__(name: str) -> Any: + try: + module_name = _EXPORT_MODULES[name] + except KeyError as exc: + raise AttributeError(f"module {__name__!r} has no attribute {name!r}") from exc + + value = getattr(import_module(module_name), name) + globals()[name] = value + return value + + +def __dir__() -> list[str]: + return sorted({*globals(), *__all__}) diff --git a/app/api/v1/endpoints/project_data.py b/app/api/v1/endpoints/project_data.py index 1d9a928..58efa1b 100644 --- a/app/api/v1/endpoints/project_data.py +++ b/app/api/v1/endpoints/project_data.py @@ -1,10 +1,9 @@ from fastapi import APIRouter, Depends, HTTPException, Path, Query from psycopg import AsyncConnection -import app.native.wndb as wndb +from app.infra.db.postgresql.scada import ScadaInfoRepository from app.infra.db.postgresql.scheme import SchemeRepository from app.auth.project_dependencies import get_project_pg_connection -from app.services import project_info router = APIRouter() @@ -26,9 +25,7 @@ async def get_scada_info_with_connection( 返回项目中所有的SCADA设备信息 """ try: - _ = conn - network_name = project_info.name - scada_data = wndb.get_all_scada_info(network_name) if network_name else [] + scada_data = await ScadaInfoRepository.get_scadas(conn) return {"success": True, "data": scada_data, "count": len(scada_data)} except Exception as e: raise HTTPException( diff --git a/app/infra/db/postgresql/scada.py b/app/infra/db/postgresql/scada.py new file mode 100644 index 0000000..ef33852 --- /dev/null +++ b/app/infra/db/postgresql/scada.py @@ -0,0 +1,51 @@ +from typing import Any + +from psycopg import AsyncConnection + + +def _optional_text(value: Any) -> str | None: + return str(value).strip() if value is not None else None + + +def _optional_float(value: Any) -> float | None: + return float(value) if value is not None else None + + +class ScadaInfoRepository: + """Read SCADA metadata from the current project's business database.""" + + @staticmethod + async def get_scadas(conn: AsyncConnection) -> list[dict[str, Any]]: + async with conn.cursor() as cur: + await cur.execute( + """ + SELECT id, + type, + associated_element_id, + api_query_id, + transmission_mode, + transmission_frequency, + reliability, + x_coor, + y_coor + FROM public.scada_info + """ + ) + records = await cur.fetchall() + + return [ + { + "id": str(record["id"]).strip(), + "type": str(record["type"]).strip().lower(), + "associated_element_id": _optional_text( + record["associated_element_id"] + ), + "api_query_id": record["api_query_id"], + "transmission_mode": record["transmission_mode"], + "transmission_frequency": record["transmission_frequency"], + "reliability": _optional_float(record["reliability"]), + "x": _optional_float(record["x_coor"]), + "y": _optional_float(record["y_coor"]), + } + for record in records + ] diff --git a/app/infra/db/timescaledb/composite_queries.py b/app/infra/db/timescaledb/composite_queries.py index a57479d..6baf715 100644 --- a/app/infra/db/timescaledb/composite_queries.py +++ b/app/infra/db/timescaledb/composite_queries.py @@ -1,18 +1,19 @@ import time -from typing import List, Optional, Any, Dict, Tuple from datetime import datetime, timedelta -from psycopg import AsyncConnection -import pandas as pd +from typing import Any, Dict, List, Optional, Tuple + import numpy as np +import pandas as pd +from psycopg import AsyncConnection + +import app.native.wndb as wndb from app.algorithms.cleaning.flow import clean_flow_data_df_kf from app.algorithms.cleaning.pressure import clean_pressure_data_df_km from app.algorithms.health.analyzer import PipelineHealthAnalyzer -import app.native.wndb as wndb - +from app.infra.db.postgresql.scada import ScadaInfoRepository from app.infra.db.timescaledb.repositories.realtime import RealtimeRepository from app.infra.db.timescaledb.repositories.scheme import SchemeRepository from app.infra.db.timescaledb.repositories.scada import ScadaRepository -from app.services import project_info class CompositeQueries: @@ -20,6 +21,13 @@ class CompositeQueries: 复合查询类，提供跨表查询功能 """ + @staticmethod + async def _get_project_scada_index( + postgres_conn: AsyncConnection, + ) -> Dict[str, Dict[str, Any]]: + scadas = await ScadaInfoRepository.get_scadas(postgres_conn) + return {scada["id"]: scada for scada in scadas} + @staticmethod async def get_scada_associated_realtime_simulation_data( timescale_conn: AsyncConnection, @@ -48,31 +56,22 @@ class CompositeQueries: ValueError: 当 SCADA 设备未找到或字段无效时 """ result = {} - # 1. 查询所有 SCADA 信息 - network_name = project_info.name - scada_infos = wndb.get_all_scada_info(network_name) if network_name else [] + scada_by_id = await CompositeQueries._get_project_scada_index(postgres_conn) for device_id in device_ids: - # 2. 根据 device_id 找到对应的 SCADA 信息 - target_scada = None - for scada in scada_infos: - if scada["id"] == device_id: - target_scada = scada - break - + target_scada = scada_by_id.get(device_id) if not target_scada: raise ValueError(f"SCADA device {device_id} not found") - # 3. 根据 type 和 associated_element_id 查询对应的模拟数据 element_id = target_scada["associated_element_id"] scada_type = target_scada["type"] - if scada_type.lower() == "pipe_flow": + if scada_type == "pipe_flow": # 查询 link 模拟数据 res = await RealtimeRepository.get_link_field_by_time_range( timescale_conn, start_time, end_time, element_id, "flow" ) - elif scada_type.lower() == "pressure": + elif scada_type == "pressure": # 查询 node 模拟数据 res = await RealtimeRepository.get_node_field_by_time_range( timescale_conn, start_time, end_time, element_id, "pressure" @@ -115,26 +114,17 @@ class CompositeQueries: ValueError: 当 SCADA 设备未找到或字段无效时 """ result = {} - # 1. 查询所有 SCADA 信息 - network_name = project_info.name - scada_infos = wndb.get_all_scada_info(network_name) if network_name else [] + scada_by_id = await CompositeQueries._get_project_scada_index(postgres_conn) for device_id in device_ids: - # 2. 根据 device_id 找到对应的 SCADA 信息 - target_scada = None - for scada in scada_infos: - if scada["id"] == device_id: - target_scada = scada - break - + target_scada = scada_by_id.get(device_id) if not target_scada: raise ValueError(f"SCADA device {device_id} not found") - # 3. 根据 type 和 associated_element_id 查询对应的模拟数据 element_id = target_scada["associated_element_id"] scada_type = target_scada["type"] - if scada_type.lower() == "pipe_flow": + if scada_type == "pipe_flow": # 查询 link 模拟数据 res = await SchemeRepository.get_link_field_by_scheme_and_time_range( timescale_conn, @@ -145,7 +135,7 @@ class CompositeQueries: element_id, "flow", ) - elif scada_type.lower() == "pressure": + elif scada_type == "pressure": # 查询 node 模拟数据 res = await SchemeRepository.get_node_field_by_scheme_and_time_range( timescale_conn, @@ -167,19 +157,19 @@ class CompositeQueries: @staticmethod async def get_realtime_simulation_data( timescale_conn: AsyncConnection, - featureInfos: List[Tuple[str, str]], + feature_infos: List[Tuple[str, str]], start_time: datetime, end_time: datetime, ) -> Dict[str, List[Dict[str, Any]]]: """ 获取 link/node 模拟值 - 根据传入的 featureInfos，找到关联的 link/node， + 根据传入的 feature_infos，找到关联的 link/node， 并根据对应的 type，查询对应的模拟数据 Args: timescale_conn: TimescaleDB 异步连接 - featureInfos: 传入的 feature 信息列表，包含 (element_id, type) + feature_infos: 传入的 feature 信息列表，包含 (element_id, type) start_time: 开始时间 end_time: 结束时间 @@ -190,20 +180,20 @@ class CompositeQueries: ValueError: 当 SCADA 设备未找到或字段无效时 """ result = {} - for feature_id, type in featureInfos: + for feature_id, feature_type in feature_infos: - if type.lower() == "pipe": + if feature_type.lower() == "pipe": # 查询 link 模拟数据 res = await RealtimeRepository.get_link_field_by_time_range( timescale_conn, start_time, end_time, feature_id, "flow" ) - elif type.lower() == "junction": + elif feature_type.lower() == "junction": # 查询 node 模拟数据 res = await RealtimeRepository.get_node_field_by_time_range( timescale_conn, start_time, end_time, feature_id, "pressure" ) else: - raise ValueError(f"Unknown type: {type}") + raise ValueError(f"Unknown type: {feature_type}") # 添加 scada_id 到每个数据项 for item in res: item["feature_id"] = feature_id @@ -213,7 +203,7 @@ class CompositeQueries: @staticmethod async def get_scheme_simulation_data( timescale_conn: AsyncConnection, - featureInfos: List[Tuple[str, str]], + feature_infos: List[Tuple[str, str]], start_time: datetime, end_time: datetime, scheme_type: str, @@ -222,12 +212,12 @@ class CompositeQueries: """ 获取 link/node scheme 模拟值 - 根据传入的 featureInfos，找到关联的 link/node， + 根据传入的 feature_infos，找到关联的 link/node， 并根据对应的 type，查询对应的模拟数据 Args: timescale_conn: TimescaleDB 异步连接 - featureInfos: 传入的 feature 信息列表，包含 (element_id, type) + feature_infos: 传入的 feature 信息列表，包含 (element_id, type) start_time: 开始时间 end_time: 结束时间 scheme_type: 工况类型 @@ -240,8 +230,8 @@ class CompositeQueries: ValueError: 当类型无效时 """ result = {} - for feature_id, type in featureInfos: - if type.lower() == "pipe": + for feature_id, feature_type in feature_infos: + if feature_type.lower() == "pipe": # 查询 link 模拟数据 res = await SchemeRepository.get_link_field_by_scheme_and_time_range( timescale_conn, @@ -252,7 +242,7 @@ class CompositeQueries: feature_id, "flow", ) - elif type.lower() == "junction": + elif feature_type.lower() == "junction": # 查询 node 模拟数据 res = await SchemeRepository.get_node_field_by_scheme_and_time_range( timescale_conn, @@ -264,7 +254,7 @@ class CompositeQueries: "pressure", ) else: - raise ValueError(f"Unknown type: {type}") + raise ValueError(f"Unknown type: {feature_type}") # 添加 feature_id 到每个数据项 for item in res: item["feature_id"] = feature_id @@ -301,33 +291,27 @@ class CompositeQueries: ValueError: 当元素类型无效时 """ - # 1. 查询所有 SCADA 信息 - network_name = project_info.name - scada_infos = wndb.get_all_scada_info(network_name) if network_name else [] - - # 2. 根据 element_type 和 element_id 找到关联的 SCADA 设备 - associated_scada = None - for scada in scada_infos: - if scada["associated_element_id"] == element_id: - associated_scada = scada - break + scada_by_id = await CompositeQueries._get_project_scada_index(postgres_conn) + associated_scada = next( + ( + scada + for scada in scada_by_id.values() + if scada["associated_element_id"] == element_id + ), + None, + ) if not associated_scada: - # 没有找到关联的 SCADA 设备 return None - # 3. 通过 SCADA device_id 获取监测数据 device_id = associated_scada["id"] - # 根据 use_cleaned 参数选择字段 data_field = "cleaned_value" if use_cleaned else "monitored_value" - # 保证 device_id 以列表形式传递 res = await ScadaRepository.get_scada_field_by_id_time_range( timescale_conn, [device_id], start_time, end_time, data_field ) - # 将 device_id 替换为 element_id 返回 return {element_id: res.get(device_id, [])} @staticmethod @@ -351,108 +335,124 @@ class CompositeQueries: end_time: 结束时间 Returns: - "success" 或错误信息 + "success" + + Raises: + ValueError: 当前项目没有可清洗设备或指定时间范围内没有监测数据 """ - try: - # 获取所有 SCADA 信息 - network_name = project_info.name - scada_infos = wndb.get_all_scada_info(network_name) if network_name else [] - # 将列表转换为字典，以 device_id 为键 - scada_device_info_dict = {info["id"]: info for info in scada_infos} + scada_by_id = await CompositeQueries._get_project_scada_index(postgres_conn) + supported_types = {"pressure", "pipe_flow", "flow"} - # 如果 device_ids 为空，则处理所有 SCADA 设备 - if not device_ids: - device_ids = list(scada_device_info_dict.keys()) + if device_ids: + device_ids = [str(device_id).strip() for device_id in device_ids] + missing_metadata_ids = [ + device_id + for device_id in device_ids + if device_id not in scada_by_id + ] + if missing_metadata_ids: + raise ValueError( + f"当前项目中有 {len(missing_metadata_ids)} 个 SCADA 设备缺少元数据" + ) - # 批量查询所有设备的数据 - data = await ScadaRepository.get_scada_field_by_id_time_range( - timescale_conn, device_ids, start_time, end_time, "monitored_value" + unsupported_ids = [ + device_id + for device_id in device_ids + if scada_by_id[device_id]["type"] not in supported_types + ] + if unsupported_ids: + raise ValueError( + f"当前项目中有 {len(unsupported_ids)} 个 SCADA 设备类型不支持清洗" + ) + else: + device_ids = [ + device_id + for device_id, info in scada_by_id.items() + if info["type"] in supported_types + ] + + if not device_ids: + raise ValueError("当前项目没有可清洗的 SCADA 设备") + + data = await ScadaRepository.get_scada_field_by_id_time_range( + timescale_conn, device_ids, start_time, end_time, "monitored_value" + ) + if not data: + raise ValueError("指定时间范围内没有 SCADA 监测数据") + + normalized_data = { + str(device_id): records for device_id, records in data.items() + } + missing_data_ids = [ + device_id for device_id in device_ids if not normalized_data.get(device_id) + ] + if missing_data_ids: + raise ValueError( + f"指定时间范围内有 {len(missing_data_ids)} 个 SCADA 设备没有监测数据" ) - if not data: - return "error: fetch none scada data" # 没有数据，直接返回 + all_records = [ + { + "time": record["time"], + "device_id": device_id, + "value": record["value"], + } + for device_id, records in normalized_data.items() + for record in records + ] + if not all_records: + raise ValueError("指定时间范围内没有 SCADA 监测数据") - # 将嵌套字典转换为 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"], - } + df_long = pd.DataFrame(all_records) + df = df_long.pivot(index="time", columns="device_id", values="value") + + pressure_ids = [ + device_id + for device_id in df.columns + if scada_by_id[device_id]["type"] == "pressure" + ] + flow_ids = [ + device_id + for device_id in df.columns + if scada_by_id[device_id]["type"] in {"pipe_flow", "flow"} + ] + + updated_rows = 0 + for grouped_ids, cleaning_function in ( + (pressure_ids, clean_pressure_data_df_km), + (flow_ids, clean_flow_data_df_kf), + ): + if not grouped_ids: + continue + + source_df = df[grouped_ids].reset_index() + cleaned_df = cleaning_function(source_df) + time_values = cleaned_df["time"].tolist() + + for device_id in grouped_ids: + if device_id not in cleaned_df.columns: + raise ValueError(f"设备 {device_id} 的清洗结果缺少数据列") + + cleaned_values = cleaned_df[device_id].tolist() + for time_value, value in zip(time_values, cleaned_values): + time_dt = ( + time_value + if isinstance(time_value, datetime) + else datetime.fromisoformat(str(time_value)) ) + await ScadaRepository.update_scada_field( + timescale_conn, + time_dt, + device_id, + "cleaned_value", + value, + ) + updated_rows += 1 - if not all_records: - return "error: fetch none scada data" # 没有数据，直接返回 + if updated_rows == 0: + raise ValueError("SCADA 数据清洗未产生任何数据库更新") - # 创建 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" - ] - - # 处理pressure数据 - if pressure_ids: - pressure_df = df[pressure_ids] - # 重置索引，将 time 变为普通列 - pressure_df = pressure_df.reset_index() - # 调用清洗方法 - cleaned_df = clean_pressure_data_df_km(pressure_df) - # 将清洗后的数据写回数据库 - 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 变为普通列 - flow_df = flow_df.reset_index() - # 调用清洗方法 - cleaned_df = clean_flow_data_df_kf(flow_df) - # 将清洗后的数据写回数据库 - 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_dt = datetime.fromisoformat(time_str) - value = cleaned_values[i] - await ScadaRepository.update_scada_field( - timescale_conn, - time_dt, - device_id, - "cleaned_value", - value, - ) - - return "success" - except Exception as e: - return f"error: {str(e)}" + return "success" @staticmethod async def predict_pipeline_health( diff --git a/tests/unit/test_postgres_scada_repository.py b/tests/unit/test_postgres_scada_repository.py new file mode 100644 index 0000000..d1f4cd7 --- /dev/null +++ b/tests/unit/test_postgres_scada_repository.py @@ -0,0 +1,62 @@ +import asyncio + +from app.infra.db.postgresql.scada import ScadaInfoRepository + + +class _FakeCursor: + def __init__(self): + self.query = None + + async def __aenter__(self): + return self + + async def __aexit__(self, exc_type, exc, tb): + return False + + async def execute(self, query): + self.query = query + + async def fetchall(self): + return [ + { + "id": " 25470001 ", + "type": " PRESSURE ", + "associated_element_id": " J1 ", + "api_query_id": "query-1", + "transmission_mode": "realtime", + "transmission_frequency": None, + "reliability": "0.95", + "x_coor": "117.1", + "y_coor": "32.9", + } + ] + + +class _FakeConnection: + def __init__(self): + self.cursor_instance = _FakeCursor() + + def cursor(self): + return self.cursor_instance + + +def test_get_scadas_normalizes_id_and_type(): + conn = _FakeConnection() + + result = asyncio.run(ScadaInfoRepository.get_scadas(conn)) + + assert result == [ + { + "id": "25470001", + "type": "pressure", + "associated_element_id": "J1", + "api_query_id": "query-1", + "transmission_mode": "realtime", + "transmission_frequency": None, + "reliability": 0.95, + "x": 117.1, + "y": 32.9, + } + ] + assert "associated_element_id" in conn.cursor_instance.query + assert "FROM public.scada_info" in conn.cursor_instance.query diff --git a/tests/unit/test_scada_cleaning.py b/tests/unit/test_scada_cleaning.py new file mode 100644 index 0000000..1b9da91 --- /dev/null +++ b/tests/unit/test_scada_cleaning.py @@ -0,0 +1,267 @@ +import asyncio +from datetime import datetime, timezone +from unittest.mock import AsyncMock + +import pandas as pd +import pytest + +from app.api.v1.endpoints import project_data +from app.infra.db.timescaledb import composite_queries + + +PROJECT_SCADA = { + "id": "fengyang-pressure-1", + "type": "pressure", + "associated_element_id": "J1", + "api_query_id": "query-1", + "transmission_mode": "realtime", + "transmission_frequency": None, + "reliability": 1.0, + "x": 117.1, + "y": 32.9, +} +START_TIME = datetime(2026, 6, 1, tzinfo=timezone.utc) +END_TIME = datetime(2026, 6, 2, tzinfo=timezone.utc) + + +def _cleaning_args(): + return ( + object(), + object(), + ["fengyang-pressure-1"], + datetime(2026, 6, 1, tzinfo=timezone.utc), + datetime(2026, 6, 2, tzinfo=timezone.utc), + ) + + +def test_clean_scada_uses_current_project_metadata(monkeypatch): + monkeypatch.setattr( + composite_queries.ScadaInfoRepository, + "get_scadas", + AsyncMock( + return_value=[{"id": "fengyang-pressure-1", "type": "pressure"}] + ), + ) + monkeypatch.setattr( + composite_queries.ScadaRepository, + "get_scada_field_by_id_time_range", + AsyncMock( + return_value={ + "fengyang-pressure-1": [ + {"time": "2026-06-01T00:00:00+08:00", "value": 26.5} + ] + } + ), + ) + update_mock = AsyncMock() + monkeypatch.setattr( + composite_queries.ScadaRepository, + "update_scada_field", + update_mock, + ) + monkeypatch.setattr( + composite_queries, + "clean_pressure_data_df_km", + lambda frame: pd.DataFrame( + { + "time": frame["time"], + "fengyang-pressure-1": frame["fengyang-pressure-1"], + } + ), + ) + + result = asyncio.run( + composite_queries.CompositeQueries.clean_scada_data(*_cleaning_args()) + ) + + assert result == "success" + update_mock.assert_awaited_once() + + +def test_clean_scada_rejects_devices_missing_from_project_metadata(monkeypatch): + monkeypatch.setattr( + composite_queries.ScadaInfoRepository, + "get_scadas", + AsyncMock(return_value=[{"id": "other-device", "type": "pressure"}]), + ) + query_mock = AsyncMock() + monkeypatch.setattr( + composite_queries.ScadaRepository, + "get_scada_field_by_id_time_range", + query_mock, + ) + + with pytest.raises(ValueError, match="缺少元数据"): + asyncio.run( + composite_queries.CompositeQueries.clean_scada_data(*_cleaning_args()) + ) + + query_mock.assert_not_awaited() + + +def test_clean_scada_rejects_zero_database_updates(monkeypatch): + monkeypatch.setattr( + composite_queries.ScadaInfoRepository, + "get_scadas", + AsyncMock( + return_value=[{"id": "fengyang-pressure-1", "type": "pressure"}] + ), + ) + monkeypatch.setattr( + composite_queries.ScadaRepository, + "get_scada_field_by_id_time_range", + AsyncMock( + return_value={ + "fengyang-pressure-1": [ + {"time": "2026-06-01T00:00:00+08:00", "value": 26.5} + ] + } + ), + ) + update_mock = AsyncMock() + monkeypatch.setattr( + composite_queries.ScadaRepository, + "update_scada_field", + update_mock, + ) + monkeypatch.setattr( + composite_queries, + "clean_pressure_data_df_km", + lambda _frame: pd.DataFrame( + {"time": [], "fengyang-pressure-1": []} + ), + ) + + with pytest.raises(ValueError, match="未产生任何数据库更新"): + asyncio.run( + composite_queries.CompositeQueries.clean_scada_data(*_cleaning_args()) + ) + + update_mock.assert_not_awaited() + + +def test_clean_scada_propagates_write_failures(monkeypatch): + monkeypatch.setattr( + composite_queries.ScadaInfoRepository, + "get_scadas", + AsyncMock( + return_value=[{"id": "fengyang-pressure-1", "type": "pressure"}] + ), + ) + monkeypatch.setattr( + composite_queries.ScadaRepository, + "get_scada_field_by_id_time_range", + AsyncMock( + return_value={ + "fengyang-pressure-1": [ + {"time": "2026-06-01T00:00:00+08:00", "value": 26.5} + ] + } + ), + ) + monkeypatch.setattr( + composite_queries.ScadaRepository, + "update_scada_field", + AsyncMock(side_effect=RuntimeError("database write failed")), + ) + monkeypatch.setattr( + composite_queries, + "clean_pressure_data_df_km", + lambda frame: frame, + ) + + with pytest.raises(RuntimeError, match="database write failed"): + asyncio.run( + composite_queries.CompositeQueries.clean_scada_data(*_cleaning_args()) + ) + + +def _patch_project_scada_metadata(monkeypatch): + monkeypatch.setattr( + composite_queries.ScadaInfoRepository, + "get_scadas", + AsyncMock(return_value=[PROJECT_SCADA.copy()]), + ) + + +def test_realtime_scada_simulation_uses_current_project_metadata(monkeypatch): + _patch_project_scada_metadata(monkeypatch) + query_mock = AsyncMock(return_value=[{"time": START_TIME, "value": 26.5}]) + monkeypatch.setattr( + composite_queries.RealtimeRepository, + "get_node_field_by_time_range", + query_mock, + raising=False, + ) + + result = asyncio.run( + composite_queries.CompositeQueries.get_scada_associated_realtime_simulation_data( + object(), object(), [PROJECT_SCADA["id"]], START_TIME, END_TIME + ) + ) + + assert result[PROJECT_SCADA["id"]][0]["scada_id"] == PROJECT_SCADA["id"] + assert query_mock.await_args.args[1:] == ( + START_TIME, + END_TIME, + "J1", + "pressure", + ) + + +def test_scheme_scada_simulation_uses_current_project_metadata(monkeypatch): + _patch_project_scada_metadata(monkeypatch) + query_mock = AsyncMock(return_value=[{"time": START_TIME, "value": 26.5}]) + monkeypatch.setattr( + composite_queries.SchemeRepository, + "get_node_field_by_scheme_and_time_range", + query_mock, + raising=False, + ) + + result = asyncio.run( + composite_queries.CompositeQueries.get_scada_associated_scheme_simulation_data( + object(), + object(), + [PROJECT_SCADA["id"]], + START_TIME, + END_TIME, + "baseline", + "scheme-1", + ) + ) + + assert result[PROJECT_SCADA["id"]][0]["scada_id"] == PROJECT_SCADA["id"] + assert query_mock.await_args.args[5:] == ("J1", "pressure") + + +def test_element_scada_query_uses_current_project_metadata(monkeypatch): + _patch_project_scada_metadata(monkeypatch) + query_mock = AsyncMock( + return_value={PROJECT_SCADA["id"]: [{"time": START_TIME, "value": 26.5}]} + ) + monkeypatch.setattr( + composite_queries.ScadaRepository, + "get_scada_field_by_id_time_range", + query_mock, + ) + + result = asyncio.run( + composite_queries.CompositeQueries.get_element_associated_scada_data( + object(), object(), "J1", START_TIME, END_TIME + ) + ) + + assert result == {"J1": [{"time": START_TIME, "value": 26.5}]} + + +def test_scada_info_endpoint_uses_current_project_connection(monkeypatch): + monkeypatch.setattr( + project_data.ScadaInfoRepository, + "get_scadas", + AsyncMock(return_value=[PROJECT_SCADA.copy()]), + ) + + result = asyncio.run(project_data.get_scada_info_with_connection(object())) + + assert result == {"success": True, "data": [PROJECT_SCADA], "count": 1}