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

feat(sensor): 同步监测点优化到客户版

Browse Source
This commit is contained in:
jiang
2026-08-04 14:55:37 +08:00
parent 4ced14ee04
commit 41bf5e9d92
10 changed files with 1473 additions and 669 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/algorithms/sensor/sensitivity.py
+879 -664
View File
File diff suppressed because it is too large Load Diff
app/api/v1/endpoints/sensor_placement.py
+22 -1
View File
@@ -2,7 +2,7 @@ import logging
from typing import Any
from urllib.parse import quote
from fastapi import APIRouter, Depends, HTTPException, Query, status
from fastapi import APIRouter, Depends, HTTPException, Path, Query, status
from fastapi.responses import StreamingResponse
from starlette.concurrency import run_in_threadpool
@@ -13,6 +13,7 @@ from app.algorithms.sensor import (
from app.auth.metadata_dependencies import get_current_metadata_user
from app.auth.project_dependencies import ProjectContext, get_project_context
from app.domain.schemas.sensor_placement import (
SensorPointResponse,
SensorPlacementExportRequest,
SensorPlacementOptimizeRequest,
SensorPlacementSchemeResponse,
@@ -24,6 +25,7 @@ from app.services.sensor_placement import (
SensorPlacementValidationError,
build_sensor_placement_workbook,
can_edit_sensor_placement,
get_sensor_placement_candidate,
get_sensor_placement_scheme,
update_sensor_placement_scheme,
)
@@ -94,6 +96,25 @@ def _get_scheme_response(
raise _service_http_error(exc) from exc
@router.get(
"/sensor-placement-candidates/{node_id}",
response_model=SensorPointResponse,
summary="获取监测点候选节点详情",
)
async def get_sensor_placement_candidate_detail(
node_id: str = Path(..., min_length=1, max_length=32),
project_context: ProjectContext = Depends(get_project_context),
) -> dict[str, Any]:
try:
return await run_in_threadpool(
get_sensor_placement_candidate,
project_context.project_code,
node_id,
)
except SensorPlacementValidationError as exc:
raise _service_http_error(exc) from exc
@router.post(
"/sensor-placement-optimization-runs",
response_model=SensorPlacementSchemeResponse,
app/domain/schemas/sensor_placement.py
+4
View File
@@ -67,6 +67,10 @@ class SensorPlacementExportRequest(BaseModel):
class SensorPointResponse(BaseModel):
node_id: str
max_pipe_diameter: float | None = Field(
...,
description="节点关联管道的最大管径，单位：毫米",
)
project_x: float
project_y: float
map_x: float
app/native/wndb/s42_sensor_placement.py
+16 -1
View File
@@ -66,8 +66,22 @@ def get_sensor_placement_nodes(
with conn.cursor(row_factory=dict_row) as cur:
cur.execute(
"""
WITH incident_pipe_diameters AS (
SELECT node_id, MAX(diameter) AS max_pipe_diameter
FROM (
SELECT node1 AS node_id, diameter
FROM pipes
WHERE node1 = ANY(%s)
UNION ALL
SELECT node2 AS node_id, diameter
FROM pipes
WHERE node2 = ANY(%s)
) AS incident_pipes
GROUP BY node_id
)
SELECT DISTINCT ON (gj.id)
gj.id AS node_id,
ipd.max_pipe_diameter,
gj.elevation,
ST_X(c.coord) AS project_x,
ST_Y(c.coord) AS project_y,
@@ -75,10 +89,11 @@ def get_sensor_placement_nodes(
ST_Y(gj.geom) AS map_y
FROM geo_junctions_mat AS gj
JOIN coordinates AS c ON c.node = gj.id
LEFT JOIN incident_pipe_diameters AS ipd ON ipd.node_id = gj.id
WHERE gj.id = ANY(%s)
ORDER BY gj.id
""",
(node_ids,),
(node_ids, node_ids, node_ids),
)
return list(cur.fetchall())
app/services/sensor_placement.py
+14
View File
@@ -80,6 +80,11 @@ def _sensor_points(
points.append(
{
"node_id": node_id,
"max_pipe_diameter": (
float(node["max_pipe_diameter"])
if node["max_pipe_diameter"] is not None
else None
),
"project_x": project_x,
"project_y": project_y,
"map_x": map_x,
@@ -92,6 +97,15 @@ def _sensor_points(
return points
def get_sensor_placement_candidate(
network: str,
node_id: str,
) -> dict[str, Any]:
"""Return the authoritative editable point data for one junction."""
return _sensor_points(network, _normalize_locations([node_id]))[0]
def validate_sensor_placement_nodes(
network: str,
sensor_location: list[str],
contracts/manifest.json
+1 -1
View File
@@ -3,7 +3,7 @@
"contracts": {
"server": {
"file": "server-v1.openapi.json",
"sha256": "9cd5b962e9556ec227c52d0dc7d4ef4af562dcea86e16877c923c37de0f4f704"
"sha256": "b003a57c9b8c1a041b644363ef58afb8bfcede1fe076ce48a190d2a1ac915e3f"
}
}
}
contracts/server-v1.openapi.json
+121
View File
@@ -2598,6 +2598,18 @@
"title": "Map Y",
"type": "number"
},
"max_pipe_diameter": {
"anyOf": [
{
"type": "number"
},
{
"type": "null"
}
],
"description": "节点关联管道的最大管径，单位：毫米",
"title": "Max Pipe Diameter"
},
"node_id": {
"title": "Node Id",
"type": "string"
@@ -2613,6 +2625,7 @@
},
"required": [
"node_id",
"max_pipe_diameter",
"project_x",
"project_y",
"map_x",
@@ -35546,6 +35559,114 @@
]
}
},
"/api/v1/sensor-placement-candidates/{node_id}": {
"get": {
"operationId": "get_sensor_placement_candidates_node_id",
"parameters": [
{
"in": "path",
"name": "node_id",
"required": true,
"schema": {
"maxLength": 32,
"minLength": 1,
"title": "Node Id",
"type": "string"
}
},
{
"in": "header",
"name": "X-Project-Id",
"required": true,
"schema": {
"title": "X-Project-Id",
"type": "string"
}
}
],
"responses": {
"200": {
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/SensorPointResponse"
}
}
},
"description": "Successful Response"
},
"401": {
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/ProblemDetails"
}
}
},
"description": "Authentication required"
},
"403": {
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/ProblemDetails"
}
}
},
"description": "Insufficient permission"
},
"404": {
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/ProblemDetails"
}
}
},
"description": "Resource not found"
},
"409": {
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/ProblemDetails"
}
}
},
"description": "Resource conflict"
},
"422": {
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/ProblemDetails"
}
}
},
"description": "Validation error"
},
"503": {
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/ProblemDetails"
}
}
},
"description": "Dependency unavailable"
}
},
"security": [
{
"OAuth2PasswordBearer": []
}
],
"summary": "获取监测点候选节点详情",
"tags": [
"Sensor Placement"
]
}
},
"/api/v1/sensor-placement-optimization-runs": {
"post": {
"operationId": "post_sensor_placement_optimization_runs",
tests/api/test_sensor_placement_endpoints.py
+17
View File
@@ -32,6 +32,7 @@ def _scheme(**overrides):
"sensor_points": [
{
"node_id": "J1",
"max_pipe_diameter": 400.0,
"project_x": 13500000.0,
"project_y": 3600000.0,
"map_x": 13500000.0,
@@ -42,6 +43,7 @@ def _scheme(**overrides):
},
{
"node_id": "J2",
"max_pipe_diameter": 300.0,
"project_x": 13500100.0,
"project_y": 3600100.0,
"map_x": 13500100.0,
@@ -114,6 +116,9 @@ def _load_module(monkeypatch):
"get_sensor_placement_scheme": lambda network, scheme_id: _scheme(
id=scheme_id
),
"get_sensor_placement_candidate": (
lambda network, node_id: _scheme()["sensor_points"][0]
),
"update_sensor_placement_scheme": (
lambda network, scheme_id, **kwargs: _scheme(
id=scheme_id,
@@ -170,6 +175,18 @@ def test_optimize_returns_created_scheme(monkeypatch):
assert captured["username"] == "alice"
def test_get_candidate_returns_maximum_incident_pipe_diameter(monkeypatch):
module = _load_module(monkeypatch)
response = _client(module).get(
"/api/v1/sensor-placement-candidates/J1",
)
assert response.status_code == 200
assert response.json()["node_id"] == "J1"
assert response.json()["max_pipe_diameter"] == 400.0
def test_optimize_rejects_unsupported_sensor_type(monkeypatch):
module = _load_module(monkeypatch)
response = _client(module).post(
tests/unit/test_sensor_placement_service.py
+8 -2
View File
@@ -29,6 +29,7 @@ def test_build_workbook_contains_engineering_columns(monkeypatch):
lambda network, locations: [
{
"node_id": "J1",
"max_pipe_diameter": 400.0,
"project_x": 13500000.0,
"project_y": 3600000.0,
"map_x": 13500000.0,
@@ -75,7 +76,7 @@ def test_build_workbook_contains_engineering_columns(monkeypatch):
assert workbook["方案信息"]["B8"].value == "未保存草稿"
def test_sensor_points_keep_engineering_coordinates_and_transform_map_coordinates(
def test_candidate_keeps_engineering_coordinates_and_transforms_map_coordinates(
monkeypatch,
):
monkeypatch.setattr(
@@ -84,6 +85,7 @@ def test_sensor_points_keep_engineering_coordinates_and_transform_map_coordinate
lambda network, node_ids: [
{
"node_id": "J1",
"max_pipe_diameter": 400.0,
"project_x": 3038.94,
"project_y": -34446.59,
"map_x": 13525191.530279,
@@ -93,10 +95,11 @@ def test_sensor_points_keep_engineering_coordinates_and_transform_map_coordinate
],
)
point = sensor_placement._sensor_points("tjwater", ["J1"])[0]
point = sensor_placement.get_sensor_placement_candidate("tjwater", "J1")
assert point["project_x"] == 3038.94
assert point["project_y"] == -34446.59
assert point["max_pipe_diameter"] == 400.0
assert point["longitude"] == pytest.approx(121.498863, abs=1e-6)
assert point["latitude"] == pytest.approx(30.924784, abs=1e-6)
@@ -133,6 +136,8 @@ def test_sensor_nodes_use_materialized_web_mercator_geometry(monkeypatch):
assert "ST_Y(c.coord)" in query
assert "ST_X(gj.geom)" in query
assert "ST_Y(gj.geom)" in query
assert "MAX(diameter) AS max_pipe_diameter" in query
assert cursor.execute.call_args.args[1] == (["J1"], ["J1"], ["J1"])
def test_workbook_escapes_formula_in_scheme_metadata(monkeypatch):
@@ -142,6 +147,7 @@ def test_workbook_escapes_formula_in_scheme_metadata(monkeypatch):
lambda network, locations: [
{
"node_id": "J1",
"max_pipe_diameter": 400.0,
"project_x": 3038.94,
"project_y": -34446.59,
"map_x": 13525191.53,
tests/unit/test_sensor_sensitivity.py
+391
View File
@@ -0,0 +1,391 @@
from pathlib import Path
import numpy as np
import pytest
import wntr
from scipy.sparse import csr_matrix, isspmatrix_csr
from scipy.sparse.csgraph import dijkstra
from app.algorithms.sensor import sensitivity
def _build_test_network() -> wntr.network.WaterNetworkModel:
wn = wntr.network.WaterNetworkModel()
wn.options.time.duration = 0
wn.add_reservoir("R1", base_head=100.0, coordinates=(-1.0, 0.0))
wn.add_junction("J0", elevation=5.0, coordinates=(0.0, 0.0))
for index in range(1, 13):
wn.add_junction(
f"J{index}",
base_demand=0.001 + index * 0.00001,
elevation=5.0 + index * 0.05,
coordinates=(float(index % 4), float(index // 4)),
)
wn.add_pipe("P0", "R1", "J0", length=100.0, diameter=0.4, roughness=110)
wn.add_pipe("P1", "J0", "J1", length=100.0, diameter=0.4, roughness=110)
for index in range(1, 11):
wn.add_pipe(
f"P{index + 1}",
f"J{index}",
f"J{index + 1}",
length=80.0 + index,
diameter=0.3,
roughness=105,
)
# J12 is connected only through a small pipe, while J11 also touches P11.
wn.add_pipe("P12", "J11", "J12", length=90.0, diameter=0.1, roughness=105)
wn.add_pipe("PX1", "J2", "J6", length=120.0, diameter=0.3, roughness=105)
wn.add_pipe("PX2", "J5", "J9", length=120.0, diameter=0.3, roughness=105)
return wn
def _prepared_selection_network(
coordinates: np.ndarray,
edges: list[tuple[int, int, float]],
) -> sensitivity._PreparedNetwork:
node_count = len(coordinates)
rows: list[int] = []
columns: list[int] = []
weights: list[float] = []
for start, end, weight in edges:
rows.extend((start, end))
columns.extend((end, start))
weights.extend((weight, weight))
coverage_graph = csr_matrix(
(weights, (rows, columns)),
shape=(node_count, node_count),
)
return sensitivity._PreparedNetwork(
node_names=tuple(f"N{index:04d}" for index in range(node_count)),
full_node_indices=np.arange(node_count, dtype=np.int64),
candidate_indices=np.arange(node_count, dtype=np.int64),
coordinates=np.asarray(coordinates, dtype=np.float64),
incidence=csr_matrix((node_count, 1), dtype=np.float64),
conductance=np.ones(1, dtype=np.float64),
roughness_response=np.ones(1, dtype=np.float64),
distance_graph=csr_matrix((node_count, node_count), dtype=np.float64),
coverage_graph=coverage_graph,
)
def test_algorithm_is_deterministic_and_runs_epanet_once(monkeypatch, tmp_path):
wn = _build_test_network()
original_run_sim = wntr.sim.EpanetSimulator.run_sim
prefixes: list[str] = []
def counted_run_sim(simulator, *args, **kwargs):
prefixes.append(str(kwargs["file_prefix"]))
return original_run_sim(simulator, *args, **kwargs)
monkeypatch.setattr(wntr.sim.EpanetSimulator, "run_sim", counted_run_sim)
monkeypatch.chdir(tmp_path)
first = sensitivity.optimize_sensor_placement(wn, sensor_num=4, min_diameter=0)
second = sensitivity.optimize_sensor_placement(wn, sensor_num=4, min_diameter=0)
assert first == second
assert len(first) == len(set(first)) == 4
assert len(prefixes) == 2
assert all(not Path(prefix).parent.exists() for prefix in prefixes)
assert not list(tmp_path.glob("temp.*"))
def test_hydraulic_simulation_keeps_only_initial_state_and_restores_duration():
wn = _build_test_network()
wn.options.time.duration = 24 * 60 * 60
results = sensitivity._run_hydraulic_simulation(wn)
assert len(results.node["head"].index) == 1
assert wn.options.time.duration == 24 * 60 * 60
def test_preparation_keeps_network_matrices_sparse():
wn = _build_test_network()
results = sensitivity._run_hydraulic_simulation(wn)
prepared = sensitivity._prepare_network(wn, results, min_diameter=0)
assert isspmatrix_csr(prepared.incidence)
assert isspmatrix_csr(prepared.distance_graph)
assert isspmatrix_csr(prepared.coverage_graph)
assert prepared.incidence.nnz <= 2 * prepared.incidence.shape[1]
assert prepared.distance_graph.nnz <= wn.num_pipes
assert prepared.coverage_graph.nnz <= 2 * wn.num_links
assert (prepared.coverage_graph != prepared.coverage_graph.T).nnz == 0
dense_incidence_bytes = int(np.prod(prepared.incidence.shape)) * 8
sparse_payload_bytes = (
prepared.incidence.data.nbytes
+ prepared.incidence.indices.nbytes
+ prepared.incidence.indptr.nbytes
)
assert sparse_payload_bytes < dense_incidence_bytes
def test_minimum_diameter_filters_installation_candidates_in_millimetres():
wn = _build_test_network()
results = sensitivity._run_hydraulic_simulation(wn)
prepared = sensitivity._prepare_network(wn, results, min_diameter=300)
candidate_names = {
prepared.node_names[index] for index in prepared.candidate_indices
}
assert "J12" not in candidate_names
assert "J11" in candidate_names
selected = sensitivity.optimize_sensor_placement(
wn,
sensor_num=4,
min_diameter=300,
)
assert set(selected) <= candidate_names
with pytest.raises(ValueError, match="候选节点少于"):
sensitivity.optimize_sensor_placement(
wn,
sensor_num=len(candidate_names) + 1,
min_diameter=300,
)
def test_sparse_estimate_preserves_dense_reference_placement_quality():
wn = _build_test_network()
results = sensitivity._run_hydraulic_simulation(wn)
prepared = sensitivity._prepare_network(wn, results, min_diameter=0)
approximate_log_sensitivity = sensitivity._estimate_log_pressure_sensitivity(
prepared
)
approximate_distance = sensitivity._estimate_hydraulic_distance_sums(prepared)
approximate_selected = sensitivity._select_sensor_nodes(
prepared,
approximate_log_sensitivity,
approximate_distance,
sensor_num=4,
)
incidence = prepared.incidence.toarray()
laplacian = (
prepared.incidence.multiply(prepared.conductance)
@ prepared.incidence.T
).toarray()
diagonal_scale = float(np.max(np.abs(np.diag(laplacian))))
laplacian += np.eye(laplacian.shape[0]) * (
diagonal_scale * np.sqrt(np.finfo(np.float64).eps)
)
response = np.linalg.solve(
laplacian,
incidence * prepared.roughness_response,
)
exact_sensitivity = np.abs(response).sum(axis=1)
exact_distances = dijkstra(
prepared.distance_graph.transpose().tocsr(),
directed=True,
indices=prepared.full_node_indices,
return_predecessors=False,
)[:, prepared.full_node_indices]
exact_distances[~np.isfinite(exact_distances)] = 0.0
exact_distance = exact_distances.sum(axis=0)
exact_selected = sensitivity._select_sensor_nodes(
prepared,
np.log(np.maximum(exact_sensitivity, np.finfo(np.float64).tiny)),
exact_distance,
sensor_num=4,
)
exact_score = exact_sensitivity * exact_distance
node_index = {
node_name: index for index, node_name in enumerate(prepared.node_names)
}
approximate_objective = sum(
exact_score[node_index[node_name]] for node_name in approximate_selected
)
exact_objective = sum(
exact_score[node_index[node_name]] for node_name in exact_selected
)
assert approximate_objective / exact_objective >= 0.95
def test_mixed_coverage_avoids_candidate_density_bias():
dense_west = np.linspace(0.0, 2.0, 200)
sparse_east = np.linspace(3.0, 10.0, 20)
x_coordinates = np.concatenate((dense_west, sparse_east))
coordinates = np.column_stack(
(x_coordinates, np.zeros(len(x_coordinates), dtype=np.float64))
)
ordered = np.argsort(x_coordinates)
edges = [
(
int(start),
int(end),
float(x_coordinates[end] - x_coordinates[start]),
)
for start, end in zip(ordered[:-1], ordered[1:])
]
prepared = _prepared_selection_network(coordinates, edges)
log_scores = np.linspace(4.0, 0.0, len(coordinates))
distance_sums = np.ones(len(coordinates), dtype=np.float64)
selected = sensitivity._select_sensor_nodes(
prepared,
log_scores,
distance_sums,
sensor_num=6,
)
name_to_position = {
name: position for position, name in enumerate(prepared.node_names)
}
selected_positions = [name_to_position[name] for name in selected]
new_metrics = sensitivity._geographic_coverage_metrics(
coordinates,
selected_positions,
)
legacy_labels, _centers = sensitivity._cluster_labels(
coordinates,
6,
random_seed=sensitivity._RANDOM_SEED + 2,
)
legacy_positions: list[int] = []
represented: set[int] = set()
for position in np.argsort(-log_scores):
label = int(legacy_labels[position])
if label in represented:
continue
represented.add(label)
legacy_positions.append(int(position))
legacy_metrics = sensitivity._geographic_coverage_metrics(
coordinates,
legacy_positions,
)
assert new_metrics[0] <= legacy_metrics[0] * 0.6
assert new_metrics[2] >= legacy_metrics[2] * 1.5
assert max(x_coordinates[selected_positions]) >= 9.0
def test_disconnected_components_each_receive_a_sensor_when_slots_allow():
coordinates = np.asarray(
[
(0.0, 0.0),
(1.0, 0.0),
(0.0, 0.01),
(1.0, 0.01),
]
)
prepared = _prepared_selection_network(
coordinates,
[(0, 1, 1.0), (2, 3, 1.0)],
)
selected = sensitivity._select_sensor_nodes(
prepared,
np.asarray([10.0, 9.0, 8.0, 7.0]),
np.ones(4),
sensor_num=2,
)
assert len(set(selected) & {"N0000", "N0001"}) == 1
assert len(set(selected) & {"N0002", "N0003"}) == 1
def test_overlapping_components_respect_global_geographic_spacing():
coordinates = np.asarray(
[
(0.0, 0.0),
(10.0, 0.0),
(0.1, 0.0),
(10.1, 0.0),
]
)
prepared = _prepared_selection_network(
coordinates,
[(0, 1, 10.0), (2, 3, 10.0)],
)
selected = sensitivity._select_sensor_nodes(
prepared,
np.asarray([10.0, 1.0, 9.0, 0.0]),
np.ones(4),
sensor_num=2,
)
selected_positions = [prepared.node_names.index(name) for name in selected]
minimum_gap = sensitivity._geographic_coverage_metrics(
coordinates,
selected_positions,
)[2]
assert minimum_gap >= 0.9
def test_component_quota_prefers_longer_networks_when_slots_are_limited():
coordinates = np.asarray(
[
(0.0, 0.0),
(10.0, 0.0),
(20.0, 0.0),
(25.0, 0.0),
(30.0, 0.0),
(31.0, 0.0),
]
)
prepared = _prepared_selection_network(
coordinates,
[(0, 1, 10.0), (2, 3, 5.0), (4, 5, 1.0)],
)
selected = sensitivity._select_sensor_nodes(
prepared,
np.asarray([1.0, 1.0, 2.0, 2.0, 100.0, 100.0]),
np.ones(6),
sensor_num=2,
)
assert set(selected) <= {"N0000", "N0001", "N0002", "N0003"}
assert len(set(selected) & {"N0000", "N0001"}) == 1
assert len(set(selected) & {"N0002", "N0003"}) == 1
def test_duplicate_coordinates_use_topology_and_return_exact_count():
coordinates = np.zeros((6, 2), dtype=np.float64)
prepared = _prepared_selection_network(
coordinates,
[(index, index + 1, 1.0) for index in range(5)],
)
log_scores = np.linspace(6.0, 1.0, 6)
first = sensitivity._select_sensor_nodes(
prepared,
log_scores,
np.ones(6),
sensor_num=4,
)
second = sensitivity._select_sensor_nodes(
prepared,
log_scores,
np.ones(6),
sensor_num=4,
)
assert first == second
assert len(first) == len(set(first)) == 4
@pytest.mark.parametrize(
("sensor_num", "min_diameter", "message"),
[
(0, 0, "监测点数量必须大于 0"),
(1, -1, "最小管径不能小于 0"),
],
)
def test_algorithm_rejects_invalid_parameters(sensor_num, min_diameter, message):
with pytest.raises(ValueError, match=message):
sensitivity.optimize_sensor_placement(
_build_test_network(),
sensor_num=sensor_num,
min_diameter=min_diameter,
)