OrgTJWater/TJWaterServerBinary
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(burst-location): use correct data sources

Browse Source 
Simulation mode reads scheme data for both burst and normal observations. Monitoring mode reuses the burst window when no normal window is provided.
This commit is contained in:
jiang
2026-07-08 17:51:08 +08:00
parent d62bcae85e
commit 5a91da0904
3 changed files with 242 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/api/v1/endpoints/burst_location.py
+4 -2
View File
@@ -29,8 +29,10 @@ class BurstLocationRequest(BaseModel):
normal_flow: dict[str, float] | list[dict[str, Any]] | None = Field(None, description="正常时的流量数据")
min_dpressure: float = Field(2.0, description="最小压力差（bar）")
basic_pressure: float = Field(10.0, description="基准压力（bar）")
scada_burst_start: datetime | None = Field(None, description="SCADA爆管开始时间")
scada_burst_end: datetime | None = Field(None, description="SCADA爆管结束时间")
scada_burst_start: datetime | None = Field(None, description="爆管/模拟方案开始时间")
scada_burst_end: datetime | None = Field(None, description="爆管/模拟方案结束时间")
scada_normal_start: datetime | None = Field(None, description="监测数据正常工况开始时间")
scada_normal_end: datetime | None = Field(None, description="监测数据正常工况结束时间")
use_scada_flow: bool = Field(False, description="是否使用SCADA流量数据")
scheme_name: str | None = Field(None, description="方案名称")
simulation_scheme_name: str | None = Field(None, description="模拟方案名称")
app/services/burst_location.py
+119 -43
View File
@@ -60,6 +60,8 @@ def run_burst_location_by_network(
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,
@@ -87,12 +89,37 @@ def run_burst_location_by_network(
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_scada_windows(
scada_burst_start=scada_burst_start,
scada_burst_end=scada_burst_end,
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:
@@ -117,15 +144,15 @@ def run_burst_location_by_network(
) = _build_observed_series_from_simulation(
network=network,
sensor_ids=selected_pressure_ids,
start_dt=burst_start_dt,
end_dt=burst_end_dt,
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="realtime",
simulation_scheme_name=None,
simulation_source="scheme",
simulation_scheme_name=simulation_scheme_name,
simulation_scheme_type=resolved_simulation_scheme_type,
)
observed_source = "simulation_scheme_burst_realtime_normal_timerange"
observed_source = "simulation_scheme_timerange"
else:
(
burst_pressure_series,
@@ -138,21 +165,27 @@ def run_burst_location_by_network(
data_type="pressure",
series_name="burst_pressure",
)
(
normal_pressure_series,
normal_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="normal_pressure",
simulation_source="realtime",
simulation_scheme_name=None,
simulation_scheme_type=resolved_simulation_scheme_type,
)
observed_source = "scada_burst_realtime_normal_timerange"
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(
@@ -179,6 +212,11 @@ def run_burst_location_by_network(
)
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。")
@@ -199,12 +237,12 @@ def run_burst_location_by_network(
_build_observed_series_from_simulation(
network=network,
sensor_ids=selected_flow_ids,
start_dt=burst_start_dt,
end_dt=burst_end_dt,
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="realtime",
simulation_scheme_name=None,
simulation_source="scheme",
simulation_scheme_name=simulation_scheme_name,
simulation_scheme_type=resolved_simulation_scheme_type,
)
)
@@ -217,19 +255,20 @@ def run_burst_location_by_network(
data_type="flow",
series_name="burst_flow",
)
normal_flow_series, normal_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="normal_flow",
simulation_source="realtime",
simulation_scheme_name=None,
simulation_scheme_type=resolved_simulation_scheme_type,
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)
@@ -281,6 +320,13 @@ def run_burst_location_by_network(
"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,
@@ -376,6 +422,23 @@ def _validate_scada_windows(
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,
@@ -392,7 +455,7 @@ def _build_observed_series_from_scada(
]
if missing_ids:
preview = ", ".join(missing_ids[:10])
raise ValueError(f"{series_name} 缺少可用 SCADA 映射: {preview}")
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(
@@ -410,7 +473,9 @@ def _build_observed_series_from_scada(
float(item["value"]) for item in records if item.get("value") is not None
]
if not numeric_values:
raise ValueError(f"{series_name} 在时间窗内无有效数据: {sensor_id}")
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))
@@ -436,7 +501,7 @@ def _build_observed_series_from_simulation(
]
if missing_ids:
preview = ", ".join(missing_ids[:10])
raise ValueError(f"{series_name} 缺少可用 SCADA 映射: {preview}")
raise ValueError(f"{_series_display_name(series_name)} 缺少可用 SCADA 映射: {preview}")
simulation_data = _query_simulation_data_by_sensor_ids(
network=network,
@@ -458,13 +523,24 @@ def _build_observed_series_from_simulation(
float(item["value"]) for item in records if item.get("value") is not None
]
if not numeric_values:
raise ValueError(f"{series_name} 在时间窗内无有效模拟数据: {sensor_id}")
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,
tests/unit/test_burst_location_service.py
+119 -17
View File
@@ -190,7 +190,7 @@ def test_run_burst_location_uses_single_timerange_with_burst_source_split(monkey
use_scada_flow=True,
)
assert result["observed_source"] == "simulation_scheme_burst_realtime_normal_timerange"
assert result["observed_source"] == "simulation_scheme_timerange"
assert result["simulation_scheme"] == {
"name": "BurstSchemeA",
"type": "burst_analysis",
@@ -199,22 +199,17 @@ def test_run_burst_location_uses_single_timerange_with_burst_source_split(monkey
assert result["flow_samples"] == {"burst": 4, "normal": 4}
assert list(captured["burst_pressure"].index) == ["J1"]
assert captured["burst_pressure"]["J1"] == pytest.approx(15.0)
assert captured["normal_pressure"]["J1"] == pytest.approx(11.0)
assert captured["normal_pressure"]["J1"] == pytest.approx(15.0)
assert captured["burst_flow"]["J2"] == pytest.approx(6.0)
assert captured["burst_flow"]["P1"] == pytest.approx(8.0)
assert captured["normal_flow"]["J2"] == pytest.approx(4.0)
assert captured["normal_flow"]["P1"] == pytest.approx(5.0)
assert captured["normal_flow"]["J2"] == pytest.approx(6.0)
assert captured["normal_flow"]["P1"] == pytest.approx(8.0)
assert all(call["scheme_name"] == "BurstSchemeA" for call in scheme_calls)
assert len(scheme_calls) == 3
assert len(scheme_calls) == 6
assert any(call["element_type"] == "node" and call["field"] == "pressure" for call in scheme_calls)
assert any(call["element_type"] == "link" and call["field"] == "flow" for call in scheme_calls)
assert any(call["element_type"] == "node" and call["field"] == "actual_demand" for call in scheme_calls)
assert len(realtime_calls) == 3
assert all(datetime.fromisoformat(call["start_time"]).hour == 0 for call in realtime_calls)
assert all(datetime.fromisoformat(call["end_time"]).hour == 1 for call in realtime_calls)
assert any(call["element_type"] == "node" and call["field"] == "pressure" for call in realtime_calls)
assert any(call["element_type"] == "link" and call["field"] == "flow" for call in realtime_calls)
assert any(call["element_type"] == "node" and call["field"] == "actual_demand" for call in realtime_calls)
assert realtime_calls == []
assert result["scada_window"] == {
"burst_start": "2025-01-01T00:00:00+00:00",
"burst_end": "2025-01-01T01:00:00+00:00",
@@ -296,7 +291,42 @@ def test_build_observed_series_from_simulation_normalizes_result_ids(monkeypatch
assert series["100026"] == pytest.approx(12.0)
def test_run_burst_location_monitoring_uses_scada_for_burst_and_realtime_for_normal(
def test_build_observed_series_from_scada_uses_chinese_error_label(monkeypatch):
module = _load_burst_location_module()
monkeypatch.setattr(
module,
"get_all_scada_info",
lambda network: [
{
"type": "pressure",
"associated_element_id": "100026",
"api_query_id": "pressure-query",
}
],
)
monkeypatch.setattr(
module.InternalQueries,
"query_scada_by_ids_timerange",
staticmethod(lambda **kwargs: {"pressure-query": []}),
)
with pytest.raises(ValueError) as exc_info:
module._build_observed_series_from_scada(
network="tjwater",
sensor_ids=["100026"],
start_dt=datetime(2025, 1, 1, 0, 0, 0, tzinfo=timezone.utc),
end_dt=datetime(2025, 1, 1, 1, 0, 0, tzinfo=timezone.utc),
data_type="pressure",
series_name="burst_pressure",
)
message = str(exc_info.value)
assert "爆管压力数据 在时间窗内无有效数据: 100026" in message
assert "burst_pressure" not in message
def test_run_burst_location_monitoring_uses_scada_for_burst_and_normal(
monkeypatch, tmp_path
):
module = _load_burst_location_module()
@@ -324,10 +354,14 @@ def test_run_burst_location_monitoring_uses_scada_for_burst_and_realtime_for_nor
def fake_scada_query(**kwargs):
scada_calls.append(kwargs)
start_hour = datetime.fromisoformat(kwargs["start_time"]).astimezone(
timezone(timedelta(hours=8))
).hour
values = [20.0, 22.0] if start_hour == 8 else [10.0, 12.0]
return {
"pressure-query": [
{"time": kwargs["start_time"], "value": 20.0},
{"time": kwargs["end_time"], "value": 22.0},
{"time": kwargs["start_time"], "value": values[0]},
{"time": kwargs["end_time"], "value": values[1]},
]
}
@@ -358,10 +392,78 @@ def test_run_burst_location_monitoring_uses_scada_for_burst_and_realtime_for_nor
burst_leakage=1.0,
scada_burst_start=datetime(2025, 1, 1, 8, 0, 0, tzinfo=timezone(timedelta(hours=8))),
scada_burst_end=datetime(2025, 1, 1, 9, 0, 0, tzinfo=timezone(timedelta(hours=8))),
scada_normal_start=datetime(2025, 1, 1, 7, 0, 0, tzinfo=timezone(timedelta(hours=8))),
scada_normal_end=datetime(2025, 1, 1, 8, 0, 0, tzinfo=timezone(timedelta(hours=8))),
)
assert result["observed_source"] == "scada_burst_realtime_normal_timerange"
assert len(scada_calls) == 1
assert len(realtime_calls) == 1
assert result["observed_source"] == "scada_burst_scada_normal_timerange"
assert len(scada_calls) == 2
assert len(realtime_calls) == 0
assert captured["burst_pressure"]["J1"] == pytest.approx(21.0)
assert captured["normal_pressure"]["J1"] == pytest.approx(11.0)
assert result["scada_window"] == {
"burst_start": "2025-01-01T00:00:00+00:00",
"burst_end": "2025-01-01T01:00:00+00:00",
"normal_start": "2024-12-31T23:00:00+00:00",
"normal_end": "2025-01-01T00:00:00+00:00",
}
def test_run_burst_location_monitoring_reuses_burst_window_for_normal(
monkeypatch, tmp_path
):
module = _load_burst_location_module()
captured = {}
scada_calls = []
monkeypatch.setattr(
module,
"get_all_scada_info",
lambda network: [
{
"type": "pressure",
"associated_element_id": "J1",
"api_query_id": "pressure-query",
}
],
)
monkeypatch.setattr(module, "_prepare_burst_inp", lambda network: str(tmp_path / "fake.inp"))
monkeypatch.setattr(
module,
"run_burst_location",
lambda **kwargs: captured.update(kwargs) or {"located_pipe": "Pipe-001"},
)
monkeypatch.setattr(
module.InternalQueries,
"query_scada_by_ids_timerange",
staticmethod(
lambda **kwargs: scada_calls.append(kwargs)
or {
"pressure-query": [
{"time": kwargs["start_time"], "value": 20.0},
{"time": kwargs["end_time"], "value": 22.0},
]
}
),
)
monkeypatch.setattr(
module.InternalQueries,
"query_realtime_simulation_by_ids_timerange",
staticmethod(lambda **kwargs: pytest.fail("monitoring mode must not query realtime simulation")),
)
result = module.run_burst_location_by_network(
network="tjwater",
username="testuser",
data_source="monitoring",
burst_leakage=1.0,
scada_burst_start=datetime(2025, 1, 1, 8, 0, 0, tzinfo=timezone(timedelta(hours=8))),
scada_burst_end=datetime(2025, 1, 1, 9, 0, 0, tzinfo=timezone(timedelta(hours=8))),
)
assert result["observed_source"] == "scada_timerange"
assert len(scada_calls) == 2
assert scada_calls[0]["start_time"] == scada_calls[1]["start_time"]
assert scada_calls[0]["end_time"] == scada_calls[1]["end_time"]
assert captured["burst_pressure"]["J1"] == pytest.approx(21.0)
assert captured["normal_pressure"]["J1"] == pytest.approx(21.0)