from __future__ import annotations import os from datetime import datetime from typing import Any import pandas as pd from app.algorithms.burst_location import run_burst_location from app.infra.db.timescaledb.internal_queries import InternalQueries from app.services.scheme_management import ( query_burst_location_scheme_detail, query_burst_location_schemes, scheme_name_exists, store_scheme_info, ) from app.services.tjnetwork import dump_inp, get_all_scada_info from app.services.time_api import extract_date, parse_utc_time, utc_now SeriesInput = pd.Series | dict[str, Any] | list[dict[str, Any]] FLOW_SCADA_TYPES = {"pipe_flow", "flow", "demand"} SIMULATION_DATA_SOURCES = {"monitoring", "simulation"} DEFAULT_SIMULATION_SCHEME_TYPE = "burst_analysis" def _normalize_series(data: SeriesInput, field_name: str) -> pd.Series: if isinstance(data, pd.Series): series = data.copy() elif isinstance(data, dict): series = pd.Series(data, dtype=float) elif isinstance(data, list): if len(data) == 0: return pd.Series(dtype=float) frame = pd.DataFrame(data) if not {"id", "value"}.issubset(frame.columns): raise ValueError(f"{field_name} list item must include 'id' and 'value'.") series = pd.Series( frame["value"].values, index=frame["id"].astype(str).values, dtype=float ) else: raise ValueError(f"Unsupported data format for {field_name}.") series.index = series.index.map(_normalize_identifier) return pd.to_numeric(series, errors="raise") def run_burst_location_by_network( *, network: str, username: str, data_source: str = "monitoring", burst_leakage: float, pressure_scada_ids: list[str] | None = None, burst_pressure: SeriesInput | None = None, normal_pressure: SeriesInput | None = None, flow_scada_ids: list[str] | None = None, burst_flow: SeriesInput | None = None, normal_flow: SeriesInput | None = None, min_dpressure: float = 2.0, basic_pressure: float = 10.0, scada_burst_start: datetime | str | None = None, scada_burst_end: datetime | str | None = None, scada_normal_start: datetime | str | None = None, scada_normal_end: datetime | str | None = None, use_scada_flow: bool = False, scheme_name: str | None = None, simulation_scheme_name: str | None = None, simulation_scheme_type: str | None = None, ) -> dict[str, Any]: if not network: raise ValueError("network is required.") normalized_data_source = _normalize_data_source( data_source, simulation_scheme_name=simulation_scheme_name ) resolved_simulation_scheme_type = ( simulation_scheme_type or DEFAULT_SIMULATION_SCHEME_TYPE ) selected_pressure_ids = ( _dedupe_ids(pressure_scada_ids) if pressure_scada_ids else _get_sensor_nodes(network, data_type="pressure") ) if not selected_pressure_ids: raise ValueError("未提供有效压力传感器，且系统未识别到可用压力传感器。") use_scada_pressure = any( value is not None for value in [ scada_burst_start, scada_burst_end, scada_normal_start, scada_normal_end, ] ) if use_scada_pressure: burst_start_dt, burst_end_dt = _validate_time_window( start_value=scada_burst_start, end_value=scada_burst_end, start_field="scada_burst_start", end_field="scada_burst_end", label=( "爆管方案时间窗" if normalized_data_source == "simulation" else "爆管时段 SCADA 时间窗" ), ) normal_start_dt: datetime | None = None normal_end_dt: datetime | None = None if scada_normal_start is not None or scada_normal_end is not None: normal_start_dt, normal_end_dt = _validate_time_window( start_value=scada_normal_start, end_value=scada_normal_end, start_field="scada_normal_start", end_field="scada_normal_end", label="正常时段 SCADA 时间窗", ) normal_pressure_from_payload = ( _normalize_series(normal_pressure, "normal_pressure") if normal_pressure is not None else None ) if normalized_data_source == "simulation": if not simulation_scheme_name: raise ValueError("模拟方案模式必须提供 simulation_scheme_name。") ( burst_pressure_series, burst_pressure_samples, ) = _build_observed_series_from_simulation( network=network, sensor_ids=selected_pressure_ids, start_dt=burst_start_dt, end_dt=burst_end_dt, data_type="pressure", series_name="burst_pressure", simulation_source="scheme", simulation_scheme_name=simulation_scheme_name, simulation_scheme_type=resolved_simulation_scheme_type, ) ( normal_pressure_series, normal_pressure_samples, ) = _build_observed_series_from_simulation( network=network, sensor_ids=selected_pressure_ids, start_dt=normal_start_dt or burst_start_dt, end_dt=normal_end_dt or burst_end_dt, data_type="pressure", series_name="normal_pressure", simulation_source="scheme", simulation_scheme_name=simulation_scheme_name, simulation_scheme_type=resolved_simulation_scheme_type, ) observed_source = "simulation_scheme_timerange" else: ( burst_pressure_series, burst_pressure_samples, ) = _build_observed_series_from_scada( network=network, sensor_ids=selected_pressure_ids, start_dt=burst_start_dt, end_dt=burst_end_dt, data_type="pressure", series_name="burst_pressure", ) if normal_pressure_from_payload is None: ( normal_pressure_series, normal_pressure_samples, ) = _build_observed_series_from_scada( network=network, sensor_ids=selected_pressure_ids, start_dt=normal_start_dt or burst_start_dt, end_dt=normal_end_dt or burst_end_dt, data_type="pressure", series_name="normal_pressure", ) observed_source = ( "scada_burst_scada_normal_timerange" if normal_start_dt is not None and normal_end_dt is not None else "scada_timerange" ) else: normal_pressure_series = normal_pressure_from_payload normal_pressure_samples = 1 observed_source = "scada_burst_payload_normal_timerange" selected_pressure_ids, burst_pressure_series, normal_pressure_series = ( _align_observed_series_pair( ids=selected_pressure_ids, burst_series=burst_pressure_series, normal_series=normal_pressure_series, data_label="压力数据", ) ) else: if burst_pressure is None or normal_pressure is None: raise ValueError( "未提供 burst_pressure/normal_pressure，且未提供完整 SCADA 时间窗参数。" ) burst_pressure_series = _normalize_series(burst_pressure, "burst_pressure") normal_pressure_series = _normalize_series(normal_pressure, "normal_pressure") burst_pressure_samples = 1 normal_pressure_samples = 1 observed_source = "request_payload" burst_start_dt = burst_end_dt = None selected_flow_ids: list[str] | None = None burst_flow_series: pd.Series | None = None normal_flow_series: pd.Series | None = None use_flow_scada_source = use_scada_pressure and ( use_scada_flow or flow_scada_ids is not None ) if use_flow_scada_source: selected_flow_ids = ( _dedupe_ids(flow_scada_ids) if flow_scada_ids is not None else _get_sensor_nodes(network, data_type="flow") ) if not selected_flow_ids: raise ValueError("未找到可用流量传感器，无法从 SCADA 查询流量数据。") normal_flow_from_payload = ( _normalize_series(normal_flow, "normal_flow") if normal_flow is not None else None ) if normalized_data_source == "simulation": if not simulation_scheme_name: raise ValueError("模拟方案模式必须提供 simulation_scheme_name。") burst_flow_series, burst_flow_samples = ( _build_observed_series_from_simulation( network=network, sensor_ids=selected_flow_ids, start_dt=burst_start_dt, end_dt=burst_end_dt, data_type="flow", series_name="burst_flow", simulation_source="scheme", simulation_scheme_name=simulation_scheme_name, simulation_scheme_type=resolved_simulation_scheme_type, ) ) normal_flow_series, normal_flow_samples = ( _build_observed_series_from_simulation( network=network, sensor_ids=selected_flow_ids, start_dt=normal_start_dt or burst_start_dt, end_dt=normal_end_dt or burst_end_dt, data_type="flow", series_name="normal_flow", simulation_source="scheme", simulation_scheme_name=simulation_scheme_name, simulation_scheme_type=resolved_simulation_scheme_type, ) ) else: burst_flow_series, burst_flow_samples = _build_observed_series_from_scada( network=network, sensor_ids=selected_flow_ids, start_dt=burst_start_dt, end_dt=burst_end_dt, data_type="flow", series_name="burst_flow", ) if normal_flow_from_payload is None: normal_flow_series, normal_flow_samples = ( _build_observed_series_from_scada( network=network, sensor_ids=selected_flow_ids, start_dt=normal_start_dt or burst_start_dt, end_dt=normal_end_dt or burst_end_dt, data_type="flow", series_name="normal_flow", ) ) else: normal_flow_series = normal_flow_from_payload normal_flow_samples = 1 selected_flow_ids, burst_flow_series, normal_flow_series = ( _align_observed_series_pair( ids=selected_flow_ids, burst_series=burst_flow_series, normal_series=normal_flow_series, data_label="流量数据", ) ) else: if flow_scada_ids is not None: selected_flow_ids = _dedupe_ids(flow_scada_ids) burst_flow_series = ( _normalize_series(burst_flow, "burst_flow") if burst_flow is not None else None ) normal_flow_series = ( _normalize_series(normal_flow, "normal_flow") if normal_flow is not None else None ) burst_flow_samples = 1 if burst_flow_series is not None else 0 normal_flow_samples = 1 if normal_flow_series is not None else 0 inp_path = _prepare_burst_inp(network) result = run_burst_location( wn_inp_path=inp_path, pressure_scada_ids=selected_pressure_ids, burst_pressure=burst_pressure_series, normal_pressure=normal_pressure_series, burst_leakage=burst_leakage, flow_scada_ids=selected_flow_ids, burst_flow=burst_flow_series, normal_flow=normal_flow_series, min_dpressure=min_dpressure, basic_pressure=basic_pressure, ) payload: dict[str, Any] = { **result, "network": network, "data_source": normalized_data_source, "pressure_scada_ids": selected_pressure_ids, "flow_scada_ids": selected_flow_ids or [], "observed_source": observed_source, "pressure_samples": { "burst": burst_pressure_samples, "normal": normal_pressure_samples, }, "flow_samples": {"burst": burst_flow_samples, "normal": normal_flow_samples}, "burst_leakage": burst_leakage, "min_dpressure": min_dpressure, "basic_pressure": basic_pressure, } if use_scada_pressure: payload["scada_window"] = { "burst_start": burst_start_dt.isoformat(), "burst_end": burst_end_dt.isoformat(), } if normal_start_dt is not None and normal_end_dt is not None: payload["scada_window"].update( { "normal_start": normal_start_dt.isoformat(), "normal_end": normal_end_dt.isoformat(), } ) if normalized_data_source == "simulation": payload["simulation_scheme"] = { "name": simulation_scheme_name, "type": resolved_simulation_scheme_type, } if scheme_name: _store_burst_scheme( network=network, scheme_name=scheme_name, username=username, payload=payload, burst_leakage=burst_leakage, min_dpressure=min_dpressure, basic_pressure=basic_pressure, ) return payload def list_burst_location_schemes( network: str, query_date: datetime | str | None = None ) -> list[dict[str, Any]]: parsed_date = extract_date(query_date, field_name="query_date") if query_date is not None else None return query_burst_location_schemes( name=network, network=network, query_date=parsed_date ) def get_burst_location_scheme_detail(network: str, scheme_name: str) -> dict[str, Any]: result = query_burst_location_scheme_detail(network, scheme_name) if not result: raise ValueError(f"未找到爆管定位方案: {scheme_name}") return result def _store_burst_scheme( *, network: str, scheme_name: str, username: str, payload: dict[str, Any], burst_leakage: float, min_dpressure: float, basic_pressure: float, ) -> None: if scheme_name_exists(network, scheme_name): raise ValueError(f"方案名称已存在: {scheme_name}") now_iso = utc_now().isoformat() scheme_detail = { "network": network, "pressure_scada_ids": payload.get("pressure_scada_ids", []), "flow_scada_ids": payload.get("flow_scada_ids", []), "observed_source": payload.get("observed_source"), "algorithm_params": { "burst_leakage": burst_leakage, "min_dpressure": min_dpressure, "basic_pressure": basic_pressure, }, "scada_window": payload.get("scada_window"), "result_summary": { "located_pipe": payload.get("located_pipe"), "simulation_times": payload.get("simulation_times"), "similarity_mode": payload.get("similarity_mode"), }, "result_payload": payload, } store_scheme_info( name=network, scheme_name=scheme_name, scheme_type="burst_location", username=username, scheme_start_time=now_iso, scheme_detail=scheme_detail, ) def _validate_scada_windows( *, scada_burst_start: datetime | str | None, scada_burst_end: datetime | str | None, ) -> tuple[datetime, datetime]: values = [scada_burst_start, scada_burst_end] if any(v is None for v in values): raise ValueError( "使用后端 SCADA 查询时，必须同时提供 scada_burst_start/scada_burst_end。" ) burst_start_dt = _to_datetime(scada_burst_start) burst_end_dt = _to_datetime(scada_burst_end) if burst_start_dt >= burst_end_dt: raise ValueError( "爆管时段 SCADA 时间窗非法：scada_burst_start 必须早于 scada_burst_end。" ) return burst_start_dt, burst_end_dt def _validate_time_window( *, start_value: datetime | str | None, end_value: datetime | str | None, start_field: str, end_field: str, label: str, ) -> tuple[datetime, datetime]: if start_value is None or end_value is None: raise ValueError(f"{label}必须同时提供 {start_field}/{end_field}。") start_dt = _to_datetime(start_value) end_dt = _to_datetime(end_value) if start_dt >= end_dt: raise ValueError(f"{label}非法：{start_field} 必须早于 {end_field}。") return start_dt, end_dt def _align_observed_series_pair( *, ids: list[str], burst_series: pd.Series, normal_series: pd.Series, data_label: str, ) -> tuple[list[str], pd.Series, pd.Series]: common_ids = [ sensor_id for sensor_id in _dedupe_ids(ids) if sensor_id in burst_series.index and sensor_id in normal_series.index ] if not common_ids: raise ValueError(f"{data_label}没有同时具备爆管时段和正常时段有效数据的点位。") return common_ids, burst_series.loc[common_ids], normal_series.loc[common_ids] def _build_observed_series_from_scada( *, network: str, sensor_ids: list[str], start_dt: datetime, end_dt: datetime, data_type: str, series_name: str, ) -> tuple[pd.Series, int]: sensor_ids = _dedupe_ids(sensor_ids) scada_mapping = _build_scada_mapping(network=network, data_type=data_type) missing_ids = [ sensor_id for sensor_id in sensor_ids if sensor_id not in scada_mapping ] if missing_ids: preview = ", ".join(missing_ids[:10]) raise ValueError(f"{_series_display_name(series_name)} 缺少可用 SCADA 映射: {preview}") query_ids = [scada_mapping[sensor_id] for sensor_id in sensor_ids] scada_data = InternalQueries.query_scada_by_ids_timerange( db_name=network, device_ids=query_ids, start_time=start_dt.isoformat(), end_time=end_dt.isoformat(), ) scada_data = _normalize_timeseries_by_id(scada_data) values: dict[str, float] = {} sample_counts: list[int] = [] for sensor_id, query_id in zip(sensor_ids, query_ids): records = scada_data.get(query_id, []) numeric_values = [ float(item["value"]) for item in records if item.get("value") is not None ] if not numeric_values: continue values[sensor_id] = float(sum(numeric_values) / len(numeric_values)) sample_counts.append(len(numeric_values)) if not values: raise ValueError( f"{_series_display_name(series_name)} 在时间窗内无有效数据: {', '.join(sensor_ids[:10])}" ) return pd.Series(values, dtype=float), min(sample_counts) def _build_observed_series_from_simulation( *, network: str, sensor_ids: list[str], start_dt: datetime, end_dt: datetime, data_type: str, series_name: str, simulation_source: str, simulation_scheme_name: str | None, simulation_scheme_type: str, ) -> tuple[pd.Series, int]: sensor_ids = _dedupe_ids(sensor_ids) sensor_metadata = _build_sensor_metadata(network=network, data_type=data_type) missing_ids = [ sensor_id for sensor_id in sensor_ids if sensor_id not in sensor_metadata ] if missing_ids: preview = ", ".join(missing_ids[:10]) raise ValueError(f"{_series_display_name(series_name)} 缺少可用 SCADA 映射: {preview}") simulation_data = _query_simulation_data_by_sensor_ids( network=network, sensor_ids=sensor_ids, sensor_metadata=sensor_metadata, start_dt=start_dt, end_dt=end_dt, data_type=data_type, simulation_source=simulation_source, simulation_scheme_name=simulation_scheme_name, simulation_scheme_type=simulation_scheme_type, ) simulation_data = _normalize_timeseries_by_id(simulation_data) values: dict[str, float] = {} sample_counts: list[int] = [] for sensor_id in sensor_ids: records = simulation_data.get(sensor_id, []) numeric_values = [ float(item["value"]) for item in records if item.get("value") is not None ] if not numeric_values: raise ValueError( f"{_series_display_name(series_name)} 在时间窗内无有效模拟数据: {sensor_id}" ) values[sensor_id] = float(sum(numeric_values) / len(numeric_values)) sample_counts.append(len(numeric_values)) return pd.Series(values, dtype=float), min(sample_counts) def _series_display_name(series_name: str) -> str: return { "burst_pressure": "爆管压力数据", "normal_pressure": "正常压力数据", "burst_flow": "爆管流量数据", "normal_flow": "正常流量数据", }.get(series_name, series_name) def _query_simulation_data_by_sensor_ids( *, network: str, sensor_ids: list[str], sensor_metadata: dict[str, dict[str, str]], start_dt: datetime, end_dt: datetime, data_type: str, simulation_source: str, simulation_scheme_name: str | None, simulation_scheme_type: str, ) -> dict[str, list[dict[str, Any]]]: if simulation_source not in {"scheme", "realtime"}: raise ValueError(f"Unsupported simulation_source: {simulation_source}") sensor_ids = _dedupe_ids(sensor_ids) result: dict[str, list[dict[str, Any]]] = { sensor_id: [] for sensor_id in sensor_ids } if data_type == "pressure": result.update( _query_simulation_values( network=network, element_ids=sensor_ids, element_type="node", field="pressure", start_dt=start_dt, end_dt=end_dt, simulation_source=simulation_source, simulation_scheme_name=simulation_scheme_name, simulation_scheme_type=simulation_scheme_type, ) ) return result if data_type != "flow": raise ValueError(f"Unsupported data_type: {data_type}") link_ids: list[str] = [] demand_ids: list[str] = [] unsupported_ids: list[str] = [] for sensor_id in sensor_ids: scada_type = sensor_metadata[sensor_id]["scada_type"] if scada_type in {"pipe_flow", "flow"}: link_ids.append(sensor_id) elif scada_type == "demand": demand_ids.append(sensor_id) else: unsupported_ids.append(f"{sensor_id}({scada_type})") if unsupported_ids: preview = ", ".join(unsupported_ids[:10]) raise ValueError(f"flow 模拟数据暂不支持以下 SCADA 类型: {preview}") if link_ids: result.update( _query_simulation_values( network=network, element_ids=link_ids, element_type="link", field="flow", start_dt=start_dt, end_dt=end_dt, simulation_source=simulation_source, simulation_scheme_name=simulation_scheme_name, simulation_scheme_type=simulation_scheme_type, ) ) if demand_ids: result.update( _query_simulation_values( network=network, element_ids=demand_ids, element_type="node", field="actual_demand", start_dt=start_dt, end_dt=end_dt, simulation_source=simulation_source, simulation_scheme_name=simulation_scheme_name, simulation_scheme_type=simulation_scheme_type, ) ) return result def _query_simulation_values( *, network: str, element_ids: list[str], element_type: str, field: str, start_dt: datetime, end_dt: datetime, simulation_source: str, simulation_scheme_name: str | None, simulation_scheme_type: str, ) -> dict[str, list[dict[str, Any]]]: element_ids = _dedupe_ids(element_ids) if not element_ids: return {} if simulation_source == "scheme": if not simulation_scheme_name: raise ValueError("读取方案模拟数据时必须提供 simulation_scheme_name。") return InternalQueries.query_scheme_simulation_by_ids_timerange( db_name=network, scheme_type=simulation_scheme_type, scheme_name=simulation_scheme_name, element_ids=element_ids, start_time=start_dt.isoformat(), end_time=end_dt.isoformat(), element_type=element_type, field=field, ) if simulation_source == "realtime": return InternalQueries.query_realtime_simulation_by_ids_timerange( db_name=network, element_ids=element_ids, start_time=start_dt.isoformat(), end_time=end_dt.isoformat(), element_type=element_type, field=field, ) raise ValueError(f"Unsupported simulation_source: {simulation_source}") def _build_sensor_metadata(network: str, data_type: str) -> dict[str, dict[str, str]]: metadata: dict[str, dict[str, str]] = {} for item in get_all_scada_info(network): scada_type = str(item.get("type", "")).lower() if data_type == "pressure": if scada_type != "pressure": continue elif data_type == "flow": if scada_type not in FLOW_SCADA_TYPES: continue else: raise ValueError(f"Unsupported data_type: {data_type}") element_id = _normalize_identifier(item.get("associated_element_id")) query_id = _normalize_identifier(item.get("api_query_id")) if element_id and query_id: metadata[element_id] = {"query_id": query_id, "scada_type": scada_type} return metadata def _build_scada_mapping(network: str, data_type: str) -> dict[str, str]: metadata = _build_sensor_metadata(network=network, data_type=data_type) return {element_id: item["query_id"] for element_id, item in metadata.items()} def _normalize_data_source( data_source: str | None, simulation_scheme_name: str | None = None ) -> str: normalized = str(data_source or "").strip().lower() if not normalized: return "simulation" if simulation_scheme_name else "monitoring" if normalized not in SIMULATION_DATA_SOURCES: allowed_sources = ", ".join(sorted(SIMULATION_DATA_SOURCES)) raise ValueError( f"Unsupported data_source: {data_source}. Allowed: {allowed_sources}" ) return normalized def _get_sensor_nodes(network: str, data_type: str) -> list[str]: mapping = _build_scada_mapping(network=network, data_type=data_type) sensor_ids = sorted(mapping.keys()) if not sensor_ids: type_name = "压力" if data_type == "pressure" else "流量" raise ValueError(f"未找到{type_name}传感器对应节点（scada_info.type）。") return sensor_ids def _dedupe_ids(ids: list[str] | None) -> list[str]: if ids is None: return [] return list( dict.fromkeys( normalized for normalized in (_normalize_identifier(item) for item in ids) if normalized ) ) def _normalize_identifier(value: Any) -> str: if value is None: return "" return str(value).strip() def _normalize_timeseries_by_id( data: dict[Any, list[dict[str, Any]]] | None, ) -> dict[str, list[dict[str, Any]]]: normalized_data: dict[str, list[dict[str, Any]]] = {} for raw_id, records in (data or {}).items(): normalized_id = _normalize_identifier(raw_id) if not normalized_id: continue normalized_data.setdefault(normalized_id, []).extend(records or []) return normalized_data def _to_datetime(value: datetime | str) -> datetime: return parse_utc_time(value) def _prepare_burst_inp(network: str) -> str: project_root = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")) db_inp_dir = os.path.join(project_root, "db_inp") os.makedirs(db_inp_dir, exist_ok=True) inp_path = os.path.join(db_inp_dir, f"{network}.burst.inp") if os.path.isfile(inp_path) and os.path.getsize(inp_path) > 0: return inp_path dump_inp(network, inp_path, "2") if not os.path.isfile(inp_path) or os.path.getsize(inp_path) <= 0: raise ValueError(f"爆管定位 INP 文件无效: {inp_path}") return inp_path