简介：本资源面向高校学生与算法开发者，提供基于Python与NSGA2算法的供水管网水质监测点布局优化完整方案，适用于毕业设计、课程设计及项目开发场景。项目针对突发污染事件下的监测点选取问题，采用整数编码的NSGA2算法，构建最短监测时间与最大监测概率双目标函数，并借助基于EPANET的WNTR库完成水力水质模拟与数据处理，实现从污染注入实验到算法迭代的完整流程。压缩包共58个文件，约746KB，包含36个Python源码文件、7个XML配置、4个JSON数据、2个INP管网模型及CSV结果文件等，涵盖水质模拟、数据处理、非支配排序、拥挤度计算、交叉变异等核心模块。已有134人学习关注。读者可获得可直接运行的源码、项目文档与清晰的目录结构，便于理解帕累托解与进化算法在管网布局中的落地方式，并在此基础上延伸扩展。

1. 供水管网水质监测点布局：一份能跑通全流程的 NSGA2 实战源码

突发污染事件下，供水管网里到底该把有限的水质传感器布在哪几个节点，才能既快又准地抓到污染团？这是环境工程和市政管网方向绕不开的优化问题。手里这份基于 Python + NSGA2 的供水管网水质监测点布局源码，解决的正是这件事：用 WNTR 做水力水质模拟，把污染注入实验的结果喂给整数编码的非支配遗传算法，迭代出帕累托前沿上的监测点组合。它适合正在做毕业设计、课程设计，或者需要一套可复现管网优化流程的开发者。整套代码把模拟、数据处理、算法迭代串成了闭环，流程基本能走通，拿来当二次开发的底座比从零搭要省太多事。

2. 先搞懂双目标与整数编码：为什么监测点布局要用 NSGA2

2.1 监测时间与监测概率为什么天然打架

水质监测点布局的核心矛盾，用一句话说清楚：你想让传感器离污染源越近越好，这样响应时间短；但污染可能发生在管网里任何一个节点，你又希望少数几个传感器能覆盖尽可能多的污染场景。前者对应最短监测时间，后者对应最大监测概率，这两个目标在数学上是负相关的——把点布在管网中心区域，平均响应快，但边缘节点的污染事件可能漏检；把点分散到各个角落，覆盖率高，可每个点的响应时间又拉长了。

传统做法是加权求和，把两个目标揉成一个适应度函数，但权重怎么定全凭经验，改一次权重就要重跑一遍，而且得到的只是某一个折中解，看不到全局的取舍关系。NSGA2 的价值在于它一次运行就能给出一整条帕累托前沿，每个解都代表一种「响应时间」和「覆盖概率」的权衡方案，决策者可以根据实际预算和风险偏好从前沿上挑点。这就是为什么这类布局问题几乎成了多目标进化算法的经典练兵场。

2.2 整数编码：监测点选择不是连续变量

很多 NSGA2 的入门示例用的是实数编码，变量在某个区间内连续取值。但监测点布局是个组合优化问题——管网里有几百个节点，每个节点只有「选」和「不选」两种状态，你要从 N 个节点里挑出 K 个。如果硬用实数编码再四舍五入，交叉变异之后会产生大量重复或越界的无效解，算法效率会被拖垮。

这份源码用的是整数编码，染色体直接存节点编号，比如[12, 45, 78, 103, 156]就代表选了这五个节点作为监测点。交叉操作在父代之间交换部分节点编号，变异操作把某个位置的节点替换成管网里的另一个随机节点。这样做的好处是解空间和实际决策空间一一对应，不会产生无效解，收敛速度也更快。代价是需要额外处理重复节点的问题——两个父代可能都选了同一个节点，交叉后子代里会出现重复编号，得在解码阶段去重。

2.3 用 WNTR 替代 EPANET 链接库的实操理由

早期做管网模拟，大家习惯调 EPANET 的 C 库或者用 EPANET 的图形界面手动跑。但 EPANET 的 Python 接口一直不太顺手，要么用epanettools这种第三方封装，要么自己写 ctypes 调动态库，环境配置能折腾掉半天。USEPA 后来开源了 WNTR（Water Network Tool for Resilience），纯 Python 实现，直接pip install wntr就能用，读.inp文件、跑水力模拟、做水质注入实验都是一行代码的事。

这份源码里simulation目录下的WaterQualitySim.py和WaterQualitySimData.py就是基于 WNTR 写的。WaterQualitySim.py负责加载管网模型、设置污染注入节点和注入时间、运行水质模拟；WaterQualitySimData.py负责从模拟结果里提取每个监测点对每次污染事件的响应时间，整理成算法能吃的矩阵。这套流程比手动调 EPANET 稳定得多，也更容易在服务器上批量跑。

3. 把源码跑起来：环境配置与模拟数据生成

3.1 依赖安装与目录结构确认

拿到压缩包解压后，先别急着跑finalMain.py。第一步是把环境配好，WNTR 依赖 numpy、pandas、networkx 这些科学计算库，NSGA2 部分还用了 matplotlib 画帕累托前沿。建议用 conda 建一个干净的虚拟环境，避免和系统里的包版本打架。

# 创建虚拟环境，Python 版本建议 3.8 到 3.10 conda create -n nsga2_water python=3.9 conda activate nsga2_water # 安装核心依赖 pip install wntr numpy pandas matplotlib networkx scipy # 如果要用 DBSCAN 做节点聚类，还需要 scikit-learn pip install scikit-learn

装完之后检查一下目录结构。根目录下SensorPlacementInDistributionNetworks-master是主工程，simulation放水质模拟相关代码，nsga2放算法核心，nodeimportance放节点重要性计算，spacealgorithm放空间聚类相关脚本。example目录里有Net3.inp和ky8.inp两个管网模型文件，Net3.inp是 EPANET 自带的经典测试网络，适合先拿来验证流程。

提示：如果pip install wntr卡在编译环节，大概率是缺少 C 编译器。Windows 上装个 Visual Studio Build Tools，Linux 上apt install build-essential就能解决。

3.2 用 WNTR 跑一次污染注入模拟

在跑完整算法之前，先单独验证水质模拟模块能不能正常工作。打开simulation/WaterQualitySim.py，核心逻辑是加载.inp文件、设置污染源、调用 WNTR 的WaterQualitySimulator。下面这段代码是我从源码里抽出来的最小可运行示例，你可以直接复制到一个新脚本里测试。

import wntr # 加载管网模型，Net3.inp 在 example 目录下 wn = wntr.network.WaterNetworkModel('example/Net3.inp') # 创建水质模拟器 wqsim = wntr.sim.WNTRSimulator(wn) # 设置污染注入：在节点 10 注入，从第 0 秒开始持续 3600 秒 injection_node = '10' wn.get_node(injection_node).quality = 100.0 # 注入浓度 # 运行水力水质模拟，时长 24 小时 results = wqsim.run_sim() # 提取节点 15 的水质浓度序列 node_quality = results.node['quality']['15'] print(node_quality.head())

这段代码的逻辑很直白：WaterNetworkModel读模型，WNTRSimulator建模拟器，run_sim()跑完整时段。关键参数是注入节点的quality属性，它设的是污染物的初始浓度，源码里用的是 100.0，你可以根据实际场景调整。results.node['quality']是一个 DataFrame，行是时间步，列是节点名，每个单元格是对应时刻的浓度值。

跑通这一步之后，WaterQualitySimData.py里的EventDerivation.py会批量遍历所有可能的注入节点，对每个节点跑一次模拟，记录每个候选监测点的首次响应时间。这个过程比较耗时，Net3 网络有 92 个节点，每个节点跑一次 24 小时模拟，全部跑完大概要十几分钟。源码里final_json.json和waterQuality.json就是预先跑好的结果缓存，如果你只是想验证算法部分，可以直接用这两个文件跳过模拟。

3.3 从模拟结果到算法输入：数据格式转换

模拟跑完之后，WaterQualitySimData.py会把结果整理成一个二维矩阵，行是污染事件（注入节点），列是候选监测点，矩阵值是响应时间。如果某个监测点对某次污染事件始终没响应，就填一个很大的数（比如 99999）表示漏检。这个矩阵就是 NSGA2 的目标函数输入。

源码里final_json.json的结构大概是这样的：

{ "events": ["10", "15", "20", "25"], "sensors": ["12", "45", "78", "103"], "response_matrix": [ [120, 340, 99999, 560], [450, 99999, 230, 780], [99999, 120, 450, 340], [230, 560, 780, 99999] ] }

events是污染注入节点列表，sensors是候选监测点列表，response_matrix[i][j]表示第 i 次污染事件被第 j 个监测点检测到的响应时间（秒）。算法在计算目标函数时，对每个解（一组监测点），取每个污染事件在所有选中监测点里的最小响应时间，然后对所有事件求平均，得到最短监测时间目标；同时统计响应时间小于阈值的事件数量，除以总事件数，得到监测概率目标。

注意：response_matrix里的 99999 是个哨兵值，表示该监测点对该事件无响应。在计算监测概率时，如果某个事件在所有选中监测点上的响应时间都是 99999，就算漏检。这个阈值在源码里默认是模拟总时长，你可以根据实际管网规模调整。

4. NSGA2 算法核心：整数编码、交叉变异与帕累托前沿

4.1 种群初始化与整数编码的实现

nsga2/population_init.py负责生成初始种群。因为是整数编码，每个个体就是一个长度为 K 的列表，K 是你要布设的监测点数量。初始化时从所有候选节点里随机抽 K 个不重复的节点，重复这个过程直到种群规模达到设定值。

import random def init_population(pop_size, num_sensors, candidate_nodes): """ 生成整数编码的初始种群 pop_size: 种群大小 num_sensors: 每个个体选几个监测点 candidate_nodes: 候选节点编号列表 """ population = [] for _ in range(pop_size): # 从候选节点里随机抽 num_sensors 个不重复的 individual = random.sample(candidate_nodes, num_sensors) population.append(individual) return population

random.sample保证了个体内不会出现重复节点，省去了后续去重的麻烦。num_sensors这个参数很关键，它决定了你要布几个点。源码里默认设的是 5，你可以根据管网规模和预算调整。候选节点列表一般就是管网里所有节点，但如果你先用nodeimportance模块算过节点重要性，可以只保留重要性排名前 50% 的节点作为候选，这样能大幅缩小搜索空间，加快收敛。

4.2 快速非支配排序与拥挤度计算

NSGA2 和普通遗传算法最大的区别就在选择环节。普通 GA 按适应度值排序，NSGA2 按非支配层级和拥挤度排序。nsga2/dominance1.py和nsga2/crowddist.py分别实现了支配关系判断和拥挤度计算。

支配关系的定义：解 A 支配解 B，当且仅当 A 在所有目标上都不差于 B，且至少在一个目标上严格优于 B。对于监测点布局，目标一是最小化平均响应时间，目标二是最大化监测概率（等价于最小化漏检率）。dominance1.py里的dominates函数就是逐目标比较。

def dominates(obj1, obj2): """ 判断 obj1 是否支配 obj2 obj1, obj2: [监测时间, 漏检率] 两个目标都是越小越好 """ not_worse = all(o1 <= o2 for o1, o2 in zip(obj1, obj2)) strictly_better = any(o1 < o2 for o1, o2 in zip(obj1, obj2)) return not_worse and strictly_better

拥挤度计算是为了保持帕累托前沿上解的多样性。如果只按非支配层级选，同一层级里所有解被视为等价，随机选可能导致解扎堆。crowddist.py对同一层级的解按每个目标排序，计算相邻解之间的目标值距离，距离越大说明该解周围越稀疏，越应该保留。

4.3 交叉、变异与主循环

nsga2/crossover.py和nsga2/mutation.py实现了整数编码下的遗传操作。交叉用的是多点交叉，随机选几个位置，交换父代之间的节点编号。变异是随机选一个位置，把该位置的节点替换成候选节点里另一个随机节点。

def crossover(parent1, parent2, crossover_rate=0.8): """多点交叉，交换部分节点编号""" if random.random() > crossover_rate: return parent1[:], parent2[:] point1 = random.randint(1, len(parent1) - 2) point2 = random.randint(point1, len(parent1) - 1) child1 = parent1[:point1] + parent2[point1:point2] + parent1[point2:] child2 = parent2[:point1] + parent1[point1:point2] + parent2[point2:] return child1, child2 def mutation(individual, candidate_nodes, mutation_rate=0.1): """随机替换一个节点""" if random.random() > mutation_rate: return individual[:] idx = random.randint(0, len(individual) - 1) new_node = random.choice(candidate_nodes) while new_node in individual: new_node = random.choice(candidate_nodes) individual[idx] = new_node return individual

交叉率默认 0.8，变异率默认 0.1，这两个参数在finalMain.py里可以调。交叉率太高会破坏优良解的结构，太低则搜索效率下降；变异率太高退化成随机搜索，太低容易早熟收敛。我一般会先跑 50 代看看帕累托前沿的分布，如果前沿上的解很稀疏，就适当提高变异率。

finalMain.py是主循环入口，流程是：初始化种群 → 计算目标函数 → 非支配排序 → 拥挤度计算 → 选择 → 交叉变异 → 合并父子代 → 环境选择 → 迭代。每一代都会把当前帕累托前沿上的解存下来，最后用matplotlib画出来。

4.4 目标函数计算与结果解读

目标函数的计算在nsga2/estimate.py里。对每个个体（一组监测点），遍历所有污染事件，从response_matrix里取出该事件在所有选中监测点上的响应时间，取最小值作为该事件的检测时间。如果最小值是 99999，说明漏检。然后：

• 目标一：所有事件检测时间的平均值（忽略漏检事件，或者把漏检事件的时间设为一个惩罚值）
• 目标二：漏检事件数除以总事件数

def evaluate(individual, response_matrix, event_count, penalty=99999): """计算一个个体的两个目标值""" total_time = 0 detected = 0 for i in range(event_count): min_time = min(response_matrix[i][j] for j in individual) if min_time < penalty: total_time += min_time detected += 1 avg_time = total_time / detected if detected > 0 else penalty miss_rate = 1 - detected / event_count return [avg_time, miss_rate]

跑完finalMain.py之后，你会得到一张帕累托前沿图，横轴是平均监测时间，纵轴是漏检率。前沿上的每个点对应一组监测点编号。源码里final_json.json会保存最终的前沿解，你可以直接打开看每个解选了哪些节点。

提示：如果帕累托前沿上的解都挤在角落，检查一下num_sensors是不是设得太小。监测点数量太少，两个目标的冲突会非常剧烈，前沿会退化成一条很短的曲线。

5. 避坑与排查：跑这套源码时最容易翻车的五个地方

5.1 WNTR 版本不兼容导致run_sim报错

现象：调用wqsim.run_sim()时抛出AttributeError: 'WaterNetworkModel' object has no attribute 'options'或者类似的属性缺失错误。

原因：WNTR 在 0.3.x 到 1.0.x 之间 API 有过一次大改，旧版代码里的wn.options在新版里变成了wn.options.time和wn.options.hydraulic分开访问。源码里有些地方用的是旧版写法。

解决：先确认版本pip show wntr。如果是 1.0 以上，把wn.options相关的调用改成wn.options.time.duration和wn.options.hydraulic.demand_model。或者直接降级到pip install wntr==0.3.0，源码就是在这个版本下测试的。

5.2 污染注入模拟跑得太慢，卡在EventDerivation.py

现象：EventDerivation.py遍历所有节点跑模拟，跑了半小时还没结束。

原因：Net3 有 92 个节点，每个节点跑一次 24 小时水质模拟，WNTR 的水质模拟比水力模拟慢一个数量级，全部跑完确实要很久。

解决：源码里已经提供了final_json.json和waterQuality.json作为缓存，直接跳过模拟步骤，在finalMain.py里把数据加载路径指向这两个文件就行。如果你非要自己跑，把模拟时长从 24 小时缩短到 6 小时，或者只选重要性排名前 30 的节点做注入实验。

5.3 交叉变异后个体出现重复节点

现象：算法跑了几代之后，某个个体里出现了两个相同的节点编号，比如[12, 45, 45, 78, 103]。

原因：多点交叉时，父代1的片段和父代2的片段拼接后，可能包含相同的节点编号。源码里的crossover函数没有做去重。

解决：在交叉之后加一步去重，把重复的节点替换成候选节点里没出现过的随机节点。或者改用基于集合的交叉策略，把父代节点取并集后再随机采样。我在自己的项目里一般会在mutation之后统一做一次去重，确保每个个体内部节点唯一。

5.4 帕累托前沿只有一个点

现象：跑完 200 代，画出来的帕累托前沿只有一个解，或者所有解的目标值完全一样。

原因：目标函数计算有问题，最常见的是response_matrix里所有值都是 99999，导致每个个体的两个目标都是惩罚值。另一个可能是num_sensors设得太大，候选节点里随便选 K 个都能覆盖所有事件，两个目标没有冲突。

解决：先打印response_matrix的前几行，确认里面有正常的响应时间值。如果全是 99999，说明模拟阶段没跑成功，检查注入节点的quality属性有没有设上。如果矩阵正常，把num_sensors从 5 降到 3 再试。

5.5 画图时中文字体显示成方块

现象：matplotlib画的帕累托前沿图，标题和轴标签里的中文全是方块。

原因：matplotlib默认字体不支持中文。

解决：在画图脚本开头加两行：

import matplotlib.pyplot as plt plt.rcParams['font.sans-serif'] = ['SimHei'] # Windows 用黑体 plt.rcParams['axes.unicode_minus'] = False

Linux 上如果没有 SimHei，换成['WenQuanYi Micro Hei']或者['Noto Sans CJK SC']。

6. 进阶技巧：用节点重要性预筛选候选集加速收敛

跑通基础流程之后，你会发现算法收敛速度是个瓶颈。Net3 有 92 个节点，候选集就是 92 个，种群规模 100，迭代 200 代，每代要算 100 次目标函数，每次目标函数要遍历 92 个事件，计算量不小。如果换成真实城市管网，节点数上千，不优化的话跑一晚上都出不来结果。

源码里nodeimportance目录下的nodeImportance.py和centrality_result.py就是干这个的。它用 networkx 计算管网拓扑的几种中心性指标：度中心性、介数中心性、接近中心性。度中心性高的节点连接管道多，介数中心性高的节点位于很多最短路径上，这些节点一旦发生污染，影响范围大，优先在这些节点附近布设监测点，能用更少的点覆盖更多事件。

我一般会这样做：先算所有节点的介数中心性，按降序排列，取前 40% 作为候选集，再跑 NSGA2。这样候选集从 92 降到 37，搜索空间小了一个数量级，收敛速度明显加快，而且最终前沿上的解质量不会差太多——因为被筛掉的都是边缘节点，它们对污染事件的响应本来就慢。

import networkx as nx import wntr wn = wntr.network.WaterNetworkModel('example/Net3.inp') G = wn.to_graph() # 转成 networkx 图 # 计算介数中心性 betweenness = nx.betweenness_centrality(G) # 按中心性降序排列，取前 40% sorted_nodes = sorted(betweenness.items(), key=lambda x: x[1], reverse=True) top_k = int(len(sorted_nodes) * 0.4) candidate_nodes = [node for node, _ in sorted_nodes[:top_k]] print(f"候选节点数从 {len(G.nodes)} 降到 {len(candidate_nodes)}")

这段代码的关键是wn.to_graph()，WNTR 的WaterNetworkModel可以直接转成 networkx 的图对象，然后调betweenness_centrality算介数中心性。top_k控制筛选比例，我试过 30% 到 50% 之间，40% 是个比较稳的值。筛得太狠（比如 10%）可能把一些关键节点漏掉，导致前沿解的质量下降；筛得太松（比如 80%）加速效果不明显。

还有一个技巧是复用模拟结果。如果你要调num_sensors跑多组实验，不需要每次都重新跑水质模拟。response_matrix只跟管网模型和污染注入方案有关，跟选几个监测点无关。把模拟结果存成 JSON，换num_sensors的时候直接加载，能省掉大量重复计算。源码里final_json.json就是这个思路，我建议你在自己的项目里也养成这个习惯——模拟和优化分离，模拟结果落盘，优化阶段只读缓存。

从那以后我每次跑管网优化，都会先把response_matrix存成带时间戳的 JSON，算法调参阶段只读缓存，只有换管网模型或者改注入方案时才重新跑模拟。这个习惯帮我省了至少几十个小时的重复计算。希望帮到你。

本文还有配套的精品资源，点击获取