From 682c26fddd26be6e168bcd02d093fe08ba597fc7 Mon Sep 17 00:00:00 2001 From: Jiang Date: Mon, 14 Sep 2026 12:30:41 +0800 Subject: [PATCH] feat(timeseries): unify element history queries --- app/api/v1/endpoints/timeseries/composite.py | 183 +----- app/domain/measurement_units.py | 13 + app/domain/schemas/scada.py | 1 + app/domain/schemas/timeseries_history.py | 88 +++ app/infra/db/postgresql/network_settings.py | 32 + app/infra/db/postgresql/scada.py | 20 +- app/services/simulation.py | 9 +- app/services/timeseries_history.py | 237 ++++++++ contracts/manifest.json | 2 +- contracts/server-v1.openapi.json | 595 ++++++++----------- resources/db_v2/DATABASE_ARCHITECTURE.md | 21 +- tests/unit/test_analysis_simulation.py | 12 +- tests/unit/test_measurement_units.py | 7 + tests/unit/test_postgres_scada_repository.py | 2 + tests/unit/test_project_scada_metadata.py | 1 + tests/unit/test_timeseries_history.py | 146 +++++ tests/unit/test_timeseries_history_schema.py | 50 ++ 17 files changed, 888 insertions(+), 531 deletions(-) create mode 100644 app/domain/measurement_units.py create mode 100644 app/domain/schemas/timeseries_history.py create mode 100644 app/infra/db/postgresql/network_settings.py create mode 100644 app/services/timeseries_history.py create mode 100644 tests/unit/test_measurement_units.py create mode 100644 tests/unit/test_timeseries_history.py create mode 100644 tests/unit/test_timeseries_history_schema.py diff --git a/app/api/v1/endpoints/timeseries/composite.py b/app/api/v1/endpoints/timeseries/composite.py index 9170a65..6259246 100644 --- a/app/api/v1/endpoints/timeseries/composite.py +++ b/app/api/v1/endpoints/timeseries/composite.py @@ -1,183 +1,34 @@ from fastapi import APIRouter, Depends, HTTPException, Query from datetime import datetime from psycopg import AsyncConnection -from uuid import UUID from app.services.timeseries_analysis import TimeseriesAnalysisService +from app.domain.schemas.timeseries_history import ( + ElementHistoryQuery, + ElementHistoryResponse, +) +from app.services.timeseries_history import TimeseriesHistoryService from .dependencies import get_timescale_connection, get_postgres_connection router = APIRouter() -@router.get("/timeseries/views/scada-simulations", summary="获取SCADA关联的模拟数据") -async def get_scada_associated_simulation_data( - start_time: datetime = Query(..., description="查询开始时间"), - end_time: datetime = Query(..., description="查询结束时间"), - device_ids: str = Query(..., description="SCADA设备ID列表，逗号分隔"), - run_id: UUID | None = Query(None, description="分析运行 ID；为空时查询实时数据"), +@router.post( + "/timeseries/views/element-history/query", + summary="批量查询管网元素历史数据", + response_model=ElementHistoryResponse, +) +async def query_element_history( + payload: ElementHistoryQuery, timescale_conn: AsyncConnection = Depends(get_timescale_connection), postgres_conn: AsyncConnection = Depends(get_postgres_connection), -): - """ - 获取SCADA关联的link/node模拟值 - - 根据传入的SCADA device_ids，找到关联的link/node， - 并根据对应的type，查询对应的模拟数据。支持查询实时或分析运行数据。 - - Args: - start_time: 查询开始时间 - end_time: 查询结束时间 - device_ids: SCADA设备ID列表，用逗号分隔 - run_id: 分析运行 ID，若为空则查询实时数据 - timescale_conn: TimescaleDB连接 - postgres_conn: PostgreSQL连接 - - Returns: - SCADA关联的模拟数据 - - Raises: - HTTPException: 当查询参数无效时返回400错误，未找到数据时返回404错误 - """ +) -> ElementHistoryResponse: try: - device_ids_list = ( - [id.strip() for id in device_ids.split(",") if id.strip()] - if device_ids - else [] + return await TimeseriesHistoryService.query( + timescale_conn, postgres_conn, payload ) - - if run_id is not None: - result = await TimeseriesAnalysisService.get_scada_associated_analysis_simulation_data( - timescale_conn, - postgres_conn, - device_ids_list, - start_time, - end_time, - run_id, - ) - else: - result = ( - await TimeseriesAnalysisService.get_scada_associated_realtime_simulation_data( - timescale_conn, - postgres_conn, - device_ids_list, - start_time, - end_time, - ) - ) - if result is None: - raise HTTPException(status_code=404, detail="No simulation data found") - return result - except ValueError as e: - raise HTTPException(status_code=400, detail=str(e)) - - -@router.get("/timeseries/views/element-simulations", summary="获取管网元素的模拟数据") -async def get_feature_simulation_data( - start_time: datetime = Query(..., description="查询开始时间"), - end_time: datetime = Query(..., description="查询结束时间"), - feature_infos: str = Query( - ..., description="特征信息，格式: id1:type1,id2:type2，type为pipe（管道）或junction（节点）" - ), - run_id: UUID | None = Query(None, description="分析运行 ID；为空时查询实时数据"), - timescale_conn: AsyncConnection = Depends(get_timescale_connection), -): - """ - 获取link/node模拟值 - - 根据传入的featureInfos，找到关联的link/node， - 并根据对应的type，查询对应的模拟数据。支持查询实时或分析运行数据。 - - Args: - start_time: 查询开始时间 - end_time: 查询结束时间 - feature_infos: 格式为 "element_id1:type1,element_id2:type2" - 例如: "P1:pipe,J1:junction" - run_id: 分析运行 ID，若为空则查询实时数据 - timescale_conn: TimescaleDB连接 - - Returns: - 管网元素的模拟数据 - - Raises: - HTTPException: 当feature_infos为空返回400错误，未找到数据返回404错误，其他错误返回400错误 - """ - try: - feature_infos_list = [] - if feature_infos: - for item in feature_infos.split(","): - item = item.strip() - if ":" in item: - element_id, element_type = item.split(":", 1) - feature_infos_list.append( - (element_id.strip(), element_type.strip()) - ) - - if not feature_infos_list: - raise HTTPException(status_code=400, detail="feature_infos cannot be empty") - - if run_id is not None: - result = await TimeseriesAnalysisService.get_analysis_simulation_data( - timescale_conn, - feature_infos_list, - start_time, - end_time, - run_id, - ) - else: - result = await TimeseriesAnalysisService.get_realtime_simulation_data( - timescale_conn, - feature_infos_list, - start_time, - end_time, - ) - - if result is None: - raise HTTPException(status_code=404, detail="No simulation data found") - return result - except ValueError as e: - raise HTTPException(status_code=400, detail=str(e)) - - -@router.get("/timeseries/views/element-scada-readings", summary="获取管网元素关联的SCADA监测数据") -async def get_element_associated_scada_data( - element_id: str = Query(..., description="管网元素ID（管道或节点）"), - start_time: datetime = Query(..., description="查询开始时间"), - end_time: datetime = Query(..., description="查询结束时间"), - use_cleaned: bool = Query(False, description="是否使用清洗后的数据"), - timescale_conn: AsyncConnection = Depends(get_timescale_connection), - postgres_conn: AsyncConnection = Depends(get_postgres_connection), -): - """ - 获取link/node关联的SCADA监测值 - - 根据传入的link/node id，匹配SCADA信息， - 如果存在关联的SCADA device_id，获取实际的监测数据。 - - Args: - element_id: 管网元素ID - start_time: 查询开始时间 - end_time: 查询结束时间 - use_cleaned: 是否使用清洗后的数据，默认为False使用原始数据 - timescale_conn: TimescaleDB连接 - postgres_conn: PostgreSQL连接 - - Returns: - 管网元素关联的SCADA监测数据 - - Raises: - HTTPException: 当查询参数无效时返回400错误，未找到关联数据返回404错误 - """ - try: - result = await TimeseriesAnalysisService.get_element_associated_scada_data( - timescale_conn, postgres_conn, element_id, start_time, end_time, use_cleaned - ) - if result is None: - raise HTTPException( - status_code=404, detail="No associated SCADA data found" - ) - return result - except ValueError as e: - raise HTTPException(status_code=400, detail=str(e)) + except ValueError as exc: + raise HTTPException(status_code=422, detail=str(exc)) from exc @router.post("/timeseries/scada-cleaning-runs", summary="清洗SCADA监测数据") diff --git a/app/domain/measurement_units.py b/app/domain/measurement_units.py new file mode 100644 index 0000000..6c2f55a --- /dev/null +++ b/app/domain/measurement_units.py @@ -0,0 +1,13 @@ +from typing import Literal + + +Metric = Literal["flow", "pressure", "velocity"] + +DISPLAY_UNITS: dict[Metric, str] = { + "flow": "m³/h", + "pressure": "m", + "velocity": "m/s", +} + +def display_unit_for_metric(metric: Metric) -> str: + return DISPLAY_UNITS[metric] diff --git a/app/domain/schemas/scada.py b/app/domain/schemas/scada.py index 1a68fa1..cc2cfef 100644 --- a/app/domain/schemas/scada.py +++ b/app/domain/schemas/scada.py @@ -9,6 +9,7 @@ class ScadaDeviceResponse(BaseModel): node_id: str | None = None link_id: str | None = None api_query_id: str | None = None + measurement_unit: str transmission_mode: str transmission_frequency: str reliability: int | None = None diff --git a/app/domain/schemas/timeseries_history.py b/app/domain/schemas/timeseries_history.py new file mode 100644 index 0000000..94f8284 --- /dev/null +++ b/app/domain/schemas/timeseries_history.py @@ -0,0 +1,88 @@ +from datetime import datetime +from enum import StrEnum +from uuid import UUID + +from pydantic import BaseModel, Field, field_validator, model_validator + + +class HistoryElementType(StrEnum): + PIPE = "pipe" + JUNCTION = "junction" + + +class HistoryMode(StrEnum): + OBSERVED = "observed" + REALTIME_COMPARISON = "realtime_comparison" + ANALYSIS_COMPARISON = "analysis_comparison" + + +class HistoryMetric(StrEnum): + FLOW = "flow" + PRESSURE = "pressure" + + +class HistorySource(StrEnum): + SCADA_RAW = "scada_raw" + SCADA_CLEANED = "scada_cleaned" + REALTIME_SIMULATION = "realtime_simulation" + ANALYSIS_SIMULATION = "analysis_simulation" + + +class ElementHistoryTarget(BaseModel): + element_id: str = Field(min_length=1, max_length=128) + element_type: HistoryElementType + device_ids: list[str] | None = Field(default=None, max_length=100) + + @field_validator("element_id") + @classmethod + def normalize_element_id(cls, value: str) -> str: + return value.strip() + + @field_validator("device_ids") + @classmethod + def normalize_device_ids(cls, value: list[str] | None) -> list[str] | None: + if value is None: + return None + normalized = list(dict.fromkeys(item.strip() for item in value if item.strip())) + if not normalized: + raise ValueError("device_ids must contain at least one device") + return normalized + + +class ElementHistoryQuery(BaseModel): + start_time: datetime + end_time: datetime + mode: HistoryMode = HistoryMode.OBSERVED + run_id: UUID | None = None + elements: list[ElementHistoryTarget] = Field(min_length=1, max_length=200) + + @model_validator(mode="after") + def validate_query(self) -> "ElementHistoryQuery": + if self.start_time >= self.end_time: + raise ValueError("start_time must be earlier than end_time") + if self.mode == HistoryMode.ANALYSIS_COMPARISON and self.run_id is None: + raise ValueError("run_id is required for analysis_comparison") + if self.mode != HistoryMode.ANALYSIS_COMPARISON and self.run_id is not None: + raise ValueError("run_id is only valid for analysis_comparison") + return self + + +class HistoryPoint(BaseModel): + time: datetime + value: float | None + + +class ElementHistorySeries(BaseModel): + element_id: str + element_type: HistoryElementType + device_id: str | None = None + metric: HistoryMetric + source: HistorySource + source_unit: str = Field(description="Unit used by the returned point values") + display_unit: str = Field(description="Recommended UI display unit") + unit_inferred: bool = False + points: list[HistoryPoint] + + +class ElementHistoryResponse(BaseModel): + series: list[ElementHistorySeries] diff --git a/app/infra/db/postgresql/network_settings.py b/app/infra/db/postgresql/network_settings.py new file mode 100644 index 0000000..e1bfc76 --- /dev/null +++ b/app/infra/db/postgresql/network_settings.py @@ -0,0 +1,32 @@ +from typing import Any + +from psycopg import AsyncConnection + + +class NetworkSettingsRepository: + @staticmethod + async def get_result_units(conn: AsyncConnection) -> dict[str, str]: + async with conn.cursor() as cur: + await cur.execute( + """ + SELECT engine_version, key, value + FROM network.simulation_settings + WHERE (engine_version = 'v3' AND key IN ('FLOW_UNITS', 'PRESSURE_UNITS')) + OR (engine_version = 'legacy' AND key IN ('UNITS', 'PRESSURE')) + ORDER BY CASE engine_version WHEN 'v3' THEN 0 ELSE 1 END + """ + ) + rows: list[dict[str, Any]] = await cur.fetchall() + + units: dict[str, str] = {} + for row in rows: + key = str(row["key"]) + metric = "flow" if key in {"FLOW_UNITS", "UNITS"} else "pressure" + units.setdefault(metric, str(row["value"]).strip()) + missing = {"flow", "pressure"} - units.keys() + if missing: + raise ValueError( + "network simulation settings missing result units: " + + ", ".join(sorted(missing)) + ) + return units diff --git a/app/infra/db/postgresql/scada.py b/app/infra/db/postgresql/scada.py index 3eafe97..9e67e73 100644 --- a/app/infra/db/postgresql/scada.py +++ b/app/infra/db/postgresql/scada.py @@ -9,7 +9,7 @@ from app.native.wndb.core.database import read_all, try_read _SCADA_VIEW_SELECT = """ SELECT id AS device_id, device_type, node_id, link_id, api_query_id, - transmission_mode, transmission_frequency, reliability, x, y, + measurement_unit, transmission_mode, transmission_frequency, reliability, x, y, ST_X(ST_Transform(geom, 4326)) AS longitude, ST_Y(ST_Transform(geom, 4326)) AS latitude FROM gis.scada_devices @@ -57,6 +57,7 @@ def _device(record: dict[str, Any]) -> dict[str, Any]: "node_id": _optional_text(record["node_id"]), "link_id": _optional_text(record["link_id"]), "api_query_id": _optional_text(record["api_query_id"]), + "measurement_unit": str(record["measurement_unit"]).strip(), "transmission_mode": record["transmission_mode"], "transmission_frequency": record["transmission_frequency"], "reliability": _optional_int(record["reliability"]), @@ -105,6 +106,22 @@ class ScadaInfoRepository: ) return {str(row["device_id"]).strip() for row in await cur.fetchall()} + @staticmethod + async def get_scadas_for_elements( + conn: AsyncConnection, + node_ids: list[str], + link_ids: list[str], + ) -> list[dict[str, Any]]: + if not node_ids and not link_ids: + return [] + async with conn.cursor() as cur: + await cur.execute( + _SCADA_VIEW_SELECT + + " WHERE node_id = ANY(%s) OR link_id = ANY(%s) ORDER BY device_id", + (node_ids, link_ids), + ) + return [_device(record) for record in await cur.fetchall()] + def get_scada_info_schema(name: str) -> dict[str, dict[str, Any]]: return { @@ -113,6 +130,7 @@ def get_scada_info_schema(name: str) -> dict[str, dict[str, Any]]: "node_id": {"type": "str", "optional": True, "readonly": True}, "link_id": {"type": "str", "optional": True, "readonly": True}, "api_query_id": {"type": "str", "optional": True, "readonly": True}, + "measurement_unit": {"type": "str", "optional": False, "readonly": True}, "transmission_mode": {"type": "str", "optional": False, "readonly": True}, "transmission_frequency": {"type": "str", "optional": False, "readonly": True}, "reliability": {"type": "int", "optional": False, "readonly": True}, diff --git a/app/services/simulation.py b/app/services/simulation.py index a5863bf..a8bb1ff 100644 --- a/app/services/simulation.py +++ b/app/services/simulation.py @@ -532,7 +532,14 @@ def run_simulation( raise RuntimeError("run_project output missing times.report_step") if not scheme_type or not scheme_name: raise ValueError("extended simulation requires analysis run type and name") - detail = scheme_detail or {} + detail = dict(scheme_detail or {}) + result_units = output_data.get("units") + if isinstance(result_units, dict): + detail["result_units"] = { + key: str(result_units[key]).strip() + for key in ("flow", "pressure") + if result_units.get(key) + } run_id = create_analysis_run( name=db_name, scheme_name=scheme_name, diff --git a/app/services/timeseries_history.py b/app/services/timeseries_history.py new file mode 100644 index 0000000..66d11f9 --- /dev/null +++ b/app/services/timeseries_history.py @@ -0,0 +1,237 @@ +from collections import defaultdict +from typing import Any + +from psycopg import AsyncConnection + +from app.domain.measurement_units import display_unit_for_metric +from app.domain.schemas.timeseries_history import ( + ElementHistoryQuery, + ElementHistoryResponse, + ElementHistorySeries, + HistoryElementType, + HistoryMetric, + HistoryPoint, + HistorySource, +) +from app.infra.db.postgresql.analysis import AnalysisRepository +from app.infra.db.postgresql.network_settings import NetworkSettingsRepository +from app.infra.db.postgresql.scada import ScadaInfoRepository +from app.infra.db.timescaledb.repositories.analysis import AnalysisResultsRepository +from app.infra.db.timescaledb.repositories.realtime import RealtimeRepository +from app.infra.db.timescaledb.repositories.scada import ScadaRepository + + +def _target_metric(element_type: HistoryElementType) -> HistoryMetric: + return ( + HistoryMetric.FLOW + if element_type == HistoryElementType.PIPE + else HistoryMetric.PRESSURE + ) + + +def _device_metric(device_type: str) -> HistoryMetric | None: + normalized = device_type.strip().lower() + if normalized in {"pipe_flow", "flow"}: + return HistoryMetric.FLOW + if normalized == "pressure": + return HistoryMetric.PRESSURE + return None + + +def _points( + rows: list[dict[str, Any]], + *, + value_field: str, +) -> list[HistoryPoint]: + return [ + HistoryPoint( + time=row["time"], + value=( + float(row[value_field]) + if row.get(value_field) is not None + else None + ), + ) + for row in rows + ] + + +class TimeseriesHistoryService: + @staticmethod + async def query( + timescale_conn: AsyncConnection, + postgres_conn: AsyncConnection, + query: ElementHistoryQuery, + ) -> ElementHistoryResponse: + pipe_ids = list( + dict.fromkeys( + target.element_id + for target in query.elements + if target.element_type == HistoryElementType.PIPE + ) + ) + junction_ids = list( + dict.fromkeys( + target.element_id + for target in query.elements + if target.element_type == HistoryElementType.JUNCTION + ) + ) + devices = await ScadaInfoRepository.get_scadas_for_elements( + postgres_conn, junction_ids, pipe_ids + ) + devices_by_element: dict[tuple[HistoryElementType, str], list[dict[str, Any]]] = ( + defaultdict(list) + ) + for device in devices: + if device.get("link_id") in pipe_ids: + devices_by_element[(HistoryElementType.PIPE, device["link_id"])].append( + device + ) + if device.get("node_id") in junction_ids: + devices_by_element[ + (HistoryElementType.JUNCTION, device["node_id"]) + ].append(device) + + selected_devices: dict[str, tuple[dict[str, Any], HistoryElementType, str]] = {} + for target in query.elements: + target_devices = devices_by_element.get( + (target.element_type, target.element_id), [] + ) + if target.device_ids is not None: + target_device_ids = {device["device_id"] for device in target_devices} + unknown = set(target.device_ids) - target_device_ids + if unknown: + raise ValueError( + f"SCADA devices do not belong to {target.element_type.value} " + f"{target.element_id}: {', '.join(sorted(unknown))}" + ) + requested = set(target.device_ids) + target_devices = [ + device + for device in target_devices + if device["device_id"] in requested + ] + expected_metric = _target_metric(target.element_type) + for device in target_devices: + if _device_metric(device["device_type"]) != expected_metric: + continue + selected_devices[device["device_id"]] = ( + device, + target.element_type, + target.element_id, + ) + + scada_rows = await ScadaRepository.get_scada_by_ids_time_range( + timescale_conn, + list(selected_devices), + query.start_time, + query.end_time, + ) + scada_by_device: dict[str, list[dict[str, Any]]] = defaultdict(list) + for row in scada_rows: + scada_by_device[str(row["device_id"])].append(row) + + series: list[ElementHistorySeries] = [] + for device_id, (device, element_type, element_id) in selected_devices.items(): + rows = scada_by_device.get(device_id, []) + if not rows: + continue + metric = _target_metric(element_type) + unit = device["measurement_unit"] + for source, field in ( + (HistorySource.SCADA_RAW, "monitored_value"), + (HistorySource.SCADA_CLEANED, "cleaned_value"), + ): + series.append( + ElementHistorySeries( + element_id=element_id, + element_type=element_type, + device_id=device_id, + metric=metric, + source=source, + source_unit=unit, + display_unit=display_unit_for_metric(metric.value), + points=_points( + rows, + value_field=field, + ), + ) + ) + + if query.mode.value == "observed": + return ElementHistoryResponse(series=series) + + current_units = await NetworkSettingsRepository.get_result_units(postgres_conn) + if query.mode.value == "analysis_comparison": + run = await AnalysisRepository.get_run(postgres_conn, query.run_id) + if run is None: + raise ValueError(f"analysis run does not exist: {query.run_id}") + parameters = run.get("parameters") or {} + recorded_units = parameters.get("result_units") + unit_inferred = not isinstance(recorded_units, dict) + result_units = recorded_units if not unit_inferred else current_units + link_values = await AnalysisResultsRepository.get_series_by_ids( + timescale_conn, + query.run_id, + "link", + pipe_ids, + query.start_time, + query.end_time, + "flow", + ) + node_values = await AnalysisResultsRepository.get_series_by_ids( + timescale_conn, + query.run_id, + "node", + junction_ids, + query.start_time, + query.end_time, + "pressure", + ) + source = HistorySource.ANALYSIS_SIMULATION + else: + unit_inferred = False + result_units = current_units + link_values = await RealtimeRepository.get_link_fields_by_ids_time_range( + timescale_conn, + query.start_time, + query.end_time, + pipe_ids, + "flow", + ) + node_values = await RealtimeRepository.get_node_fields_by_ids_time_range( + timescale_conn, + query.start_time, + query.end_time, + junction_ids, + "pressure", + ) + source = HistorySource.REALTIME_SIMULATION + + for target in query.elements: + metric = _target_metric(target.element_type) + rows = ( + link_values.get(target.element_id, []) + if target.element_type == HistoryElementType.PIPE + else node_values.get(target.element_id, []) + ) + if not rows: + continue + source_unit = str(result_units.get(metric.value, "")).strip() + series.append( + ElementHistorySeries( + element_id=target.element_id, + element_type=target.element_type, + metric=metric, + source=source, + source_unit=source_unit, + display_unit=display_unit_for_metric(metric.value), + unit_inferred=unit_inferred, + points=_points( + rows, + value_field="value", + ), + ) + ) + return ElementHistoryResponse(series=series) diff --git a/contracts/manifest.json b/contracts/manifest.json index f704c5d..49cf8d2 100644 --- a/contracts/manifest.json +++ b/contracts/manifest.json @@ -3,7 +3,7 @@ "contracts": { "server": { "file": "server-v1.openapi.json", - "sha256": "fb720e3009948c3cb2f37674d0bbdbb7b6136973223eccca89880475697435bc" + "sha256": "b565d841061c9091f48ff3118bcc0cbb1b918b0cb8c2316d177570e8b2d8ba29" } } } diff --git a/contracts/server-v1.openapi.json b/contracts/server-v1.openapi.json index 47842b5..5b68dce 100644 --- a/contracts/server-v1.openapi.json +++ b/contracts/server-v1.openapi.json @@ -982,6 +982,224 @@ "title": "BurstLocationRequestRest", "type": "object" }, + "ElementHistoryQuery": { + "properties": { + "elements": { + "items": { + "$ref": "#/components/schemas/ElementHistoryTarget" + }, + "maxItems": 200, + "minItems": 1, + "title": "Elements", + "type": "array" + }, + "end_time": { + "format": "date-time", + "title": "End Time", + "type": "string" + }, + "mode": { + "$ref": "#/components/schemas/HistoryMode", + "default": "observed" + }, + "run_id": { + "anyOf": [ + { + "format": "uuid", + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Run Id" + }, + "start_time": { + "format": "date-time", + "title": "Start Time", + "type": "string" + } + }, + "required": [ + "start_time", + "end_time", + "elements" + ], + "title": "ElementHistoryQuery", + "type": "object" + }, + "ElementHistoryResponse": { + "properties": { + "series": { + "items": { + "$ref": "#/components/schemas/ElementHistorySeries" + }, + "title": "Series", + "type": "array" + } + }, + "required": [ + "series" + ], + "title": "ElementHistoryResponse", + "type": "object" + }, + "ElementHistorySeries": { + "properties": { + "device_id": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Device Id" + }, + "display_unit": { + "description": "Recommended UI display unit", + "title": "Display Unit", + "type": "string" + }, + "element_id": { + "title": "Element Id", + "type": "string" + }, + "element_type": { + "$ref": "#/components/schemas/HistoryElementType" + }, + "metric": { + "$ref": "#/components/schemas/HistoryMetric" + }, + "points": { + "items": { + "$ref": "#/components/schemas/HistoryPoint" + }, + "title": "Points", + "type": "array" + }, + "source": { + "$ref": "#/components/schemas/HistorySource" + }, + "source_unit": { + "description": "Unit used by the returned point values", + "title": "Source Unit", + "type": "string" + }, + "unit_inferred": { + "default": false, + "title": "Unit Inferred", + "type": "boolean" + } + }, + "required": [ + "element_id", + "element_type", + "metric", + "source", + "source_unit", + "display_unit", + "points" + ], + "title": "ElementHistorySeries", + "type": "object" + }, + "ElementHistoryTarget": { + "properties": { + "device_ids": { + "anyOf": [ + { + "items": { + "type": "string" + }, + "maxItems": 100, + "type": "array" + }, + { + "type": "null" + } + ], + "title": "Device Ids" + }, + "element_id": { + "maxLength": 128, + "minLength": 1, + "title": "Element Id", + "type": "string" + }, + "element_type": { + "$ref": "#/components/schemas/HistoryElementType" + } + }, + "required": [ + "element_id", + "element_type" + ], + "title": "ElementHistoryTarget", + "type": "object" + }, + "HistoryElementType": { + "enum": [ + "pipe", + "junction" + ], + "title": "HistoryElementType", + "type": "string" + }, + "HistoryMetric": { + "enum": [ + "flow", + "pressure" + ], + "title": "HistoryMetric", + "type": "string" + }, + "HistoryMode": { + "enum": [ + "observed", + "realtime_comparison", + "analysis_comparison" + ], + "title": "HistoryMode", + "type": "string" + }, + "HistoryPoint": { + "properties": { + "time": { + "format": "date-time", + "title": "Time", + "type": "string" + }, + "value": { + "anyOf": [ + { + "type": "number" + }, + { + "type": "null" + } + ], + "title": "Value" + } + }, + "required": [ + "time", + "value" + ], + "title": "HistoryPoint", + "type": "object" + }, + "HistorySource": { + "enum": [ + "scada_raw", + "scada_cleaned", + "realtime_simulation", + "analysis_simulation" + ], + "title": "HistorySource", + "type": "string" + }, "JsonValue": {}, "LeakageIdentifyRequestRest": { "properties": { @@ -2409,6 +2627,10 @@ ], "title": "Longitude" }, + "measurement_unit": { + "title": "Measurement Unit", + "type": "string" + }, "node_id": { "anyOf": [ { @@ -2465,6 +2687,7 @@ "required": [ "device_id", "device_type", + "measurement_unit", "transmission_mode", "transmission_frequency" ], @@ -35146,58 +35369,10 @@ ] } }, - "/api/v1/timeseries/views/element-scada-readings": { - "get": { - "description": "获取link/node关联的SCADA监测值\n\n根据传入的link/node id，匹配SCADA信息，\n如果存在关联的SCADA device_id，获取实际的监测数据。\n\nArgs:\n element_id: 管网元素ID\n start_time: 查询开始时间\n end_time: 查询结束时间\n use_cleaned: 是否使用清洗后的数据，默认为False使用原始数据\n timescale_conn: TimescaleDB连接\n postgres_conn: PostgreSQL连接\n \nReturns:\n 管网元素关联的SCADA监测数据\n \nRaises:\n HTTPException: 当查询参数无效时返回400错误，未找到关联数据返回404错误", - "operationId": "get_timeseries_views_element_scada_readings", + "/api/v1/timeseries/views/element-history/query": { + "post": { + "operationId": "post_timeseries_views_element_history_query", "parameters": [ - { - "description": "管网元素ID（管道或节点）", - "in": "query", - "name": "element_id", - "required": true, - "schema": { - "description": "管网元素ID（管道或节点）", - "title": "Element Id", - "type": "string" - } - }, - { - "description": "查询开始时间", - "in": "query", - "name": "start_time", - "required": true, - "schema": { - "description": "查询开始时间", - "format": "date-time", - "title": "Start Time", - "type": "string" - } - }, - { - "description": "查询结束时间", - "in": "query", - "name": "end_time", - "required": true, - "schema": { - "description": "查询结束时间", - "format": "date-time", - "title": "End Time", - "type": "string" - } - }, - { - "description": "是否使用清洗后的数据", - "in": "query", - "name": "use_cleaned", - "required": false, - "schema": { - "default": false, - "description": "是否使用清洗后的数据", - "title": "Use Cleaned", - "type": "boolean" - } - }, { "in": "header", "name": "X-Project-Id", @@ -35208,12 +35383,22 @@ } } ], + "requestBody": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ElementHistoryQuery" + } + } + }, + "required": true + }, "responses": { "200": { "content": { "application/json": { "schema": { - "$ref": "#/components/schemas/JsonValue" + "$ref": "#/components/schemas/ElementHistoryResponse" } } }, @@ -35285,311 +35470,7 @@ "OAuth2PasswordBearer": [] } ], - "summary": "获取管网元素关联的SCADA监测数据", - "tags": [ - "TimescaleDB - Composite" - ] - } - }, - "/api/v1/timeseries/views/element-simulations": { - "get": { - "description": "获取link/node模拟值\n\n根据传入的featureInfos，找到关联的link/node，\n并根据对应的type，查询对应的模拟数据。支持查询实时或分析运行数据。\n\nArgs:\n start_time: 查询开始时间\n end_time: 查询结束时间\n feature_infos: 格式为 \"element_id1:type1,element_id2:type2\"\n 例如: \"P1:pipe,J1:junction\"\n run_id: 分析运行 ID，若为空则查询实时数据\n timescale_conn: TimescaleDB连接\n \nReturns:\n 管网元素的模拟数据\n \nRaises:\n HTTPException: 当feature_infos为空返回400错误，未找到数据返回404错误，其他错误返回400错误", - "operationId": "get_timeseries_views_element_simulations", - "parameters": [ - { - "description": "查询开始时间", - "in": "query", - "name": "start_time", - "required": true, - "schema": { - "description": "查询开始时间", - "format": "date-time", - "title": "Start Time", - "type": "string" - } - }, - { - "description": "查询结束时间", - "in": "query", - "name": "end_time", - "required": true, - "schema": { - "description": "查询结束时间", - "format": "date-time", - "title": "End Time", - "type": "string" - } - }, - { - "description": "特征信息，格式: id1:type1,id2:type2，type为pipe（管道）或junction（节点）", - "in": "query", - "name": "feature_infos", - "required": true, - "schema": { - "description": "特征信息，格式: id1:type1,id2:type2，type为pipe（管道）或junction（节点）", - "title": "Feature Infos", - "type": "string" - } - }, - { - "description": "分析运行 ID；为空时查询实时数据", - "in": "query", - "name": "run_id", - "required": false, - "schema": { - "anyOf": [ - { - "format": "uuid", - "type": "string" - }, - { - "type": "null" - } - ], - "description": "分析运行 ID；为空时查询实时数据", - "title": "Run Id" - } - }, - { - "in": "header", - "name": "X-Project-Id", - "required": true, - "schema": { - "title": "X-Project-Id", - "type": "string" - } - } - ], - "responses": { - "200": { - "content": { - "application/json": { - "schema": { - "$ref": "#/components/schemas/JsonValue" - } - } - }, - "description": "Successful Response" - }, - "401": { - "content": { - "application/json": { - "schema": { - "$ref": "#/components/schemas/ProblemDetails" - } - } - }, - "description": "Authentication required" - }, - "403": { - "content": { - "application/json": { - "schema": { - "$ref": "#/components/schemas/ProblemDetails" - } - } - }, - "description": "Insufficient permission" - }, - "404": { - "content": { - "application/json": { - "schema": { - "$ref": "#/components/schemas/ProblemDetails" - } - } - }, - "description": "Resource not found" - }, - "409": { - "content": { - "application/json": { - "schema": { - "$ref": "#/components/schemas/ProblemDetails" - } - } - }, - "description": "Resource conflict" - }, - "422": { - "content": { - "application/json": { - "schema": { - "$ref": "#/components/schemas/ProblemDetails" - } - } - }, - "description": "Validation error" - }, - "503": { - "content": { - "application/json": { - "schema": { - "$ref": "#/components/schemas/ProblemDetails" - } - } - }, - "description": "Dependency unavailable" - } - }, - "security": [ - { - "OAuth2PasswordBearer": [] - } - ], - "summary": "获取管网元素的模拟数据", - "tags": [ - "TimescaleDB - Composite" - ] - } - }, - "/api/v1/timeseries/views/scada-simulations": { - "get": { - "description": "获取SCADA关联的link/node模拟值\n\n根据传入的SCADA device_ids，找到关联的link/node，\n并根据对应的type，查询对应的模拟数据。支持查询实时或分析运行数据。\n\nArgs:\n start_time: 查询开始时间\n end_time: 查询结束时间\n device_ids: SCADA设备ID列表，用逗号分隔\n run_id: 分析运行 ID，若为空则查询实时数据\n timescale_conn: TimescaleDB连接\n postgres_conn: PostgreSQL连接\n \nReturns:\n SCADA关联的模拟数据\n \nRaises:\n HTTPException: 当查询参数无效时返回400错误，未找到数据时返回404错误", - "operationId": "get_timeseries_views_scada_simulations", - "parameters": [ - { - "description": "查询开始时间", - "in": "query", - "name": "start_time", - "required": true, - "schema": { - "description": "查询开始时间", - "format": "date-time", - "title": "Start Time", - "type": "string" - } - }, - { - "description": "查询结束时间", - "in": "query", - "name": "end_time", - "required": true, - "schema": { - "description": "查询结束时间", - "format": "date-time", - "title": "End Time", - "type": "string" - } - }, - { - "description": "SCADA设备ID列表，逗号分隔", - "in": "query", - "name": "device_ids", - "required": true, - "schema": { - "description": "SCADA设备ID列表，逗号分隔", - "title": "Device Ids", - "type": "string" - } - }, - { - "description": "分析运行 ID；为空时查询实时数据", - "in": "query", - "name": "run_id", - "required": false, - "schema": { - "anyOf": [ - { - "format": "uuid", - "type": "string" - }, - { - "type": "null" - } - ], - "description": "分析运行 ID；为空时查询实时数据", - "title": "Run Id" - } - }, - { - "in": "header", - "name": "X-Project-Id", - "required": true, - "schema": { - "title": "X-Project-Id", - "type": "string" - } - } - ], - "responses": { - "200": { - "content": { - "application/json": { - "schema": { - "$ref": "#/components/schemas/JsonValue" - } - } - }, - "description": "Successful Response" - }, - "401": { - "content": { - "application/json": { - "schema": { - "$ref": "#/components/schemas/ProblemDetails" - } - } - }, - "description": "Authentication required" - }, - "403": { - "content": { - "application/json": { - "schema": { - "$ref": "#/components/schemas/ProblemDetails" - } - } - }, - "description": "Insufficient permission" - }, - "404": { - "content": { - "application/json": { - "schema": { - "$ref": "#/components/schemas/ProblemDetails" - } - } - }, - "description": "Resource not found" - }, - "409": { - "content": { - "application/json": { - "schema": { - "$ref": "#/components/schemas/ProblemDetails" - } - } - }, - "description": "Resource conflict" - }, - "422": { - "content": { - "application/json": { - "schema": { - "$ref": "#/components/schemas/ProblemDetails" - } - } - }, - "description": "Validation error" - }, - "503": { - "content": { - "application/json": { - "schema": { - "$ref": "#/components/schemas/ProblemDetails" - } - } - }, - "description": "Dependency unavailable" - } - }, - "security": [ - { - "OAuth2PasswordBearer": [] - } - ], - "summary": "获取SCADA关联的模拟数据", + "summary": "批量查询管网元素历史数据", "tags": [ "TimescaleDB - Composite" ] diff --git a/resources/db_v2/DATABASE_ARCHITECTURE.md b/resources/db_v2/DATABASE_ARCHITECTURE.md index 3029f63..ee4ee6f 100644 --- a/resources/db_v2/DATABASE_ARCHITECTURE.md +++ b/resources/db_v2/DATABASE_ARCHITECTURE.md @@ -1,6 +1,6 @@ # TJWater 数据库改造说明与当前结构 -> 本文记录截至 2026-09-10 的数据库实际状态。结构、约束、行数、TimescaleDB chunk 和策略均直接读取数据库，不以仓库中的 SQL 脚本为依据。文中不包含主机、端口、账号、密码或 DSN。 +> 本文记录截至 2026-09-14 的数据库实际状态。结构、约束、行数、TimescaleDB chunk 和策略均直接读取数据库，不以仓库中的 SQL 脚本为依据。文中不包含主机、端口、账号、密码或 DSN。 ## 改造范围与当前状态 @@ -20,10 +20,11 @@ - `system_hub.public` 补充了项目数据库外键、数据库路由约束、连接池约束、必要的非空约束，以及 5 张表和 44 个字段的中文数据库注释。 - 用户角色和项目角色仍是可扩展字符串，没有增加枚举检查约束。 - `audit_logs.user_id` 和 `audit_logs.project_id` 仍为逻辑关联，没有增加外键。 -- 业务库 48 个表或物化视图、192 个字段，以及时序库 7 张表、47 个字段均已写入中文数据库注释。 +- 业务库 48 个表或物化视图、194 个字段，以及时序库 7 张表、47 个字段均已写入中文数据库注释。 - 后端已对接新 schema。WNDB、PostgreSQL 管理连接和同步 TimescaleDB 访问均使用有界连接池，闲置项目按最近使用顺序回收；动态异步池采用代际切换，配置更新不会中断旧借用或阻塞新请求。 - 后端批量元素查询读取 GIS 物化视图，模型增删改和 INP 导入提交后执行并发刷新；批量事务只刷新一次。 - `pattern_values`、`pattern_flow_samples`、`curve_points`、`demands` 和 `link_vertices` 的顺序号按所属父对象编号，主键已改为父对象 ID 与 `sequence_no` 的复合键。 +- 5 个活跃业务库、5 个对应管网模板库和 `tjwater_v2_schema_template` 均已增加 `asset.scada_devices.measurement_unit`；压力设备回填为 `m`，流量设备回填为 `m3/h`，`gis.scada_devices` 物化视图同步暴露该字段。 逻辑项目 `tjwater_v2` 当前为 `active`。 @@ -359,15 +360,27 @@ TimescaleDB 的空结构模板为 `tjwater_v2_timescale_template`，普通连接 方案查询统一读取 `/api/v1/analysis/runs`，详情和时序结果按 UUID `run_id` 关联。时间轴读取 `/api/v1/timeseries/analysis/runs/{run_id}/values`，爆管定位的模拟数据源也传递 `simulation_run_id`。监测点优化使用 `/api/v1/sensor-placement-runs`。前端不再调用旧的 `/schemes`、`/timeseries/schemes` 和 `/sensor-placement-schemes` 接口。 +管网元素历史查询统一使用 `POST /api/v1/timeseries/views/element-history/query`。单次请求批量返回元素关联的原始/清洗 SCADA、实时模拟或指定分析运行时序，并显式返回 `source_unit`、`display_unit` 和 `unit_inferred`。点值仍使用 `source_unit`，前端由共享单位模块统一换算为流量 `m³/h`、压力 `m` 和流速 `m/s`。旧的 `element-simulations`、`element-scada-readings` 和 `scada-simulations` 组合接口已移除。 + +当前业务库 `network.simulation_settings` 中的模型结果源单位如下。`LPS` 表示 L/s，`MLD` 表示百万升/日；公制模型的流速源单位为 m/s。前端不得再假定所有项目都是 `LPS`，而应读取项目设置并通过共享单位模块换算。 + +| 业务库 | 流量源单位 | 压力源单位 | 前端显示单位 | +| --- | --- | --- | --- | +| `fengyang_v2` | `LPS` | `METERS` | m³/h、m、m/s | +| `md_v2` | `LPS` | `METERS` | m³/h、m、m/s | +| `szh_v2` | `MLD` | `METERS` | m³/h、m、m/s | +| `tjwater_v2` | `LPS` | `METERS` | m³/h、m、m/s | +| `zjb` | `LPS` | `METERS` | m³/h、m、m/s | + 模型修改提交后，后端先调用 `gis.refresh_all_materialized_views(boolean)` 刷新数据库查询层。GeoWebCache 不会感知 PostgreSQL 物化视图刷新，因此 7 个新图层配置了 300 秒的服务端和客户端缓存有效期，前端最迟在 5 分钟后读到新瓦片。部署或批量迁移完成后仍可执行一次图层缓存清空，避免等待已有瓦片自然过期。 ### asset：SCADA 设备配置 -`asset.scada_devices` 保存设备类型、采集接口标识、传输模式、频率、可靠性和可选几何。每台设备必须关联一个节点或一条连接，不能同时关联两者。设备的历史测量值不放在业务库，保存在时序库 `scada.measurements`。 +`asset.scada_devices` 保存设备类型、采集接口标识、传输模式、频率、可靠性、测量源单位 `measurement_unit` 和可选几何。单位字段非空且不允许空字符串。每台设备必须关联一个节点或一条连接，不能同时关联两者。设备的历史测量值不放在业务库，保存在时序库 `scada.measurements`。 ### analysis：分析运行和非时序结果 -`analysis.runs` 表示一次实际执行，保存运行名称、类型、创建人、开始时间、状态和参数。当前没有单独的 `scenarios` 模板表。 +`analysis.runs` 表示一次实际执行，保存运行名称、类型、创建人、开始时间、状态和参数。新执行会在 `parameters.result_units` 中保存 EPANET 输出的流量和压力源单位，避免日后调整模型设置时误读历史值。旧运行没有该元数据时，API 使用当前模型单位并返回 `unit_inferred=true`。当前没有单独的 `scenarios` 模板表。 每次执行都会创建新的 `run_id`，名称和业务时间相同也不会覆盖旧结果。扩展仿真开始写结果前，业务库先记录 `running`；时序库写入完成后更新为 `completed`，写入失败则保留为 `failed`。业务库和时序库据此共用同一个执行标识。 diff --git a/tests/unit/test_analysis_simulation.py b/tests/unit/test_analysis_simulation.py index e6042cf..c15cfaa 100644 --- a/tests/unit/test_analysis_simulation.py +++ b/tests/unit/test_analysis_simulation.py @@ -159,6 +159,7 @@ def test_extended_simulation_stores_results_by_run_id(monkeypatch): lambda name: json.dumps( { "output": { + "units": {"flow": "LPS", "pressure": "MTR"}, "times": {"num_periods": 2, "report_step": 900}, "node_results": [{"node": "J1", "result": [{}, {}]}], "link_results": [{"link": "P1", "result": [{}, {}]}], @@ -167,7 +168,12 @@ def test_extended_simulation_stores_results_by_run_id(monkeypatch): ), ) lifecycle_calls: list[tuple] = [] - monkeypatch.setattr(simulation, "create_analysis_run", lambda **kwargs: run_id) + create_calls: list[dict] = [] + monkeypatch.setattr( + simulation, + "create_analysis_run", + lambda **kwargs: (create_calls.append(kwargs), run_id)[1], + ) monkeypatch.setattr( simulation, "update_analysis_run", @@ -195,6 +201,10 @@ def test_extended_simulation_stores_results_by_run_id(monkeypatch): assert kwargs["db_name"] == "demo" assert returned_run_id == run_id assert lifecycle_calls[-1][1]["status"] == "completed" + assert create_calls[0]["scheme_detail"]["result_units"] == { + "flow": "LPS", + "pressure": "MTR", + } assert transaction_calls == [("begin", "demo"), ("end", "demo")] refresh_mock.assert_called_once_with("demo") diff --git a/tests/unit/test_measurement_units.py b/tests/unit/test_measurement_units.py new file mode 100644 index 0000000..b08a9e4 --- /dev/null +++ b/tests/unit/test_measurement_units.py @@ -0,0 +1,7 @@ +from app.domain.measurement_units import display_unit_for_metric + + +def test_display_units_are_stable_ui_units(): + assert display_unit_for_metric("flow") == "m³/h" + assert display_unit_for_metric("pressure") == "m" + assert display_unit_for_metric("velocity") == "m/s" diff --git a/tests/unit/test_postgres_scada_repository.py b/tests/unit/test_postgres_scada_repository.py index bd0ef93..cbee847 100644 --- a/tests/unit/test_postgres_scada_repository.py +++ b/tests/unit/test_postgres_scada_repository.py @@ -29,6 +29,7 @@ class _FakeCursor: "node_id": " J1 ", "link_id": None, "api_query_id": "query-1", + "measurement_unit": "m", "transmission_mode": "realtime", "transmission_frequency": None, "reliability": "95", @@ -63,6 +64,7 @@ def test_get_scadas_normalizes_id_and_type(): "node_id": "J1", "link_id": None, "api_query_id": "query-1", + "measurement_unit": "m", "transmission_mode": "realtime", "transmission_frequency": None, "reliability": 95, diff --git a/tests/unit/test_project_scada_metadata.py b/tests/unit/test_project_scada_metadata.py index 57650b8..ffd847a 100644 --- a/tests/unit/test_project_scada_metadata.py +++ b/tests/unit/test_project_scada_metadata.py @@ -13,6 +13,7 @@ PROJECT_SCADA = { "node_id": "J1", "link_id": None, "api_query_id": "query-1", + "measurement_unit": "m", "transmission_mode": "realtime", "transmission_frequency": None, "reliability": 1.0, diff --git a/tests/unit/test_timeseries_history.py b/tests/unit/test_timeseries_history.py new file mode 100644 index 0000000..31aa6f8 --- /dev/null +++ b/tests/unit/test_timeseries_history.py @@ -0,0 +1,146 @@ +import asyncio +from datetime import UTC, datetime, timedelta +from unittest.mock import AsyncMock + +import pytest + +from app.domain.schemas.timeseries_history import ElementHistoryQuery +from app.services import timeseries_history + + +START = datetime(2026, 9, 1, tzinfo=UTC) +END = START + timedelta(hours=1) + + +def test_realtime_history_batches_devices_and_converts_units(monkeypatch): + monkeypatch.setattr( + timeseries_history.ScadaInfoRepository, + "get_scadas_for_elements", + AsyncMock( + return_value=[ + { + "device_id": "F-1", + "device_type": "pipe_flow", + "link_id": "P-1", + "node_id": None, + "measurement_unit": "m3/h", + }, + { + "device_id": "P-1A", + "device_type": "pressure", + "link_id": None, + "node_id": "J-1", + "measurement_unit": "m", + }, + { + "device_id": "P-1B", + "device_type": "pressure", + "link_id": None, + "node_id": "J-1", + "measurement_unit": "m", + }, + ] + ), + ) + scada_query = AsyncMock( + return_value=[ + { + "time": START, + "device_id": "F-1", + "monitored_value": 36.0, + "cleaned_value": 35.0, + }, + { + "time": START, + "device_id": "P-1A", + "monitored_value": 20.0, + "cleaned_value": 21.0, + }, + { + "time": START, + "device_id": "P-1B", + "monitored_value": 22.0, + "cleaned_value": 23.0, + }, + ] + ) + monkeypatch.setattr( + timeseries_history.ScadaRepository, + "get_scada_by_ids_time_range", + scada_query, + ) + monkeypatch.setattr( + timeseries_history.NetworkSettingsRepository, + "get_result_units", + AsyncMock(return_value={"flow": "MLD", "pressure": "METERS"}), + ) + monkeypatch.setattr( + timeseries_history.RealtimeRepository, + "get_link_fields_by_ids_time_range", + AsyncMock(return_value={"P-1": [{"time": START, "value": 2.0}]}), + ) + monkeypatch.setattr( + timeseries_history.RealtimeRepository, + "get_node_fields_by_ids_time_range", + AsyncMock(return_value={"J-1": [{"time": START, "value": 24.0}]}), + ) + + result = asyncio.run( + timeseries_history.TimeseriesHistoryService.query( + object(), + object(), + ElementHistoryQuery( + start_time=START, + end_time=END, + mode="realtime_comparison", + elements=[ + {"element_id": "P-1", "element_type": "pipe"}, + {"element_id": "J-1", "element_type": "junction"}, + ], + ), + ) + ) + + assert scada_query.await_count == 1 + assert set(scada_query.await_args.args[1]) == {"F-1", "P-1A", "P-1B"} + assert {item.device_id for item in result.series if item.device_id} == { + "F-1", + "P-1A", + "P-1B", + } + simulation = { + (item.element_id, item.metric.value): item + for item in result.series + if item.source.value == "realtime_simulation" + } + assert simulation[("P-1", "flow")].points[0].value == 2.0 + assert simulation[("P-1", "flow")].source_unit == "MLD" + assert simulation[("J-1", "pressure")].points[0].value == 24.0 + assert all(item.display_unit in {"m³/h", "m"} for item in result.series) + + +def test_requested_device_must_belong_to_element(monkeypatch): + monkeypatch.setattr( + timeseries_history.ScadaInfoRepository, + "get_scadas_for_elements", + AsyncMock(return_value=[]), + ) + + with pytest.raises(ValueError, match="do not belong"): + asyncio.run( + timeseries_history.TimeseriesHistoryService.query( + object(), + object(), + ElementHistoryQuery( + start_time=START, + end_time=END, + elements=[ + { + "element_id": "J-1", + "element_type": "junction", + "device_ids": ["missing"], + } + ], + ), + ) + ) diff --git a/tests/unit/test_timeseries_history_schema.py b/tests/unit/test_timeseries_history_schema.py new file mode 100644 index 0000000..774735d --- /dev/null +++ b/tests/unit/test_timeseries_history_schema.py @@ -0,0 +1,50 @@ +from datetime import UTC, datetime, timedelta +from uuid import uuid4 + +import pytest +from pydantic import ValidationError + +from app.domain.schemas.timeseries_history import ElementHistoryQuery + + +def _query(**overrides): + start = datetime(2026, 1, 1, tzinfo=UTC) + values = { + "start_time": start, + "end_time": start + timedelta(hours=1), + "mode": "observed", + "elements": [{"element_id": " P1 ", "element_type": "pipe"}], + } + values.update(overrides) + return ElementHistoryQuery(**values) + + +def test_history_query_normalizes_targets_and_device_ids(): + query = _query( + elements=[ + { + "element_id": " P1 ", + "element_type": "pipe", + "device_ids": [" D1 ", "D1", "D2"], + } + ] + ) + + assert query.elements[0].element_id == "P1" + assert query.elements[0].device_ids == ["D1", "D2"] + + +def test_analysis_comparison_requires_run_id(): + with pytest.raises(ValidationError, match="run_id is required"): + _query(mode="analysis_comparison") + + +def test_other_modes_reject_run_id(): + with pytest.raises(ValidationError, match="only valid"): + _query(run_id=uuid4()) + + +def test_history_query_rejects_reversed_time_range(): + start = datetime(2026, 1, 1, tzinfo=UTC) + with pytest.raises(ValidationError, match="start_time must be earlier"): + _query(start_time=start, end_time=start)