企业官网建设流程全解析

车间 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, field

from typing import List, Optional, Tuple

from enum import Enum, auto

import numpy as np

import csv

from pathlib import Path

from datetime import datetime, timedelta

from collections import namedtuple

class DataFlag(Enum):

"""数据状态标记"""

VALID = "有效"

MISSING = "缺失"

OUT_OF_RANGE = "超量程"

CALIBRATION = "校准中"

MAINTENANCE = "维护中"

SUSPECTED = "可疑"

class ExceedanceLevel(Enum):

"""超标级别"""

NORMAL = "正常"

EXCEED = "超标"

SEVERE = "严重超标"

@dataclass(frozen=True)

class VOCRecord:

"""单条 VOC 监测记录 —— 值对象"""

timestamp: datetime # 采样时间

concentration: float # VOC 浓度 (mg/m³)

flag: DataFlag = DataFlag.VALID

sensor_id: str = "VOC_01"

zone: str = "Zone_A"

@dataclass(frozen=True)

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,flag

2024-03-01 08:00:00,45.2,VOC_01,Zone_A,VALID

2024-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', encoding='utf-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(

timestamp=ts,

concentration=conc,

flag=flag,

sensor_id=row.get('sensor_id', 'VOC_01'),

zone=row.get('zone', 'Zone_A')

)

self.records.append(record)

return self.records

def _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 = config

def validate(self, records: List[VOCRecord]) -> List[VOCRecord]:

"""

执行有效性校验

Args:

records: 原始记录列表

Returns:

有效记录列表

"""

valid = []

for i, r in enumerate(records):

# 规则1: 标记必须是 VALID

if 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倍间隔

continue

valid.append(r)

return valid

def check_completeness(self, records: List[VOCRecord],

expected_count: int) -> float:

"""

检查数据完整率

Args:

records: 有效记录列表

expected_count: 期望记录数

Returns:

完整率 (0~1)

"""

if expected_count == 0:

return 0.0

return min(1.0, len(records) / expected_count)

4.5 超标事件检测器（核心算法）

class ExceedanceDetector:

"""

超标事件检测器 —— 状态模式

检测逻辑:

1. 遍历有效数据，标记每点的超标级别

2. 连续超标时段合并（中断容忍 ≤ max_gap）

3. 输出超标事件列表

超标事件 = namedtuple('Event', ['start', 'end', 'level', 'peak', 'duration'])

"""

Event = namedtuple('Event', ['start', 'end', 'level', 'peak', 'duration', 'avg_conc'])

def __init__(self, config: ComplianceConfig):

self.cfg = config

self.events: List[namedtuple] = []

def detect(self, records: List[VOCRecord]) -> List[namedtuple]:

"""

检测所有超标事件

Args:

records: 有效记录列表

Returns:

超标事件列表

"""

self.events.clear()

if not records:

return []

current_event = None

for 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.timestamp

current_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[i+1].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(

start=current_event['start'],

end=current_event['end'],

level=current_event['level'],

peak=round(current_event['peak'], 2),

duration=round(duration / 60, 2), # 分钟

avg_conc=round(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(

start=current_event['start'],

end=current_event['end'],

level=current_event['level'],

peak=round(current_event['peak'], 2),

duration=round(duration / 60, 2),

avg_conc=round(avg, 2)

)

self.events.append(event)

return self.events

def _classify(self, concentration: float) -> ExceedanceLevel:

"""浓度分级"""

if concentration > self.cfg.limit_severe:

return ExceedanceLevel.SEVERE

elif concentration > self.cfg.limit_normal:

return ExceedanceLevel.EXCEED

else:

return ExceedanceLevel.NORMAL

4.6 合规统计计算器

class StatisticsCalculator:

"""

合规统计计算器

计算:

- 排放达标率

- 超标频次（分级别）

- 累计超标时长

- 最长连续超标时长

- 峰值浓度

"""

def __init__(self, config: ComplianceConfig):

self.cfg = config

def 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), default=0.0)

# 峰值

peak_conc = max((r.concentration for r in records), default=0.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'] += 1

if r.concentration > self.cfg.limit_normal:

hourly[hour_key]['exceed'] += 1

return hourly

def _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" #{i+1:<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 plt

import matplotlib.dates as mdates

times = [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-', linewidth=0.8, label='VOC 浓度', alpha=0.8)

# 限值线

ax.axhline(y=config.limit_normal, color='orange', linestyle='--',

linewidth=1.5, label=f'排放限值 ({config.limit_normal} mg/m³)')

ax.axhline(y=config.limit_severe, color='red', linestyle='--',

linewidth=1.5, label=f'严重超标线 ({config.limit_severe} mg/m³)')

# 超标区域着色

exceed_mask = np.array(concentrations) > config.limit_normal

if any(exceed_mask):

ax.fill_between(times, 0, concentrations, where=exceed_mask,

color='red', alpha=0.3, label='超标区域')

# 格式化

ax.set_xlabel('时间')

ax.set_ylabel('VOC 浓度 (mg/m³)')

ax.set_title('车间 VOC 浓度监测趋势')

ax.legend(loc='upper right')

ax.grid(True, alpha=0.3)

ax.xaxis.set_major_formatter(mdates.DateFormatter('%m-%d %H:%M'))

plt.xticks(rotation=45)

plt.tight_layout()

plt.savefig(output_path, dpi=150)

plt.close()

except ImportError:

print("⚠️ matplotlib 未安装，跳过可视化"

利用AI解决实际问题，如果你觉得这个工具好用，欢迎关注长安牧笛！