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

refactor(sensor): 优化灵敏度监测点布置算法

Browse Source
This commit is contained in:
jiang
2026-08-03 18:34:13 +08:00
parent 87d922ea61
commit b0e9d480ef
3 changed files with 1540 additions and 664 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
docs/sensor-sensitivity-optimization.md
+270
View File
@@ -0,0 +1,270 @@
# 监测点灵敏度算法优化说明
本文记录压力监测点布置算法的改造方案、数学含义、复杂度变化和本地验证结果。实现位于 `app/algorithms/sensor/sensitivity.py`，对外兼容入口仍为 `get_ID(name, sensor_num, min_diameter)`。
## 改造目标
原算法根据节点压力对管道粗糙度变化的灵敏度，以及节点到全网的水力距离，给每个候选节点评分。空间聚类用于限制监测点过度集中。业务评分可以写成：
$$
score_i = sensitivity_i \times \sum_j distance(i, j)
$$
本次改造保留这套评分语义和空间覆盖原则，主要处理以下问题：
- 稠密的节点与节点、节点与管道、管道与管道矩阵占用大量内存。
- 显式计算行列式、逆矩阵和伪逆，计算量随节点数快速增长。
- 从每个节点执行最短路径并保存全节点距离矩阵，时间和内存均为平方级。
- 一次任务重复运行 EPANET，并保留算法没有使用的扩展时序结果。
- 聚类没有固定随机状态，相同输入可能返回不同结果。
- KMeans 按候选节点密度最小化平方距离，密集管网会系统性获得更多监测点，不能保证地图或管网路径覆盖均匀。
- WNTR 管径使用米，接口参数使用毫米，旧候选过滤没有统一单位。
## 稀疏表示替代稠密矩阵
管网的节点通常只连接少量管道。新实现使用 SciPy CSR 或 CSC 矩阵保存节点与管道关联矩阵、水力有向图和水力雅可比矩阵，只记录真实存在的连接。
准备阶段生成一个只读的 `_PreparedNetwork`，其中包括：
- 参与分析的节点及其在完整模型中的索引。
- 符合安装条件的候选节点索引。
- 归一化前的二维坐标。
- 稀疏节点与管道关联矩阵。
- 管道导通系数和粗糙度响应系数。
- 按实际流向构建的稀疏水力距离图。
- 按初始运行状态构建的无向物理覆盖图。
水库、水箱、泵和阀门的边界节点继续沿用原算法的排除规则。与水源直接相连的管道不参加扰动计算。平行有向边只保留权重最小的边，以符合最短路径语义。
主体存储由平方级降到接近 `O(n + m)`。稀疏 LU 分解可能产生填充，其内存仍取决于管网拓扑，但实现不会再主动分配完整的 `n × n` 或 `n × m` 数组。
## 用 Cauchy 投影估计压力灵敏度
旧算法先显式计算压力对管道粗糙度的响应矩阵：
$$
J = X^{-1} A S
$$
其中，`X` 是节点水力雅可比矩阵，`A` 是节点与管道关联矩阵，`S` 是管道粗糙度响应矩阵。节点灵敏度是 `J` 对应行的 L1 范数。完整的 `J` 为 `节点数 × 管道数`，大模型无法保存。
新算法使用 Cauchy 分布的 1-stable 特性估计每一行的 L1 范数。设 `R` 为 Cauchy 随机投影矩阵，只需求解：
$$
Y = X^{-1} A S R
$$
对节点 `i` 而言，`Y` 中每个投影值服从以 `||J_i||_1` 为尺度的 Cauchy 分布。实现使用投影绝对值的对数几何均值估计 `log(||J_i||_1)`，不保存完整灵敏度矩阵。
当前固定参数如下：
| 参数 | 数值 | 用途 |
|---|---:|---|
| 随机种子 | 42 | 保证结果可复现 |
| Cauchy 投影数 | 256 | 控制灵敏度估计精度 |
| 投影批大小 | 16 | 限制投影中间矩阵内存 |
| 雅可比正则化 | `max(abs(diag(X))) × sqrt(eps)` | 改善接近奇异矩阵的稳定性 |
| 稀疏排序 | `MMD_AT_PLUS_A` | 减少 LU 分解填充 |
水力雅可比矩阵只进行一次稀疏 LU 分解，256 次投影复用该分解结果。批处理期间最多保留 16 列投影数据。
## 用空间代表点估计水力距离总和
旧算法从每个节点执行一次最短路径，并保存 `n × n` 的水力距离矩阵。新算法根据节点坐标构建最多 256 个空间代表点：
1. 使用固定随机状态的 `MiniBatchKMeans` 对节点坐标分组。
2. 每组选择最靠近聚类中心的节点作为代表点。
3. 代表点权重等于该组包含的节点数。
4. 在反向水力图上从代表点执行有向 Dijkstra，得到原图中各节点到代表点的距离。
节点 `i` 的全网水力距离总和估计为：
$$
distance\_sum_i \approx \sum_{l \in landmarks} weight_l \times distance(i, l)
$$
Dijkstra 每批处理 16 个代表点，内存中只保留当前距离块。不可达距离按原算法约定计为 0，避免断开分支得到无穷评分。
## 评分约束下的混合覆盖选点
最终排序使用对数形式：
$$
log\_score_i = log\_sensitivity_i + \log(distance\_sum_i)
$$
对数是单调函数，因此该排序等价于比较 `sensitivity_i × distance_sum_i`，同时可以避免大数乘法溢出。
最终选点不再对全部候选节点执行 KMeans。KMeans 的目标函数按候选节点数计权，节点密集区域即使地理范围较小，也会获得更多聚类中心。本次改为评分约束下的最远空白区优先策略。
候选坐标使用同一个尺度因子归一化，保留管网原始长宽比。初始状态下开启的管道按物理长度构建无向覆盖图，开启的泵和阀门作为点连接；关闭连接会形成独立区域。监测点名额首先按各连通区域的有效管道长度分配：名额足够时每个区域至少一个，其余名额按最大余数法分配，并受该区域候选数量限制。
选点按区域名额从多到少处理，使主干管网先形成覆盖骨架。第一个区域的首点取综合评分最高的候选；后续区域的首点也必须考虑此前所有区域的已选点，先进入全局地图空白度前 70% 的候选集，再比较综合评分。这一约束避免两个拓扑断开但地图上重叠或相邻的区域各自选择一个近邻高分点。
设全局已选集合为 `S`，其余候选的最近地图距离和本连通区域内的最近管网路径距离分别为：
$$
g_i = \min_{s \in S} ||x_i-x_s||, \qquad
t_i = \min_{s \in S} shortest\_path(i, s)
$$
两类距离分别除以当前未选候选中的最大值，混合空白度取两者较大值：
$$
coverage_i = \max\left(\frac{g_i}{\max g}, \frac{t_i}{\max t}\right)
$$
每轮只保留空白度达到当前最大值 70% 的候选，再从中选择综合评分最高的节点。地图距离始终相对全部已选区域更新，管网路径距离在当前连通区域内更新。这样地图或管网路径中任一维度仍有明显空白时，该区域都不会被高密度节点误判为已经覆盖。综合评分相同时按节点 ID 排序，结果数量、唯一性和可复现性保持不变。
实现只维护候选节点的最近距离数组。每新增一个监测点，执行一次单源 Dijkstra 并增量更新，不构造候选节点两两距离矩阵。
所有模型使用同一套算法和相同参数，不根据节点数量切换实现。
## EPANET 只计算初始状态
后续计算只读取水力结果的第一个时刻。新实现会临时将 `wn.options.time.duration` 设置为 0，只运行一次 EPANET 初始状态模拟，并在结束或异常后恢复原始时长。
EPANET 中间文件写入独立的 `TemporaryDirectory`，任务结束后自动清理。并发请求不再共用工作目录下的 `temp.*` 文件。
旧调用链最多重复运行约四次 EPANET，新调用链只运行一次，也不会分配没有使用的完整时序结果。
## 最小管径规则
`min_diameter` 的接口单位是毫米，WNTR 中的管径单位是米。准备阶段使用以下换算：
$$
diameter\_mm = diameter\_m \times 1000
$$
节点连接的管道中，只要至少一根达到最小管径，该节点就可以作为安装候选。小管径管道仍参加全网水力和灵敏度计算，管径条件只限制监测点安装位置。
当候选节点少于请求的监测点数量时，算法会返回包含候选数量和请求数量的明确错误。
## 复杂度变化
下表中的 `n` 为参与分析的节点数，`m` 为参与分析的管道数，`k=256` 为灵敏度投影数，`l≤256` 为水力距离代表点数，`b=16` 为批大小。
| 环节 | 原实现 | 新实现 |
|---|---|---|
| 节点与管道关系 | 稠密 `n × m` | CSR 稀疏矩阵，约 `O(n + m)` |
| 节点关系与距离 | 多个稠密 `n × n` 矩阵 | 稀疏有向图和流式距离块 |
| 灵敏度 | 稠密行列式、逆矩阵或伪逆，时间接近 `O(n³)` | 一次稀疏 LU 分解和 `k` 次稀疏求解 |
| 灵敏度结果 | 保存完整 `n × m` 响应矩阵 | 保存 `n` 个对数灵敏度和当前投影批 |
| 水力距离 | 从全部节点执行最短路径并保存 `n × n` 结果 | 从至多 `l` 个代表点执行 Dijkstra，每批 `b` 个 |
| 最终选点 | 按节点密度分配的完整 KMeans | 连通区域配额和 70% 混合覆盖，线性距离数组 |
| 水力模拟 | 调用链中重复执行，并可能保存完整时序 | 一次初始状态模拟 |
稀疏 LU 的时间和内存不能简单视为线性，其填充程度受网络拓扑影响。当前压力测试覆盖到 20.3 万原始节点，不能据此保证任意更大或连接更稠密的模型都保持相同比例。
## 精度取舍
新实现不逐元素生成旧算法的完整精确矩阵，而是估计最终排序需要的两个统计量：
- Cauchy 投影估计压力响应矩阵每一行的 L1 范数。
- 加权空间代表点估计节点到全网的水力距离总和。
单元测试使用小型模型构造完整稠密参考结果，要求近似方案选点的精确参考目标值不低于精确方案的 95%。本地模型对比结果如下：
| 模型 | 近似选点的精确参考目标比 |
|---|---:|
| `fengxian.inp` | 98.80% |
| MD 模型 | 99.71% |
固定随机种子和固定采样数保证同一模型、监测点数量和最小管径得到相同结果。近似排序仍可能与完整稠密算法不同，特别是多个候选节点得分接近时。
混合覆盖会主动放弃部分集中在同一区域的高分节点。单点综合评分之和因此不是唯一质量指标，还需要同时检查未覆盖半径、最小点间距和入选节点的评分百分位。
## 资源压力测试记录
以下结果来自 2026-08-03 的本地验证。环境为 Linux、Python 3.12、Conda `server` 环境，主机物理内存约 30 GiB。测试参数统一为 20 个监测点、最小管径 0。
外部看门狗使用以下停止条件：
- 算法进程树 RSS 达到 7.5 GiB。
- 系统可用内存低于 6 GiB。
- 单模型运行超过 600 秒。
- 子进程虚拟地址空间硬限制为 8 GiB。
任一条件满足时，看门狗会终止整个进程组。三次压力测试均未触发停止条件。
| 模型 | 原始节点 | 实际分析节点 | 实际分析管道 | 算法流水线耗时 | 峰值 RSS |
|---|---:|---:|---:|---:|---:|
| `temp/leakage/temp_3698123.inp` | 31,143 | 28,723 | 29,974 | 2.77 秒 | 467 MiB |
| `inp/jbh.inp` | 94,049 | 69,949 | 82,369 | 8.92 秒 | 980 MiB |
| `inp/Todo/v-16常熟模型.inp` | 203,569 | 145,948 | 176,512 | 20.00 秒 | 1.84 GiB |
实际分析节点少于原始节点，是因为水库、水箱、泵和阀门边界节点按算法规则排除。20.3 万节点模型原文件使用非标准的 `REPORTING TIMESTEP`，并且缺少 `[END]`。压力测试只在隔离临时副本中将其规范化，没有修改原文件。
20.3 万节点模型各阶段耗时如下：
| 阶段 | 耗时 |
|---|---:|
| 模型加载 | 6.63 秒 |
| EPANET 初始状态模拟 | 8.30 秒 |
| 稀疏数据准备 | 3.00 秒 |
| 压力灵敏度估计 | 0.76 秒 |
| 水力距离估计 | 0.94 秒 |
| 监测点选择 | 0.37 秒 |
### `tjwater` 分布优化前后对比
2026-08-03 使用 `db_inp/tjwater.db.inp`、20 个监测点、最小管径 0 进行同机对比。进程通过 systemd scope 限制在 2.5 GiB 内存，并设置 90 秒超时；两次运行均未触发保护。覆盖距离使用保持长宽比后的模型坐标，按管网最大轴跨度归一化。
| 指标 | KMeans 选点 | 70% 混合覆盖 | 变化 |
|---|---:|---:|---:|
| 实际分析/候选节点 | 87,877 | 87,877 | 不变 |
| 最大地图覆盖半径 | 0.176786 | 0.173941 | -1.61% |
| P95 地图覆盖半径 | 0.122093 | 0.090686 | -25.72% |
| 最小监测点间距 | 0.000559 | 0.040887 | 约 73.2 倍 |
| 综合评分中位百分位 | 98.76% | 95.41% | -3.35 个百分点 |
| 端到端耗时 | 12.34 秒 | 11.71 秒 | -5.11% |
| 峰值 RSS | 913 MiB | 932 MiB | +19 MiB |
结果达到预定验收条件：P95 覆盖半径下降超过 15%，评分中位百分位下降少于 10 个百分点，运行时间没有增加，峰值内存增加远低于 256 MiB。
### 2026-08-03：跨连通区域共享全局间距
首次混合覆盖结果中，节点 `121302` 与 `110874` 分属两个不可达连通区域，但地图距离只有约 550.47 个模型单位。旧的分区独立首点规则分别选中了两个区域的最高分节点，形成视觉近邻。加入跨区域全局地图间距后，`121302` 被替换，最小归一化点间距由 0.013685 进一步提高到 0.040887。
本次调整保留连通区域名额，改变区域之间互不感知的选点方式。区域按名额从多到少处理，主干管网先形成覆盖骨架。第一个区域仍从综合评分最高的候选开始；后续区域选择首点时，先计算该区域所有候选到全局已选点的最近地图距离，只保留达到本区域最大空白距离 70% 的候选，再比较灵敏度综合评分。区域内部后续选点继续使用地图距离与管网路径距离的混合空白度。
这项约束只影响跨区域首点选择，不改变灵敏度估计、水力距离估计、区域名额、最小管径规则和公开 API。合成回归模型会构造两个地图上重叠但拓扑断开的区域，确保算法不会再次分别选择两个相邻高分点。
## 测试与回归验证
新增测试位于 `tests/unit/test_sensor_sensitivity.py`，覆盖以下行为：
- 相同输入返回相同节点，并且每次调用只运行一次 EPANET。
- EPANET 只保留初始状态，模型原始模拟时长能够恢复。
- 关联矩阵和距离图保持 CSR 稀疏格式。
- 最小管径按毫米过滤安装候选。
- 近似方案在稠密参考目标上的结果不低于 95%。
- 高密西部、低密东部的合成模型不再按候选节点密度分配名额。
- 地理位置相邻但拓扑断开的区域在名额允许时分别获得监测点。
- 地图上重叠的独立区域共享全局地理间距，不能各自选择相邻的首个高分点。
- 名额不足时优先覆盖有效管道长度更大的连通区域。
- 重复坐标依靠管网路径距离继续选点，并保持数量和确定性。
- 非法监测点数量和最小管径返回明确错误。
- 模拟结束后不残留共享 `temp.*` 文件。
回归命令：
```bash
conda run -n server python -m pytest \
tests/unit \
tests/auth \
tests/api/test_sensor_placement_endpoints.py \
tests/api/test_simulation_endpoints.py \
-q
```
2026-08-03 的执行结果为 `142 passed, 2 skipped, 7 warnings`。`git diff --check` 同时通过。
## 实现边界
- 256 次投影和 256 个代表点是当前质量与性能验证后的固定参数。调整参数需要重新运行稠密参考质量测试和大模型压力测试。
- 稀疏 LU 对高连接度或拓扑特殊的模型可能产生更多填充，应继续用进程级资源保护运行未知大模型。
- 算法使用单个初始水力状态。如果业务目标改为覆盖全天多个工况，需要先定义多工况评分和结果合并规则，不能直接恢复长时段模拟后沿用当前评分。
- 节点坐标用于代表点构建和地图覆盖，假定其能表达一致的平面相对距离；缺少有效二维坐标的模型会返回错误。
- 物理覆盖图使用初始水力状态。全天工况中频繁开闭的阀门或泵需要在多工况方案中重新定义连通区域合并规则。
- 当前压力测试验证到 203,569 个原始节点，没有验证 30 万节点模型。
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,
)