jiang
5966d039de
Reorganize algorithm packages by business responsibility, move orchestration into services, and keep database access behind pooled repositories. Harden analysis API validation, remove unsafe legacy simulation endpoints, and add regression and architecture boundary coverage. BREAKING CHANGE: legacy algorithm module paths and obsolete simulation endpoints are removed.
96 lines
2.7 KiB
Python
96 lines
2.7 KiB
Python
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from app.algorithms.pressure_sensor_placement.kmeans_placement import (
|
|
optimize_sensor_placement,
|
|
)
|
|
|
|
|
|
class FakeNetwork:
|
|
junction_name_list = ["J1", "J2", "J3"]
|
|
pipe_name_list = ["P1", "P2"]
|
|
|
|
def __init__(self):
|
|
self._nodes = {
|
|
"J1": SimpleNamespace(coordinates=(0.0, 0.0)),
|
|
"J2": SimpleNamespace(coordinates=(1.0, 0.0)),
|
|
"J3": SimpleNamespace(coordinates=(2.0, 0.0)),
|
|
}
|
|
self._links = {
|
|
"P1": SimpleNamespace(
|
|
diameter=0.4, start_node_name="J1", end_node_name="J2"
|
|
),
|
|
"P2": SimpleNamespace(
|
|
diameter=0.1, start_node_name="J2", end_node_name="J3"
|
|
),
|
|
}
|
|
|
|
def get_node(self, node_id):
|
|
return self._nodes[node_id]
|
|
|
|
def get_link(self, link_id):
|
|
return self._links[link_id]
|
|
|
|
|
|
class DuplicateNearestNodeNetwork:
|
|
junction_name_list = ["J1", "J2", "J3", "J4"]
|
|
pipe_name_list = ["P1", "P2", "P3"]
|
|
|
|
def __init__(self):
|
|
self._nodes = {
|
|
"J1": SimpleNamespace(
|
|
coordinates=(-0.2967811321994103, 0.4867937179527957)
|
|
),
|
|
"J2": SimpleNamespace(
|
|
coordinates=(1.8589649622870763, -1.0070372175769569)
|
|
),
|
|
"J3": SimpleNamespace(
|
|
coordinates=(-1.4882734126175363, -1.6220731365802619)
|
|
),
|
|
"J4": SimpleNamespace(
|
|
coordinates=(1.203441836600899, 2.2320704059758474)
|
|
),
|
|
}
|
|
self._links = {
|
|
"P1": SimpleNamespace(
|
|
diameter=0.4, start_node_name="J1", end_node_name="J2"
|
|
),
|
|
"P2": SimpleNamespace(
|
|
diameter=0.4, start_node_name="J2", end_node_name="J3"
|
|
),
|
|
"P3": SimpleNamespace(
|
|
diameter=0.4, start_node_name="J3", end_node_name="J4"
|
|
),
|
|
}
|
|
|
|
def get_node(self, node_id):
|
|
return self._nodes[node_id]
|
|
|
|
def get_link(self, link_id):
|
|
return self._links[link_id]
|
|
|
|
|
|
def test_kmeans_filters_candidates_by_minimum_pipe_diameter():
|
|
selected = optimize_sensor_placement(
|
|
FakeNetwork(), sensor_count=2, min_diameter_mm=300
|
|
)
|
|
|
|
assert set(selected) == {"J1", "J2"}
|
|
|
|
|
|
def test_kmeans_rejects_sensor_count_above_eligible_candidates():
|
|
with pytest.raises(ValueError, match="候选节点数量"):
|
|
optimize_sensor_placement(
|
|
FakeNetwork(), sensor_count=3, min_diameter_mm=300
|
|
)
|
|
|
|
|
|
def test_kmeans_returns_unique_monitoring_nodes():
|
|
selected = optimize_sensor_placement(
|
|
DuplicateNearestNodeNetwork(), sensor_count=2, min_diameter_mm=300
|
|
)
|
|
|
|
assert len(selected) == 2
|
|
assert len(set(selected)) == 2
|