
关键词:恒温恒湿净化设备、组态联动、温湿度阈值、自动启停逻辑、Modbus TCP、SCADA、环境控制、防抖算法、死区控制 标签:#物联网 #Modbus #TCP/IP #UDP #POE供电 #以太网温湿度传感器 #网口温湿度变送器 #机房监控
在前几篇中,我们已经完成了以太网温湿度传感器的硬件选型、固件开发、网络架构和点位布设。采集到数据之后,下一步就是让数据驱动控制——当库房温湿度偏离设定范围时,自动启停恒温恒湿净化设备,维持环境稳定。
传统做法是人工巡检+手动开关,存在明显短板:
维度 | 人工控制 | 组态联动自动控制 |
|---|---|---|
响应速度 | 分钟~小时级 | 秒级 |
控制精度 | 依赖人员经验 | 可精确到 ±0.1℃ / ±1%RH |
运行时间 | 固定时段 | 按需启停,节能 |
可追溯性 | 纸质记录 | 完整日志+趋势分析 |
多设备协同 | 难以协调 | 统一调度,避免冲突 |
异常发现 | 滞后 | 实时告警 |
核心需求:基于实时温湿度数据,通过可编程逻辑实现多台恒温恒湿净化设备的自动启停、轮换运行和故障保护。
┌─────────────────────────────────────────────────────────────────────┐
│ 组态监控层(SCADA/HMI) │
├─────────────────────────────────────────────────────────────────────┤
│ ┌─────────────────┐ ┌─────────────────┐ ┌────────────────────┐ │
│ │ 实时数据库 │ │ 报警引擎 │ │ 趋势分析与报表 │ │
│ │ (Redis/InfluxDB)│ │ (阈值判定) │ │ (历史数据) │ │
│ └────────┬────────┘ └────────┬────────┘ └─────────┬──────────┘ │
│ │ │ │ │
│ ┌────────▼─────────────────────▼──────────────────────▼──────────┐│
│ │ 联动逻辑引擎(核心控制层) ││
│ │ ┌─────────────┐ ┌─────────────┐ ┌────────────────────────┐ ││
│ │ │ 阈值判定模块 │ │ 防抖/死区 │ │ 设备调度模块 │ ││
│ │ │ (比较逻辑) │ │ (滤波算法) │ │ (轮值/互斥/优先级) │ ││
│ │ └─────────────┘ └─────────────┘ └────────────────────────┘ ││
│ │ ┌─────────────────────────────────────────────────────────────┐││
│ │ │ 通信协议适配层 │││
│ │ │ Modbus TCP (南向: 设备) | Modbus TCP (北向: SCADA) │││
│ │ │ SNMP (传感器) | MQTT (云平台) │││
│ │ └─────────────────────────────────────────────────────────────┘││
│ └───────────────────────────────┬─────────────────────────────────┘│
│ │ │
│ ┌───────────────────────────────▼─────────────────────────────────┐│
│ │ 设备控制层(PLC / 嵌入式控制器) ││
│ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌───────────────────┐ ││
│ │ │ 精密空调 │ │ 除湿机 │ │ 加湿器 │ │ 净化/新风系统 │ ││
│ │ │ (AC-01) │ │ (DEH-01) │ │ (HUM-01) │ │ (AIR-01) │ ││
│ │ └──────────┘ └──────────┘ └──────────┘ └───────────────────┘ ││
│ └──────────────────────────────────────────────────────────────────┘│
│ │
│ ┌──────────────────────────────────────────────────────────────────┐│
│ │ 数据采集层(传感器网络) ││
│ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌───────────────────┐ ││
│ │ │ 传感器01 │ │ 传感器02 │ │ 传感器03 │ │ ... 传感器N │ ││
│ │ │ (UDP+SNMP)│ │ (UDP+SNMP)│ │ (UDP+SNMP)│ │ (UDP+SNMP) │ ││
│ │ └──────────┘ └──────────┘ └──────────┘ └───────────────────┘ ││
│ └──────────────────────────────────────────────────────────────────┘│
└─────────────────────────────────────────────────────────────────────┘部署方式 | 适用场景 | 优点 | 缺点 |
|---|---|---|---|
PLC 内置逻辑 | 小型系统(<10台设备) | 实时性强,可靠性高 | 逻辑修改需重新编程 |
边缘网关 | 中型系统(10~50台设备) | 灵活,支持多种协议 | 网关故障影响全局 |
SCADA 软逻辑 | 大型系统(>50台设备) | 可视化组态,易维护 | 依赖上位机运行 |
混合模式 | 关键系统 | PLC做硬保护,SCADA做优化调度 | 复杂度高 |
推荐方案:PLC 负责底层设备保护和紧急停机逻辑(硬实时),边缘网关/SCADA 负责基于温湿度阈值的自动启停调度(软实时)。
最基础的阈值模型是双阈值(上限/下限):
温度控制模型:
T_high_alarm = 24℃ → 触发高温告警,启动制冷
T_high = 23.5℃ → 启动制冷阈值(低于此值停止制冷)
T_low = 14.5℃ → 启动制热阈值
T_low_alarm = 14℃ → 触发低温告警,启动制热
湿度控制模型:
H_high_alarm = 60%RH → 触发高湿告警,启动除湿
H_high = 58%RH → 启动除湿阈值
H_low = 47%RH → 启动加湿阈值
H_low_alarm = 45%RH → 触发低湿告警,启动加湿死区是避免设备频繁启停的关键参数:
死区宽度 = |T_high - T_high_alarm| = |23.5 - 24| = 0.5℃
工作原理:
当 T = 24.1℃ → 启动制冷
当 T = 23.4℃ → 停止制冷
在 23.5℃ ~ 24℃ 之间,设备保持当前状态不变
这 0.5℃ 的死区防止了温度在阈值附近波动时设备频繁切换死区宽度选择:
参数 | 推荐死区 | 理由 |
|---|---|---|
温度 | 0.5~1.0℃ | 精密空调启停一次约5分钟,死区太小会导致频繁启停 |
湿度 | 2~3%RH | 湿度变化比温度慢,死区可以稍大 |
露点 | 1.0℃ | 露点变化与温度相关,需匹配温度死区 |
单个传感器故障可能导致误动作,采用多数表决 + 异常剔除策略:
"""
multi_sensor_decision.py - 多传感器融合判定
"""
import numpy as np
from typing import List, Dict
class MultiSensorDecision:
"""
基于多个传感器的温湿度数据,做出控制决策
"""
def __init__(self, temp_high=23.5, temp_low=14.5,
hum_high=58.0, hum_low=47.0,
outlier_threshold=2.0):
self.temp_high = temp_high
self.temp_low = temp_low
self.hum_high = hum_high
self.hum_low = hum_low
self.outlier_threshold = outlier_threshold # 偏离中位数的最大允许偏差
def decide_temperature(self, readings: List[float]) -> Dict:
"""
基于多传感器温度读数做决策
Args:
readings: 多个传感器的温度读数列表
Returns:
{
'action': 'cool' | 'heat' | 'none',
'valid_count': 有效传感器数量,
'median': 中位数,
'outliers': 异常传感器索引列表
}
"""
if not readings:
return {'action': 'none', 'valid_count': 0, 'median': None, 'outliers': []}
# 异常剔除(基于中位数绝对偏差 MAD)
median = np.median(readings)
mad = np.median(np.abs(np.array(readings) - median))
valid_readings = []
outliers = []
for i, r in enumerate(readings):
if mad > 0 and abs(r - median) > self.outlier_threshold * mad:
outliers.append(i)
else:
valid_readings.append(r)
if not valid_readings:
return {'action': 'none', 'valid_count': 0, 'median': median, 'outliers': outliers}
# 使用有效读数的中位数做决策
decision_temp = np.median(valid_readings)
if decision_temp > self.temp_high:
action = 'cool'
elif decision_temp < self.temp_low:
action = 'heat'
else:
action = 'none'
return {
'action': action,
'valid_count': len(valid_readings),
'median': decision_temp,
'outliers': outliers
}
def decide_humidity(self, readings: List[float]) -> Dict:
"""湿度决策,逻辑与温度相同"""
if not readings:
return {'action': 'none', 'valid_count': 0, 'median': None, 'outliers': []}
median = np.median(readings)
mad = np.median(np.abs(np.array(readings) - median))
valid_readings = []
outliers = []
for i, r in enumerate(readings):
if mad > 0 and abs(r - median) > self.outlier_threshold * mad:
outliers.append(i)
else:
valid_readings.append(r)
if not valid_readings:
return {'action': 'none', 'valid_count': 0, 'median': median, 'outliers': outliers}
decision_hum = np.median(valid_readings)
if decision_hum > self.hum_high:
action = 'dehumidify'
elif decision_hum < self.hum_low:
action = 'humidify'
else:
action = 'none'
return {
'action': action,
'valid_count': len(valid_readings),
'median': decision_hum,
'outliers': outliers
}
# 使用示例
if __name__ == "__main__":
decision_maker = MultiSensorDecision()
# 模拟6个传感器的温度读数(℃)
temp_readings = [23.8, 24.1, 23.9, 24.0, 23.7, 35.2] # 最后一个为异常值
result = decision_maker.decide_temperature(temp_readings)
print(f"温度决策: {result['action']}, 有效传感器: {result['valid_count']}, "
f"中位数: {result['median']:.1f}℃, 异常传感器: {result['outliers']}")
# 湿度读数
hum_readings = [57.5, 58.2, 57.8, 58.0, 57.6, 57.9]
result = decision_maker.decide_humidity(hum_readings)
print(f"湿度决策: {result['action']}, 有效传感器: {result['valid_count']}, "
f"中位数: {result['median']:.1f}%RH")防止传感器瞬时波动导致设备频繁启停:
"""
debounce.py - 时间防抖算法
"""
import time
from enum import Enum
from dataclasses import dataclass
class DeviceState(Enum):
OFF = 0
ON = 1
@dataclass
class DebounceConfig:
on_delay: float # 满足条件后延迟多久才启动(秒)
off_delay: float # 不满足条件后延迟多久才停止(秒)
min_run_time: float = 300 # 最短运行时间(秒),防止频繁启停
min_off_time: float = 120 # 最短停机时间(秒)
class DebounceController:
"""
时间防抖控制器
"""
def __init__(self, config: DebounceConfig, name: str = ""):
self.config = config
self.name = name
self.current_state = DeviceState.OFF
self.target_state = DeviceState.OFF
self.state_change_time = 0
self.last_state_change = 0
def update(self, should_be_on: bool, current_time: float = None) -> DeviceState:
"""
更新控制状态
Args:
should_be_on: 基于阈值的判定结果(True=应该启动,False=应该停止)
current_time: 当前时间戳
Returns:
实际应输出给设备的状态
"""
if current_time is None:
current_time = time.time()
# 更新目标状态
if should_be_on:
if self.target_state != DeviceState.ON:
self.target_state = DeviceState.ON
self.state_change_time = current_time
else:
if self.target_state != DeviceState.OFF:
self.target_state = DeviceState.OFF
self.state_change_time = current_time
# 检查是否满足状态切换条件
time_in_target = current_time - self.state_change_time
time_since_last_change = current_time - self.last_state_change
if self.target_state == DeviceState.ON and self.current_state == DeviceState.OFF:
# 需要启动
if time_in_target >= self.config.on_delay:
# 检查最短停机时间
if time_since_last_change >= self.config.min_off_time:
self.current_state = DeviceState.ON
self.last_state_change = current_time
print(f"[{self.name}] 启动 (延迟 {time_in_target:.0f}s)")
elif self.target_state == DeviceState.OFF and self.current_state == DeviceState.ON:
# 需要停止
if time_in_target >= self.config.off_delay:
# 检查最短运行时间
if time_since_last_change >= self.config.min_run_time:
self.current_state = DeviceState.OFF
self.last_state_change = current_time
print(f"[{self.name}] 停止 (延迟 {time_in_target:.0f}s)")
return self.current_state
# 使用示例
if __name__ == "__main__":
config = DebounceConfig(
on_delay=60, # 满足条件60秒后才启动(防止瞬时波动)
off_delay=120, # 不满足条件120秒后才停止(保持稳定运行)
min_run_time=300, # 最短运行5分钟
min_off_time=120 # 最短停机2分钟
)
controller = DebounceController(config, name="精密空调-01")
# 模拟时间线
t = 0
should_on_sequence = [
(0, False), # 初始:关闭
(30, True), # 30s: 温度超标,请求启动
(60, True), # 60s: 仍超标
(61, False), # 61s: 温度恢复(瞬时波动)
(90, True), # 90s: 再次超标
(150, True), # 150s: 持续超标,应该启动
(400, False), # 400s: 温度恢复,请求停止
(520, False), # 520s: 持续正常,应该停止
]
for ts, should_on in should_on_sequence:
t = ts
state = controller.update(should_on, t)
print(f" t={t}s: should_on={should_on}, actual_state={state.name}")对传感器原始数据进行平滑处理:
"""
window_filter.py - 滑动窗口滤波
"""
import numpy as np
from collections import deque
class SlidingWindowFilter:
"""
滑动窗口滤波器,去除传感器噪声
"""
def __init__(self, window_size=10, method='median'):
"""
Args:
window_size: 窗口大小(样本数)
method: 'mean' | 'median' | 'trimmed'(去极值均值)
"""
self.window_size = window_size
self.method = method
self.buffer = deque(maxlen=window_size)
def add_sample(self, value: float) -> float:
"""
添加新样本,返回滤波后的值
"""
self.buffer.append(value)
if len(self.buffer) < self.window_size // 2:
return value # 数据不足时返回原始值
data = np.array(self.buffer)
if self.method == 'mean':
return float(np.mean(data))
elif self.method == 'median':
return float(np.median(data))
elif self.method == 'trimmed':
# 去掉最高和最低10%的值后求平均
trimmed = np.sort(data)[int(len(data)*0.1):int(len(data)*0.9)]
return float(np.mean(trimmed))
else:
return float(data[-1]) # 返回最新值多台相同设备需要轮流工作,延长寿命:
"""
rotation_scheduler.py - 设备轮值调度
"""
import time
from typing import List, Dict
from dataclasses import dataclass, field
@dataclass
class Device:
id: str
name: str
total_run_time: float = 0.0 # 累计运行时间(秒)
last_start_time: float = 0.0
is_running: bool = False
priority: int = 0 # 优先级(数字越小优先级越高)
class RotationScheduler:
"""
多设备轮值调度器
"""
def __init__(self, devices: List[Device], max_concurrent: int = 2):
"""
Args:
devices: 设备列表
max_concurrent: 最大同时运行数量
"""
self.devices = devices
self.max_concurrent = max_concurrent
self._sorted_by_runtime = sorted(devices, key=lambda d: d.total_run_time)
def select_devices(self, need_count: int) -> List[Device]:
"""
选择需要启动的设备
Args:
need_count: 需要启动的设备数量
Returns:
选中的设备列表
"""
# 按累计运行时间排序(运行最少的优先)
sorted_devices = sorted(self.devices, key=lambda d: d.total_run_time)
# 从运行最少的设备中选择
selected = []
for dev in sorted_devices:
if len(selected) >= need_count:
break
if not dev.is_running: # 不重复启动已在运行的设备
selected.append(dev)
return selected
def update_runtime(self, current_time: float):
"""
更新所有运行设备的累计运行时间
"""
for dev in self.devices:
if dev.is_running and dev.last_start_time > 0:
dev.total_run_time += current_time - dev.last_start_time
dev.last_start_time = current_time
def start_device(self, dev: Device, current_time: float):
"""启动设备"""
dev.is_running = True
dev.last_start_time = current_time
def stop_device(self, dev: Device, current_time: float):
"""停止设备"""
if dev.is_running:
dev.total_run_time += current_time - dev.last_start_time
dev.is_running = False
dev.last_start_time = 0
# 使用示例
if __name__ == "__main__":
devices = [
Device("AC-01", "精密空调1", total_run_time=12000), # 已运行较多
Device("AC-02", "精密空调2", total_run_time=8000), # 中等
Device("AC-03", "精密空调3", total_run_time=3000), # 最少
Device("AC-04", "精密空调4", total_run_time=15000), # 最多
]
scheduler = RotationScheduler(devices, max_concurrent=2)
# 需要启动2台设备
selected = scheduler.select_devices(need_count=2)
print("选中的设备:")
for dev in selected:
print(f" {dev.name} (累计运行: {dev.total_run_time}s)")
scheduler.start_device(dev, time.time())某些设备不能同时运行(如制冷和制热):
"""
device_mutex.py - 设备互斥与优先级管理
"""
from enum import Enum
from typing import Dict, List
class DeviceType(Enum):
COOL = "cool" # 制冷
HEAT = "heat" # 制热
DEHUMIDIFY = "dehumidify" # 除湿
HUMIDIFY = "humidify" # 加湿
PURIFY = "purify" # 净化
# 互斥关系表
MUTEX_TABLE = {
DeviceType.COOL: [DeviceType.HEAT], # 制冷和制热互斥
DeviceType.HEAT: [DeviceType.COOL], # 制热和制冷互斥
DeviceType.DEHUMIDIFY: [DeviceType.HUMIDIFY], # 除湿和加湿互斥
DeviceType.HUMIDIFY: [DeviceType.DEHUMIDIFY], # 加湿和除湿互斥
}
# 优先级(数字越小优先级越高)
PRIORITY_TABLE = {
DeviceType.COOL: 1, # 高温对档案损害更大,优先级最高
DeviceType.DEHUMIDIFY: 2, # 高湿导致霉变
DeviceType.HEAT: 3, # 低温影响较小
DeviceType.HUMIDIFY: 4, # 低湿影响最小
DeviceType.PURIFY: 5, # 净化可以随时运行
}
class DeviceMutexManager:
"""
设备互斥管理器
"""
def __init__(self):
self.running_devices: Dict[str, DeviceType] = {} # device_id -> type
def can_start(self, device_id: str, device_type: DeviceType) -> bool:
"""
检查设备是否可以启动
"""
# 检查是否与已运行设备互斥
if device_type in MUTEX_TABLE:
for mutex_type in MUTEX_TABLE[device_type]:
for running_id, running_type in self.running_devices.items():
if running_type == mutex_type:
# 检查优先级
if PRIORITY_TABLE.get(device_type, 99) < PRIORITY_TABLE.get(running_type, 99):
# 新设备优先级更高,可以抢占
return True
else:
# 新设备优先级更低,不能启动
return False
return True
def start_device(self, device_id: str, device_type: DeviceType) -> bool:
"""
尝试启动设备
Returns: True=成功, False=被互斥规则阻止
"""
if not self.can_start(device_id, device_type):
return False
# 如果新设备优先级更高,停止被互斥的已运行设备
if device_type in MUTEX_TABLE:
for mutex_type in MUTEX_TABLE[device_type]:
to_stop = []
for running_id, running_type in self.running_devices.items():
if running_type == mutex_type:
to_stop.append(running_id)
for rid in to_stop:
self.stop_device(rid)
self.running_devices[device_id] = device_type
return True
def stop_device(self, device_id: str):
"""停止设备"""
if device_id in self.running_devices:
del self.running_devices[device_id]
def get_running_devices(self) -> Dict[str, DeviceType]:
"""获取当前运行的所有设备"""
return self.running_devices.copy()┌─────────────────────────────────────────────────────────────────────┐
│ 恒温恒湿净化设备组态监控画面 │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ ┌───────────────────────────────────────────────────────────────┐ │
│ │ 库房平面图 + 实时温湿度分布 │ │
│ │ ┌──────┐ ┌──────┐ ┌──────┐ │ │
│ │ │23.5℃│ │24.1℃│ │23.8℃│ 颜色编码: │ │
│ │ │52%RH│ │55%RH│ │51%RH│ 🟢 正常 🟡 预警 🔴 告警 │ │
│ │ └──────┘ └──────┘ └──────┘ │ │
│ └───────────────────────────────────────────────────────────────┘ │
│ │
│ ┌───────────────────────────────────────────────────────────────┐ │
│ │ 设备状态面板 │ │
│ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────────────┐│ │
│ │ │ AC-01 │ │ AC-02 │ │ DEH-01 │ │ HUM-01 ││ │
│ │ │ 🟢 运行 │ │ ⚪ 待机 │ │ 🟢 运行 │ │ ⚪ 待机 ││ │
│ │ │ 制冷中 │ │ 累计8.5h │ │ 除湿中 │ │ 累计3.2h ││ │
│ │ └──────────┘ └──────────┘ └──────────┘ └──────────────────┘│ │
│ └───────────────────────────────────────────────────────────────┘ │
│ │
│ ┌───────────────────────────────────────────────────────────────┐ │
│ │ 趋势曲线(最近24小时) │ │
│ │ 温度: ╭─╮╭─╮ ╭─╮ │ │
│ │ ╰─╯╰─╯────╯ ╰─ 24℃───────────────── │ │
│ │ 23.5℃─────────── 目标范围 │ │
│ │ 湿度: ╭──╮ │ │
│ │ ╰──╯──╯╭──╮ 58%RH───────────────── │ │
│ │ ╰──╯ 47%RH─────────── 目标范围 │ │
│ └───────────────────────────────────────────────────────────────┘ │
│ │
│ ┌───────────────────────────────────────────────────────────────┐ │
│ │ 控制面板 │ │
│ │ [手动/自动] 切换 设定值: 温度 14~24℃ 湿度 45~60%RH │ │
│ │ 当前模式: 自动 死区: 温度 0.5℃ 湿度 2%RH │ │
│ │ 防抖: ON (60s/120s) │ │
│ └───────────────────────────────────────────────────────────────┘ │
│ │
│ ┌───────────────────────────────────────────────────────────────┐ │
│ │ 报警列表(最近10条) │ │
│ │ 2024-03-15 14:23:05 [高温] 传感器03 温度 24.3℃ > 24℃ │ │
│ │ 2024-03-15 14:25:12 [恢复] 传感器03 温度 23.8℃ 恢复正常 │ │
│ │ 2024-03-15 15:01:33 [设备] AC-01 启动 (温度超标) │ │
│ └───────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────┘"""
control_logic_config.py - 联动逻辑配置
"""
# 阈值配置表
THRESHOLD_CONFIG = {
"archive_room_A": {
"temperature": {
"high_alarm": 24.0, # 高温告警
"high": 23.5, # 启动制冷
"low": 14.5, # 启动制热
"low_alarm": 14.0, # 低温告警
"dead_band": 0.5 # 死区
},
"humidity": {
"high_alarm": 60.0, # 高湿告警
"high": 58.0, # 启动除湿
"low": 47.0, # 启动加湿
"low_alarm": 45.0, # 低湿告警
"dead_band": 2.0 # 死区
}
}
}
# 防抖配置表
DEBOUNCE_CONFIG = {
"cool": {
"on_delay": 60, # 60秒后启动制冷
"off_delay": 120, # 120秒后停止制冷
"min_run_time": 300,# 最短运行5分钟
"min_off_time": 120 # 最短停机2分钟
},
"heat": {
"on_delay": 60,
"off_delay": 120,
"min_run_time": 300,
"min_off_time": 120
},
"dehumidify": {
"on_delay": 120, # 湿度变化慢,延迟更长
"off_delay": 180,
"min_run_time": 600,
"min_off_time": 300
},
"humidify": {
"on_delay": 120,
"off_delay": 180,
"min_run_time": 600,
"min_off_time": 300
}
}
# 设备配置表
DEVICE_CONFIG = {
"AC-01": {
"name": "精密空调-01",
"type": "cool",
"modbus_addr": 1,
"control_register": 40001,
"status_register": 30001,
"priority": 1
},
"AC-02": {
"name": "精密空调-02",
"type": "cool",
"modbus_addr": 2,
"control_register": 40001,
"status_register": 30001,
"priority": 1
},
"DEH-01": {
"name": "除湿机-01",
"type": "dehumidify",
"modbus_addr": 3,
"control_register": 40001,
"status_register": 30001,
"priority": 2
}
}恒温恒湿净化设备的组态联动开发,核心在于阈值模型、防抖算法、设备调度三者的有机结合。阈值模型定义了"什么时候该动作",防抖算法确保"不因瞬时波动而误动作",设备调度解决"多台设备如何协调工作"。在实际部署中,建议采用PLC做硬保护、SCADA/边缘网关做软调度的混合架构,既保证可靠性又兼顾灵活性。联动逻辑上线后,需要持续运行观察至少一个月,根据实际数据微调参数,才能达到最优控制效果。
原创声明:本文系作者授权腾讯云开发者社区发表,未经许可,不得转载。
如有侵权,请联系 cloudcommunity@tencent.com 删除。