From f25423dec47d6610297dbc92c53e5733e7e3f9d0 Mon Sep 17 00:00:00 2001 From: Huarch Date: Tue, 3 Nov 2026 17:26:38 +0800 Subject: [PATCH] feat(benchmark): add tjwater2 burst calibration design --- docs/tjwater2-v2-burst-simulation-design.md | 67 ++++ .../tjwater2-v2-burst-simulation.json | 137 +++++++++ scripts/benchmark/simulation_design.py | 185 +++++++++++ scripts/check_burst_simulation_design.py | 288 ++++++++++++++++++ 4 files changed, 677 insertions(+) create mode 100644 docs/tjwater2-v2-burst-simulation-design.md create mode 100644 resources/benchmarks/tjwater2-v2-burst-simulation.json create mode 100644 scripts/benchmark/simulation_design.py create mode 100644 scripts/check_burst_simulation_design.py diff --git a/docs/tjwater2-v2-burst-simulation-design.md b/docs/tjwater2-v2-burst-simulation-design.md new file mode 100644 index 0000000..6af9d25 --- /dev/null +++ b/docs/tjwater2-v2-burst-simulation-design.md @@ -0,0 +1,67 @@ +# tjwater2_v2 爆管模拟校准设计 + +状态：参数与只读校验已准备，尚未运行水力模拟。目标项目为 `tjwater2_v2`（演示项目2（新版）），不是 `tjwater_v2`。 + +## 目标 + +先用现有管网、阀门和监测点选择可观测的事故管段，再在隔离评测数据库中校准破口面积。首版使用管道截面积的 1/32、1/16、1/8；1/8 是上限。若 1/8 仍不能形成明确降压信号，该管段退出 SCADA 爆管用例，不继续扩大破口。 + +模拟调用链为： + +```text +只读源数据快照 -> 隔离 BizDB/模型模板 -> 正常对照模拟 + -> 爆管三档校准 -> 选择最小通过面积 + -> 保存正式分析结果 -> 映射 SCADA 数据 +``` + +当前 `temporary_project_database` 会从项目模型模板创建临时数据库，同时短暂禁止模板连接并终止模板会话。因此即使 `run_burst_simulation` 不保存结果，也不能直接针对原项目执行。正式运行必须把只读逻辑快照恢复到独立评测数据库及其模板，Server、worker 和 Agent 全部使用评测路由。 + +## 方法参数 + +`run_burst_simulation` 接收带时区的 `start_time`、60 至 86400 秒的 `duration`、1 至 20 个输出节点、至多 20 条输出管段、1 至 20 条爆管管段，以及与管段一一对应的正数孔口面积 `orifice_areas_cm2`。本次每个调用只模拟一条爆管管段，等待预算使用 240 秒；排队成功只表示已接受，仍需查询计算任务和最终结果。 + +`tjwater2_v2` 使用 LPS 流量单位、米制压力、0.5 emitter 指数，水力和报告步长均为 15 分钟。压力设备值使用 m，流量设备值使用 m³/h。3600 秒模拟必须产生 0、15、30、45、60 分钟五个结果时刻；缺少任一时刻即失败。爆管服务把 emitter 放在管段末端 junction，切换临时模型为 PDA，并设置 `REQUIRED PRESSURE=10 m`。 + +孔口面积是输入，不是漏水流量。服务的 emitter 系数为： + +```text +coefficient[L/s/sqrt(m)] = 0.65 * area[cm²] / 10000 * sqrt(19.6) * 1000 +``` + +正式运行后，定位使用的泄漏量按事故节点相对正常对照的 demand 增量计算，并从 L/s 除以 1000 转成 m³/s。流量监测设备单位为 m³/h，模型流量结果为 L/s，映射 SCADA 时乘以 3.6。 + +## 管段与场景 + +主场景管段 `541720` 为 DN800、长 483.038 m，破口节点 `502046`。附近两个压力点沿管网约 1.2 km，最近流量点约 1.2 km；两座边界阀可隔离 11 个节点。其破口面积为 157.079633、314.159265、628.318531 cm²。 + +清单同时保存 SCADA 设备 ID 与其实际水力模型对象 ID，不能假定两者相同。例如压力设备 `447382`、`444985` 分别绑定节点 `7543252226207002012`、`14141146287589625673`，流量设备 `542121` 绑定管段 `pipe-eaa5c512bd8f4649b37bdbe3d`。模拟输出使用模型对象 ID，得分与数据映射使用设备 ID。 + +启动校验还要求所选压力设备到绑定节点、流量设备到绑定管段的本地坐标距离均不超过 1 m。全量复核中待人工确认的 `542017`、`547063` 和 `DEMO-WQ-551571` 不进入本套爆管 benchmark；不得为通过校验而自动吸附设备坐标或猜测改绑对象。 + +次级场景管段 `520323` 为 DN300、长 33.5951 m，破口节点 `91055`。最近压力点沿管网约 262 m，五座边界阀可隔离 19 个节点。其破口面积为 22.089323、44.178647、88.357293 cm²。 + +两条管段分别在北京时间 2026-10-08 02:00 和 12:00 模拟，形成 B01 至 B04 四个物理场景。每个场景都必须有同时间、同持续时长、相同模型状态的无爆管正常对照。严重档先验证是否可观测；通过后依次测试中度和轻度，最终选择满足要求的最小面积。最坏情况下执行 12 次爆管校准模拟，正常对照按项目、时间和模型版本复用。 + +## 通过条件 + +一个面积只有同时满足以下条件才可冻结： + +1. 至少一个指定压力点连续两个 15 分钟时刻较正常对照下降 30%。 +2. 第二压力点下降至少 10%，或指定流量点相对正常对照变化至少 20%。 +3. 正常对照不触发相同规则。 +4. 指定监测节点不出现负压，模拟正常收敛，所有请求输出完整。 +5. 同一管段的夜间和白天场景使用同一面积；两者都通过时才能进入正式 benchmark。 + +压力与流量变化都按相同时间戳的正常对照计算：`abs(事故值 - 正常值) / max(abs(正常值), 1e-6)`。压力规则只取下降方向，流量规则取绝对变化。不得用其他时段均值代替同时间戳对照。 + +压力下降、流量变化、实际泄漏量、计算状态和缺失输出都保存到校准报告。失败的档位保留，不重试到通过。若严重档失败，该管段标记为不适用；若较小档位失败而较大档位通过，选择最小通过档位。 + +## 校验与后续执行 + +清单保存在 `resources/benchmarks/tjwater2-v2-burst-simulation.json`。以下命令使用本地占位 context handle，把全部 12 组计划参数送入当前 `BurstSimulationInput` 做契约校验，再只读检查 Metadata 项目身份、当前 BizDB、模型模板、管段属性、模型单位、步长、监测点、设备到绑定对象的距离、边界阀和既有 emitter；它不创建运行上下文，也不调用模拟工具： + +```bash +conda run -n server python scripts/check_burst_simulation_design.py +``` + +隔离评测数据库准备完成后，runner 先按清单调用非持久 `run_burst_simulation` 校准。面积冻结后，通过 `save_simulation_scenario` 在评测数据库保存全量节点和管段结果，生成固定的运行 ID 和数据摘要。模型分析结果与模拟 SCADA 保留明确来源，不标记为实测数据。 diff --git a/resources/benchmarks/tjwater2-v2-burst-simulation.json b/resources/benchmarks/tjwater2-v2-burst-simulation.json new file mode 100644 index 0000000..67adf25 --- /dev/null +++ b/resources/benchmarks/tjwater2-v2-burst-simulation.json @@ -0,0 +1,137 @@ +{ + "version": "burst-simulation-design@1", + "suite_id": "tjwater2-v2-burst-calibration", + "project_code": "tjwater2_v2", + "project_id": "871d7538-589c-47de-a6d5-3185ccb9f0e1", + "model_expectations": { + "node_count": 92392, + "pipe_count": 91049, + "valve_count": 4489, + "flow_unit": "LPS", + "pressure_unit": "METERS", + "pressure_measurement_unit": "m", + "flow_measurement_unit": "m3/h", + "emitter_exponent": 0.5, + "hydraulic_timestep_seconds": 900, + "report_timestep_seconds": 900 + }, + "duration_seconds": 3600, + "event_times": [ + "2026-10-08T02:00:00+08:00", + "2026-10-08T12:00:00+08:00" + ], + "severity_levels": [ + {"level": "light", "cross_section_fraction": 0.03125}, + {"level": "medium", "cross_section_fraction": 0.0625}, + {"level": "severe", "cross_section_fraction": 0.125} + ], + "calibration_order": ["severe", "medium", "light"], + "selection_rule": { + "primary_pressure_drop_percent": 30, + "corroborating_pressure_drop_percent": 10, + "flow_change_percent": 20, + "consecutive_samples": 2, + "expected_sample_count": 5, + "relative_change_epsilon": 0.000001, + "normal_control_must_not_trigger": true, + "negative_monitor_pressure_allowed": false, + "choose_smallest_passing_area": true + }, + "binding_validation": { + "max_pressure_distance_m": 1.0, + "max_flow_distance_m": 1.0 + }, + "cases": [ + {"case_id": "B01", "pipe_id": "541720", "occurred_at": "2026-10-08T02:00:00+08:00"}, + {"case_id": "B02", "pipe_id": "541720", "occurred_at": "2026-10-08T12:00:00+08:00"}, + {"case_id": "B03", "pipe_id": "520323", "occurred_at": "2026-10-08T02:00:00+08:00"}, + {"case_id": "B04", "pipe_id": "520323", "occurred_at": "2026-10-08T12:00:00+08:00"} + ], + "pipes": [ + { + "role": "primary", + "pipe_id": "541720", + "diameter_mm": 800, + "length_m": 483.038, + "start_node_id": "532586", + "emitter_node_id": "502046", + "longitude": 121.483586, + "latitude": 30.996219, + "candidate_orifice_areas_cm2": { + "light": 157.079633, + "medium": 314.159265, + "severe": 628.318531 + }, + "pressure_bindings": [ + {"device_id": "447382", "node_id": "7543252226207002012"}, + {"device_id": "444985", "node_id": "14141146287589625673"}, + {"device_id": "100026", "node_id": "100026"}, + {"device_id": "115039", "node_id": "115039"}, + {"device_id": "107672", "node_id": "107672"} + ], + "flow_bindings": [ + {"device_id": "542121", "link_id": "pipe-eaa5c512bd8f4649b37bdbe3d"}, + {"device_id": "449417", "link_id": "449417"}, + {"device_id": "516234", "link_id": "516234"} + ], + "boundary_valve_ids": [ + "valve-6fd58ab5b5714fbab738a544", + "valve-8aea6a1876d8475b9f35a845" + ], + "output_node_ids": [ + "12290448554375617026", "1371402016460607520", "400946", "413270", + "414851", "465564", "466633", "482531", "493166", "502046", + "532586", "7543252226207002012", "14141146287589625673", + "100026", "115039", "107672" + ], + "output_link_ids": [ + "541720", "pipe-eaa5c512bd8f4649b37bdbe3d", "449417", "516234", + "valve-6fd58ab5b5714fbab738a544", + "valve-8aea6a1876d8475b9f35a845" + ] + }, + { + "role": "secondary", + "pipe_id": "520323", + "diameter_mm": 300, + "length_m": 33.5951, + "start_node_id": "91054", + "emitter_node_id": "91055", + "longitude": 121.468305, + "latitude": 30.932363, + "candidate_orifice_areas_cm2": { + "light": 22.089323, + "medium": 44.178647, + "severe": 88.357293 + }, + "pressure_bindings": [ + {"device_id": "91100", "node_id": "91100"}, + {"device_id": "98367", "node_id": "98367"}, + {"device_id": "69693", "node_id": "69693"}, + {"device_id": "78085", "node_id": "78085"}, + {"device_id": "105833", "node_id": "105833"} + ], + "flow_bindings": [ + {"device_id": "527865", "link_id": "527865"}, + {"device_id": "489082", "link_id": "489082"}, + {"device_id": "531564", "link_id": "531564"} + ], + "boundary_valve_ids": [ + "valve-18b95b02a7564e7dac737102", + "valve-3a8fd38228d74583a08b00d9", + "valve-3b453e4e901049d1a1346689", + "valve-ad39de2f15f24ade9d18c836", + "valve-fc1164434a954f418a6f11d6" + ], + "output_node_ids": ["91054", "91055", "91100", "98367", "69693", "78085", "105833"], + "output_link_ids": [ + "520323", "527865", "489082", "531564", + "valve-18b95b02a7564e7dac737102", + "valve-3a8fd38228d74583a08b00d9", + "valve-3b453e4e901049d1a1346689", + "valve-ad39de2f15f24ade9d18c836", + "valve-fc1164434a954f418a6f11d6" + ] + } + ] +} diff --git a/scripts/benchmark/simulation_design.py b/scripts/benchmark/simulation_design.py new file mode 100644 index 0000000..297ad61 --- /dev/null +++ b/scripts/benchmark/simulation_design.py @@ -0,0 +1,185 @@ +"""Strict contracts for a burst-simulation calibration design.""" +from __future__ import annotations + +import math +from datetime import datetime +from pathlib import Path +from typing import Literal +from uuid import UUID + +from pydantic import BaseModel, ConfigDict, Field, model_validator + + +Severity = Literal["light", "medium", "severe"] + + +class StrictModel(BaseModel): + model_config = ConfigDict(extra="forbid", allow_inf_nan=False) + + +class ModelExpectations(StrictModel): + node_count: int = Field(gt=0) + pipe_count: int = Field(gt=0) + valve_count: int = Field(gt=0) + flow_unit: Literal["LPS"] + pressure_unit: Literal["METERS"] + pressure_measurement_unit: Literal["m"] + flow_measurement_unit: Literal["m3/h"] + emitter_exponent: float = Field(gt=0) + hydraulic_timestep_seconds: int = Field(gt=0) + report_timestep_seconds: int = Field(gt=0) + + +class SeverityLevel(StrictModel): + level: Severity + cross_section_fraction: float = Field(gt=0, le=0.125) + + +class SelectionRule(StrictModel): + primary_pressure_drop_percent: float = Field(gt=0, le=100) + corroborating_pressure_drop_percent: float = Field(gt=0, le=100) + flow_change_percent: float = Field(gt=0, le=1000) + consecutive_samples: int = Field(ge=1, le=20) + expected_sample_count: int = Field(ge=2, le=100) + relative_change_epsilon: float = Field(gt=0, le=0.001) + normal_control_must_not_trigger: Literal[True] + negative_monitor_pressure_allowed: Literal[False] + choose_smallest_passing_area: Literal[True] + + +class BindingValidation(StrictModel): + max_pressure_distance_m: float = Field(gt=0, le=10) + max_flow_distance_m: float = Field(gt=0, le=10) + + +class SimulationCase(StrictModel): + case_id: str = Field(pattern=r"^B[0-9]{2}$") + pipe_id: str = Field(min_length=1) + occurred_at: datetime + + @model_validator(mode="after") + def aware_time(self): + if self.occurred_at.tzinfo is None: + raise ValueError("case time must include a timezone") + return self + + +class PressureBinding(StrictModel): + device_id: str = Field(min_length=1) + node_id: str = Field(min_length=1) + + +class FlowBinding(StrictModel): + device_id: str = Field(min_length=1) + link_id: str = Field(min_length=1) + + +class PipeCalibration(StrictModel): + role: Literal["primary", "secondary"] + pipe_id: str = Field(min_length=1) + diameter_mm: float = Field(gt=0) + length_m: float = Field(gt=0) + start_node_id: str = Field(min_length=1) + emitter_node_id: str = Field(min_length=1) + longitude: float = Field(ge=-180, le=180) + latitude: float = Field(gt=-90, lt=90) + candidate_orifice_areas_cm2: dict[Severity, float] + pressure_bindings: list[PressureBinding] = Field(min_length=1, max_length=20) + flow_bindings: list[FlowBinding] = Field(min_length=1, max_length=20) + boundary_valve_ids: list[str] = Field(min_length=1, max_length=20) + output_node_ids: list[str] = Field(min_length=1, max_length=20) + output_link_ids: list[str] = Field(default_factory=list, max_length=20) + + @model_validator(mode="after") + def complete_outputs(self): + if set(self.candidate_orifice_areas_cm2) != {"light", "medium", "severe"}: + raise ValueError("all three severity areas are required") + pressure_device_ids = [value.device_id for value in self.pressure_bindings] + pressure_node_ids = [value.node_id for value in self.pressure_bindings] + flow_device_ids = [value.device_id for value in self.flow_bindings] + flow_link_ids = [value.link_id for value in self.flow_bindings] + for values, label in ( + (pressure_device_ids, "pressure devices"), + (pressure_node_ids, "pressure nodes"), + (flow_device_ids, "flow devices"), + (flow_link_ids, "flow links"), + (self.boundary_valve_ids, "boundary valves"), + (self.output_node_ids, "output nodes"), + (self.output_link_ids, "output links"), + ): + if len(values) != len(set(values)): + raise ValueError(f"{label} must be distinct") + if not {self.start_node_id, self.emitter_node_id, *pressure_node_ids}.issubset( + self.output_node_ids + ): + raise ValueError("pipe endpoints and pressure nodes must be returned") + if not {self.pipe_id, *flow_link_ids, *self.boundary_valve_ids}.issubset( + self.output_link_ids + ): + raise ValueError("target, flow and boundary links must be returned") + return self + + +class BurstSimulationDesign(StrictModel): + version: Literal["burst-simulation-design@1"] + suite_id: Literal["tjwater2-v2-burst-calibration"] + project_code: Literal["tjwater2_v2"] + project_id: UUID + model_expectations: ModelExpectations + duration_seconds: int = Field(ge=60, le=86400) + event_times: list[datetime] = Field(min_length=1, max_length=8) + severity_levels: list[SeverityLevel] = Field(min_length=3, max_length=3) + calibration_order: list[Severity] = Field(min_length=3, max_length=3) + selection_rule: SelectionRule + binding_validation: BindingValidation + cases: list[SimulationCase] = Field(min_length=1, max_length=20) + pipes: list[PipeCalibration] = Field(min_length=1, max_length=4) + + @model_validator(mode="after") + def coherent_design(self): + if any(value.tzinfo is None for value in self.event_times): + raise ValueError("event times must include a timezone") + if len(set(self.event_times)) != len(self.event_times): + raise ValueError("event times must be distinct") + levels = {entry.level: entry.cross_section_fraction for entry in self.severity_levels} + if set(levels) != {"light", "medium", "severe"}: + raise ValueError("light, medium and severe levels are required") + if not levels["light"] < levels["medium"] < levels["severe"]: + raise ValueError("severity fractions must increase") + expected_levels = {"light": 1 / 32, "medium": 1 / 16, "severe": 1 / 8} + if any( + not math.isclose(levels[level], fraction, rel_tol=0, abs_tol=1e-12) + for level, fraction in expected_levels.items() + ): + raise ValueError("severity fractions must remain 1/32, 1/16 and 1/8") + if self.calibration_order != ["severe", "medium", "light"]: + raise ValueError("calibration order must be severe, medium, light") + report_step = self.model_expectations.report_timestep_seconds + if self.duration_seconds % report_step: + raise ValueError("duration must contain whole report timesteps") + expected_samples = self.duration_seconds // report_step + 1 + if self.selection_rule.expected_sample_count != expected_samples: + raise ValueError("expected sample count does not match duration and report step") + pipes = {pipe.pipe_id: pipe for pipe in self.pipes} + if len(pipes) != len(self.pipes): + raise ValueError("pipe IDs must be distinct") + if len({case.case_id for case in self.cases}) != len(self.cases): + raise ValueError("case IDs must be distinct") + expected_cases = {(pipe_id, at) for pipe_id in pipes for at in self.event_times} + actual_cases = {(case.pipe_id, case.occurred_at) for case in self.cases} + if actual_cases != expected_cases: + raise ValueError("cases must cover every pipe and event time exactly once") + for pipe in self.pipes: + cross_section_cm2 = math.pi * (pipe.diameter_mm / 10) ** 2 / 4 + for level, fraction in levels.items(): + expected = cross_section_cm2 * fraction + actual = pipe.candidate_orifice_areas_cm2[level] + if not math.isclose(actual, expected, rel_tol=0, abs_tol=0.001): + raise ValueError( + f"{pipe.pipe_id} {level} area does not match its diameter fraction" + ) + return self + + +def load_simulation_design(path: Path) -> BurstSimulationDesign: + return BurstSimulationDesign.model_validate_json(path.read_text(encoding="utf-8")) diff --git a/scripts/check_burst_simulation_design.py b/scripts/check_burst_simulation_design.py new file mode 100644 index 0000000..70d0c63 --- /dev/null +++ b/scripts/check_burst_simulation_design.py @@ -0,0 +1,288 @@ +#!/usr/bin/env python3 +"""Read-only validation of the tjwater2_v2 burst-simulation design.""" +from __future__ import annotations + +import argparse +import asyncio +import sys +import traceback +from datetime import timedelta +from pathlib import Path + +from psycopg import connect +from psycopg.conninfo import conninfo_to_dict, make_conninfo +from psycopg.rows import dict_row + + +def readonly(dsn: str): + return connect( + make_conninfo( + dsn, + options="-c default_transaction_read_only=on -c statement_timeout=20000", + application_name="tjwater-burst-design-readonly", + connect_timeout="5", + ), + row_factory=dict_row, + ) + + +async def resolve_project(design): + from sqlalchemy import text + + from app.infra.db.metadb.database import SessionLocal, engine + from app.infra.db.metadb.repositories.metadata_repository import MetadataRepository + + try: + async with SessionLocal() as session: + await session.execute( + text("SET TRANSACTION ISOLATION LEVEL REPEATABLE READ, READ ONLY") + ) + repo = MetadataRepository(session) + project = await repo.get_project_by_code(design.project_code) + if ( + project is None + or project.id != design.project_id + or project.status != "active" + ): + raise ValueError("project identity or active status does not match") + biz = await repo.get_project_db_routing(project.id, "biz_data") + iot = await repo.get_project_db_routing(project.id, "iot_data") + if biz is None or iot is None: + raise ValueError("project database routing is incomplete") + await session.rollback() + return biz.dsn, iot.dsn + finally: + await engine.dispose() + + +def validate_database(connection, design, *, template: bool) -> None: + connection.execute("SET TRANSACTION ISOLATION LEVEL REPEATABLE READ, READ ONLY") + if connection.execute("SHOW transaction_read_only").fetchone()["transaction_read_only"] != "on": + raise ValueError("database transaction is not read-only") + expected = design.model_expectations + counts = connection.execute( + "SELECT (SELECT count(*) FROM network.nodes) AS nodes, " + "(SELECT count(*) FROM network.pipes) AS pipes, " + "(SELECT count(*) FROM network.links WHERE link_type='valve') AS valves" + ).fetchone() + if (counts["nodes"], counts["pipes"], counts["valves"]) != ( + expected.node_count, + expected.pipe_count, + expected.valve_count, + ): + raise ValueError("model counts do not match the frozen design") + + settings = { + row["key"]: row["value"] + for row in connection.execute( + "SELECT key,value FROM network.simulation_settings " + "WHERE engine_version='legacy' AND key=ANY(%s)", + (["UNITS", "PRESSURE", "EMITTER EXPONENT"],), + ) + } + times = { + row["key"]: row["value"] + for row in connection.execute( + "SELECT key,value FROM network.time_settings " + "WHERE key=ANY(%s)", + (["HYDRAULIC TIMESTEP", "REPORT TIMESTEP"],), + ) + } + if settings != { + "UNITS": expected.flow_unit, + "PRESSURE": expected.pressure_unit, + "EMITTER EXPONENT": f"{expected.emitter_exponent:.4f}", + }: + raise ValueError("simulation settings do not match the frozen design") + if times != {"HYDRAULIC TIMESTEP": "0:15:00", "REPORT TIMESTEP": "0:15:00"}: + raise ValueError("simulation timesteps do not match the frozen design") + + for pipe in design.pipes: + row = connection.execute( + "SELECT l.start_node_id,l.end_node_id,p.length,p.diameter,p.status " + "FROM network.links l JOIN network.pipes p ON p.link_id=l.id WHERE l.id=%s", + (pipe.pipe_id,), + ).fetchone() + if row is None: + raise ValueError(f"pipe {pipe.pipe_id} is missing") + if ( + str(row["start_node_id"]) != pipe.start_node_id + or str(row["end_node_id"]) != pipe.emitter_node_id + or abs(float(row["length"]) - pipe.length_m) > 0.001 + or abs(float(row["diameter"]) - pipe.diameter_mm) > 0.001 + or str(row["status"]).upper() != "OPEN" + ): + raise ValueError(f"pipe {pipe.pipe_id} properties changed") + node_rows = connection.execute( + "SELECT id,node_type FROM network.nodes WHERE id=ANY(%s)", + (pipe.output_node_ids,), + ).fetchall() + if {str(row["id"]) for row in node_rows} != set(pipe.output_node_ids): + raise ValueError(f"pipe {pipe.pipe_id} output nodes are incomplete") + node_types = {str(row["id"]): str(row["node_type"]) for row in node_rows} + if node_types.get(pipe.emitter_node_id) != "junction": + raise ValueError(f"pipe {pipe.pipe_id} emitter endpoint is not a junction") + link_rows = connection.execute( + "SELECT id,link_type FROM network.links WHERE id=ANY(%s)", + (pipe.output_link_ids,), + ).fetchall() + if {str(row["id"]) for row in link_rows} != set(pipe.output_link_ids): + raise ValueError(f"pipe {pipe.pipe_id} output links are incomplete") + link_types = {str(row["id"]): str(row["link_type"]) for row in link_rows} + if any(link_types.get(value) != "valve" for value in pipe.boundary_valve_ids): + raise ValueError(f"pipe {pipe.pipe_id} boundary valve changed") + emitter = connection.execute( + "SELECT coefficient FROM network.emitters WHERE junction_id=%s", + (pipe.emitter_node_id,), + ).fetchone() + if emitter is not None: + raise ValueError(f"pipe {pipe.pipe_id} endpoint already has an emitter") + + if not template: + pressure_bindings = { + ( + str(row["device_id"]), + str(row["node_id"]), + str(row["measurement_unit"]), + str(row["transmission_frequency"]), + ) + for row in connection.execute( + "SELECT device_id,node_id,measurement_unit,transmission_frequency " + "FROM asset.scada_devices " + "WHERE device_type='pressure' AND device_id=ANY(%s)", + ([value.device_id for value in pipe.pressure_bindings],), + ) + } + flow_bindings = { + ( + str(row["device_id"]), + str(row["link_id"]), + str(row["measurement_unit"]), + str(row["transmission_frequency"]), + ) + for row in connection.execute( + "SELECT device_id,link_id,measurement_unit,transmission_frequency " + "FROM asset.scada_devices " + "WHERE device_type IN ('pipe_flow','flow') AND device_id=ANY(%s)", + ([value.device_id for value in pipe.flow_bindings],), + ) + } + expected_pressure_bindings = { + ( + value.device_id, + value.node_id, + expected.pressure_measurement_unit, + str(timedelta(seconds=expected.report_timestep_seconds)), + ) + for value in pipe.pressure_bindings + } + expected_flow_bindings = { + ( + value.device_id, + value.link_id, + expected.flow_measurement_unit, + str(timedelta(seconds=expected.report_timestep_seconds)), + ) + for value in pipe.flow_bindings + } + if pressure_bindings != expected_pressure_bindings: + raise ValueError(f"pipe {pipe.pipe_id} pressure bindings changed") + if flow_bindings != expected_flow_bindings: + raise ValueError(f"pipe {pipe.pipe_id} flow bindings changed") + pressure_distances = { + str(row["device_id"]): row["distance_m"] + for row in connection.execute( + "SELECT d.device_id,ST_Distance(d.geom,n.geom) AS distance_m " + "FROM asset.scada_devices d " + "JOIN gis.node_geometries n ON n.node_id=d.node_id " + "WHERE d.device_id=ANY(%s)", + ([value.device_id for value in pipe.pressure_bindings],), + ) + } + flow_distances = { + str(row["device_id"]): row["distance_m"] + for row in connection.execute( + "SELECT d.device_id," + "ST_Distance(d.geom,ST_Transform(p.geom,900914)) AS distance_m " + "FROM asset.scada_devices d " + "JOIN gis.pipes p ON p.id=d.link_id " + "WHERE d.device_id=ANY(%s)", + ([value.device_id for value in pipe.flow_bindings],), + ) + } + if len(pressure_distances) != len(pipe.pressure_bindings) or any( + distance is None + or float(distance) > design.binding_validation.max_pressure_distance_m + for distance in pressure_distances.values() + ): + raise ValueError(f"pipe {pipe.pipe_id} pressure binding geometry changed") + if len(flow_distances) != len(pipe.flow_bindings) or any( + distance is None + or float(distance) > design.binding_validation.max_flow_distance_m + for distance in flow_distances.values() + ): + raise ValueError(f"pipe {pipe.pipe_id} flow binding geometry changed") + connection.rollback() + + +def validate_tool_inputs(design) -> None: + from app.mcp.extended_models import BurstSimulationInput + + pipes = {pipe.pipe_id: pipe for pipe in design.pipes} + for case in design.cases: + pipe = pipes[case.pipe_id] + for severity in design.calibration_order: + BurstSimulationInput( + context_handle="benchmark-design-context-handle-0001", + timeout_seconds=240, + start_time=case.occurred_at, + duration=design.duration_seconds, + output_node_ids=pipe.output_node_ids, + output_link_ids=pipe.output_link_ids, + burst_pipe_ids=[pipe.pipe_id], + orifice_areas_cm2=[pipe.candidate_orifice_areas_cm2[severity]], + ) + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "--design", + type=Path, + default=Path("resources/benchmarks/tjwater2-v2-burst-simulation.json"), + ) + args = parser.parse_args() + sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + + try: + from scripts.benchmark.simulation_design import load_simulation_design + + design = load_simulation_design(args.design) + validate_tool_inputs(design) + biz_dsn, _ = asyncio.run(resolve_project(design)) + physical_name = conninfo_to_dict(biz_dsn).get("dbname") + if not physical_name: + raise ValueError("business route has no database name") + template_dsn = make_conninfo(biz_dsn, dbname=physical_name + "_template") + with readonly(biz_dsn) as connection: + validate_database(connection, design, template=False) + with readonly(template_dsn) as connection: + validate_database(connection, design, template=True) + except Exception as exc: + frames = traceback.extract_tb(exc.__traceback__) + location = frames[-1] if frames else None + suffix = f" at {location.filename}:{location.lineno}" if location else "" + print(f"burst simulation design check failed ({type(exc).__name__}){suffix}", file=sys.stderr) + return 1 + + maximum_runs = len(design.cases) * len(design.severity_levels) + print( + f"burst simulation design ready: {len(design.pipes)} pipes, " + f"{len(design.cases)} cases, at most {maximum_runs} calibration runs; " + "tool inputs, SCADA bindings, BizDB and model template were read-only verified" + ) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())