fix(burst-location): correct normal data window

app/algorithms/burst_location/burst_location.py

@@ -121,7 +121,7 @@ def run_burst_location(
     basic_pressure: float = 10.0,
     n_workers: int = DEFAULT_N_WORKERS,
     partition_on_full_graph: bool = True,
-    visualize_partition: bool = True,
+    visualize_partition: bool = False,
     visualize_pause_seconds: float = 0.3,
     final_candidates_csv_path: (
         str | None

app/services/burst_location.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import os
-from datetime import datetime
+from datetime import datetime, timedelta
 from typing import Any
 
 import pandas as pd
@@ -124,6 +124,8 @@ def run_burst_location_by_network(
         if normalized_data_source == "simulation":
             if not simulation_scheme_name:
                 raise ValueError("模拟方案模式必须提供 simulation_scheme_name。")
+            normal_start_dt = burst_start_dt
+            normal_end_dt = burst_end_dt
             (
                 burst_pressure_series,
                 burst_pressure_samples,
@@ -144,16 +146,21 @@ def run_burst_location_by_network(
             ) = _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,
+                start_dt=normal_start_dt,
+                end_dt=normal_end_dt,
                 data_type="pressure",
                 series_name="normal_pressure",
-                simulation_source="scheme",
-                simulation_scheme_name=simulation_scheme_name,
+                simulation_source="realtime",
+                simulation_scheme_name=None,
                 simulation_scheme_type=resolved_simulation_scheme_type,
             )
-            observed_source = "simulation_scheme_timerange"
+            observed_source = "simulation_scheme_burst_realtime_normal_timerange"
         else:
+            if normal_pressure_from_payload is None and (
+                normal_start_dt is None or normal_end_dt is None
+            ):
+                normal_start_dt = burst_start_dt - timedelta(days=1)
+                normal_end_dt = burst_end_dt - timedelta(days=1)
             (
                 burst_pressure_series,
                 burst_pressure_samples,
@@ -172,20 +179,24 @@ def run_burst_location_by_network(
                 ) = _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,
+                    start_dt=normal_start_dt,
+                    end_dt=normal_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"
-                )
+                observed_source = "scada_burst_scada_normal_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(
@@ -237,16 +248,21 @@ def run_burst_location_by_network(
                 _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,
+                    start_dt=normal_start_dt,
+                    end_dt=normal_end_dt,
                     data_type="flow",
                     series_name="normal_flow",
-                    simulation_source="scheme",
-                    simulation_scheme_name=simulation_scheme_name,
+                    simulation_source="realtime",
+                    simulation_scheme_name=None,
                     simulation_scheme_type=resolved_simulation_scheme_type,
                 )
             )
         else:
+            if normal_flow_from_payload is None and (
+                normal_start_dt is None or normal_end_dt is None
+            ):
+                normal_start_dt = burst_start_dt - timedelta(days=1)
+                normal_end_dt = burst_end_dt - timedelta(days=1)
             burst_flow_series, burst_flow_samples = _build_observed_series_from_scada(
                 network=network,
                 sensor_ids=selected_flow_ids,
@@ -260,8 +276,8 @@ def run_burst_location_by_network(
                     _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,
+                        start_dt=normal_start_dt,
+                        end_dt=normal_end_dt,
                         data_type="flow",
                         series_name="normal_flow",
                     )
@@ -269,6 +285,14 @@ def run_burst_location_by_network(
             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)
@@ -297,6 +321,7 @@ def run_burst_location_by_network(
         normal_flow=normal_flow_series,
         min_dpressure=min_dpressure,
         basic_pressure=basic_pressure,
+        visualize_partition=False,
     )
 
     payload: dict[str, Any] = {
@@ -439,6 +464,23 @@ def _validate_time_window(
     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,
@@ -473,11 +515,13 @@ 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_display_name(series_name)} 在时间窗内无有效数据: {sensor_id}"
-            )
+            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)
 

tests/unit/test_burst_location_service.py

@@ -190,29 +190,41 @@ def test_run_burst_location_uses_single_timerange_with_burst_source_split(monkey
         use_scada_flow=True,
     )
 
-    assert result["observed_source"] == "simulation_scheme_timerange"
+    assert result["observed_source"] == "simulation_scheme_burst_realtime_normal_timerange"
     assert result["simulation_scheme"] == {
         "name": "BurstSchemeA",
         "type": "burst_analysis",
     }
     assert result["pressure_samples"] == {"burst": 4, "normal": 4}
     assert result["flow_samples"] == {"burst": 4, "normal": 4}
+    assert captured["visualize_partition"] is False
     assert list(captured["burst_pressure"].index) == ["J1"]
     assert captured["burst_pressure"]["J1"] == pytest.approx(15.0)
-    assert captured["normal_pressure"]["J1"] == pytest.approx(15.0)
+    assert captured["normal_pressure"]["J1"] == pytest.approx(11.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(6.0)
-    assert captured["normal_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 all(call["scheme_name"] == "BurstSchemeA" for call in scheme_calls)
-    assert len(scheme_calls) == 6
+    assert len(scheme_calls) == 3
     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 realtime_calls == []
+    assert len(realtime_calls) == 3
+    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 {call["start_time"] for call in scheme_calls + realtime_calls} == {
+        "2025-01-01T00:00:00+00:00"
+    }
+    assert {call["end_time"] for call in scheme_calls + realtime_calls} == {
+        "2025-01-01T01:00:00+00:00"
+    }
     assert result["scada_window"] == {
         "burst_start": "2025-01-01T00:00:00+00:00",
         "burst_end": "2025-01-01T01:00:00+00:00",
+        "normal_start": "2025-01-01T00:00:00+00:00",
+        "normal_end": "2025-01-01T01:00:00+00:00",
     }
 
 
@@ -326,6 +338,51 @@ def test_build_observed_series_from_scada_uses_chinese_error_label(monkeypatch):
     assert "burst_pressure" not in message
 
 
+def test_build_observed_series_from_scada_skips_missing_sensor_values(monkeypatch):
+    module = _load_burst_location_module()
+
+    monkeypatch.setattr(
+        module,
+        "get_all_scada_info",
+        lambda network: [
+            {"type": "pressure", "associated_element_id": "J1", "api_query_id": "q1"},
+            {"type": "pressure", "associated_element_id": "J2", "api_query_id": "q2"},
+            {"type": "pressure", "associated_element_id": "J3", "api_query_id": "q3"},
+        ],
+    )
+    monkeypatch.setattr(
+        module.InternalQueries,
+        "query_scada_by_ids_timerange",
+        staticmethod(
+            lambda **kwargs: {
+                "q1": [
+                    {"time": kwargs["start_time"], "value": 10.0},
+                    {"time": kwargs["end_time"], "value": 12.0},
+                ],
+                "q2": [],
+                "q3": [
+                    {"time": kwargs["start_time"], "value": None},
+                    {"time": kwargs["end_time"], "value": 18.0},
+                ],
+            }
+        ),
+    )
+
+    series, sample_count = module._build_observed_series_from_scada(
+        network="tjwater",
+        sensor_ids=["J1", "J2", "J3"],
+        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",
+    )
+
+    assert list(series.index) == ["J1", "J3"]
+    assert series["J1"] == pytest.approx(11.0)
+    assert series["J3"] == pytest.approx(18.0)
+    assert sample_count == 1
+
+
 def test_run_burst_location_monitoring_uses_scada_for_burst_and_normal(
     monkeypatch, tmp_path
 ):
@@ -409,7 +466,7 @@ def test_run_burst_location_monitoring_uses_scada_for_burst_and_normal(
     }
 
 
-def test_run_burst_location_monitoring_reuses_burst_window_for_normal(
+def test_run_burst_location_monitoring_defaults_normal_window_to_previous_day(
     monkeypatch, tmp_path
 ):
     module = _load_burst_location_module()
@@ -433,18 +490,26 @@ def test_run_burst_location_monitoring_reuses_burst_window_for_normal(
         "run_burst_location",
         lambda **kwargs: captured.update(kwargs) or {"located_pipe": "Pipe-001"},
     )
+
+    def fake_scada_query(**kwargs):
+        scada_calls.append(kwargs)
+        start_time = datetime.fromisoformat(kwargs["start_time"])
+        values = (
+            [20.0, 22.0]
+            if start_time.date().isoformat() == "2025-01-01"
+            else [10.0, 12.0]
+        )
+        return {
+            "pressure-query": [
+                {"time": kwargs["start_time"], "value": values[0]},
+                {"time": kwargs["end_time"], "value": values[1]},
+            ]
+        }
+
     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},
-                ]
-            }
-        ),
+        staticmethod(fake_scada_query),
     )
     monkeypatch.setattr(
         module.InternalQueries,
@@ -461,9 +526,165 @@ def test_run_burst_location_monitoring_reuses_burst_window_for_normal(
         scada_burst_end=datetime(2025, 1, 1, 9, 0, 0, tzinfo=timezone(timedelta(hours=8))),
     )
 
-    assert result["observed_source"] == "scada_timerange"
+    assert result["observed_source"] == "scada_burst_scada_normal_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 datetime.fromisoformat(scada_calls[1]["start_time"]) == (
+        datetime.fromisoformat(scada_calls[0]["start_time"]) - timedelta(days=1)
+    )
+    assert datetime.fromisoformat(scada_calls[1]["end_time"]) == (
+        datetime.fromisoformat(scada_calls[0]["end_time"]) - timedelta(days=1)
+    )
     assert captured["burst_pressure"]["J1"] == pytest.approx(21.0)
-    assert captured["normal_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-31T00:00:00+00:00",
+        "normal_end": "2024-12-31T01:00:00+00:00",
+    }
+
+
+def test_run_burst_location_monitoring_flow_uses_previous_day_normal_window(
+    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",
+            },
+            {
+                "type": "pipe_flow",
+                "associated_element_id": "P1",
+                "api_query_id": "flow-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"},
+    )
+
+    def fake_scada_query(**kwargs):
+        scada_calls.append(kwargs)
+        is_burst_day = (
+            datetime.fromisoformat(kwargs["start_time"]).date().isoformat()
+            == "2025-01-01"
+        )
+        if kwargs["device_ids"] == ["pressure-query"]:
+            values = [20.0, 22.0] if is_burst_day else [10.0, 12.0]
+            query_id = "pressure-query"
+        else:
+            values = [7.0, 9.0] if is_burst_day else [3.0, 5.0]
+            query_id = "flow-query"
+        return {
+            query_id: [
+                {"time": kwargs["start_time"], "value": values[0]},
+                {"time": kwargs["end_time"], "value": values[1]},
+            ]
+        }
+
+    monkeypatch.setattr(
+        module.InternalQueries,
+        "query_scada_by_ids_timerange",
+        staticmethod(fake_scada_query),
+    )
+
+    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))),
+        use_scada_flow=True,
+    )
+
+    assert result["observed_source"] == "scada_burst_scada_normal_timerange"
+    assert len(scada_calls) == 4
+    for burst_call, normal_call in [
+        (scada_calls[0], scada_calls[1]),
+        (scada_calls[2], scada_calls[3]),
+    ]:
+        assert datetime.fromisoformat(normal_call["start_time"]) == (
+            datetime.fromisoformat(burst_call["start_time"]) - timedelta(days=1)
+        )
+        assert datetime.fromisoformat(normal_call["end_time"]) == (
+            datetime.fromisoformat(burst_call["end_time"]) - timedelta(days=1)
+        )
+    assert captured["burst_pressure"]["J1"] == pytest.approx(21.0)
+    assert captured["normal_pressure"]["J1"] == pytest.approx(11.0)
+    assert captured["burst_flow"]["P1"] == pytest.approx(8.0)
+    assert captured["normal_flow"]["P1"] == pytest.approx(4.0)
+
+
+def test_run_burst_location_monitoring_aligns_partial_scada_data(
+    monkeypatch, tmp_path
+):
+    module = _load_burst_location_module()
+    captured = {}
+
+    monkeypatch.setattr(
+        module,
+        "get_all_scada_info",
+        lambda network: [
+            {"type": "pressure", "associated_element_id": "J1", "api_query_id": "q1"},
+            {"type": "pressure", "associated_element_id": "J2", "api_query_id": "q2"},
+            {"type": "pressure", "associated_element_id": "J3", "api_query_id": "q3"},
+        ],
+    )
+    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"},
+    )
+
+    def fake_scada_query(**kwargs):
+        start_hour = datetime.fromisoformat(kwargs["start_time"]).astimezone(
+            timezone(timedelta(hours=8))
+        ).hour
+        if start_hour == 8:
+            return {
+                "q1": [{"time": kwargs["start_time"], "value": 20.0}],
+                "q2": [{"time": kwargs["start_time"], "value": 30.0}],
+                "q3": [],
+            }
+        return {
+            "q1": [{"time": kwargs["start_time"], "value": 10.0}],
+            "q2": [],
+            "q3": [{"time": kwargs["start_time"], "value": 12.0}],
+        }
+
+    monkeypatch.setattr(
+        module.InternalQueries,
+        "query_scada_by_ids_timerange",
+        staticmethod(fake_scada_query),
+    )
+
+    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))),
+        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["pressure_scada_ids"] == ["J1"]
+    assert captured["pressure_scada_ids"] == ["J1"]
+    assert list(captured["burst_pressure"].index) == ["J1"]
+    assert list(captured["normal_pressure"].index) == ["J1"]
+    assert captured["burst_pressure"]["J1"] == pytest.approx(20.0)
+    assert captured["normal_pressure"]["J1"] == pytest.approx(10.0)