From f5f972bbec0913f71892374000b12f06eba7685d Mon Sep 17 00:00:00 2001 From: Huarch Date: Sun, 15 Nov 2026 23:36:29 +0800 Subject: [PATCH] feat(api): add compact hydraulic playback endpoints --- app/api/v1/endpoints/timeseries/playback.py | 69 +++++++ app/api/v1/router.py | 2 + app/services/hydraulic_playback.py | 184 ++++++++++++++++++ .../api/test_hydraulic_playback_endpoints.py | 79 ++++++++ tests/unit/test_hydraulic_playback.py | 93 +++++++++ 5 files changed, 427 insertions(+) create mode 100644 app/api/v1/endpoints/timeseries/playback.py create mode 100644 app/services/hydraulic_playback.py create mode 100644 tests/api/test_hydraulic_playback_endpoints.py create mode 100644 tests/unit/test_hydraulic_playback.py diff --git a/app/api/v1/endpoints/timeseries/playback.py b/app/api/v1/endpoints/timeseries/playback.py new file mode 100644 index 0000000..8129f67 --- /dev/null +++ b/app/api/v1/endpoints/timeseries/playback.py @@ -0,0 +1,69 @@ +"""Project-authorized compact hydraulic playback transport.""" + +from collections import OrderedDict +import asyncio +import time + +from fastapi import APIRouter, Depends, HTTPException, Query, Response +from psycopg import AsyncConnection + +from app.auth.project_dependencies import ProjectContext, get_project_context, get_project_pg_connection +from app.services.hydraulic_playback import MAX_FRAME_COUNT, PlaybackGeometry, read_frames, read_geometry +from app.domain.time import parse_utc_time +from .dependencies import get_timescale_connection + +router = APIRouter(prefix="/timeseries/realtime/playback") +# Bounded per-process cache, private to the authorized project and database route. +_cache: OrderedDict[tuple, tuple[float, PlaybackGeometry]] = OrderedDict() +_load_lock = asyncio.Lock() + + +async def geometry_for_project(ctx: ProjectContext, conn: AsyncConnection) -> PlaybackGeometry: + key = (str(ctx.project_id), conn.info.host, conn.info.port, conn.info.dbname) + cached = _cache.get(key) + if cached and time.monotonic() - cached[0] < 60: + _cache.move_to_end(key) + return cached[1] + async with _load_lock: + cached = _cache.get(key) + if cached and time.monotonic() - cached[0] < 60: + return cached[1] + geometry = await read_geometry(conn) + _cache[key] = (time.monotonic(), geometry) + _cache.move_to_end(key) + while len(_cache) > 4: + _cache.popitem(last=False) + return geometry + + +@router.get("/geometry", summary="读取带版本的水力专题层静态几何") +async def playback_geometry(ctx: ProjectContext = Depends(get_project_context), + conn: AsyncConnection = Depends(get_project_pg_connection)): + try: + geometry = await geometry_for_project(ctx, conn) + except ValueError as exc: + raise HTTPException(422, str(exc)) from exc + return Response(geometry.content, media_type="application/octet-stream", + headers={"Cache-Control": "private, no-store", "ETag": f'"{geometry.version}"'}) + + +@router.get("/frames", summary="读取索引一致的 Float32 水力时间块") +async def playback_frames( + start_time: str, + model_version: str = Query(min_length=24, max_length=24), + frame_count: int = Query(default=4, ge=1, le=MAX_FRAME_COUNT), + step_seconds: int = Query(default=900, ge=60, le=3600), + ctx: ProjectContext = Depends(get_project_context), + pg_conn: AsyncConnection = Depends(get_project_pg_connection), + ts_conn: AsyncConnection = Depends(get_timescale_connection), +): + try: + start = parse_utc_time(start_time, field_name="start_time") + geometry = await geometry_for_project(ctx, pg_conn) + if geometry.version != model_version: + raise HTTPException(409, "管网模型已更新，请重新加载专题层") + content = await read_frames(ts_conn, geometry, start, frame_count, step_seconds) + except ValueError as exc: + raise HTTPException(422, str(exc)) from exc + return Response(content, media_type="application/octet-stream", + headers={"Cache-Control": "private, no-store"}) diff --git a/app/api/v1/router.py b/app/api/v1/router.py index 3c5311f..f0fdf75 100644 --- a/app/api/v1/router.py +++ b/app/api/v1/router.py @@ -47,6 +47,7 @@ from app.api.v1.endpoints.timeseries import ( analysis as ts_analysis, composite as ts_composite, realtime as ts_realtime, + playback as ts_playback, scada as ts_scada, ) from app.auth.permissions import ( @@ -191,6 +192,7 @@ api_router.include_router( # TimescaleDB data for endpoint_router, tag in ( (ts_realtime.router, "TimescaleDB - Realtime"), + (ts_playback.router, "TimescaleDB - Hydraulic Playback"), (ts_analysis.router, "TimescaleDB - Analysis"), ): api_router.include_router( diff --git a/app/services/hydraulic_playback.py b/app/services/hydraulic_playback.py new file mode 100644 index 0000000..4135f3a --- /dev/null +++ b/app/services/hydraulic_playback.py @@ -0,0 +1,184 @@ +"""Versioned static geometry and compact, indexed hydraulic playback frames.""" + +from array import array +import asyncio +from dataclasses import dataclass +from datetime import datetime, timedelta +import hashlib +import json +import math +import struct +import sys +from typing import Any + +from psycopg import AsyncConnection + +from app.infra.db.postgresql.network_settings import NetworkSettingsRepository + +WORLD_METERS = 40075016.68557849 +MAX_FRAME_COUNT = 8 + + +def packet(header: dict[str, Any], values: array) -> bytes: + """Four-byte aligned JSON header followed by little-endian Float32 values.""" + encoded = json.dumps(header, separators=(",", ":"), ensure_ascii=False).encode() + encoded += b" " * (-len(encoded) % 4) + if sys.byteorder != "little": + values = array("f", values) + values.byteswap() + return struct.pack(" tuple[float, float]: + latitude = max(-85.05112878, min(85.05112878, latitude)) + return ((longitude + 180) / 360, + (1 - math.asinh(math.tan(math.radians(latitude))) / math.pi) / 2) + + +@dataclass(frozen=True) +class PlaybackGeometry: + version: str + node_ids: tuple[str, ...] + pipe_ids: tuple[str, ...] + node_index: dict[str, int] + pipe_index: dict[str, int] + content: bytes + flow_factor: float + pressure_factor: float + velocity_factor: float + + +def result_factors(units: dict[str, str]) -> tuple[float, float, float]: + """Match EPANET outputs to m³/h, metres of water, and m/s.""" + flow = units["flow"].upper() + flow_factors = {"LPS": 3.6, "LPM": .06, "MLD": 1000 / 24, + "CMH": 1, "CMD": 1 / 24, "CFS": .028316846592 * 3600, + "GPM": .003785411784 * 60, "MGD": 3785.411784 / 24, + "IMGD": 4546.09 / 24, "AFD": 1233.48183754752 / 24} + pressure = units["pressure"].upper() + pressure_factors = {"M": 1, "METERS": 1, "PSI": .7032496149, "KPA": .1019716213} + if flow not in flow_factors or pressure not in pressure_factors: + raise ValueError("不支持的模拟结果单位") + return (flow_factors[flow], pressure_factors[pressure], + .3048 if flow in {"CFS", "GPM", "MGD", "IMGD", "AFD"} else 1) + + +def line_paths(geometry: dict) -> list: + if geometry["type"] == "LineString": + return [geometry["coordinates"]] + if geometry["type"] == "MultiLineString": + return geometry["coordinates"] + if geometry["type"] == "GeometryCollection": + # PostGIS LineMerge returns an empty collection for zero-length lines. + return [path for child in geometry["geometries"] for path in line_paths(child)] + raise ValueError("管线几何必须为线类型") + + +def build_geometry(nodes: list[dict], pipes: list[dict], units: dict[str, str]) -> PlaybackGeometry: + if not nodes or not pipes: + raise ValueError("管网没有可发布的节点或管线几何") + node_ids = tuple(str(row["id"]) for row in nodes) + pipe_ids = tuple(str(row["id"]) for row in pipes) + node_positions = [mercator(float(row["longitude"]), float(row["latitude"])) for row in nodes] + positions_by_id = dict(zip(node_ids, node_positions)) + origin = [(min(p[i] for p in node_positions) + max(p[i] for p in node_positions)) / 2 for i in (0, 1)] + values = array("f", (coordinate - origin[axis] for position in node_positions for axis, coordinate in enumerate(position))) + segment_count = 0 + for index, row in enumerate(pipes): + geometry = row["geometry"] + if isinstance(geometry, str): + geometry = json.loads(geometry) + paths = line_paths(geometry) + start_node = positions_by_id.get(str(row.get("start_node_id", ""))) + if start_node and paths and paths[0] and paths[-1]: + first = mercator(*paths[0][0][:2]) + last = mercator(*paths[-1][-1][:2]) + if math.dist(last, start_node) < math.dist(first, start_node): + paths = [list(reversed(path)) for path in reversed(paths)] + distance = 0.0 + for path in paths: + positions = [mercator(float(p[0]), float(p[1])) for p in path] + for start, end in zip(positions, positions[1:]): + length = math.hypot(end[0] - start[0], end[1] - start[1]) + if length <= 0: + continue + # Instanced segment: endpoints relative to origin, pipe index, + # cumulative distance, segment length, physical diameter (mm). + values.extend((start[0] - origin[0], start[1] - origin[1], + end[0] - origin[0], end[1] - origin[1], + index, distance, length, float(row["diameter"] or 0))) + distance += length + segment_count += 1 + factors = result_factors(units) + descriptor = json.dumps([node_ids, pipe_ids, origin, units], separators=(",", ":")).encode() + version = hashlib.sha256(descriptor + values.tobytes()).hexdigest()[:24] + header = {"kind": "geometry", "protocol": 1, "modelVersion": version, + "nodeIds": node_ids, "pipeIds": pipe_ids, "origin": origin, + "segmentCount": segment_count, "units": {"pressure": "m", "flow": "m³/h", "velocity": "m/s"}} + return PlaybackGeometry(version, node_ids, pipe_ids, + {key: i for i, key in enumerate(node_ids)}, + {key: i for i, key in enumerate(pipe_ids)}, + packet(header, values), *factors) + + +async def read_geometry(conn: AsyncConnection) -> PlaybackGeometry: + async with conn.cursor() as cursor: + await cursor.execute(""" + SELECT n.id, ST_X(ST_Transform(g.geom,4326)) AS longitude, + ST_Y(ST_Transform(g.geom,4326)) AS latitude + FROM gis.network_nodes n JOIN gis.node_geometries g ON g.node_id=n.id + WHERE g.geom IS NOT NULL AND NOT ST_IsEmpty(g.geom) ORDER BY n.id + """) + nodes = await cursor.fetchall() + await cursor.execute(""" + SELECT id, start_node_id, diameter, ST_AsGeoJSON(ST_Transform(ST_LineMerge(geom),4326))::json AS geometry + FROM gis.pipes WHERE geom IS NOT NULL AND NOT ST_IsEmpty(geom) ORDER BY id + """) + pipes = await cursor.fetchall() + units = await NetworkSettingsRepository.get_result_units(conn) + return await asyncio.to_thread(build_geometry, nodes, pipes, units) + + +def finite(value: Any, factor: float) -> float: + if value is None: + return math.nan + result = float(value) * factor + return result if math.isfinite(result) and abs(result) <= 3.402823466e38 else math.nan + + +async def read_frames(conn: AsyncConnection, geometry: PlaybackGeometry, start: datetime, + frame_count: int, step_seconds: int) -> bytes: + times = [start + timedelta(seconds=step_seconds * i) for i in range(frame_count)] + node_count, pipe_count = len(geometry.node_ids), len(geometry.pipe_ids) + stride = node_count + 2 * pipe_count + values = array("f", [math.nan]) * (stride * frame_count) + time_index = {value: i for i, value in enumerate(times)} + valid_counts = {metric: [0] * frame_count for metric in ("pressure", "flow", "velocity")} + async with conn.cursor() as cursor: + await cursor.execute(""" + SELECT time, node_id, pressure FROM realtime.node_results + WHERE time = ANY(%s) ORDER BY time, node_id + """, (times,)) + async for row in cursor: + index = geometry.node_index.get(str(row["node_id"])) + if index is not None: + frame = time_index[row["time"]] + values[frame * stride + index] = finite(row["pressure"], geometry.pressure_factor) + valid_counts["pressure"][frame] += math.isfinite(values[frame * stride + index]) + await cursor.execute(""" + SELECT time, link_id, flow, velocity FROM realtime.link_results + WHERE time = ANY(%s) ORDER BY time, link_id + """, (times,)) + async for row in cursor: + index = geometry.pipe_index.get(str(row["link_id"])) + if index is not None: + frame = time_index[row["time"]] + offset = frame * stride + node_count + values[offset + index] = finite(row["flow"], geometry.flow_factor) + values[offset + pipe_count + index] = finite(row["velocity"], geometry.velocity_factor) + valid_counts["flow"][frame] += math.isfinite(values[offset + index]) + valid_counts["velocity"][frame] += math.isfinite(values[offset + pipe_count + index]) + return packet({"kind": "frames", "protocol": 1, "modelVersion": geometry.version, + "nodeCount": node_count, "pipeCount": pipe_count, + "times": [int(t.timestamp() * 1000) for t in times], "validCounts": valid_counts, + "available": [valid_counts["pressure"][i] > 0 and valid_counts["flow"][i] > 0 for i in range(frame_count)]}, values) diff --git a/tests/api/test_hydraulic_playback_endpoints.py b/tests/api/test_hydraulic_playback_endpoints.py new file mode 100644 index 0000000..6a9d098 --- /dev/null +++ b/tests/api/test_hydraulic_playback_endpoints.py @@ -0,0 +1,79 @@ +from types import SimpleNamespace +from uuid import uuid4 +import asyncio + +from fastapi.testclient import TestClient +import pytest + +from app.api.v1.endpoints.timeseries import playback +from app.auth.project_dependencies import ProjectContext +from tests.conftest import build_test_app + + +def client(monkeypatch): + context = ProjectContext(project_id=uuid4(), project_code="test", user_id=uuid4(), project_role="viewer") + app = build_test_app(playback.router, "/api/v1") + app.dependency_overrides[playback.get_project_context] = lambda: context + app.dependency_overrides[playback.get_project_pg_connection] = lambda: object() + app.dependency_overrides[playback.get_timescale_connection] = lambda: object() + async def geometry(*_): + return SimpleNamespace(version="a" * 24, content=b"geometry") + monkeypatch.setattr(playback, "geometry_for_project", geometry) + return TestClient(app) + + +def test_geometry_is_private_and_frames_reject_a_different_model_version(monkeypatch): + http = client(monkeypatch) + response = http.get("/api/v1/timeseries/realtime/playback/geometry") + assert response.status_code == 200 + assert response.headers["cache-control"] == "private, no-store" + assert response.content == b"geometry" + response = http.get("/api/v1/timeseries/realtime/playback/frames", params={ + "start_time": "2026-07-16T00:00:00+08:00", "model_version": "b" * 24, + }) + assert response.status_code == 409 + + +@pytest.mark.parametrize("params", [ + {"start_time": "2026-07-16T00:00:00"}, + {"start_time": "2026-07-16T00:00:00Z", "frame_count": 9}, + {"start_time": "2026-07-16T00:00:00Z", "step_seconds": 0}, +]) +def test_frames_require_timezone_and_bound_the_block_size(monkeypatch, params): + http = client(monkeypatch) + response = http.get("/api/v1/timeseries/realtime/playback/frames", params={"model_version": "a" * 24, **params}) + assert response.status_code == 422 + + +def test_frames_pass_exact_utc_time_and_counts_to_reader(monkeypatch): + http = client(monkeypatch) + async def frames(_connection, geometry, start, count, step): + assert geometry.version == "a" * 24 + assert start.isoformat() == "2026-07-15T16:00:00+00:00" + assert count == 4 and step == 900 + return b"frames" + monkeypatch.setattr(playback, "read_frames", frames) + response = http.get("/api/v1/timeseries/realtime/playback/frames", params={ + "start_time": "2026-07-16T00:00:00+08:00", "model_version": "a" * 24, + }) + assert response.status_code == 200 and response.content == b"frames" + + +def test_cache_separates_projects_and_database_routes(monkeypatch): + playback._cache.clear() + calls = [] + async def read(conn): + calls.append(conn.info.dbname) + return SimpleNamespace(version=conn.info.dbname) + monkeypatch.setattr(playback, "read_geometry", read) + ctx = SimpleNamespace(project_id=uuid4()) + async def run(): + for name in ["one", "one", "two"]: + connection = SimpleNamespace(info=SimpleNamespace(host="host", port=5432, dbname=name)) + result = await playback.geometry_for_project(ctx, connection) + assert result.version == name + other = SimpleNamespace(project_id=uuid4()) + await playback.geometry_for_project(other, SimpleNamespace(info=SimpleNamespace(host="host", port=5432, dbname="one"))) + asyncio.run(run()) + assert calls == ["one", "two", "one"] + playback._cache.clear() diff --git a/tests/unit/test_hydraulic_playback.py b/tests/unit/test_hydraulic_playback.py new file mode 100644 index 0000000..d37228f --- /dev/null +++ b/tests/unit/test_hydraulic_playback.py @@ -0,0 +1,93 @@ +import asyncio +from datetime import datetime, timezone +import json +import math +import struct + +import pytest + +from app.services.hydraulic_playback import build_geometry, read_frames, result_factors + + +def fixture_geometry(longitude=121.5): + return build_geometry([ + {"id": "J1", "longitude": longitude, "latitude": 30.8}, + {"id": "J2", "longitude": longitude + .001, "latitude": 30.8}, + ], [{"id": "P1", "start_node_id": "J1", "diameter": 600, + "geometry": {"type": "LineString", "coordinates": [[longitude + .001, 30.8], [longitude, 30.8]]}}], + {"flow": "LPS", "pressure": "METERS"}) + + +def unpack(content): + length = struct.unpack_from(" J2 + assert values[6:8] == values[2:4] + assert fixture_geometry().version == geometry.version + assert fixture_geometry(121.6).version != geometry.version + + +def test_unit_factors_preserve_signed_flow_and_reject_unknown_units(): + assert result_factors({"flow": "LPS", "pressure": "METERS"}) == (3.6, 1, 1) + flow, pressure, velocity = result_factors({"flow": "GPM", "pressure": "PSI"}) + assert flow == pytest.approx(.003785411784 * 60) + assert pressure == pytest.approx(.7032496149) + assert velocity == .3048 + with pytest.raises(ValueError): + result_factors({"flow": "unknown", "pressure": "METERS"}) + + +class Cursor: + def __init__(self, time): + self.time = time + self.rows = [] + async def __aenter__(self): + return self + async def __aexit__(self, *_): + pass + async def execute(self, query, params): + assert len(params[0]) == 2 + if "node_results" in query: + self.rows = [{"time": self.time, "node_id": "J1", "pressure": 25}, + {"time": self.time, "node_id": "J2", "pressure": None}] + else: + self.rows = [{"time": self.time, "link_id": "P1", "flow": -10, "velocity": .7}] + def __aiter__(self): + self.iterator = iter(self.rows) + return self + async def __anext__(self): + try: + return next(self.iterator) + except StopIteration: + raise StopAsyncIteration + + +class Connection: + def __init__(self, time): + self.time = time + def cursor(self): + return Cursor(self.time) + + +def test_frames_pack_exact_sample_times_signed_values_and_missing_values_without_extrapolation(): + time = datetime(2026, 7, 15, 16, tzinfo=timezone.utc) + content = asyncio.run(read_frames(Connection(time), fixture_geometry(), time, 2, 900)) + header, values = unpack(content) + assert header["times"] == [1784131200000, 1784132100000] + assert header["available"] == [True, False] + assert header["validCounts"]["pressure"] == [1, 0] + assert values[0] == 25 + assert math.isnan(values[1]) + assert values[2] == -36 + assert values[3] == pytest.approx(.7) + assert all(math.isnan(value) for value in values[4:])