1. 项目概述
三伏天车间里热得能煎鸡蛋,但环保监测再也不是让人头疼的问题。这套基于西门子CO在线监控的环境监测系统,彻底改变了传统人工记录的低效模式。作为一名在工业自动化领域摸爬滚打多年的工程师,我想分享这个项目的完整实现过程,从硬件选型到软件实现,再到实际应用中的那些"血泪教训"。
这个系统的核心价值在于实现了环境参数的实时监测、异常报警和数据分析。通过西门子S7-1200 PLC作为数据采集中枢,配合各类传感器和4G工业路由器,构建了一个稳定可靠的监测网络。系统上线后,不仅让环保检查变得轻松自如,还意外发现了不少生产过程中的隐藏问题。
2. 硬件系统搭建
2.1 核心硬件选型
选择硬件设备时,我们主要考虑三个因素:工业环境适应性、通讯协议兼容性和维护便利性。经过多次实地测试,最终确定的配置清单如下:
-
西门子S7-1214C DC/DC/DC PLC:这款PLC堪称工业界的"小钢炮",工作温度范围-20℃~60℃,完全适应车间恶劣环境。选择DC电源版本是为了避免交流电源干扰问题。
-
三合一环境传感器:集成了PM2.5、温度和湿度监测功能,采用Modbus RTU协议,测量范围PM2.5(0-1000μg/m³)、温度(-40~80℃)、湿度(0-100%RH)。特别选择了带防尘罩的工业级型号。
-
4G工业路由器:采用华为MH5000模组,支持全网通,内置看门狗功能,断网自动恢复。实测在车间最角落的厕所都能保持稳定连接。
-
定制控制柜:1.5mm厚冷轧钢板材质,IP54防护等级,内部配置了导轨安装座、接线端子和散热风扇。柜门上加装了触摸屏用于本地监控。
提示:传感器安装位置的选择直接影响数据准确性。我们通过多次测试,最终确定距离地面1.5米、远离门窗和通风口的位置为最佳监测点。
2.2 PLC数据块配置
PLC程序采用TIA Portal V16开发,数据块设计遵循以下原则:
- 相关参数集中存放
- 保留足够的扩展空间
- 变量命名直观易懂
核心数据块配置如下:
stl复制DATA_BLOCK "EnvMonitor_DB"
{ S7_Optimized_Access := 'TRUE' }
VERSION : 0.1
NON_RETAIN
VAR
// 环境数据区
Temperature : REAL; // 单位℃
Humidity : REAL; // 单位%
PM2_5 : INT; // 单位μg/m³
CO_Concentration : REAL; // 单位ppm
// 设备状态区
Sensor_Status : WORD; // 位0:PM2.5状态 位1:温度状态...
Comm_Error : BYTE; // 通讯错误码
TimeStamp : DT; // 数据时间戳
// 报警阈值区
PM2_5_Alarm : INT := 150; // PM2.5报警阈值
CO_Alarm : REAL := 30.0; // CO报警阈值
END_VAR
3. 数据采集与处理
3.1 PLC通讯连接实现
我们使用Python的python-snap7库与PLC建立通讯,关键点在于处理工业环境下的不稳定连接。以下是优化后的连接代码:
python复制import snap7
import time
from typing import Optional
class PLCConnector:
def __init__(self, ip: str, rack=0, slot=1):
self.ip = ip
self.rack = rack
self.slot = slot
self.plc = snap7.client.Client()
self._connection_timeout = 10
self._reconnect_interval = [1, 3, 5] # 重试间隔(秒)
def connect(self) -> bool:
for retry, interval in enumerate(self._reconnect_interval):
try:
self.plc.connect(self.ip, self.rack, self.slot)
if self.plc.get_connected():
print(f"成功连接到PLC {self.ip}")
return True
except Exception as e:
print(f"第{retry+1}次连接失败: {str(e)}")
time.sleep(interval)
print(f"无法连接到PLC {self.ip}")
return False
def read_env_data(self) -> Optional[dict]:
if not self.plc.get_connected():
if not self.connect():
return None
try:
# 读取DB2从字节0开始的22字节数据
data = self.plc.db_read(2, 0, 22)
# 使用snap7.util方法解析数据
return {
'temperature': snap7.util.get_real(data, 0),
'humidity': snap7.util.get_real(data, 4),
'pm25': snap7.util.get_int(data, 8),
'co': snap7.util.get_real(data, 10),
'status': snap7.util.get_word(data, 14),
'timestamp': snap7.util.get_dt(data, 16)
}
except Exception as e:
print(f"读取数据失败: {str(e)}")
self.plc.disconnect()
return None
3.2 数据清洗与校验
采集到的原始数据需要经过严格的质量控制流程:
- 范围校验:剔除明显超出物理可能的数值
- 突变检测:使用Z-score方法识别异常波动
- 时间对齐:补偿网络延迟造成的时间偏差
- 传感器补偿:根据温度对PM2.5读数进行修正
python复制import pandas as pd
import numpy as np
class DataCleaner:
@staticmethod
def basic_validation(df: pd.DataFrame) -> pd.DataFrame:
"""基础数据验证"""
df = df[
(df['PM2_5'] >= 0) & (df['PM2_5'] <= 1000) &
(df['Temperature'] >= -20) & (df['Temperature'] <= 80) &
(df['Humidity'] >= 0) & (df['Humidity'] <= 100)
]
return df.dropna()
@staticmethod
def detect_anomalies(df: pd.DataFrame, window=5, threshold=3) -> pd.DataFrame:
"""使用Z-score检测异常值"""
for col in ['PM2_5', 'Temperature', 'Humidity']:
rolling_mean = df[col].rolling(window=window).mean()
rolling_std = df[col].rolling(window=window).std()
df[f'{col}_Zscore'] = (df[col] - rolling_mean) / rolling_std
df = df[df[f'{col}_Zscore'].abs() < threshold]
return df
@staticmethod
def temperature_compensation(df: pd.DataFrame) -> pd.DataFrame:
"""温度补偿修正"""
df['PM2_5_Corrected'] = df['PM2_5'] * (
1 + 0.015 * (df['Temperature'] - 25) -
0.0002 * (df['Temperature'] - 25)**2
)
return df
4. 数据存储与可视化
4.1 时序数据库设计
我们选用InfluxDB 2.0作为时序数据库,其数据模型设计如下:
python复制from influxdb_client import InfluxDBClient, Point
from influxdb_client.client.write_api import SYNCHRONOUS
class InfluxDBWriter:
def __init__(self, url, token, org, bucket):
self.client = InfluxDBClient(url=url, token=token, org=org)
self.write_api = self.client.write_api(write_options=SYNCHRONOUS)
self.bucket = bucket
self.org = org
def write_env_data(self, data: dict):
p = Point("environment")
.tag("location", "workshop_A")
.tag("device", "monitor_01")
.field("temperature", data['temperature'])
.field("humidity", data['humidity'])
.field("pm25", data['pm25'])
.field("co", data['co'])
.time(data['timestamp'])
try:
self.write_api.write(bucket=self.bucket, org=self.org, record=p)
except Exception as e:
print(f"写入InfluxDB失败: {str(e)}")
4.2 可视化看板实现
使用Grafana构建的监控看板包含以下关键组件:
- 实时曲线图:展示各参数的实时变化趋势
- 状态指示器:用颜色直观显示是否超标
- 历史对比:支持选择不同时间段数据对比
- 报警统计:按天/周/月统计报警次数
python复制import plotly.graph_objects as go
from plotly.subplots import make_subplots
class DashboardGenerator:
@staticmethod
def create_realtime_figure(df: pd.DataFrame):
fig = make_subplots(rows=2, cols=1, shared_xaxes=True)
# 温湿度曲线
fig.add_trace(
go.Scatter(x=df['time'], y=df['temperature'], name="温度(℃)"),
row=1, col=1
)
fig.add_trace(
go.Scatter(x=df['time'], y=df['humidity'], name="湿度(%)"),
row=1, col=1
)
# PM2.5和CO曲线
fig.add_trace(
go.Scatter(x=df['time'], y=df['pm25'], name="PM2.5(μg/m³)"),
row=2, col=1
)
fig.add_trace(
go.Scatter(x=df['time'], y=df['co'], name="CO(ppm)"),
row=2, col=1
)
fig.update_layout(
height=800,
title_text="车间环境监测实时数据",
hovermode="x unified"
)
return fig
5. 报警与通知系统
5.1 多级报警策略
我们设计了三级报警机制:
- 提醒级:参数超过警戒值80%时,记录日志
- 警告级:超过标准值,触发现场声光报警
- 严重级:持续超标,自动通知相关负责人
报警规则配置示例:
yaml复制alarm_rules:
pm25:
warning: 75
critical: 150
duration: 300 # 持续5分钟超标才触发
co:
warning: 15
critical: 30
duration: 180
5.2 微信通知实现
通过企业微信API实现报警通知:
python复制import requests
import json
class WeChatNotifier:
def __init__(self, corpid, corpsecret, agentid):
self.base_url = "https://qyapi.weixin.qq.com"
self.corpid = corpid
self.corpsecret = corpsecret
self.agentid = agentid
self.access_token = None
self.token_expire = 0
def _get_token(self):
url = f"{self.base_url}/cgi-bin/gettoken"
params = {
"corpid": self.corpid,
"corpsecret": self.corpsecret
}
response = requests.get(url, params=params)
result = response.json()
self.access_token = result["access_token"]
self.token_expire = time.time() + result["expires_in"] - 300
def send_alert(self, user_ids, content):
if not self.access_token or time.time() > self.token_expire:
self._get_token()
url = f"{self.base_url}/cgi-bin/message/send"
params = {"access_token": self.access_token}
data = {
"touser": "|".join(user_ids),
"msgtype": "text",
"agentid": self.agentid,
"text": {"content": content},
"safe": 0
}
response = requests.post(url, params=params, json=data)
return response.json()
6. 系统部署与运维
6.1 现场安装要点
-
传感器安装:
- 避免靠近门窗、通风口
- 与污染源保持适当距离
- 多个传感器之间保持3米以上间距
-
电气接线:
- 信号线与动力线分开走线
- 使用屏蔽双绞线
- 做好接地处理
-
网络配置:
- 为工业设备划分独立VLAN
- 配置QoS保证监测数据优先传输
- 设置防火墙规则限制访问
6.2 日常维护经验
-
传感器校准:
- PM2.5传感器每3个月进行一次标准气体校准
- 温湿度传感器每年送检一次
-
数据备份策略:
- 实时数据保留3个月
- 小时均值保留1年
- 日均值保留5年
-
常见故障处理:
- 通讯中断:检查交换机、路由器状态
- 数据异常:检查传感器是否脏污
- PLC故障:检查程序循环时间是否过长
7. 项目收益与改进方向
系统上线后取得了显著效果:
- 环保检查响应时间从原来的2小时缩短到即时响应
- 设备异常停机时间减少37%
- 发现并解决了3处隐蔽的污染源
- 年节约人工检测成本约15万元
未来改进方向:
- 增加AI异常检测算法
- 与生产系统联动实现智能调节
- 扩展监测参数(如VOCs、噪声等)
- 部署移动监测终端
这套系统的成功实施证明,工业物联网技术在环境监测领域大有可为。通过合理的设计和扎实的实施,完全可以将枯燥的环保监测变成智能化、可视化的管理工具。