fix: make burst impact coverage envelopes exclusive

app/services/burst_impact.py

@@ -139,9 +139,12 @@ def estimate_burst_impact(network: str, start: datetime, simulation: dict[str, A
         return output
     groups = [{key: group[key] for key in ("group_id", "title", "node_ids")}
               for group in impact["grades"]]
-    geometry = generate_node_coverage(network, groups)
+    # GRADES are ordered from low to high. Keep the numerical classes intact
+    # while removing overlap between their independently derived envelopes.
+    geometry = generate_node_coverage(network, groups, exclusive_groups=True)
     impact["coverage_status"] = geometry["coverage"]["status"]
     geometry["coverage"]["boundary"] = impact["boundary"]
+    geometry["coverage"]["boundary"] += "分档包络互斥展示，重叠区域优先显示较高降压档，不改变节点分档或表示精确水力等值边界。"
     for feature in geometry["map_payload"]["features"]:
         feature["properties"].update(
             impact_phase=impact["phase"], pressure_drop_unit="m",

app/services/node_coverage.py

@@ -6,7 +6,7 @@ from typing import Any
 
 from pyproj import CRS, Transformer
 from shapely import MultiPoint, concave_hull, get_num_coordinates
-from shapely.geometry import mapping
+from shapely.geometry import GeometryCollection, mapping, shape
 from shapely.ops import transform, unary_union
 
 from app.native.wndb.gis.network_views import get_node_coverage_links, get_node_geographic_positions
@@ -17,6 +17,47 @@ SIMPLIFY_METERS = 2.0
 MAX_COVERAGE_NODES = 10000
 
 
+def exclusive_coverage_features(features: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    """Clip display envelopes, with later groups taking precedence; retain input order.
+
+    This changes only derived geometry, never node membership or model values.
+    No post-clipping buffer or simplification may reintroduce an overlap.
+    """
+    occupied = GeometryCollection()
+    result = []
+    summaries = {}
+    for feature in features:
+        properties = feature["properties"]
+        summaries.update(properties.get("coverage_group_summary", {}))
+    for feature in features:
+        properties = feature["properties"]
+        summaries[properties["coverage_group"]] = {
+            key: properties.get(key) for key in ("node_count", "mapped_node_count", "missing_node_count")
+        }
+    for feature in reversed(features):
+        original = shape(feature["geometry"])
+        visible = original.difference(occupied)
+        occupied = unary_union([occupied, original])
+        if visible.is_empty:
+            continue
+        if (visible.geom_type not in {"Polygon", "MultiPolygon"} or not visible.is_valid
+                or get_num_coordinates(visible) > 20000):
+            raise ValueError("Invalid exclusive derived coverage")
+        result.append({
+            **feature,
+            "properties": {**feature["properties"], "overlap_policy": "higher_priority_group_first"},
+            "geometry": mapping(visible),
+        })
+    visible_groups = {feature["properties"]["coverage_group"] for feature in result}
+    summaries = {key: {**value, "display_area_available": key in visible_groups}
+                 for key, value in summaries.items()}
+    for feature in result:
+        # Retain counts even when that group's entire display envelope was
+        # clipped away. Overlay transport deliberately sends only GeoJSON.
+        feature["properties"]["coverage_group_summary"] = summaries
+    return list(reversed(result))
+
+
 def _components(node_ids: list[str], links: list[dict[str, Any]], closed: set[str]) -> list[list[str]]:
     parent = {node_id: node_id for node_id in node_ids}
 
@@ -41,6 +82,7 @@ def build_node_coverage(
     links: list[dict[str, Any]], *, split_by_connectivity: bool = False,
     excluded_link_ids: list[str] | None = None,
     concave_ratio: float = CONCAVE_RATIO, buffer_m: float = BUFFER_METERS,
+    exclusive_groups: bool = False,
 ) -> dict[str, Any]:
     """Keep caller-defined comparison groups and derive geometry without assigning business meaning."""
     coordinates = {}
@@ -99,6 +141,12 @@ def build_node_coverage(
         }
         features.append(feature)
 
+    if exclusive_groups:
+        features = exclusive_coverage_features(features)
+        visible_groups = {feature["properties"]["coverage_group"] for feature in features}
+        for group_id, metadata in groups_metadata.items():
+            metadata["display_area_available"] = group_id in visible_groups
+
     return {
         "status": "partial" if any(group["missing_node_ids"] for group in groups_metadata.values())
         else "ready" if features else "empty",
@@ -106,7 +154,8 @@ def build_node_coverage(
         "parameters": {"concave_ratio": concave_ratio, "buffer_m": buffer_m,
                        "split_by_connectivity": split_by_connectivity,
                        "excluded_link_ids": sorted(closed),
-                       "simplify_m": simplify_m, "projection": "local_azimuthal_equidistant"},
+                       "simplify_m": simplify_m, "projection": "local_azimuthal_equidistant",
+                       "exclusive_groups": exclusive_groups},
         "output_crs": "EPSG:4326", "display_only": True,
         "position_source": "gis.node_geometries",
         "boundary": "节点空间覆盖示意；不推断故障、污染、停水、水力影响或用户服务边界。",

docs/burst-response-workflow.md

@@ -79,7 +79,7 @@ SCADA作为事件影响的辅助证据，按目标测点和事件／对照窗口
 
 在筛选最多20个展示节点之前，以原生模拟全网节点的起算首帧为计算对象，严格读取同一时刻已有在线模拟压力。不同压力单位换算为米水头，下降量为在线压力减爆管压力，互斥划分 `1≤Δ<3 m`、`3≤Δ<5 m`、`Δ≥5 m`。小于1米与缺少可比数据分别计数；错时、缺测、无效值或未知单位不补零、不取邻近值，不从代表测点外推全网。需要 `simulation.run`、`scada.view`，影响比较另需 `simulation.view`。
 
-输出 `impact` 保留每个可比节点的在线压力、爆管压力、下降量、分组、缺测清单、比较时刻和读取来源；`coverage` 和 `map_payload` 使用实际GIS节点坐标生成三档空间包络。缺坐标、覆盖节点超过10000或几何超预算时明确报告，不截取前若干节点冒充完整范围。地图采用黄／橙／红，图例保留每档已绘制／分组节点数。
+输出 `impact` 保留每个可比节点的在线压力、爆管压力、下降量、分组、缺测清单、比较时刻和读取来源；`coverage` 和 `map_payload` 使用实际GIS节点坐标生成三档空间包络。缺坐标、覆盖节点超过10000或几何超预算时明确报告，不截取前若干节点冒充完整范围。爆管分档包络按高降压档优先裁剪为互斥展示面，保留整体覆盖并集、原始节点分组和计数；低档包络被完全覆盖时保留其节点元数据，不输出空几何。该裁剪只处理空间包络的展示重叠，不把节点包络变为精确水力等值区。通用节点分组及关阀覆盖仍保持原有生成语义。地图三档采用蓝／橙／紫，避免与关阀覆盖使用的红色混淆，图例保留每档已绘制／分组节点数。
 
 爆管模型使用 PDA 需水方式和原有10模型压力单位的服务压力设定；在线模型方式、当前模型时态及在线压力单位推定来源一并保留。在线记录没有独立版本化单位元数据，不能把此估计称为严格历史同模型反事实或现场停水边界。
 

tests/unit/test_burst_impact.py

@@ -22,11 +22,11 @@ def inputs(monkeypatch):
     read = Mock(side_effect=lambda ids, *a, **kw: {
         node: [{"time": START.isoformat(), "value": 30}] for node in ids})
     monkeypatch.setattr(service.InternalQueries, "query_realtime_simulation_by_ids_timerange", read)
-    def coverage(network, groups):
+    def coverage(network, groups, **parameters):
         ids = sorted({node for group in groups for node in group["node_ids"]})
         data = build_node_coverage(groups, [
             {"id": node, "longitude": 121.5 + i * .001, "latitude": 30.9}
-            for i, node in enumerate(ids)], [])
+            for i, node in enumerate(ids)], [], **parameters)
         payload = data.pop("map_payload")
         return {"coverage": data, "map_payload": payload}
     monkeypatch.setattr(service, "generate_node_coverage", coverage)
@@ -44,6 +44,7 @@ def test_thresholds_are_exclusive_with_no_rounding_and_no_closure(inputs):
     inputs.assert_called_once_with([f"J{i}" for i in range(8)], START, START, "node", "pressure", db_name="own")
     assert len(result["map_payload"]["features"]) == 3
     assert all(feature["properties"]["model_estimate"] for feature in result["map_payload"]["features"])
+    assert result["coverage"]["parameters"]["exclusive_groups"] is True
 
 
 @pytest.mark.parametrize("unit,factor", [("kPa", 1 / 9.80665), ("PSI", 6894.757293168 / 9806.65),

tests/unit/test_node_coverage.py

@@ -3,11 +3,12 @@ from datetime import datetime, timezone
 import pytest
 from fastapi.encoders import jsonable_encoder
 from pydantic import ValidationError
-from shapely.geometry import Point, shape
+from shapely.geometry import Point, box, mapping, shape
+from shapely.ops import unary_union
 
 from app.domain.schemas.operational_data import TaskDataSnapshot
 from app.mcp.analysis_models import NodeCoverageInput
-from app.services.node_coverage import build_node_coverage
+from app.services.node_coverage import build_node_coverage, exclusive_coverage_features
 from app.services.operational_presentation import resolve_overlay
 
 
@@ -32,6 +33,44 @@ def test_coverage_can_compare_arbitrary_groups_without_assigning_isolation_seman
     assert coverage["parameters"]["split_by_connectivity"] is False
 
 
+def test_exclusive_envelopes_keep_union_and_counts_with_higher_groups_taking_precedence():
+    originals = [box(0, 0, 4, 4), box(2, 0, 6, 4), box(3, 1, 5, 3)]
+    features = [{"type": "Feature", "id": str(i), "properties": {
+        "coverage_group": str(i), "node_count": 10 + i, "mapped_node_count": 10 + i,
+    }, "geometry": mapping(geometry)} for i, geometry in enumerate(originals)]
+    result = exclusive_coverage_features(features)
+    visible = [shape(feature["geometry"]) for feature in result]
+    assert len(result) == 3
+    assert all(a.intersection(b).area == 0 for i, a in enumerate(visible) for b in visible[i + 1:])
+    assert unary_union(visible).equals(unary_union(originals))
+    assert visible[-1].equals(originals[-1])
+    assert [feature["properties"]["node_count"] for feature in result] == [10, 11, 12]
+    assert [feature["properties"]["mapped_node_count"] for feature in result] == [10, 11, 12]
+    assert all(feature["properties"]["overlap_policy"] == "higher_priority_group_first" for feature in result)
+    assert shape(features[0]["geometry"]).equals(originals[0])  # Never rewrite saved inputs.
+
+
+def test_covered_grade_keeps_node_metadata_without_invalid_empty_geometry():
+    positions = [position("A", 117.24), position("B", 117.24)]
+    groups = [group(["A"], "low"), group(["B"], "high")]
+    ordinary = build_node_coverage(groups, positions, [])
+    assert len(ordinary["map_payload"]["features"]) == 2
+    exclusive = build_node_coverage(groups, positions, [], exclusive_groups=True)
+    assert [feature["id"] for feature in exclusive["map_payload"]["features"]] == ["high"]
+    assert exclusive["groups"]["low"]["node_ids"] == ["A"]
+    assert exclusive["groups"]["low"]["mapped_node_count"] == 1
+    assert exclusive["groups"]["low"]["display_area_available"] is False
+    assert exclusive["groups"]["high"]["display_area_available"] is True
+    summaries = exclusive["map_payload"]["features"][0]["properties"]["coverage_group_summary"]
+    assert summaries["low"]["node_count"] == 1
+    assert summaries["low"]["mapped_node_count"] == 1
+    assert summaries["low"]["display_area_available"] is False
+    # A second application must not drop the summary for the clipped group.
+    repeated = exclusive_coverage_features(exclusive["map_payload"]["features"])
+    assert repeated[0]["properties"]["coverage_group_summary"] == summaries
+    assert exclusive["status"] == "ready"
+
+
 def test_connectivity_option_does_not_bridge_disconnected_groups():
     positions = [position("A", 117.24), position("B", 117.241),
                  position("C", 117.26), position("D", 117.261)]