半夜两点值班电话响了说是某片区的供水压力突然往下掉调度大屏上已经标红。我到现场一查压力变送器没坏线路也通问题出在上位机——原始值翻倍了1.2MPa的管网压力被解成了2.4MPa。这种问题在供水管网漏损监测项目里太常见了代码写得不够稳预警系统会在真正漏损时沉默在风平浪静时乱叫。这门课的6.6节正好讲到漏损监测模块的编码我就把实际项目里写的压力数据解析和漏损预警代码拆开聊一聊重点说两块数据怎么解析才可靠预警怎么判断才不误报。适合正在做智慧水务相关系统的工程师参考也适合想把管网数据处理逻辑彻底搞懂的同学看看。1. 模块定位与整体设计思路1.1 为什么漏损监测要先盯压力数据管网漏损的检测手段很多流量、水质、压力、噪声都可以拿来做文章。但实际项目里压力数据永远是优先级最高的那一路。原因很直接压力变送器便宜、布点多、采样频率高而且管网发生爆管或者泄漏时压力变化是物理上最先出现的信号——管线一破水压立刻往下掉这个响应是秒级的比流量计发现的异常快得多。我做过几个漏损监测项目甲方最关心的不是“你能不能测出流量差”而是“你能不能在我接到市民投诉之前就告诉我哪个片区压力异常”。压力数据就是用来抢这个提前量的。所以在整个漏损监测模块里代码架构上我把压力数据解析放在最底层把漏损预警算法放在上层中间用一套清洗干净的标准化数据结构衔接。底层解析错了上层算法再漂亮都是白搭。1.2 模块拆成三层每层职责单一这个模块在实际编码时我按三个层次来拆的采集接入层负责跟现场RTU远程终端单元、压力变送器通信处理Modbus RTU/TCP、DL/T 645或其他厂家私有协议把原始字节收上来。解析规约层把收到的字节流做校验、切片、字节序转换、量程换算输出统一的、带时间戳和状态位的压力数值。漏损预警层接收标准化压力数据跑阈值判断、持续压降检测、夜间最小压力分析输出不同等级的漏损预警事件。这个拆分不是为了好看是为了排查问题的时候不抓瞎。曾经有个项目半夜预警误报现场查了半天结果发现是解析层里某个厂家变送器的量程配置写错了跟算法没有半点关系。如果当时所有代码糊在一个文件里这个问题至少要排查两三个小时分层之后定位到解析层十分钟就结束了。1.3 数据链路和字段设计数据从现场到预警模块的链路大概是这样的压力变送器 - RTUModbus采集 - 前置机/网关 - MQTT/HTTP上传 - 解析服务 - 时序数据库 - 预警计算服务 - 值班平台/手机推送解析服务输出的标准化数据结构我在代码里定义成这样dataclass class PressurePoint: device_id: str # 设备编号比如 PS-DS-014 ts: int # Unix时间戳秒级 pressure: float # 换算后的压力值单位MPa quality: int # 质量码0正常, 1超量程, 2校验失败, 3通信中断 raw_value: int # 原始整型值便于回查质量码这个字段特别关键后面预警算法跑的时候只认quality 0的数据其他状态一律跳过。不设计这个字段的话通信闪断恢复瞬间的异常值就会被当成真实压力突变误报就这么来的。2. 压力数据解析的完整实现2.1 Modbus RTU数据帧结构剖析现场压力变送器九成以上走Modbus RTU协议很多是读保持寄存器功能码0x03。一帧完整的请求/响应大概是这样的地址码1字节比如0x01表示1号变送器功能码1字节0x03起始寄存器地址2字节比如0x0000寄存器数量2字节比如0x0001或0x0002有些仪表用两个寄存器存一个floatCRC校验2字节低字节在前响应帧就是地址码1字节功能码1字节数据字节数1字节数据区N字节CRC校验2字节举个例子某个压力变送器返回的原始字节是01 03 02 0A F0 39 84其中0A F0是寄存器里的原始值十六进制转十进制等于0x0AF0 2800。如果这台仪表的量程是0~1.6MPa对应原始值范围0~16000那换算系数就是1.6 / 16000 0.0001最终压力就是2800 * 0.0001 0.28MPa。写解析代码的时候第一步千万不要直接做换算先把CRC校验和数据帧结构检查过了再说。2.2 字节序、缩放系数和量程配置压力解析最容易踩坑的就是字节序和缩放系数。同一个传感器A厂家的float可能是大端序B厂家可能是小端序有的用int16有的用int32有的是寄存器值直接对应实际值乘以1000有的是乘以10000。这些规则不统一代码里必须做成配置项。我之前接过一个项目现场有两种压力变送器一种是国内厂商的量程0~1.6MPa原始值0~16000系数0.0001另一种是进口的量程0~100psi用的是两个寄存器拼接的float。如果不区分型号用一种解析逻辑去跑轻则数值翻倍重则直接出来负压。最后编码的时候我把每台设备的解析参数都做成数据库配置表字段像这样设备型号协议类型寄存器起始地址数据长度字节序数值类型量程下限量程上限缩放系数单位国产某型Modbus RTU0x00002字节大端int160160000.0001MPa进口某型Modbus RTU0x00004字节小端float00.6891.0MPa代码里统一走配置驱动不写死任何设备的参数。这样新接入一台变送器的时候运维人员只需要在后台填一行配置不用改代码重新发布项目上线之后省了太多事。2.3 压力数据过滤的三个坑解析出数值之后不能直接往库里写还要过一道数据过滤逻辑。我在实际项目里总结出来三个必须处理的坑第一个是超量程值。变送器断线或者损坏的时候寄存器会返回满量程或者0值这种值一看就不正常直接打quality 1标记不参与后续计算。第二个是瞬时跳变。通信干扰可能导致某一个采样点突然从0.4跳到1.2又跳回来这种数据不能简单删掉应该打上可疑标记让预警算法在一个时间窗口里综合判断而不是看单点突变就报预警。第三个是长时间恒定值。压力变送器卡死或者RTU程序异常的时候数据会连续几百个采样点都等于同一个值。这个隐蔽性很强因为数值本身合法可就是不动。我在代码里加了“连续相同值计数”超过一定阈值就标记设备状态异常。2.4 解析代码示例我用Python实现了一个简化的解析服务重点代码如下。实际项目里用的是异步框架但核心逻辑是一样的。import struct from dataclasses import dataclass class PressureParser: def __init__(self, device_config): # device_config 包含协议、字节序、量程、系数等参数 self.cfg device_config def validate_crc(self, frame: bytes) - bool: # 这里用查表法计算CRC16实际代码会贴完整的CRC表 crc 0xFFFF for byte in frame[:-2]: crc ^ byte for _ in range(8): if crc 0x0001: crc (crc 1) ^ 0xA001 else: crc 1 return (crc 0xFF) frame[-2] and ((crc 8) 0xFF) frame[-1] def parse_modbus_rtu(self, frame: bytes) - PressurePoint: if len(frame) 5 or frame[1] ! 0x03: return PressurePoint(self.cfg[device_id], 0, 0.0, 2, 0) if not self.validate_crc(frame): return PressurePoint(self.cfg[device_id], 0, 0.0, 2, 0) # 取出数据区 byte_count frame[2] data frame[3:3 byte_count] if self.cfg[data_length] 2: if self.cfg[byte_order] big: raw_value struct.unpack(H, data[:2])[0] else: raw_value struct.unpack(H, data[:2])[0] else: # 4字节float fmt f if self.cfg[byte_order] big else f raw_value struct.unpack(fmt, data[:4])[0] # 量程与系数换算 lower self.cfg[range_lower] upper self.cfg[range_upper] ratio self.cfg[scale_factor] pressure raw_value * ratio if pressure lower or pressure upper: quality 1 # 超量程 else: quality 0 return PressurePoint( device_idself.cfg[device_id], tsself.cfg.get(current_ts, 0), pressureround(pressure, 4), qualityquality, raw_valueraw_value )注意CRC校验函数里的多项式0xA001是Modbus协议的标准值每个字节都要计算8位遍历整帧不含CRC本身。如果你从网上抄了一个CRC函数没有仔细核对解析出来的数据可能是错的。这个代码里最有用的经验是preview阶段就把不同设备型号的样本帧收集起来各存20帧以上写一个单元测试反复跑解析逻辑确保换了设备型号也不会出问题。我在某水厂项目里就是因为提前做了这个测试规避了进口变送器float字节序不一致导致的大面积错数。3. 漏损预警算法设计既要灵敏又要克制3.1 阈值不是拍脑袋定的是算出来的漏损预警最基础的手段就是阈值判断但很多项目直接把阈值设成“压力低于0.2MPa就报警”这种做法在真实管网上完全不可行。我曾经接过一个供水分区白天高峰时段压力降到0.25MPa很正常凌晨用水量小、压力能恢复到0.4MPa以上如果统一用0.2MPa当阈值白天管网的降压运行就会疯狂误报值班人员要么被骚扰到关掉告警要么对告警麻木真正漏损来的时候反而没人当回事。正确的做法是分时段阈值。把一天按小时分成24个时段取过去14天每个时段的历史压力数据用P5分位数即该时段内5%的时间压力低于此值作为下限阈值再用P95分位数作为上限阈值用于检测压力异常升高。这样出来的阈值是一条动态曲线能适应白天和夜间的供水规律。分段阈值的历史数据计算我一般这么写import numpy as np def compute_thresholds(history_data, device_ids, hour, percentile_low5, percentile_high95): history_data: 某设备连续N天、某个小时的所有压力数据 返回 (下限阈值, 上限阈值) values np.array(history_data) low np.percentile(values, percentile_low) high np.percentile(values, percentile_high) return round(low, 3), round(high, 3)阈值表需要定期自动更新比如每周一凌晨重新计算一次保证季节性用水变化能被覆盖到。有些项目还把天气因素加进去比如降雨天和高温天分别建不同的阈值曲线高维度的阈值模型效果更好但因为要接天气预报数据实施成本高一些。3.2 持续压降检测滑动窗口斜率滑动单点压力低于阈值只能发提醒真正要定位漏损得看压力的“趋势”。管网爆管时压力不是瞬间掉到零而是一个持续下降的过程区别只是下降速率不同。小漏可能一天只掉0.03MPa大爆管可能几分钟就掉0.3MPa。所以漏损预警算法里我实现了一个基于滑动窗口的线性回归下降检测器。算法思路是取最近N个采样点比如15分钟内的15个点用最小二乘法拟合一条直线得到斜率k。如果k为负且绝对值超过设定档位同时当前压力已经低于分时段阈值就触发相应等级的预警。斜率分档我一般分三档变化速率15分钟压降初步判断预警等级慢速压降≥ 0.01 MPa疑似小漏或阀门异常蓝色提醒中速压降≥ 0.04 MPa可能支管泄漏黄色预警快速压降≥ 0.10 MPa疑似主管爆管红色预警3.3 夜间最小压力分析漏损识别的“照妖镜”白天管网用水复杂漏损信号容易被用水波动掩盖但凌晨2点到4点这段绝大多数居民和工商业用户都停了管网压力理论上应该比较稳定地处于高位。如果这时候压力异常偏低漏损嫌疑就非常大。夜间最小压力模块的代码不复杂但要配合历史基线。基本逻辑是把凌晨2:00-4:00的压力序列取中位数跟最近7天同时间段的基线做差值。差值超过0.03MPa且持续两个晚上的直接发黄色预警超过0.05MPa的直接发红色预警。我做了个小技巧计算夜间压力序列时把水泵切换、水厂调流带来的瞬时波动剔除掉只取稳定段的值不然换泵那几分钟的水锤波形会干扰判断。剔除的方式很简单——取压力序列的P25和P75之间范围再做平均可以滤掉两端的突变尖峰。3.4 多信号融合不靠单一压力信号下定论压力信号单独用误报率始终压不下来因为管网上不止漏损一个因素会导致压力波动水泵启停、大用户开关阀门、其他水厂并网这些都是干扰源。所以我在这套模块里做了信号融合层。融合逻辑很朴素压力跌落预警触发后程序会去取同一分区或上下游关联设备的流量数据和泵状态数据如果是泵突然停机的正常调节流量会同步下降压力也会降但这是一次正常操作不需要报漏损。如果是爆管漏损压力下降的同时泵站流量通常会上升因为泵要努力维持压力但控不住。如果是关联分区调流导致压力下降流量呈现的规律又会不一样。代码里用了一个简单的判定表逻辑清晰、便于值班人员理解压力趋势下游流量趋势泵运行状态判定结果持续下降上升正常运行疑似漏损红色预警持续下降下降泵已停止主动停泵不预警持续下降平稳压力控制阀动作疑似调流黄色提示单点抖动平稳平稳数据波动不预警这个判定表用代码实现起来就是一个简单的状态机跑在预警服务里每个周期评估一次把误报率降下来一半左右。4. 预警判断代码与工程落地的关键环节4.1 预警判断服务完整代码框架下面是我在一套预警服务里写的核心判断逻辑简单版但架构是完整的。import time import numpy as np class LeakWarningEngine: def __init__(self, threshold_service, device_repo): self.threshold_service threshold_service # 动态阈值服务 self.device_repo device_repo # 设备与关联数据访问 self.buffer {} # 设备ID - 最近N个压力点 def add_pressure_point(self, point: PressurePoint): if point.quality ! 0: return # 质量异常的数据直接丢弃不参与预警计算 if point.device_id not in self.buffer: self.buffer[point.device_id] [] self.buffer[point.device_id].append(point) # 只保留最近60个点对应15分钟采样频率 self.buffer[point.device_id] self.buffer[point.device_id][-60:] def detect(self, point: PressurePoint): series self.buffer.get(point.device_id, []) if len(series) 15: return None pressures [p.pressure for p in series] slopes self._calc_slope(pressures[-15:]) current_pressure pressures[-1] # 动态阈值判断 hour time.localtime(point.ts).tm_hour low_th, high_th self.threshold_service.get_threshold(point.device_id, hour) if abs(slopes) 0.0005: return None # 趋势平稳不报警 # 持续压降判断 if slopes -0.001: # 15分钟压降超过约0.015MPa if current_pressure low_th: return self._make_alarm(point.device_id, yellow, slopes, current_pressure) if slopes -0.005: # 快速压降 return self._make_alarm(point.device_id, red, slopes, current_pressure) return None def _calc_slope(self, values): # 使用最小二乘法计算斜率 x np.arange(len(values)) y np.array(values, dtypefloat) slope np.polyfit(x, y, 1)[0] return slope def _make_alarm(self, device_id, level, slope, pressure): # 组装预警事件写入数据库对外推送 return { device_id: device_id, level: level, slope: round(slope, 5), pressure: round(pressure, 3), ts: int(time.time()) }这个框架最值得注意的地方是预警事件不是直接返回就完了而是先入库再通过消息队列推给值班平台预警模块本身不负责发送短信和打电话解耦之后运维人员调推送逻辑的时候不用动预警主体代码。4.2 采样频率、滑动窗口与延迟的取舍压力采样频率决定了预警延迟和误报率之间的平衡。我在实际项目里的经验值是采样周期30秒检测窗口15分钟。窗口太短数据波动大、斜率噪声太强窗口太长发现爆管的时候可能已经过去半小时了。项目初期我试过5分钟窗口、30秒采样结果因为管网本身有压力脉动斜率经常算出来偏大误报率高到离谱。后来把窗口调到15分钟同时加入一个“连续3个窗口斜率都为负”的确认条件误报明显减少。确认条件增加了大约5-8分钟延迟但对于绝大多数漏损场景这个延迟完全可接受毕竟漏损不是几秒钟就会造成巨大损失的稳定可靠远比敏捷更重要。4.3 历史数据回放测试上线前必须做的验证预警算法写完之后不能直接接真实数据上线。我强烈建议先拿历史数据做回放测试。方法是把过去30天的历史压力数据按时间顺序重新喂给预警引擎看它触发了哪些预警事件再跟现场真实发生的爆管/维修工单做比对计算出误报率和漏报率。我做这个回放测试的时候发现了一个很有意思的事阈值曲线用P5分位数设出来之后理论上每小时内会有5%的数据低于阈值一天下来就会触发几十次阈值报警。所以阈值报警根本不能单独作为最终预警必须配合“持续压降”和“夜间最小压力分析”三条证据链至少满足两条才对外推送。回放测试就是干这个用的——把组合规则调好确保在真实数据上跑出来的告警密度每台设备每天不超过2次。4.4 数据存储和查询性能压力数据点每天的量非常大。一个中等城市供水分区有500个压力测点、30秒一个点、一天就是144万个点。如果预警服务每次阈值判断都去数据库里现查历史数据数据库必挂。我的做法是解析服务把数据同时写入时序数据库预警引擎内部维护内存缓冲窗口只查最近60个点所有关联判定用内存数据不查库。时序数据库的存储策略也需要注意原始压力数据保留30天用于回放和算法调参再往上做5分钟聚合表用于月度分析和报表超过一年的数据基本不用了直接清理或者归档到冷存储。千万别把原始数据永久留着浪费存储查询也慢。5. 现场调试与常见问题排查实录5.1 没有现场数据源时怎么模拟测试开发阶段往往等不到现场设备装好所以解析服务和预警算法需要一边开发一边测试。我通常自己写一个模拟数据源程序读一份真实设备的历史数据文件按时间间隔逐个推送出来模拟RTU的实时上报行为用来验证整个链路的正确性。模拟数据源还可以配置异常注入比如每隔100个点插入一个跳变值、每隔500个点模拟一次通信中断、随机修改字节序等用来测试解析服务的容错逻辑和预警算法对异常数据的抗干扰能力。这套模拟器你留着后面设备联调、问题复现都还能派上用场。5.2 联调阶段的常见问题速查表项目现场联调的时候我整理过一个常见问题速查表基本覆盖了绝大多数情况现象常见原因排查思路解析出的压力值是实际值的10倍缩放系数配错0.001写成0.01对比原始值和现场压力表读数解析出的压力是负数字节序不对int16高低字节反了打印原始字节手工算一遍十六进制所有设备CRC校验失败网关协议地址或串口参数与设备不一致检查波特率、数据位、校验位偶尔一个点突然跳高又回落通信干扰或没有配置数据过滤看质量码检查信号线和屏蔽层接地预警总在整点前后触发水厂调流或泵站调度导致压力规律变化检查阈值曲线是否跟实际供水规律匹配夜间报警密集水塔/高位水池夜间自然补水降压加入泵状态或水位信号做关联判定5.3 一次印象深刻的误报排查某项目上线之后一周内连着报了几次黄色预警位置都在同一个小区的测压点。每次都是晚上10点左右准时报警持续半小时后自动恢复。现场查了压力变送器没问题查了解析日志数据质量码也正常。后来我调了关联流量数据分析发现这个测压点位于一个二次供水泵房的出水口附近。晚上10点正好是小区物业设定的低峰降压时段泵房变频泵自动降低出水压力测压点压力自然跟着往下掉。这个压力下降是正常的运行策略不是漏损。最后处理方案很直接在这个测压点的配置里加上“关联二次供水泵房运行状态”的标记泵房主动降压时段内该测压点不参与漏损预警计算同时把夜间最小压力的判定调整到凌晨2点到4点避开影响窗口。这个问题在速查表里也补上了对应条目后面的项目直接参考。5.4 预警阈值更新的运维节奏动态阈值不是设置一次就一劳永逸的。建议每周自动重算一次阈值表同时关注几个指标每台设备每日触发预警次数、值班人员确认的有效预警占比、维修工单与预警的匹配率。如果有效占比太低说明阈值太敏感要把分位数从P5调到P3甚至P2如果漏报率升高再适当回调。我个人的经验是一个成熟的漏损监测系统上线稳定运行一个月之后黄色以上预警的有效率应该能维持在30%以上。低于这个数字优先怀疑阈值配置和信号融合条件而不是急着去换更复杂的机器学习模型。规则写得清楚、阈值算得准比盲目上模型有效得多。5.5 小技巧把误报样本积累成回归测试集每次预警误报被确认之后我建议顺手把当时的原始数据快照保存下来标注为“正常场景”。时间长了就积累成一个回归测试集。修改解析逻辑或者调整预警算法后先跑一遍回归测试确保之前修掉的误报不会卷土重来。这个习惯看起来简单但能在后续升级时帮你省下大量排障时间。我见过不止一个项目升级完算法修了A类误报结果B类误报暴增因为没有回归测试集根本不知道改动影响到哪里。现在我的项目里一定保留这样一个样本库每次发布前必跑。写这块代码的时候我心里最深的感触是漏损监测系统的难点真的不在代码本身而在于代码跟真实管网数据之间较劲的过程。压力解析看似简单却经常被各种型号差异、通信干扰、量程配置绊倒漏损预警看似可以天天调阈值实际上需要你不断拿历史数据回放验证用真实运维反馈修正。先把数据解析的每个环节做扎实再把误报率压下去比堆多少花哨算法都管用。如果你正准备写类似的模块我的建议就一句话留足时间跟现场数据耗这钱花得值。