简介本资源面向高校学生与算法开发者提供基于Python与NSGA2算法的供水管网水质监测点布局优化完整方案适用于毕业设计、课程设计及项目开发场景。项目针对突发污染事件下的监测点选取问题采用整数编码的NSGA2算法构建最短监测时间与最大监测概率双目标函数并借助基于EPANET的WNTR库完成水力水质模拟与数据处理实现从污染注入实验到算法迭代的完整流程。压缩包共58个文件约746KB包含36个Python源码文件、7个XML配置、4个JSON数据、2个INP管网模型及CSV结果文件等涵盖水质模拟、数据处理、非支配排序、拥挤度计算、交叉变异等核心模块。已有134人学习关注。读者可获得可直接运行的源码、项目文档与清晰的目录结构便于理解帕累托解与进化算法在管网布局中的落地方式并在此基础上延伸扩展。1. 供水管网水质监测点布局一份能跑通全流程的 NSGA2 实战源码突发污染事件下供水管网里到底该把有限的水质传感器布在哪几个节点才能既快又准地抓到污染团这是环境工程和市政管网方向绕不开的优化问题。手里这份基于 Python NSGA2 的供水管网水质监测点布局源码解决的正是这件事用 WNTR 做水力水质模拟把污染注入实验的结果喂给整数编码的非支配遗传算法迭代出帕累托前沿上的监测点组合。它适合正在做毕业设计、课程设计或者需要一套可复现管网优化流程的开发者。整套代码把模拟、数据处理、算法迭代串成了闭环流程基本能走通拿来当二次开发的底座比从零搭要省太多事。2. 先搞懂双目标与整数编码为什么监测点布局要用 NSGA22.1 监测时间与监测概率为什么天然打架水质监测点布局的核心矛盾用一句话说清楚你想让传感器离污染源越近越好这样响应时间短但污染可能发生在管网里任何一个节点你又希望少数几个传感器能覆盖尽可能多的污染场景。前者对应最短监测时间后者对应最大监测概率这两个目标在数学上是负相关的——把点布在管网中心区域平均响应快但边缘节点的污染事件可能漏检把点分散到各个角落覆盖率高可每个点的响应时间又拉长了。传统做法是加权求和把两个目标揉成一个适应度函数但权重怎么定全凭经验改一次权重就要重跑一遍而且得到的只是某一个折中解看不到全局的取舍关系。NSGA2 的价值在于它一次运行就能给出一整条帕累托前沿每个解都代表一种「响应时间」和「覆盖概率」的权衡方案决策者可以根据实际预算和风险偏好从前沿上挑点。这就是为什么这类布局问题几乎成了多目标进化算法的经典练兵场。2.2 整数编码监测点选择不是连续变量很多 NSGA2 的入门示例用的是实数编码变量在某个区间内连续取值。但监测点布局是个组合优化问题——管网里有几百个节点每个节点只有「选」和「不选」两种状态你要从 N 个节点里挑出 K 个。如果硬用实数编码再四舍五入交叉变异之后会产生大量重复或越界的无效解算法效率会被拖垮。这份源码用的是整数编码染色体直接存节点编号比如[12, 45, 78, 103, 156]就代表选了这五个节点作为监测点。交叉操作在父代之间交换部分节点编号变异操作把某个位置的节点替换成管网里的另一个随机节点。这样做的好处是解空间和实际决策空间一一对应不会产生无效解收敛速度也更快。代价是需要额外处理重复节点的问题——两个父代可能都选了同一个节点交叉后子代里会出现重复编号得在解码阶段去重。2.3 用 WNTR 替代 EPANET 链接库的实操理由早期做管网模拟大家习惯调 EPANET 的 C 库或者用 EPANET 的图形界面手动跑。但 EPANET 的 Python 接口一直不太顺手要么用epanettools这种第三方封装要么自己写 ctypes 调动态库环境配置能折腾掉半天。USEPA 后来开源了 WNTRWater 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 python3.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 ToolsLinux 上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 populationrandom.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_rate0.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_rate0.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, penalty99999): 计算一个个体的两个目标值 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 wntr0.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] FalseLinux 上如果没有 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(), keylambda x: x[1], reverseTrue) 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算法调参阶段只读缓存只有换管网模型或者改注入方案时才重新跑模拟。这个习惯帮我省了至少几十个小时的重复计算。希望帮到你。本文还有配套的精品资源点击获取