jiang
87824f3b3c
Simulation mode reads scheme data for both burst and normal observations. Monitoring mode reuses the burst window when no normal window is provided.
760 lines
28 KiB
Python
760 lines
28 KiB
Python
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"
|
|
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
|
|
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 _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:
|
|
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 _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
|