refactor(db): isolate hydraulic playback data access

app/api/v1/endpoints/timeseries/playback.py

@@ -13,6 +13,8 @@ from app.domain.time import parse_utc_time
 from .dependencies import get_timescale_connection
 
 router = APIRouter(prefix="/timeseries/realtime/playback")
+BINARY_RESPONSE = {200: {"description": "Versioned little-endian Float32 packet",
+                         "content": {"application/octet-stream": {"schema": {"type": "string", "format": "binary"}}}}}
 # Bounded per-process cache, private to the authorized project and database route.
 _cache: OrderedDict[tuple, tuple[float, PlaybackGeometry]] = OrderedDict()
 _load_lock = asyncio.Lock()
@@ -36,7 +38,8 @@ async def geometry_for_project(ctx: ProjectContext, conn: AsyncConnection) -> Pl
         return geometry
 
 
-@router.get("/geometry", summary="读取带版本的水力专题层静态几何")
+@router.get("/geometry", summary="读取带版本的水力专题层静态几何",
+            response_class=Response, responses=BINARY_RESPONSE)
 async def playback_geometry(ctx: ProjectContext = Depends(get_project_context),
                             conn: AsyncConnection = Depends(get_project_pg_connection)):
     try:
@@ -47,7 +50,8 @@ async def playback_geometry(ctx: ProjectContext = Depends(get_project_context),
                     headers={"Cache-Control": "private, no-store", "ETag": f'"{geometry.version}"'})
 
 
-@router.get("/frames", summary="读取索引一致的 Float32 水力时间块")
+@router.get("/frames", summary="读取索引一致的 Float32 水力时间块",
+            response_class=Response, responses=BINARY_RESPONSE)
 async def playback_frames(
     start_time: str,
     model_version: str = Query(min_length=24, max_length=24),

app/infra/db/postgresql/hydraulic_playback.py

@@ -0,0 +1,24 @@
+"""Published network geometry used to build indexed hydraulic playback packets."""
+
+from typing import Any
+
+from psycopg import AsyncConnection
+
+
+class HydraulicGeometryRepository:
+    @staticmethod
+    async def get_rows(conn: AsyncConnection) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
+        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()
+        return nodes, pipes

app/infra/db/timescaledb/repositories/hydraulic_playback.py

@@ -0,0 +1,29 @@
+"""Stream exact playback samples without materializing full-network JSON results."""
+
+from collections.abc import AsyncIterator
+from datetime import datetime
+from typing import Any
+
+from psycopg import AsyncConnection
+
+
+class HydraulicPlaybackRepository:
+    @staticmethod
+    async def iter_node_pressures(conn: AsyncConnection, times: list[datetime]) -> AsyncIterator[dict[str, Any]]:
+        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:
+                yield row
+
+    @staticmethod
+    async def iter_pipe_metrics(conn: AsyncConnection, times: list[datetime]) -> AsyncIterator[dict[str, Any]]:
+        async with conn.cursor() as cursor:
+            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:
+                yield row

app/services/hydraulic_playback.py

@@ -14,6 +14,8 @@ from typing import Any
 from psycopg import AsyncConnection
 
 from app.infra.db.postgresql.network_settings import NetworkSettingsRepository
+from app.infra.db.postgresql.hydraulic_playback import HydraulicGeometryRepository
+from app.infra.db.timescaledb.repositories.hydraulic_playback import HydraulicPlaybackRepository
 
 WORLD_METERS = 40075016.68557849
 MAX_FRAME_COUNT = 8
@@ -122,19 +124,7 @@ def build_geometry(nodes: list[dict], pipes: list[dict], units: dict[str, str])
 
 
 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()
+    nodes, pipes = await HydraulicGeometryRepository.get_rows(conn)
     units = await NetworkSettingsRepository.get_result_units(conn)
     return await asyncio.to_thread(build_geometry, nodes, pipes, units)
 
@@ -154,30 +144,21 @@ async def read_frames(conn: AsyncConnection, geometry: PlaybackGeometry, start:
     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])
+    async for row in HydraulicPlaybackRepository.iter_node_pressures(conn, times):
+        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])
+    async for row in HydraulicPlaybackRepository.iter_pipe_metrics(conn, times):
+        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,

contracts/manifest.json

@@ -3,7 +3,7 @@
   "contracts": {
     "server": {
       "file": "server-v1.openapi.json",
-      "sha256": "9801087d40ac0e1f67871f9a1db850cbcce47614cceec0a7e8771cdf58385f53"
+      "sha256": "04aeed6f1a7414b37267d84141d036d00d2f8579ee28af37590bd4b92137e712"
     }
   }
 }

contracts/server-v1.openapi.json

@@ -50912,6 +50912,246 @@
         ]
       }
     },
+    "/api/v1/timeseries/realtime/playback/frames": {
+      "get": {
+        "operationId": "get_timeseries_realtime_playback_frames",
+        "parameters": [
+          {
+            "in": "query",
+            "name": "start_time",
+            "required": true,
+            "schema": {
+              "title": "Start Time",
+              "type": "string"
+            }
+          },
+          {
+            "in": "query",
+            "name": "model_version",
+            "required": true,
+            "schema": {
+              "maxLength": 24,
+              "minLength": 24,
+              "title": "Model Version",
+              "type": "string"
+            }
+          },
+          {
+            "in": "query",
+            "name": "frame_count",
+            "required": false,
+            "schema": {
+              "default": 4,
+              "maximum": 8,
+              "minimum": 1,
+              "title": "Frame Count",
+              "type": "integer"
+            }
+          },
+          {
+            "in": "query",
+            "name": "step_seconds",
+            "required": false,
+            "schema": {
+              "default": 900,
+              "maximum": 3600,
+              "minimum": 60,
+              "title": "Step Seconds",
+              "type": "integer"
+            }
+          },
+          {
+            "in": "header",
+            "name": "X-Project-Id",
+            "required": true,
+            "schema": {
+              "title": "X-Project-Id",
+              "type": "string"
+            }
+          }
+        ],
+        "responses": {
+          "200": {
+            "content": {
+              "application/octet-stream": {
+                "schema": {
+                  "format": "binary",
+                  "type": "string"
+                }
+              }
+            },
+            "description": "Versioned little-endian Float32 packet"
+          },
+          "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": "读取索引一致的 Float32 水力时间块",
+        "tags": [
+          "TimescaleDB - Hydraulic Playback"
+        ]
+      }
+    },
+    "/api/v1/timeseries/realtime/playback/geometry": {
+      "get": {
+        "operationId": "get_timeseries_realtime_playback_geometry",
+        "parameters": [
+          {
+            "in": "header",
+            "name": "X-Project-Id",
+            "required": true,
+            "schema": {
+              "title": "X-Project-Id",
+              "type": "string"
+            }
+          }
+        ],
+        "responses": {
+          "200": {
+            "content": {
+              "application/octet-stream": {
+                "schema": {
+                  "format": "binary",
+                  "type": "string"
+                }
+              }
+            },
+            "description": "Versioned little-endian Float32 packet"
+          },
+          "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 - Hydraulic Playback"
+        ]
+      }
+    },
     "/api/v1/timeseries/realtime/records": {
       "get": {
         "description": "查询指定时间点的实时属性值。query_time 必须显式带时区；允许传 UTC+8，服务端会先归一化为 UTC 再执行查询。",

tests/api/test_hydraulic_playback_endpoints.py

@@ -77,3 +77,10 @@ def test_cache_separates_projects_and_database_routes(monkeypatch):
     asyncio.run(run())
     assert calls == ["one", "two", "one"]
     playback._cache.clear()
+
+
+def test_contract_advertises_binary_packets_instead_of_json():
+    schema = build_test_app(playback.router, "/api/v1").openapi()
+    for suffix in ("geometry", "frames"):
+        content = schema["paths"][f"/api/v1/timeseries/realtime/playback/{suffix}"]["get"]["responses"]["200"]["content"]
+        assert content == {"application/octet-stream": {"schema": {"type": "string", "format": "binary"}}}