"""Read project data through metadata routing, in repeatable read-only transactions.""" from __future__ import annotations import gzip import hashlib import itertools import json import math from contextlib import contextmanager from datetime import datetime, timedelta, timezone from pathlib import Path from uuid import NAMESPACE_URL, uuid5 import psycopg from psycopg.conninfo import conninfo_to_dict, make_conninfo from psycopg.rows import dict_row from scripts.benchmark.contracts import ( BurstCase, BurstSuite, canonical_json, complete_devices, fingerprint, isolation_reference, ) async def resolve_sources(suite: BurstSuite) -> tuple[dict, dict[str, str]]: # Import lazily: validating a suite must not load environment credentials. 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(suite.project_code) if project is None or project.id != suite.project_id or project.status != "active": raise ValueError("benchmark project identity or active status does not match") routes = {} for role in ("biz_data", "iot_data"): route = await repo.get_project_db_routing(project.id, role) if route is None: raise ValueError("benchmark project database routing is incomplete") routes[role] = route.dsn identity = {"project_id": str(project.id), "project_code": project.code, "name": project.name} await session.rollback() return identity, routes finally: await engine.dispose() @contextmanager def readonly_source(dsn: str): # Replace inherited options: never rely on application credentials being read-only. safe_dsn = make_conninfo(dsn, options=( "-c default_transaction_read_only=on -c statement_timeout=20000 " "-c lock_timeout=2000 -c idle_in_transaction_session_timeout=600000" ), application_name="tjwater-benchmark-source-readonly", connect_timeout="5") with psycopg.connect(safe_dsn, row_factory=dict_row) as connection: connection.execute("SET TRANSACTION ISOLATION LEVEL REPEATABLE READ, READ ONLY") if connection.execute("SHOW transaction_read_only").fetchone()["transaction_read_only"] != "on": raise ValueError("source transaction is not read-only") try: yield connection finally: connection.rollback() def source_identity(connection, dsn: str) -> dict: values = conninfo_to_dict(dsn) row = connection.execute( "SELECT current_database() AS database, current_setting('server_version_num') AS server_version, " "transaction_timestamp() AS observed_at" ).fetchone() if row["database"] != values.get("dbname"): raise ValueError("connected source differs from the metadata route") return { **row, "read_only": True, "isolation": "repeatable read", # Equal database names on different servers are not equal routes. "route_sha256": fingerprint({key: values.get(key) for key in ("host", "port", "dbname")}), } def read_model(connection, cases: list[BurstCase]) -> dict: links = connection.execute( "SELECT l.id, l.link_type::text AS kind, l.start_node_id, l.end_node_id, " "COALESCE(s.status::text,p.status::text) AS status " "FROM network.links l LEFT JOIN network.pipes p ON p.link_id=l.id " "LEFT JOIN network.link_initial_settings s ON s.link_id=l.id ORDER BY l.id" ).fetchall() for link in links: for key in ("id", "start_node_id", "end_node_id"): link[key] = str(link[key]) sources = [str(row["id"]) for row in connection.execute( "SELECT id FROM network.nodes WHERE node_type::text IN ('reservoir','tank') ORDER BY id" ).fetchall()] devices = connection.execute( "SELECT device_id,device_type,node_id,link_id,api_query_id,measurement_unit, " "transmission_frequency::text AS transmission_frequency FROM asset.scada_devices " "WHERE device_type::text IN ('pressure','pipe_flow','flow') ORDER BY device_id" ).fetchall() for device in devices: for key in ("device_id", "node_id", "link_id"): if device[key] is not None: device[key] = str(device[key]) analyses = connection.execute( "SELECT run_id,name,run_type,status,parameters FROM analysis.runs WHERE name=ANY(%s) " "ORDER BY name,created_at,run_id", ([case.reference_analysis for case in cases],), ).fetchall() return {"links": links, "source_nodes": sources, "devices": devices, "reference_analyses": analyses} def readings(connection, device_ids: list[str], start: datetime, end: datetime) -> list[dict]: return connection.execute( "SELECT time,device_id,monitored_value FROM scada.measurements " "WHERE device_id=ANY(%s) AND time>=%s AND time<=%s ORDER BY time,device_id", (device_ids, start, end), ).fetchall() def long_history_capability(connection, suite: BurstSuite, case: BurstCase, devices: list[dict]) -> dict: interval = suite.sampling_interval_minutes expected = suite.detector_days * 1440 // interval start = case.occurred_at - timedelta(days=suite.detector_days) + timedelta(minutes=interval) mapped = {device["device_id"]: device["node_id"] for device in devices if device["device_type"] == "pressure" and device["node_id"] and device["api_query_id"]} rows = connection.execute( "SELECT device_id,count(*) AS count,count(DISTINCT time) AS unique_count, " "count(*) FILTER (WHERE monitored_value IS NOT NULL " "AND monitored_value::text NOT IN ('NaN','Infinity','-Infinity') " "AND mod(extract(epoch FROM (time-%s::timestamptz)),%s)=0) AS valid_count " "FROM scada.measurements WHERE device_id=ANY(%s) AND time>=%s AND time<=%s " "GROUP BY device_id ORDER BY device_id", (case.occurred_at, interval * 60, list(mapped), start, case.occurred_at), ).fetchall() eligible = sorted({mapped[row["device_id"]] for row in rows if row["count"] == row["unique_count"] == row["valid_count"] == expected}) return {"available": len(eligible) >= 5, "expected_samples_per_device": expected, "complete_pressure_nodes": eligible, "tool_node_limit": 20, "required_for": "optional run_burst_detection, not the short-window SCADA trigger"} def trigger_preview(suite: BurstSuite, case: BurstCase, rows: list[dict], devices: list[dict], *, normal=False) -> dict: """Exercise the current pure detector; do not tick replay, clean TSDB or create tasks.""" from app.domain.schemas.scada_replay import ScadaReplayCreate from app.services.scada_replay_processing import detect_frame, prepare_cleaning start = case.occurred_at - (timedelta(days=1) if normal else timedelta()) pressure = [device for device in devices if device["device_type"] == "pressure" and device["measurement_unit"] == "m"] ids = {device["device_id"] for device in pressure} rule = suite.trigger_rule if len(ids) < rule.minimum_devices: return {"raised": False, "reason": "insufficient pressure devices with unit m"} config = ScadaReplayCreate( request_id=uuid5(NAMESPACE_URL, f"benchmark:{case.case_id}:{normal}"), start_time=start, end_time=start + timedelta(minutes=suite.event_minutes + suite.sampling_interval_minutes), device_ids=sorted(ids), rules=[{ "rule_id": "benchmark-pressure-drop", "device_ids": sorted(ids), "kind": "relative_change", "direction": "decrease", "unit": "m", "threshold": rule.pressure_drop_percent, "window_samples": rule.baseline_samples, "minimum_samples": rule.baseline_samples, "consecutive_samples": rule.consecutive_samples, "minimum_devices": rule.minimum_devices, }], ) history: dict[str, list[float]] = {device: [] for device in ids} for row in rows: value = row["monitored_value"] if row["device_id"] in ids and row["time"] < start and value is not None and math.isfinite(float(value)): history[row["device_id"]].append(float(value)) document = {"replay_id": str(config.request_id), "devices": pressure, "histories": {key: values[-rule.baseline_samples:] for key, values in history.items()}} changes = [] for at, group in itertools.groupby(rows, key=lambda row: row["time"]): if at < start: continue frame = [{**row, "source_hash": fingerprint([row["time"], row["device_id"], row["monitored_value"]])} for row in group if row["device_id"] in ids] document, events = detect_frame(document, prepare_cleaning(frame, config), config, at) changes.extend(event for event in events if event.get("category") != "data_quality") raised = [event for event in changes if event["change"] == "raised"] return {"raised": bool(raised), "first_signal_at": raised[0]["occurred_at"] if raised else None, "event_count": len(raised), "warm_pressure_devices": sum(len(v) >= rule.baseline_samples for v in history.values()), "scope": "pure short-window rule preview; not worker/Agent end-to-end acceptance"} def inspect_case(iot_connection, suite: BurstSuite, case: BurstCase, model: dict) -> dict: devices = model["devices"] ids = [device["device_id"] for device in devices] warmup = timedelta(minutes=suite.trigger_rule.baseline_samples * suite.sampling_interval_minutes) end = case.occurred_at + timedelta(minutes=suite.event_minutes) event_rows = readings(iot_connection, ids, case.occurred_at - warmup, end) normal_rows = readings(iot_connection, ids, case.occurred_at - timedelta(days=1) - warmup, end - timedelta(days=1)) complete = set(complete_devices(event_rows, case.occurred_at, end, suite.sampling_interval_minutes)) complete &= set(complete_devices(normal_rows, case.occurred_at - timedelta(days=1), end - timedelta(days=1), suite.sampling_interval_minutes)) pressure_ids = {device["device_id"] for device in devices if device["device_type"] == "pressure"} matching = [run for run in model["reference_analyses"] if run["name"] == case.reference_analysis and run["run_type"] == "burst_analysis" and run["status"] == "completed" and case.pipe_id in [str(value) for value in run["parameters"].get("burst_ID", [])]] topology = isolation_reference(model["links"], model["source_nodes"], case.pipe_id) blockers = [] if case.observation_provenance != "declared_historical_replay": blockers.append("event_observations_require_isolated_generation") if not matching: blockers.append("matching_completed_reference_analysis_missing") if len(complete & pressure_ids) < 5: blockers.append("fewer_than_five_complete_event_and_control_pressure_devices") if topology["status"] in {"target_missing", "sources_missing"}: blockers.append(topology["status"]) event_signal = trigger_preview(suite, case, event_rows, devices) normal_signal = trigger_preview(suite, case, normal_rows, devices, normal=True) scada_blockers = list(blockers) if not event_signal["raised"]: scada_blockers.append("short_window_event_trigger_not_raised") if normal_signal["raised"]: scada_blockers.append("normal_control_false_positive") return { "case_id": case.case_id, "pipe_id": case.pipe_id, "occurred_at": case.occurred_at, "observation_provenance": case.observation_provenance, "data_candidate_available": not blockers, "blockers": blockers, "complete_event_and_control_pressure_devices": sorted(complete & pressure_ids), "window_sha256": fingerprint({"event": event_rows, "normal": normal_rows}), "optional_long_history_detection": long_history_capability(iot_connection, suite, case, devices), "event_signal_preview": event_signal, "normal_signal_preview": normal_signal, "topology_reference": topology, "variants": [ {"case_id": case.case_id + "-WO", "data_candidate_available": not blockers, "blockers": list(blockers)}, {"case_id": case.case_id + "-SCADA", "data_candidate_available": not scada_blockers, "blockers": scada_blockers}, ], "acceptance_pending": ["numerical_reference_review", "approved_equivalent_candidates", "public_leakage_input_or_declared_location_limitation", "isolated_runtime_and_cutoff_verification"], } def export_readings(connection, path: Path, suite: BurstSuite, cases: list[BurstCase], devices: list[dict]) -> dict: """Export only burst-related devices and the selected history/event window.""" start = min(case.occurred_at for case in cases) - timedelta(days=suite.detector_days) end = max(case.occurred_at for case in cases) + timedelta(minutes=suite.event_minutes) ids = [device["device_id"] for device in devices] count = 0 digest = hashlib.sha256() path.touch(mode=0o600, exist_ok=False) with path.open("wb") as raw, gzip.GzipFile(fileobj=raw, mode="wb", mtime=0) as output: with connection.cursor(name="benchmark_raw_export") as cursor: cursor.execute( "SELECT time,device_id,monitored_value FROM scada.measurements " "WHERE device_id=ANY(%s) AND time>=%s AND time<=%s ORDER BY time,device_id", (ids, start, end), ) for row in cursor: count += 1 if count > 1_000_000: raise ValueError("selected observation export exceeds one million records") line = (canonical_json(row) + "\n").encode() digest.update(line) output.write(line) return {"file": str(path.name), "records": count, "content_sha256": digest.hexdigest(), "start": start, "end": end, "origin": "existing_database_readonly_export", "visibility": "private; contains future samples and must not be exposed directly to Agent"}