简介：这是一份面向城市燃气行业从业者、公共事业管理者及数字化转型研究人员的行业实践报告，围绕燃气企业如何借力数字化重塑战略、运营与服务展开，可为城燃企业制定转型路线提供参考。资源包共1个文件，为PDF格式，约4MB，内容以图文并茂的幻灯片式报告呈现，便于通读与摘取要点。报告以北京市燃气集团为样本，梳理其管网规模、用户体量等基础情况，并延伸到Web2.0、电子商务、数字营销等发展阶段带来的运营方式变化，剖析客户期望提升、气源多样化、输配网络复杂化等现实挑战。重点部分给出数字化转型模型与落地路径：数据作为企业神经中枢驱动组织与流程整合，智能决策、智慧运营、智能管网、智能服务协同推进；IT架构从集中式分层走向分布式云端，依托物联网计量、大数据与AI实现需求预测、智能调度与自动应急；同时覆盖行业云平台的SaaS、PaaS、IaaS能力以及企业文化、数据安全等配套转型，并附有掌上营业厅、亦庄示范片区等实践案例。目前已有100余人学习，适合快速建立城燃数字化运营的整体认知框架。
1. 城市燃气数字化运营实践：先看清要解决的三个真问题
很多燃气公司的数字化第一版做得挺漂亮，调度大厅大屏铺满管线走向和实时压力，运行两周就被调度员退回原形：GIS 里的管线连不成拓扑，SCADA 点位和数据中台对不上号，报警铃一天响三百次没人看。问题不在可视化，而在三件事没做扎实——表计与调压站数据能不能稳定落库、竣工图能不能变成参与水力计算的连通图、报警到工单能不能形成闭环。城市燃气数字化运营实践这个题目，落地顺序基本就是这个：先建统一时序底座，把远传表、调压站、阀室的数据收干净；再把图纸转成可计算的管网拓扑；接着把泄漏判断从单点阈值升级成多判据联动；最后才是看板和报表。下面给的 SQL、Python 和参数表都可以直接拿去试，适合管网运行、调度、信息化岗的技术人员。
2. 城市燃气数据底座：采集协议选型、时序建模与质量校验
2.1 燃气数据源盘点与采集协议选型
城市燃气一次要接进来的数据源，实时性跨度有三个数量级，硬塞进同一条链路一定出问题。调压站 RTU 是秒级，工商业远传表是分钟级，居民 NB 表基本是日级批量，阴极保护桩一小时一次。常见做法是按周期分三条通道：秒级走消息总线或 IEC 104 直采，分钟级走 MQTT 汇聚，日级走运营商平台接口或文件交换。
| 数据源 | 典型协议/接口 | 采样周期 | 主要用途 |
|---|---|---|---|
| 调压站、阀室 RTU | Modbus TCP、IEC 60870-5-104 | 1～5 s | 进出口压力、瞬时流量、阀位 |
| 工商业远传表 | CJ/T 188、Modbus RTU、MQTT | 5～15 min | 用气量、表状态、阀控 |
| 居民 NB-IoT 表 | 运营商平台 API、MQTT | 1 次/日～1 次/小时 | 抄表结算、异常用气识别 |
| 阴极保护测试桩 | Modbus RTU + 无线 DTU | 1 次/小时 | 管地电位、保护电流 |
| 可燃气体探测终端 | MQTT、私有 TCP | 30 s～5 min | 甲烷浓度、现场报警 |
选型上有个容易踩的坑：把 CJ/T 188 这种表计本地协议直接往采集网关里塞，再做协议转换入 MQTT。表面省事，实际是表计应答超时、帧校验失败全压在网关侧，出问题很难定位到哪一块表。我一般会让网关只做透传，协议解析放在采集服务里，表计异常能精确落到具体表号。
2.2 设备—测点—工况三层建模
把所有字段堆进一张宽表的做法，撑不过半年。压力量、流量、浓度、电位量纲完全不同，加一个设备类型就要加一批列，查询语句越来越长。稳妥的是三层：device记设备台账，point记测点定义，measurement只存时间戳和值，工况信息（比如调压站运行方式、管网冬夏季模式）单独放一张维度表。
-- 设备台账：一台调压站 RTU、一块远传表都是一条记录 CREATE TABLE gas_device ( device_id text PRIMARY KEY, -- 全局唯一，建议 站点码:设备类型:序号 device_name text NOT NULL, device_type text NOT NULL, -- RTU / METER / CP / DETECTOR area_code text NOT NULL, -- 所属片区，用于工单派发 install_ts timestamptz, online boolean DEFAULT true ); -- 测点定义：一个设备下挂多个测点，量纲和量程写在这里 CREATE TABLE gas_point ( device_id text NOT NULL REFERENCES gas_device(device_id), point_code text NOT NULL, -- P_IN / P_OUT / Q / CH4 unit text NOT NULL, -- kPa / m3/h / %LEL range_low double precision, range_high double precision, PRIMARY KEY (device_id, point_code) );时间戳统一按 UTC 存储，展示层再转本地时间。有公司直接存本地时间，跨片区一汇总，曲线就串位，后面查泄漏时段的压力曲线怎么都对不上。
2.3 用 MQTT 加 TimescaleDB 落通一条压力数据链路
数据底座的最小可用形态，就是一条从 MQTT 主题到超表的通路。主题层级建议固定成gas/{station}/{device}/{point}，靠主题直接切分站点和设备，别把设备号再塞进 payload，不然订阅粒度做不细。
import json import paho.mqtt.client as mqtt import psycopg2 from psycopg2.extras import execute_values conn = psycopg2.connect("dbname=gas user=ops host=127.0.0.1") BATCH, FLUSH_SIZE = [], 200 def flush(): """批量写入，避免每条消息一次事务把数据库打满""" if not BATCH: return with conn.cursor() as cur: execute_values(cur, """ INSERT INTO gas_measurement (ts, device_id, point_code, value, quality) VALUES %s """, BATCH) conn.commit() BATCH.clear() def on_connect(client, userdata, flags, rc): # QoS 1：至少一次，燃气压力数据丢点比重复更麻烦 client.subscribe("gas/+/+/+", qos=1) def on_message(client, userdata, msg): _, station, device, point = msg.topic.split("/") payload = json.loads(msg.payload) ts = payload["ts"] # ISO8601，带时区 value = float(payload["value"]) quality = int(payload.get("quality", 0)) # 0 正常 1 可疑 2 无效 BATCH.append((ts, f"{station}:{device}", point, value, quality)) if len(BATCH) >= FLUSH_SIZE: flush() client = mqtt.Client(client_id="gas-collector-01") client.on_connect = on_connect client.on_message = on_message client.connect("mqtt.internal", 1883, 60) client.loop_forever()FLUSH_SIZE是吞吐和延迟的平衡点，200 条在秒级采集下大约 1～2 秒落一次盘；如果站点数超过两千，可以降到 100 并配合连接池。quality字段必须留，量程越界和跳变的数据不能直接丢，标成可疑值留着，后面查模型误差时要用。
CREATE TABLE gas_measurement ( ts timestamptz NOT NULL, device_id text NOT NULL, point_code text NOT NULL, value double precision, quality smallint NOT NULL DEFAULT 0 ); SELECT create_hypertable('gas_measurement', 'ts', chunk_time_interval => INTERVAL '7 days'); CREATE INDEX idx_gas_dev_point_ts ON gas_measurement (device_id, point_code, ts DESC); -- 超过 30 天的数据压缩，压力曲线查询基本只走最近区间 ALTER TABLE gas_measurement SET ( timescaledb.compress, timescaledb.compress_segmentby = 'device_id, point_code', timescaledb.compress_orderby = 'ts DESC' ); SELECT add_compression_policy('gas_measurement', INTERVAL '30 days');分区跨度取 7 天，是按单站点日均 8 万条估的，一个 chunk 大约 60 万行，查询和压缩都还顺手。站点规模翻倍就缩到 3 天，别等到单 chunk 上千万行才动。
2.4 燃气数据质量的 3 个必查校验
数据进来不校验，水力模型算出来的结果没人敢信。量程、跳变、卡死这三类占了实际异常的大头，用 SQL 就能日常跑。
-- 校验一：量程越界。中压 A 管网压力正常区间按 0.05～0.4 MPa 收 SELECT device_id, point_code, count(*) AS bad_cnt FROM gas_measurement WHERE point_code LIKE 'P_%' AND (value < 50 OR value > 400) -- 存的是 kPa AND ts > now() - INTERVAL '1 day' GROUP BY 1, 2 ORDER BY bad_cnt DESC; -- 校验二：跳变。相邻两点变化超过量程的 20% 视为可疑 SELECT ts, device_id, point_code, value, value - lag(value) OVER w AS delta FROM gas_measurement WHERE ts > now() - INTERVAL '1 day' WINDOW w AS (PARTITION BY device_id, point_code ORDER BY ts) QUALIFY abs(value - lag(value) OVER w) > 0.2 * 350;-- 校验三：卡死。30 分钟滚动窗口内值完全相同的点数超过阈值 SELECT * FROM ( SELECT ts, device_id, point_code, value, count(*) OVER ( PARTITION BY device_id, point_code, value ORDER BY ts RANGE BETWEEN INTERVAL '30 minutes' PRECEDING AND CURRENT ROW ) AS same_cnt FROM gas_measurement WHERE ts > now() - INTERVAL '6 hours' ) t WHERE same_cnt >= 30;注意：卡死判定不能只比相邻两点，调压器稳定运行时相邻值本来就一样，必须用滚动窗口统计持续时长，否则每天都会刷出一堆假异常。
3. 城市燃气管网拓扑数字化：从竣工图到可计算的水力模型
3.1 从 CAD 竣工图到拓扑连通图的处理流程
图纸到模型的转换，卡点从来不是格式，而是连通性。CAD 里两条管线看着接上了，端点坐标可能差 0.2 米；GIS 里一条 LineString 穿过三通却没在交点处打断。这两种情况直接导进水力计算程序，结果是建成一堆互不相连的孤立管网。
from shapely.geometry import LineString from shapely.ops import unary_union def build_topology(lines, snap_tol=0.5): """ lines : LineString 列表，来自 CAD/GIS 导出的管段 snap_tol : 端点吸附容差，单位与坐标系一致，投影坐标下取 0.5 米 """ fixed = [] for ls in lines: # 端点坐标吸附到容差网格，消除肉眼看不出的错位 coords = [(round(x / snap_tol) * snap_tol, round(y / snap_tol) * snap_tol) for x, y in ls.coords] fixed.append(LineString(coords)) # 在真实交点处打断，交叉但未打断的线段会被切开 merged = unary_union(fixed) segs = list(merged.geoms) if merged.geom_type == "MultiLineString" else [merged] edges = [] for i, s in enumerate(segs): (x1, y1), (x2, y2) = s.coords[0], s.coords[-1] edges.append({ "edge_id": f"E{i:05d}", "n1": f"{x1:.1f},{y1:.1f}", # 用坐标串当节点 ID，省一次映射 "n2": f"{x2:.1f},{y2:.1f}", "length": round(s.length, 3), # 单位 m }) return edges吸附容差 0.5 米是经验值，来自竣工图常见的测绘误差量级。调大到 2 米会把相邻的平行支管误并成一条，调小到 0.05 米则连不上。打断后的节点 ID 用坐标串，简单直接，但记住后续必须做一次坐标精度归一，不然1.0和1.00会生成两个节点。
3.2 用 Hardy Cross 在 Python 里跑通一个燃气环网
低压燃气管网的水力计算，工程上最常用的是环网校正法。单环和多环都能算，代码量小，结果和商业软件差得不多，适合先跑通再谈精度。管段压降用达西公式推导出的形式：h = R · Q · |Q|，其中R = 8λLρ / (π²d⁵)。
import math RHO = 0.75 # 天然气密度 kg/m3，低压管网按常量处理 MU = 1.1e-5 # 动力粘度 Pa·s def resistance(L, d, Q0, K=0.0001): """L 管长 m；d 内径 m；Q0 设计流量 m3/s；K 绝对粗糙度 m""" Re = 4 * RHO * Q0 / (math.pi * d * MU) lam = 0.11 * (K / d + 68 / Re) ** 0.25 # 阿里特舒里近似式 return 8 * lam * L * RHO / (math.pi ** 2 * d ** 5) def hardy_cross(pipe_R, loops, Q, tol=1e-9, max_iter=100): """ pipe_R : 各管段阻力系数列表 loops : [[(管段索引, 沿环方向 +1/-1), ...], ...] Q : 初始流量列表，m3/s，符号表示与管段定义方向是否一致 """ Q = list(Q) for it in range(max_iter): max_dq = 0.0 for loop in loops: num = sum(s * pipe_R[i] * Q[i] * abs(Q[i]) for i, s in loop) den = sum(2 * pipe_R[i] * abs(Q[i]) for i, s in loop) if den < 1e-12: continue dq = -num / den for i, s in loop: Q[i] += s * dq max_dq = max(max_dq, abs(dq)) if max_dq < tol: return Q, it + 1 return Q, max_iter # 一个单环算例：三条管段首尾相接，节点 1 进气，节点 2、3 各带负荷 pipe_R = [ resistance(L=300, d=0.100, Q0=0.02), resistance(L=250, d=0.080, Q0=0.01), resistance(L=400, d=0.080, Q0=0.01), ] loops = [[(0, 1), (1, 1), (2, 1)]] Q_init = [0.025, 0.008, 0.007] Q, iters = hardy_cross(pipe_R, loops, Q_init) print(f"迭代 {iters} 次收敛，管段流量 {[round(q, 5) for q in Q]} m3/s")loops里每个元组第二个值表示管段方向与环方向是否一致，这是符号容易搞错的地方：环方向顺时针定义后，与环同向取 +1，反向取 -1。迭代次数在 10 次以内收敛是正常水平，超过 30 次没收敛，先查初始流量有没有给成负值，再查是不是把不同管径的管段阻力系数写反了。
提示：
resistance里的 λ 是按设计流量算的定值，实际迭代中流量在变，λ 也会变。要求高时可以每轮迭代重算一次 λ，但低压庭院管网的流量波动范围小，固定 λ 的偏差通常在 3% 以内。
3.3 水力计算的关键参数取值表
参数取值直接决定结果能不能用。下面这组是按常见做法整理的，实际项目仍以设计院给定的工况表和 CJJ 33 为准。
| 参数 | 低压庭院管 | 中压 A 支管 | 中压 A 干管 | 说明 |
|---|---|---|---|---|
| 绝对粗糙度 K | 0.0001 m | 0.0001 m | 0.0002 m | 钢管焊接后取小值，老旧管网取大值 |
| 局部阻力占比 | 5%～10% | 10%～15% | 15%～20% | 按沿程阻力乘系数，不单独建局部节点 |
| 节点最小流量 | 0.5 m³/h | 2 m³/h | 5 m³/h | 低于此值的小用户可合并到邻近节点 |
| 计算温度 | 20 ℃ | 20 ℃ | 20 ℃ | 密度按对应温度修正 |
| 收敛容差 | 1e-9 m³/s | 1e-9 m³/s | 1e-9 m³/s | 单环网几十步内收敛 |
老旧管网粗糙度取大值，是因为内壁腐蚀和凝析液会显著增加阻力。有项目直接套用新管的 0.0001 m，算出来末端压力比实测高 8%，后来把 K 调到 0.0003 m 才对上。
3.4 用 SCADA 实测数据反查模型误差
模型建完必须拿实测数据校一遍，不然就是一张好看的图。做法是取调压站出口压力和几个装表节点的压力，按同一时刻和计算值比对，看相对误差的分布。
| 校验项 | 计算方式 | 可接受水平 | 超限时先查什么 |
|---|---|---|---|
| 节点压力相对误差 | (计算值 − 实测值)/实测值 | 优于 10% | 标高差是否录入、K 取值 |
| 管段流量误差 | (计算值 − 表计值)/表计值 | 优于 15% | 大用户流量是否漏录 |
| 全网压降误差 | 首末端压差对比 | 优于 12% | 局部阻力系数、管径录入错误 |
误差超限时，先查节点标高。低压管网里 10 米标高差对应约 0.08 kPa 的静压差，看着不大，但在末端压力只有 1.5 kPa 的工况下已经是 5% 的误差。
4. 城市燃气泄漏预警与工单闭环：报警规则怎么设才不误报
4.1 单点阈值报警为什么在低压管网里失灵
固定阈值报警的假设是压力稳定，而燃气管网恰恰不稳定。冬季晚高峰用气量大，末端压力自然下探到接近阈值；调压器在切换工作状态时会带来短时抖动；传感器漂移又会把基线整体抬高或压低。这三种情况混合在一起，报警量在冬季能翻三倍，调度员很快就不看了。
真正要区分的是三类变化：用气负荷引起的缓慢下探、调压器动作引起的短时抖动、破漏引起的持续下降。前两类都不该报警，第三类才是目标。区分依据不在单点值，而在下降速率、持续时长和流量是否脱离正常区间。
4.2 压力、流量、时间三类判据的滑动窗口实现
把三个判据做成与门，误报率能降一个量级。窗口长度取 15 分钟，是因为低压管网的破漏通常在 10 分钟内就能形成可观测的压降。
import pandas as pd def detect_leak(df, win="15min", drop_kpa=0.35, slope_kpa_min=0.02, min_flow=15.0, std_limit=0.05): """ df : 单测点时序，列为 ts / pressure_kpa / flow_m3h win: 滚动窗口长度 """ d = df.set_index("ts").sort_index() d["p_ma"] = d["pressure_kpa"].rolling(win, min_periods=5).mean() d["p_std"] = d["pressure_kpa"].rolling(win, min_periods=5).std() dt_min = d.index.to_series().diff().dt.total_seconds() / 60.0 d["slope"] = d["p_ma"].diff() / dt_min # kPa/min d["alarm"] = ( (d["p_ma"].diff(3) < -drop_kpa) & # 判据一：窗口内累计压降超阈值 (d["slope"] < -slope_kpa_min) & # 判据二：下降速率超阈值 (d["flow_m3h"] > min_flow) & # 判据三：排除夜间正常小流量 (d["p_std"] < std_limit) # 判据四：排除调压器动作的抖动 ) return d # 阈值整定参考：drop_kpa 取正常日波动幅度的 1.5 倍 # slope_kpa_min 取正常下探速率的 2 倍，min_flow 取片区夜均流量的 1.2 倍四个判据里，p_std这个条件最容易被忽略，但作用很直接：调压器动作时压力标准差会明显抬升，加这一条能把这类误报压掉大半。阈值整定不建议拍脑袋，取最近 30 天正常工况的分位数来定，drop_kpa用日波动幅度的 1.5 倍，min_flow用片区夜均流量的 1.2 倍，这两个数在不同片区差别很大。
注意：
min_flow不能直接抄别的片区。工商业用户占比高的片区夜均流量可能只有 3 m³/h，居民区反而更高，抄错会让判据三永远不成立。
4.3 报警到工单的自动派发与结果回流
预警只有传到人手里才算数。派单逻辑要解决的是别把同一个片区同时派给三个人，也别在没人空闲时把工单丢了。
INSERT INTO work_order (order_id, alarm_id, area_code, crew_id, created_at, status) SELECT gen_random_uuid(), a.alarm_id, a.area_code, (SELECT crew_id FROM crew WHERE area_code = a.area_code AND status = 'IDLE' ORDER BY last_order_at NULLS FIRST LIMIT 1), now(), 'DISPATCHED' FROM alarm a WHERE a.status = 'NEW' AND a.level >= 2 AND NOT EXISTS ( SELECT 1 FROM work_order w WHERE w.area_code = a.area_code AND w.status IN ('DISPATCHED', 'ACCEPTED') );NOT EXISTS那段是并发保护，同一片区只要有未完结工单就不再派新单。ORDER BY last_order_at NULLS FIRST让从没接过单的班组优先，避免老班组越派越多。
处置结果必须回流。现场填的"未见异常""阀门操作""确认泄漏"要写回alarm表，这些标记是后面调阈值的唯一依据。没有回流，阈值永远只能靠拍脑袋。
4.4 误报率与漏报率的量化评估
上线前先定指标，否则没法判断这套规则到底行不行。泄漏事件总数可以拿历史工单和抢修记录做标注集，一般能凑出上百条正样本。
| 指标 | 计算方式 | 目标值 | 说明 |
|---|---|---|---|
| 误报率 | 无效工单 / 总工单 | < 15% | 高于 25% 调度很快就不看了 |
| 漏报率 | 未识别事件 / 实际事件 | < 2% | 靠历史抢修记录做回溯统计 |
| 平均确认时长 | 派单到现场反馈 | < 20 min | 反映派单路径是否顺畅 |
| 一次派对率 | 无需转派的工单占比 | > 85% | 低于此值说明片区划分有问题 |
回溯验证时，把规则套到历史数据上跑一遍，统计每天的报警条数。日均报警条数超过 20 条而片区只有 3 个班组，基本可以判定阈值偏松。
5. 城市燃气数字化运营上线的调优技巧：从"能看"到"敢用"
新模型最忌讳直接切生产。稳妥的做法是影子运行：新规则和历史报警并行跑 30 天，两条线的差异逐日比对，差异集中的时段往往对应着某个片区的特殊工况，比如工业用户夜班生产、锅炉房季节性启停。
压力基线按季节和小时分别建，比固定阈值管用得多。取过去 14 天同一小时的压力 20 分位作为动态基线，用分位数而不是均值，能绕开个别异常点的干扰。
import pandas as pd def hourly_baseline(df, days=14, q=0.20): """按小时取压力低分位作为基线，返回 0~23 点的基线字典""" d = df.set_index("ts").sort_index() d = d[d.index > d.index.max() - pd.Timedelta(days=days)] base = ( d.groupby([d.index.hour, d.index.date])["pressure_kpa"] .quantile(q) .groupby(level=0) .median() ) return base.round(3).to_dict() # 报警阈值改成随基线浮动：低于基线 0.35 kPa 才进入判据一动态基线让阈值跟着季节走，冬季基线自然下移，报警量不会因为负荷变化而暴涨。上线前还有几项检查值得走一遍流程：
| 检查项 | 判据 | 常见失分点 |
|---|---|---|
| 数据断点 | 24 小时内缺测率 < 1% | 网关重启后未补传 |
| 时区一致 | 全链路统一 UTC 存储 | 展示层二次转换出错 |
| 模型误差 | 节点压力误差 < 10% | 节点标高漏录 |
| 报警压测 | 回灌 30 天历史数据 | 未测过峰值写入 |
| 派单并发 | 同片区不重复派单 | 缺并发保护条件 |
回灌压测最容易走过场，把 30 天历史数据按真实时间间隔重放一遍，能同时验证采集链路吞吐、报警规则稳定性和派单并发保护。我一般会在回灌时把时间间隔压缩到 1/10，看数据库写入延迟和压缩任务的资源占用，压缩策略配得不合理时，这一步会明显暴露出来。
还有一个容易被忽略的点：测点编码一旦对外发布就不要改。营收、计量、抢修几个系统都在引用，改一次编码要停一次数据服务。真要调整，走别名表过渡，让point_code保持稳定，新老编码在别名表里做映射。
本文还有配套的精品资源，点击获取
