OrgTJWater/TJWaterServerBinary
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(timeseries): unify element history queries

Browse Source
This commit is contained in:
jiang
2026-09-14 12:30:41 +08:00
parent 0685f6dd17
commit e154676332
17 changed files with 888 additions and 531 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/api/v1/endpoints/timeseries/composite.py
+17 -166
View File
@@ -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监测数据")
app/domain/measurement_units.py
+13
View File
@@ -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]
app/domain/schemas/scada.py
+1
View File
@@ -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
app/domain/schemas/timeseries_history.py
+88
View File
@@ -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]
app/infra/db/postgresql/network_settings.py
+32
View File
@@ -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
app/infra/db/postgresql/scada.py
+19 -1
View File
@@ -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},
app/services/simulation.py
+8 -1
View File
@@ -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,
app/services/timeseries_history.py
+237
View File
@@ -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)
contracts/manifest.json
+1 -1
View File
@@ -3,7 +3,7 @@
"contracts": {
"server": {
"file": "server-v1.openapi.json",
"sha256": "fb720e3009948c3cb2f37674d0bbdbb7b6136973223eccca89880475697435bc"
"sha256": "b565d841061c9091f48ff3118bcc0cbb1b918b0cb8c2316d177570e8b2d8ba29"
}
}
}
contracts/server-v1.openapi.json
+238 -357
View File
@@ -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"
]
resources/db_v2/DATABASE_ARCHITECTURE.md
+17 -4
View File
@@ -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`。业务库和时序库据此共用同一个执行标识。
tests/unit/test_analysis_simulation.py
+11 -1
View File
@@ -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")
tests/unit/test_measurement_units.py
+7
View File
@@ -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"
tests/unit/test_postgres_scada_repository.py
+2
View File
@@ -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,
tests/unit/test_project_scada_metadata.py
+1
View File
@@ -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,
tests/unit/test_timeseries_history.py
+146
View File
@@ -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"],
}
],
),
)
)
tests/unit/test_timeseries_history_schema.py
+50
View File
@@ -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)