档案库房多台恒湿消毒净化一体机集群联动控制落地实践

逻辑：以全库房最差区域为准，按负荷分配设备

算法：
  1. 采集所有一体机的温湿度读数
  2. 计算全库房平均湿度 + 最大偏差
  3. 判定目标：将湿度拉回设定区间（如45~55%RH）
  4. 分配策略：
     - 湿度偏高的区域，就近设备优先启动
     - 负荷大的区域分配2台设备，负荷小的分配1台
     - 剩余设备进入待机（保留1台作为冗余）

逻辑：相同功能的设备按时间片轮转，避免单台过度老化

策略：
  - 除湿/加湿/净化模块每24小时轮换一次主设备
  - 轮换时采用"软切换"：新主设备先启动，30秒后旧主设备再停止
  - 运行小时数实时累计，优先让"跑得少"的设备上岗

逻辑：某台设备故障时，邻近设备自动提频补偿

流程：
  故障检测（通信超时3次或故障码非0）
      ↓
  标记故障设备，触发告警
      ↓
  计算该区域负荷缺口
      ↓
  邻近设备按距离+余量排序，选择最优设备提频
      ↓
  提频设备满负荷仍不足 → 启动备用设备
      ↓
  故障恢复后，逐步回退到均衡状态（避免反复切换）

逻辑：消毒/UV模块与人员、新风、排风强联锁

规则（最高优先级，可抢占其他模式）：
  □ 门磁开启 → 立即关闭消毒/UV模块
  □ 人体感应触发 → 立即关闭消毒/UV模块
  □ 消毒运行时 → 关闭新风、关闭排风、启动内循环
  □ 消毒结束 → 延时10分钟通风后再恢复常态模式
  □ 消毒时段默认设定在无人时段（23:00~05:00）

# cluster_controller.py - 集群联动控制器（精简版）

import time
from dataclasses import dataclass, field
from typing import Dict, List
from enum import Enum

class ClusterMode(Enum):
    BALANCE = "balance"      # 全局均衡
    ROUND_ROBIN = "robin"    # 轮值运行
    FAILSAFE = "failsafe"    # 故障接管
    DISINFECT = "disinfect"  # 消毒联动

@dataclass
class UnitState:
    unit_id: str
    zone: str
    temperature: float = 0.0
    humidity: float = 0.0
    pm25: int = 0
    run_mode: int = 0        # 0待机 1自动 2除湿 3加湿 4净化
    fan_speed: int = 0
    run_hours: int = 0
    online: bool = True
    fault_code: int = 0

class ClusterController:
    def __init__(self, units: List[UnitState], target_hum=50.0,
                 hum_tolerance=5.0, switch_interval=3600):
        self.units = {u.unit_id: u for u in units}
        self.target_hum = target_hum
        self.hum_tolerance = hum_tolerance
        self.switch_interval = switch_interval
        self.current_mode = ClusterMode.BALANCE
        self._last_switch = time.time()

    def evaluate(self) -> Dict:
        """全局评估，输出各设备控制指令"""
        # 过滤在线且无故障的设备
        healthy = [u for u in self.units.values() if u.online and u.fault_code == 0]
        if not healthy:
            return {"mode": "ALL_DOWN", "actions": {}}

        # 计算全局指标
        avg_hum = sum(u.humidity for u in healthy) / len(healthy)
        worst = max(healthy, key=lambda u: abs(u.humidity - self.target_hum))

        actions = {}
        self.current_mode = ClusterMode.BALANCE

        # 湿度偏差判定
        hum_dev = avg_hum - self.target_hum

        if hum_dev > self.hum_tolerance:
            # 需要除湿：按运行时长排序，选择"最闲"的设备
            candidates = sorted(healthy, key=lambda u: u.run_hours)
            primary = candidates[0]
            actions[primary.unit_id] = {"mode": 2, "fan_speed": 70}  # 除湿
            # 偏差大则启用第二台
            if hum_dev > self.hum_tolerance * 1.5 and len(candidates) > 1:
                actions[candidates[1].unit_id] = {"mode": 2, "fan_speed": 60}

        elif hum_dev < -self.hum_tolerance:
            # 需要加湿
            candidates = sorted(healthy, key=lambda u: u.run_hours)
            actions[candidates[0].unit_id] = {"mode": 3, "fan_speed": 60}  # 加湿

        # 净化：基于PM2.5最高的区域
        max_pm25 = max(healthy, key=lambda u: u.pm25)
        if max_pm25.pm25 > 75:
            actions[max_pm25.unit_id] = actions.get(max_pm25.unit_id, {})
            actions[max_pm25.unit_id]["purify"] = True

        # 轮值切换检查
        if time.time() - self._last_switch > self.switch_interval:
            self._do_round_robin(healthy)
            self._last_switch = time.time()

        return {
            "mode": self.current_mode.value,
            "avg_humidity": round(avg_hum, 1),
            "worst_zone": worst.zone,
            "worst_dev": round(worst.humidity - self.target_hum, 1),
            "actions": actions
        }

    def _do_round_robin(self, healthy: List[UnitState]):
        """轮值切换：把主运行权交给运行时长最短的设备"""
        min_hours = min(healthy, key=lambda u: u.run_hours)
        # 软切换：先启动新主设备，再停止旧主设备
        # （实际实现中需配合设备启停延时）
        print(f"[轮值] 主设备切换至 {min_hours.unit_id} (运行时长最少: {min_hours.run_hours}h)")

    def trigger_failsafe(self, failed_unit_id: str) -> Dict:
        """故障接管：为故障设备区域分配替代设备"""
        if failed_unit_id not in self.units:
            return {}
        failed = self.units[failed_unit_id]
        healthy = [u for u in self.units.values()
                   if u.online and u.fault_code == 0 and u.unit_id != failed_unit_id]

        # 按距离+余量排序（简化：按unit_id相邻度）
        healthy.sort(key=lambda u: abs(int(u.unit_id[-1]) - int(failed_unit_id[-1])))
        actions = {}
        if healthy:
            actions[healthy[0].unit_id] = {"fan_speed": min(100, healthy[0].fan_speed + 20)}
        return {"mode": "failsafe", "failed": failed_unit_id, "actions": actions}

    def trigger_disinfect(self, zone: str, door_open: bool, person_detected: bool) -> Dict:
        """消毒联动：安全优先"""
        if door_open or person_detected:
            self.current_mode = ClusterMode.DISINFECT
            return {
                "mode": "disinfect_stop",
                "reason": "door_open" if door_open else "person_detected",
                "actions": {"disinfect": "off", "fresh_air": "on", "exhaust": "on"}
            }
        return {"mode": "disinfect_ready", "actions": {}}

集群控制器决策
    ↓
生成 {unit_id: {mode, fan_speed, purify, ...}}
    ↓
指令校验（防抖：与当前状态一致则不下发）
    ↓
按从站地址逐台下发（Modbus TCP / RTU）
    ↓
下发后50ms回读确认
    ↓
回读一致 → 确认生效
回读不一致 → 重试（最多3次）
    ↓
更新设备状态表 + 记录联动日志

原则：按物理位置和负荷特征分组，而非简单按数量平均

示例（300㎡库房，6台一体机）：

Zone-A（靠外墙，湿度偏高，3台）：
  UNIT-01  靠门区域（人员进出、湿度波动大）
  UNIT-02  外墙中段
  UNIT-03  外墙拐角（历史最湿区域）

Zone-B（中心区域，相对稳定，2台）：
  UNIT-04  中央通道
  UNIT-05  密集架中部

Zone-C（靠窗，温度波动，1台）：
  UNIT-06  南窗下方

联动关系：
  Zone-A内部均衡（湿度偏高，优先除湿）
  Zone-B为基准参考区
  Zone-C跟随温度波动
  跨Zone：Zone-A除湿时，Zone-B保持待机不反向加湿

集成方式：集群控制器作为独立服务运行

输入：订阅平台MQTT主题（各设备实时状态）
输出：通过Modbus TCP/RTU下发控制指令
日志：联动事件写入平台数据库

关键：控制器故障时，所有设备自动回退到本地面板自控
     → 控制器不在线 ≠ 设备失控

Step 1: 单机验证
  每台设备手动切换模式（除湿/加湿/净化），确认响应正常

Step 2: 联动模拟
  用脚本模拟各区域温湿度数据，观察控制器决策是否合理

Step 3: 渐进式上线
  先只开启"全局均衡"，跑1周观察
  再加"轮值运行"，跑1周
  最后开启"故障接管"和"消毒联动"

Step 4: 异常注入测试
  手动断开一台设备网络 → 验证接管
  打开消毒间门 → 验证联锁
  模拟湿度超标 → 验证分配策略