机房监控系统集成：POE 温湿度传感器 TCP 协议与本地记录仪联动实战

┌─────────────────────────────────────────────────────────────┐
│ 机房监控网络（管理VLAN）                                      │
│                                                             │
│  ┌──────────────┐     TCP 502      ┌──────────────────┐    │
│  │ POE 温湿度   │◄────────────────►│ 边缘网关/        │    │
│  │ 传感器       │   Modbus TCP     │ 监控服务器        │    │
│  │ (内置记录仪) │                   │ (Zabbix/自研)    │    │
│  └──────┬───────┘                   └────────┬─────────┘    │
│         │ PoE (802.3af)                      │             │
│  ┌──────┴───────┐                   ┌────────┴─────────┐    │
│  │ PoE 交换机   │◄─────────────────►│ 数据库           │    │
│  │ (网管型)     │   管理/监控       │ (时序数据库)     │    │
│  └──────────────┘                   └──────────────────┘    │
└─────────────────────────────────────────────────────────────┘

import socket
import struct
import time

class SensorRecorder:
    def __init__(self, ip, port=502, unit_id=1):
        self.ip = ip
        self.port = port
        self.unit_id = unit_id

    def _modbus_read(self, reg_addr, count):
        """构造 Modbus TCP 读请求"""
        transaction_id = int(time.time()) % 65535
        protocol_id = 0
        length = 6  # 后续字节数
        func_code = 0x04  # 读输入寄存器
        # 简化：实际需构造完整 MBAP + PDU
        # 这里用 pymodbus 示意
        pass

    def _modbus_write(self, reg_addr, values):
        """构造 Modbus TCP 写请求（0x10 写多个寄存器）"""
        pass

    def configure_recorder(self, interval_sec=60, mode=1):
        """
        interval_sec: 记录间隔（秒）
        mode: 0=停止, 1=循环覆盖, 2=非循环(满则停)
        """
        # 写记录间隔
        self._write_holding(0x0010, interval_sec)
        # 写记录模式
        self._write_holding(0x0011, mode)
        # 启动记录
        self._write_holding(0x0011, mode)  # 确保写入生效

    def get_recorder_status(self):
        """读取记录仪状态"""
        stored_points = self._read_holding(0x0012, 2)  # UINT32
        oldest_ts = self._read_holding(0x0014, 2)      # UINT32
        latest_ts = self._read_holding(0x0016, 2)      # UINT32
        return {
            "stored_points": stored_points,
            "oldest_ts": oldest_ts,
            "latest_ts": latest_ts,
            "interval": self._read_holding(0x0010, 1)
        }

IP 地址：静态分配，与机柜位置绑定
端口：502（Modbus TCP 默认）
Unit ID：1（多探头级联时递增）
TCP 连接超时：5s
TCP 重试次数：3
连接保持：长连接（心跳保活，间隔30s）

class EdgeGateway:
    def __init__(self, sensors, recorder, mqtt_client):
        self.sensors = sensors          # 传感器列表
        self.recorder = recorder        # 本地记录仪（SQLite/时序DB）
        self.mqtt = mqtt_client         # 上行通道
        self.poll_interval = 30         # 轮询间隔（秒）
        self.network_ok = True

    def poll_loop(self):
        while True:
            for sensor in self.sensors:
                try:
                    # 读取实时数据
                    temp = sensor.read_register(0x0000) / 10.0
                    humi = sensor.read_register(0x0001) / 10.0
                    status = sensor.read_register(0x0003)

                    # 本地记录（无论网络状态）
                    self.recorder.insert(
                        sensor_id=sensor.id,
                        timestamp=time.time(),
                        temp=temp,
                        humi=humi,
                        status=status
                    )

                    # 网络正常时上行
                    if self.network_ok:
                        self.mqtt.publish(
                            f"archive/env/{sensor.id}",
                            json.dumps({
                                "temp": temp,
                                "humi": humi,
                                "status": status,
                                "ts": time.time()
                            })
                        )

                    # 联动判断（本地闭环）
                    self._check_local_alarm(sensor, temp, humi)

                except (socket.timeout, ConnectionRefusedError):
                    # 传感器通信失败
                    self._handle_sensor_offline(sensor)

            time.sleep(self.poll_interval)

    def _check_local_alarm(self, sensor, temp, humi):
        """本地联动判断，不依赖云端"""
        if temp > 28:
            sensor.trigger_relay(1, "close")  # 启动备用风扇
        elif temp < 18:
            sensor.trigger_relay(2, "close")  # 启动加热

def on_network_restored(self):
        """网络恢复回调"""
        self.network_ok = True

        # 查询记录仪中未上传的数据
        for sensor in self.sensors:
            try:
                # 读取传感器端记录仪状态
                stored = sensor.get_recorder_status()
                if stored["stored_points"] == 0:
                    continue

                # 读取历史数据（从最早到最新）
                # 分批读取，避免单次报文过大
                batch_size = 100  # 每批100个点
                total = stored["stored_points"]
                uploaded = 0

                while uploaded < total:
                    # 计算读取范围
                    start_addr = 0x1000 + (uploaded * 2)  # 每点占2个寄存器
                    count = min(batch_size * 2, (total - uploaded) * 2)

                    # 读取历史数据
                    raw = sensor.read_registers(start_addr, count)

                    # 解析为温湿度对
                    points = self._parse_history(raw)

                    # 批量上行
                    for pt in points:
                        self.mqtt.publish(
                            f"archive/env/{sensor.id}/history",
                            json.dumps(pt)
                        )
                        uploaded += 1

                # 补传完成，通知平台
                self.mqtt.publish(
                    f"archive/env/{sensor.id}/backfill_complete",
                    json.dumps({
                        "from": stored["oldest_ts"],
                        "to": stored["latest_ts"],
                        "points": total
                    })
                )

            except Exception as e:
                # 补传失败，下次重试
                logger.error(f"Backfill failed for {sensor.id}: {e}")

def handle_backfill(mqtt_client, db):
    """监控平台侧处理补传数据"""
    def on_message(topic, payload):
        if "/history" in topic:
            data = json.loads(payload)
            sensor_id = topic.split("/")[3]

            # 写入时序数据库（InfluxDB/TDengine）
            db.write_point(
                measurement="env_history",
                tags={"sensor_id": sensor_id},
                fields={"temp": data["temp"], "humi": data["humi"]},
                timestamp=data["ts"]  # 使用传感器端时间戳
            )

        elif "/backfill_complete" in topic:
            # 补传完成，触发数据完整性校验
            data = json.loads(payload)
            verify_integrity(data["sensor_id"], data["from"], data["to"])

    mqtt_client.subscribe("archive/env/+/history", on_message)
    mqtt_client.subscribe("archive/env/+/backfill_complete", on_message)

继电器1：温度高 → 闭合（启动风扇）
继电器2：湿度高 → 闭合（启动除湿机）
继电器3：温度低 → 闭合（启动加热）

阈值配置（通过 Modbus TCP 写入）：
  温度上限：28℃ → 0x0020
  温度下限：18℃ → 0x0021
  湿度上限：65%RH → 0x0022
  湿度下限：35%RH → 0x0023
  回滞量：1℃ / 3%RH → 0x0024

def l2_coordination(sensors, unit_controller):
    """
    多传感器融合判断，避免单点误报
    """
    # 读取所有传感器数据
    readings = [s.read_current() for s in sensors]

    # 异常值剔除（IQR法）
    valid = iqr_outlier_detect(readings)

    if len(valid) < len(sensors) * 0.6:
        # 超过40%传感器异常，可能是通信故障，不触发联动
        return

    # 加权融合
    fused_temp = weighted_fusion(valid, weights)
    fused_humi = weighted_fusion(valid, weights)

    # 趋势判断（不只看当前值，还看变化率）
    temp_rate = (fused_temp - prev_temp) / poll_interval * 60  # ℃/min

    if fused_temp > 26 and temp_rate > 0.2:
        # 温度偏高且上升快，提前启动机组
        unit_controller.set_mode("cool", advance=True)
    elif fused_temp < 20 and temp_rate < -0.2:
        unit_controller.set_mode("heat", advance=True)

def verify_integrity(sensor_id, ts_from, ts_to):
    """
    校验指定时间范围内的数据完整性
    """
    # 查询数据库中该传感器的时间序列
    query = f"""
    SELECT timestamp, temp, humi 
    FROM env_data 
    WHERE sensor_id='{sensor_id}' 
    AND timestamp BETWEEN {ts_from} AND {ts_to}
    ORDER BY timestamp
    """

    rows = db.query(query)

    # 检查是否有缺失
    expected_interval = 60  # 秒
    gaps = []
    for i in range(1, len(rows)):
        gap = rows[i].timestamp - rows[i-1].timestamp
        if gap > expected_interval * 1.5:  # 允许50%误差
            gaps.append({
                "from": rows[i-1].timestamp,
                "to": rows[i].timestamp,
                "duration": gap
            })

    if gaps:
        # 有数据缺口，记录告警
        logger.warning(f"Data gaps detected for {sensor_id}: {gaps}")
        # 可选：向传感器请求补传缺口数据
        request_gap_backfill(sensor_id, gaps)
    else:
        logger.info(f"Data integrity OK for {sensor_id}")

┌────────────────────────────────────────────────────┐
│ 数据完整性报告 - 传感器 R201-A01                    │
├────────────────────────────────────────────────────┤
│ 统计周期：2025-01-15 00:00 ~ 23:59                │
│ 预期数据点：1440（1分钟间隔）                       │
│ 实际数据点：1438                                   │
│ 完整率：99.86%                                     │
│                                                    │
│ 数据缺口：                                         │
│   [14:23:00 ~ 14:24:00] 缺失1个点（网络抖动）     │
│   [22:15:00 ~ 22:16:00] 缺失1个点（传感器重启）   │
│                                                    │
│ 补传状态：已自动补传（22:15缺口）                  │
│ 14:23缺口：传感器端记录仪无数据（传感器重启期间）  │
└────────────────────────────────────────────────────┘