python的工业过程控制场景模拟第四十六篇:读取车间VOC监测数据，统计超标时长，生成环保合规报表。

车间 VOC 监测数据合规分析系统 —— 基于 OOP 的环保报表自动生成实战环保检查最怕的不是超标本身而是拿不出证据。环保局上门要看三个月的历史趋势、超标时长、整改措施——你翻出一堆 Excel 日志发现中间缺了好几天的数据传感器校准记录也找不到。这时候你就知道平时不做自动化合规报表检查时就是在赌运气。—— 哈尔滨工程大学《工业过程控制》课程核心思想延伸一、实际应用场景描述在化工、涂装、印刷、制药等行业挥发性有机物VOC排放受到严格法规管控。典型的车间 VOC 在线监测架构如下┌──────────────────────────────────────────────┐│ 车间 VOC 在线监测系统 ││ ││ 采样探头 ──→ 气相色谱/光离子化检测器 ││ │ ││ ▼ ││ 浓度数据 (mg/m³) ││ ↓ 4-20mA / RS485 / TCP ││ ││ ┌──────────────────────────────────────────┐ ││ │ 本地数采网关 │ ││ │ 采样周期: 1次/分钟 │ ││ │ 存储: 本地 SQLite 上传云端 │ ││ └──────────────────────────────────────────┘ ││ ││ 排放标准: GB 37822-2019 ││ 限值: 80 mg/m³ (非甲烷总烃) ││ 超标告警: 80 mg/m³ ││ 严重超标: 120 mg/m³ │└──────────────────────────────────────────────┘环保合规的核心要求合规事项 法规依据 具体要求排放限值 GB 37822-2019 NMHC ≤ 80 mg/m³监测频次 HJ 1013-2018 连续自动监测≥ 1 次/分钟数据留存 排污许可条例 原始数据保存 ≥ 5 年超标报告 环保部门要求 超标时段、峰值、持续时间、原因分析设备校准 HJ 1013-2018 定期校准记录可追溯哈尔滨工程大学《工业过程控制》课程在第十二章安全与环保监控系统中强调了监测数据的合规管理环保监测不仅是技术问题更是管理问题。一个合格的环保监控系统需要具备数据完整性校验、异常事件自动标记、以及符合法规要求的报表输出能力。缺失数据比超标数据更危险——因为它意味着你无法证明自己合规。二、引入痛点2.1 现场的真实困境场景 现场发生了什么 根因检查被罚 环保局说我们上个月超标 47 次但我们自己记录只有 3 次 统计口径不一致数据缺失 周末停机期间传感器断电数据断了两天 缺失数据未标记手工报表 每月花 3 天整理 Excel 图表 没有自动化工具校准混淆 传感器漂移导致假超标算不算违规 未区分有效/无效数据整改无据 超标了但不知道是哪个工段排放的 缺少分区溯源2.2 核心矛盾环保合规报表不是好看的趋势图而是具有法律效力的证据链。它需要回答四个问题什么时候超的标超了多少持续了多久原因是什么更重要的是它必须能区分真超标和传感器故障/校准/断电导致的异常数据。2.3 我们要解决什么用一段 Python 程序构建一个车间 VOC 监测数据合规分析系统实现1. 监测数据加载 —— 从 CSV 读取 VOC 浓度时序数据2. 数据有效性校验 —— 标记缺失、异常、校准期间的数据3. 超标统计 —— 分级统计超标次数、时长、峰值4. 合规报表生成 —— 按法规要求格式输出月度/季度报告5. 可视化趋势 —— 浓度曲线 限值线 超标标记6. 面向对象设计 —— 分层清晰可扩展三、核心逻辑讲解3.1 理论基础环保数据质量管理本工具基于哈工程《工业过程控制》第十二章安全与环保监控系统① 数据有效性判定有效数据条件:1. 浓度值在合理量程内 (0 ~ 200 mg/m³)2. 采样间隔正常 (≤ 2 分钟)3. 非校准/维护时段4. 非断电/通信中断时段无效数据标记:- MISSING: 数据缺失- CALIBRATION: 校准期间- MAINTENANCE: 设备维护- OUT_OF_RANGE: 超出量程② 超标分级级别 条件 监管要求正常 ≤ 80 mg/m³ 正常运行超标 80 ~ 120 mg/m³ 记录并报告严重超标 120 mg/m³ 立即停产整改③ 超标时长计算连续超标时段 从首次超标到恢复正常的连续时间段中断容忍: 如果中断 ≤ 5 分钟视为同一超标事件④ 合规指标指标 计算方式排放达标率 有效数据中达标点数 / 总有效点数 × 100%超标频次 统计期内超标事件次数最长连续超标 单次超标事件的最长持续时间累计超标时长 所有超标事件的时长之和3.2 系统数据流┌──────────────────────────────┐│ VOC 监测 CSV 数据 ││ (timestamp, voc_mg_m3, flag) │└──────────────┬───────────────┘│┌──────────────▼───────────────┐│ ① 数据加载 有效性校验 ││ 标记缺失/异常/校准 │└──────────────┬───────────────┘│┌──────────────▼───────────────┐│ ② 超标事件检测 ││ 滑动窗口 连续时段合并 │└──────────────┬───────────────┘│┌──────────────▼───────────────┐│ ③ 统计计算 ││ 达标率/频次/时长/峰值 │└──────────────┬───────────────┘│┌──────────────▼───────────────┐│ ④ 合规报表生成 ││ 法规格式 签名栏 │└──────────────┬───────────────┘│┌──────────────▼───────────────┐│ ⑤ 趋势可视化 ││ 浓度曲线 超标标记 │└──────────────────────────────┘四、代码讲解面向对象设计4.1 类结构总览类名 职责 设计模式VOCRecord 单条 VOC 监测记录dataclass 值对象ComplianceConfig 合规限值配置值对象 值对象DataFlag 数据状态枚举 枚举DataLoader CSV 数据加载与解析 封装DataValidator 数据有效性校验器 策略模式ExceedanceDetector 超标事件检测器 状态模式StatisticsCalculator 合规统计计算器 封装ReportGenerator 合规报表生成器 模板方法TrendVisualizer 趋势可视化器 封装VOCMonitoringSystem 系统编排器聚合根 聚合根4.2 数据模型层from dataclasses import dataclass, fieldfrom typing import List, Optional, Tuplefrom enum import Enum, autoimport numpy as npimport csvfrom pathlib import Pathfrom datetime import datetime, timedeltafrom collections import namedtupleclass DataFlag(Enum):数据状态标记VALID 有效MISSING 缺失OUT_OF_RANGE 超量程CALIBRATION 校准中MAINTENANCE 维护中SUSPECTED 可疑class ExceedanceLevel(Enum):超标级别NORMAL 正常EXCEED 超标SEVERE 严重超标dataclass(frozenTrue)class VOCRecord:单条 VOC 监测记录 —— 值对象timestamp: datetime # 采样时间concentration: float # VOC 浓度 (mg/m³)flag: DataFlag DataFlag.VALIDsensor_id: str VOC_01zone: str Zone_Adataclass(frozenTrue)class ComplianceConfig:合规限值配置limit_normal: float 80.0 # 排放标准限值 (mg/m³)limit_severe: float 120.0 # 严重超标限值 (mg/m³)sampling_interval: int 60 # 正常采样间隔 (秒)max_gap: int 300 # 最大允许中断时间 (秒, 5分钟)min_valid_rate: float 0.75 # 最低有效数据率 (75%)reporting_period: str monthly # 报告周期4.3 数据加载器class DataLoader:VOC 监测数据加载器CSV 格式:timestamp,concentration,sensor_id,zone,flag2024-03-01 08:00:00,45.2,VOC_01,Zone_A,VALID2024-03-01 08:01:00,52.8,VOC_01,Zone_A,VALID...支持多种时间格式和标记字段def __init__(self):self.records: List[VOCRecord] []def load_csv(self, file_path: str, flag_col: str flag) - List[VOCRecord]:从 CSV 加载数据Args:file_path: CSV 文件路径flag_col: 状态标记列名Returns:记录列表self.records.clear()with open(file_path, r, encodingutf-8) as f:reader csv.DictReader(f)for row in reader:# 解析时间try:ts datetime.strptime(row[timestamp], %Y-%m-%d %H:%M:%S)except ValueError:continue# 解析浓度try:conc float(row[concentration])except ValueError:conc -1.0 # 标记为无效# 解析标记flag_str row.get(flag_col, VALID).strip().upper()flag self._parse_flag(flag_str)record VOCRecord(timestampts,concentrationconc,flagflag,sensor_idrow.get(sensor_id, VOC_01),zonerow.get(zone, Zone_A))self.records.append(record)return self.recordsdef _parse_flag(self, flag_str: str) - DataFlag:解析标记字符串mapping {VALID: DataFlag.VALID,MISSING: DataFlag.MISSING,OUT_OF_RANGE: DataFlag.OUT_OF_RANGE,CALIBRATION: DataFlag.CALIBRATION,MAINTENANCE: DataFlag.MAINTENANCE,SUSPECTED: DataFlag.SUSPECTED}return mapping.get(flag_str, DataFlag.VALID)4.4 数据有效性校验器class DataValidator:数据有效性校验器 —— 策略模式校验规则:1. 浓度值必须在合理范围内2. 采样间隔不能超过阈值3. 标记为非 VALID 的数据直接排除def __init__(self, config: ComplianceConfig):self.cfg configdef validate(self, records: List[VOCRecord]) - List[VOCRecord]:执行有效性校验Args:records: 原始记录列表Returns:有效记录列表valid []for i, r in enumerate(records):# 规则1: 标记必须是 VALIDif r.flag ! DataFlag.VALID:continue# 规则2: 浓度值合理 (0 且 200)if r.concentration 0 or r.concentration 200:continue# 规则3: 采样间隔检查 (如果不是第一条)if i 0:interval (r.timestamp - records[i-1].timestamp).total_seconds()if interval self.cfg.max_gap * 2: # 允许2倍间隔continuevalid.append(r)return validdef check_completeness(self, records: List[VOCRecord],expected_count: int) - float:检查数据完整率Args:records: 有效记录列表expected_count: 期望记录数Returns:完整率 (0~1)if expected_count 0:return 0.0return min(1.0, len(records) / expected_count)4.5 超标事件检测器核心算法class ExceedanceDetector:超标事件检测器 —— 状态模式检测逻辑:1. 遍历有效数据标记每点的超标级别2. 连续超标时段合并中断容忍 ≤ max_gap3. 输出超标事件列表超标事件 namedtuple(Event, [start, end, level, peak, duration])Event namedtuple(Event, [start, end, level, peak, duration, avg_conc])def __init__(self, config: ComplianceConfig):self.cfg configself.events: List[namedtuple] []def detect(self, records: List[VOCRecord]) - List[namedtuple]:检测所有超标事件Args:records: 有效记录列表Returns:超标事件列表self.events.clear()if not records:return []current_event Nonefor i, r in enumerate(records):level self._classify(r.concentration)if level ! ExceedanceLevel.NORMAL:# 超标中if current_event is None:# 新事件开始current_event {start: r.timestamp,end: r.timestamp,level: level,peak: r.concentration,readings: [r.concentration]}else:# 更新当前事件current_event[end] r.timestampcurrent_event[peak] max(current_event[peak], r.concentration)current_event[readings].append(r.concentration)else:# 正常状态if current_event is not None:# 检查是否与下一个超标点间隔太大if i len(records) - 1:gap (records[i1].timestamp - r.timestamp).total_seconds()if gap self.cfg.max_gap:# 结束当前事件duration (current_event[end] - current_event[start]).total_seconds()avg np.mean(current_event[readings])event self.Event(startcurrent_event[start],endcurrent_event[end],levelcurrent_event[level],peakround(current_event[peak], 2),durationround(duration / 60, 2), # 分钟avg_concround(avg, 2))self.events.append(event)current_event None# 处理最后一个事件if current_event is not None:duration (current_event[end] - current_event[start]).total_seconds()avg np.mean(current_event[readings])event self.Event(startcurrent_event[start],endcurrent_event[end],levelcurrent_event[level],peakround(current_event[peak], 2),durationround(duration / 60, 2),avg_concround(avg, 2))self.events.append(event)return self.eventsdef _classify(self, concentration: float) - ExceedanceLevel:浓度分级if concentration self.cfg.limit_severe:return ExceedanceLevel.SEVEREelif concentration self.cfg.limit_normal:return ExceedanceLevel.EXCEEDelse:return ExceedanceLevel.NORMAL4.6 合规统计计算器class StatisticsCalculator:合规统计计算器计算:- 排放达标率- 超标频次分级别- 累计超标时长- 最长连续超标时长- 峰值浓度def __init__(self, config: ComplianceConfig):self.cfg configdef calculate(self, records: List[VOCRecord],events: List[ExceedanceDetector.Event]) - dict:计算所有合规指标Args:records: 有效记录列表events: 超标事件列表Returns:统计结果字典if not records:return self._empty_stats()total len(records)exceed_count sum(1 for r in records if r.concentration self.cfg.limit_normal)severe_count sum(1 for r in records if r.concentration self.cfg.limit_severe)compliance_rate (total - exceed_count) / total * 100.0# 按级别统计事件exceed_events [e for e in events if e.level ExceedanceLevel.EXCEED]severe_events [e for e in events if e.level ExceedanceLevel.SEVERE]# 时长统计total_exceed_minutes sum(e.duration for e in exceed_events)total_severe_minutes sum(e.duration for e in severe_events)max_single_duration max((e.duration for e in events), default0.0)# 峰值peak_conc max((r.concentration for r in records), default0.0)peak_time next((r.timestamp for r in records if r.concentration peak_conc), None)# 按小时统计超标频次用于趋势分析hourly_exceed self._hourly_stats(records)return {total_readings: total,valid_readings: total,compliance_rate: round(compliance_rate, 2),exceed_count: len(exceed_events),severe_count: len(severe_events),total_exceed_minutes: round(total_exceed_minutes, 1),total_severe_minutes: round(total_severe_minutes, 1),max_single_duration: round(max_single_duration, 1),peak_concentration: round(peak_conc, 2),peak_time: peak_time.strftime(%Y-%m-%d %H:%M) if peak_time else N/A,hourly_exceed: hourly_exceed}def _hourly_stats(self, records: List[VOCRecord]) - dict:按小时统计超标次数hourly {}for r in records:hour_key r.timestamp.strftime(%Y-%m-%d %H:00)if hour_key not in hourly:hourly[hour_key] {total: 0, exceed: 0}hourly[hour_key][total] 1if r.concentration self.cfg.limit_normal:hourly[hour_key][exceed] 1return hourlydef _empty_stats(self) - dict:return {total_readings: 0, valid_readings: 0, compliance_rate: 0.0,exceed_count: 0, severe_count: 0, total_exceed_minutes: 0,total_severe_minutes: 0, max_single_duration: 0,peak_concentration: 0, peak_time: N/A, hourly_exceed: {}}4.7 合规报表生成器class ReportGenerator:合规报表生成器 —— 模板方法模式生成符合环保部门要求的月度/季度报告def generate_monthly_report(self, stats: dict, events: List,period: str 2024年3月) - str:生成月度合规报告Args:stats: StatisticsCalculator.calculate() 的结果events: 超标事件列表period: 报告期Returns:格式化报告文本lines [ * 65,f VOC 排放合规监测月报 ({period}), * 65,, 【基本信息】,f 监测点位: Zone_A VOC在线监测仪,f 排放限值: {stats.get(normal_limit, 80.0)} mg/m³,f 严重超标限值: {stats.get(severe_limit, 120.0)} mg/m³,f 有效数据量: {stats[valid_readings]} 条,, 【合规概况】,f 排放达标率: {stats[compliance_rate]}%,f 超标事件数: {stats[exceed_count]} 次,f 严重超标事件: {stats[severe_count]} 次,f 累计超标时长: {stats[total_exceed_minutes]} 分钟,f 最长单次超标: {stats[max_single_duration]} 分钟,, 【峰值信息】,f 最高浓度: {stats[peak_concentration]} mg/m³,f 出现时间: {stats[peak_time]},,]# 超标事件明细if events:lines.append( 【超标事件明细】)lines.append( - * 57)lines.append(f {序号:4} {开始时间:16} {结束时间:16} {级别:10} {峰值:8})lines.append( - * 57)for i, e in enumerate(events[:20]): # 最多显示20条start_str e.start.strftime(%m-%d %H:%M)end_str e.end.strftime(%m-%d %H:%M)lines.append(f #{i1:3} {start_str:16} {end_str:16} {e.level.value:10} {e.peak:8.1f})if len(events) 20:lines.append(f ... 还有 {len(events)-20} 条记录未显示)else:lines.append( 【超标事件明细】)lines.append( ✓ 本月无超标事件)lines.append()lines.append( 【备注】)lines.append( 本报告由 VOC 合规分析系统自动生成)lines.append( 操作人员签字: _______________)lines.append( 审核人员签字: _______________)lines.append( * 65)return \n.join(lines)4.8 趋势可视化器class TrendVisualizer:趋势可视化器生成:- 浓度时间序列图- 限值参考线- 超标区域着色def plot(self, records: List[VOCRecord], config: ComplianceConfig,output_path: str voc_trend.png):绘制 VOC 浓度趋势图Args:records: 有效记录列表config: 合规配置output_path: 输出图片路径try:import matplotlib.pyplot as pltimport matplotlib.dates as mdatestimes [r.timestamp for r in records]concentrations [r.concentration for r in records]fig, ax plt.subplots(figsize(14, 6))# 绘制浓度曲线ax.plot(times, concentrations, b-, linewidth0.8, labelVOC 浓度, alpha0.8)# 限值线ax.axhline(yconfig.limit_normal, colororange, linestyle--,linewidth1.5, labelf排放限值 ({config.limit_normal} mg/m³))ax.axhline(yconfig.limit_severe, colorred, linestyle--,linewidth1.5, labelf严重超标线 ({config.limit_severe} mg/m³))# 超标区域着色exceed_mask np.array(concentrations) config.limit_normalif any(exceed_mask):ax.fill_between(times, 0, concentrations, whereexceed_mask,colorred, alpha0.3, label超标区域)# 格式化ax.set_xlabel(时间)ax.set_ylabel(VOC 浓度 (mg/m³))ax.set_title(车间 VOC 浓度监测趋势)ax.legend(locupper right)ax.grid(True, alpha0.3)ax.xaxis.set_major_formatter(mdates.DateFormatter(%m-%d %H:%M))plt.xticks(rotation45)plt.tight_layout()plt.savefig(output_path, dpi150)plt.close()except ImportError:print(⚠️ matplotlib 未安装跳过可视化利用AI解决实际问题如果你觉得这个工具好用欢迎关注长安牧笛