From 5a91da09044dcde698c4015055ddb01e1ce02bf5 Mon Sep 17 00:00:00 2001 From: Jiang Date: Wed, 8 Jul 2026 17:51:08 +0800 Subject: [PATCH] fix(burst-location): use correct data sources Simulation mode reads scheme data for both burst and normal observations. Monitoring mode reuses the burst window when no normal window is provided. --- app/api/v1/endpoints/burst_location.py | 6 +- app/services/burst_location.py | 162 ++++++++++++++++------ tests/unit/test_burst_location_service.py | 136 +++++++++++++++--- 3 files changed, 242 insertions(+), 62 deletions(-) diff --git a/app/api/v1/endpoints/burst_location.py b/app/api/v1/endpoints/burst_location.py index bc4023c..0bb36e2 100644 --- a/app/api/v1/endpoints/burst_location.py +++ b/app/api/v1/endpoints/burst_location.py @@ -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="模拟方案名称") diff --git a/app/services/burst_location.py b/app/services/burst_location.py index 95acb16..3925b80 100644 --- a/app/services/burst_location.py +++ b/app/services/burst_location.py @@ -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, diff --git a/tests/unit/test_burst_location_service.py b/tests/unit/test_burst_location_service.py index c62c460..a5f00e0 100644 --- a/tests/unit/test_burst_location_service.py +++ b/tests/unit/test_burst_location_service.py @@ -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)