智慧档案环境感知实战：温湿度监控与恒温恒湿设备协同控制指南

def fuse_readings(points):
    temps = [p["temp"] for p in points if p["valid"]]
    humis = [p["humi"] for p in points if p["valid"]]
    
    # 异常值剔除（IQR法）
    temps = remove_outliers_iqr(temps)
    humis = remove_outliers_iqr(humis)
    
    return {
        "avg_temp": mean(temps),
        "max_temp": max(temps),
        "min_temp": min(temps),
        "temp_spread": max(temps) - min(temps),  # 均匀性指标
        "avg_humi": mean(humis),
        "max_humi": max(humis),
        "min_humi": min(humis),
        "worst_point": find_worst(points),  # 最差测点ID
        "uniformity": 1.0 - (max(temps)-min(temps))/10.0  # 0-1
    }

┌────────────────────────────────────────────────────────────┐
│  感知层：分散式以太网变送器（空间分布）                      │
│  ↓ 采集                                                    │
│  边缘网关（融合+决策）                                      │
│  ├── 策略引擎：感知数据 → 机组设定值/模式                   │
│  ├── 反馈校验：机组实际状态 vs 期望状态                     │
│  └── 安全联锁：互锁、滞回、防振荡                           │
│  ↓ Modbus TCP / DO                                         │
│  执行层：恒温恒湿机组 + 辅助设备                             │
│  ├── 机组控制器（PID内环）                                  │
│  ├── 除湿机/加湿机（启停）                                  │
│  └── 新风机/风阀（开度）                                    │
└────────────────────────────────────────────────────────────┘

def calc_unit_setpoint(fused, unit_ret, cfg):
    """
    fused: 融合后的库内参数
    unit_ret: 机组回风参数
    cfg: 配置（目标范围、死区、最大修正量）
    """
    # 偏差 = 库内均值 - 回风值
    temp_bias = fused["avg_temp"] - unit_ret["temp"]
    humi_bias = fused["avg_humi"] - unit_ret["humi"]
    
    # 动态修正量（限幅）
    temp_correction = clamp(temp_bias * cfg["bias_gain"], 
                           -cfg["max_temp_correction"], 
                           cfg["max_temp_correction"])
    humi_correction = clamp(humi_bias * cfg["bias_gain"], 
                           -cfg["max_humi_correction"], 
                           cfg["max_humi_correction"])
    
    target_temp = cfg["target_temp"] - temp_correction
    target_humi = cfg["target_humi"] - humi_correction
    
    return target_temp, target_humi, temp_bias, humi_bias

def worst_point_control(fused, cfg):
    """基于最差测点的保护逻辑"""
    actions = []
    
    # 温度保护
    if fused["max_temp"] > cfg["temp_high"]:
        actions.append(("cool", fused["max_temp"] - cfg["temp_high"]))
    if fused["min_temp"] < cfg["temp_low"]:
        actions.append(("heat", cfg["temp_low"] - fused["min_temp"]))
    
    # 湿度保护
    if fused["max_humi"] > cfg["humi_high"]:
        actions.append(("dehumid", fused["max_humi"] - cfg["humi_high"]))
    if fused["min_humi"] < cfg["humi_low"]:
        actions.append(("humidify", cfg["humi_low"] - fused["min_humi"]))
    
    # 均匀性保护：温差过大说明气流组织有问题
    if fused["temp_spread"] > cfg["max_spread"]:
        actions.append(("fan_speed_up", fused["temp_spread"]))
    
    return actions

需求汇总：
  库房A: 需要制冷（28℃）
  库房B: 需要除湿（65%RH）
  库房C: 正常

决策：
  优先满足最紧迫需求（温度>湿度，或按档案类型权重）
  若冲突不可调和（如A需制冷同时B需制热），启用分区风阀或独立机组

def zone_valve_control(zones, unit_capacity):
    """多区风阀分配"""
    total_demand = sum(z["demand"] for z in zones)
    if total_demand <= unit_capacity:
        # 按需分配
        for z in zones:
            z["valve"] = z["demand"] / total_demand
    else:
        # 容量不足，按优先级分配
        zones_sorted = sorted(zones, key=lambda z: z["priority"], reverse=True)
        remaining = unit_capacity
        for z in zones_sorted:
            alloc = min(z["demand"], remaining)
            z["valve"] = alloc / z["demand"] if z["demand"] > 0 else 0
            remaining -= alloc

策略：
  室外温度 < 库内温度 - 2℃ 且室外湿度在舒适区 → 增大新风比
  室外温度 > 库内温度 + 2℃ → 减小新风，回风循环
  室外湿度 > 设定上限 + 5% → 限制新风，防湿负荷
  室外湿度 < 设定下限 - 5% → 限制新风，防干燥

偏差标定记录表：
  测点ID | 距回风口距离 | 平均偏差 | 最大偏差 | 备注
  s01    | 2m           | +0.3℃   | +0.8℃   | 送风口下
  s02    | 8m           | +1.8℃   | +3.2℃   | 远角
  s03    | 靠西墙       | +2.1℃   | +4.5℃   | 西晒

步骤1：手动模式，机组固定设定运行，观察库内分布
步骤2：开启感知融合，但不下发控制，对比融合值与机组回风
步骤3：开启设定值修正，小增益（0.2）起步，观察24h
步骤4：逐步增大增益至系统稳定无振荡
步骤5：注入扰动（开门、模拟越限），验证响应
步骤6：故障注入（通信中断、单点失效），验证兜底

def fuse_readings(points):
    temps = [p["temp"] for p in points if p["valid"]]
    humis = [p["humi"] for p in points if p["valid"]]
    
    # 异常值剔除（IQR法）
    temps = remove_outliers_iqr(temps)
    humis = remove_outliers_iqr(humis)
    
    return {
        "avg_temp": mean(temps),
        "max_temp": max(temps),
        "min_temp": min(temps),
        "temp_spread": max(temps) - min(temps),  # 均匀性指标
        "avg_humi": mean(humis),
        "max_humi": max(humis),
        "min_humi": min(humis),
        "worst_point": find_worst(points),  # 最差测点ID
        "uniformity": 1.0 - (max(temps)-min(temps))/10.0  # 0-1
    }

┌────────────────────────────────────────────────────────────┐
│  感知层：分散式以太网变送器（空间分布）                      │
│  ↓ 采集                                                    │
│  边缘网关（融合+决策）                                      │
│  ├── 策略引擎：感知数据 → 机组设定值/模式                   │
│  ├── 反馈校验：机组实际状态 vs 期望状态                     │
│  └── 安全联锁：互锁、滞回、防振荡                           │
│  ↓ Modbus TCP / DO                                         │
│  执行层：恒温恒湿机组 + 辅助设备                             │
│  ├── 机组控制器（PID内环）                                  │
│  ├── 除湿机/加湿机（启停）                                  │
│  └── 新风机/风阀（开度）                                    │
└────────────────────────────────────────────────────────────┘

def calc_unit_setpoint(fused, unit_ret, cfg):
    """
    fused: 融合后的库内参数
    unit_ret: 机组回风参数
    cfg: 配置（目标范围、死区、最大修正量）
    """
    # 偏差 = 库内均值 - 回风值
    temp_bias = fused["avg_temp"] - unit_ret["temp"]
    humi_bias = fused["avg_humi"] - unit_ret["humi"]
    
    # 动态修正量（限幅）
    temp_correction = clamp(temp_bias * cfg["bias_gain"], 
                           -cfg["max_temp_correction"], 
                           cfg["max_temp_correction"])
    humi_correction = clamp(humi_bias * cfg["bias_gain"], 
                           -cfg["max_humi_correction"], 
                           cfg["max_humi_correction"])
    
    target_temp = cfg["target_temp"] - temp_correction
    target_humi = cfg["target_humi"] - humi_correction
    
    return target_temp, target_humi, temp_bias, humi_bias

def worst_point_control(fused, cfg):
    """基于最差测点的保护逻辑"""
    actions = []
    
    # 温度保护
    if fused["max_temp"] > cfg["temp_high"]:
        actions.append(("cool", fused["max_temp"] - cfg["temp_high"]))
    if fused["min_temp"] < cfg["temp_low"]:
        actions.append(("heat", cfg["temp_low"] - fused["min_temp"]))
    
    # 湿度保护
    if fused["max_humi"] > cfg["humi_high"]:
        actions.append(("dehumid", fused["max_humi"] - cfg["humi_high"]))
    if fused["min_humi"] < cfg["humi_low"]:
        actions.append(("humidify", cfg["humi_low"] - fused["min_humi"]))
    
    # 均匀性保护：温差过大说明气流组织有问题
    if fused["temp_spread"] > cfg["max_spread"]:
        actions.append(("fan_speed_up", fused["temp_spread"]))
    
    return actions

需求汇总：
  库房A: 需要制冷（28℃）
  库房B: 需要除湿（65%RH）
  库房C: 正常

决策：
  优先满足最紧迫需求（温度>湿度，或按档案类型权重）
  若冲突不可调和（如A需制冷同时B需制热），启用分区风阀或独立机组

def zone_valve_control(zones, unit_capacity):
    """多区风阀分配"""
    total_demand = sum(z["demand"] for z in zones)
    if total_demand <= unit_capacity:
        # 按需分配
        for z in zones:
            z["valve"] = z["demand"] / total_demand
    else:
        # 容量不足，按优先级分配
        zones_sorted = sorted(zones, key=lambda z: z["priority"], reverse=True)
        remaining = unit_capacity
        for z in zones_sorted:
            alloc = min(z["demand"], remaining)
            z["valve"] = alloc / z["demand"] if z["demand"] > 0 else 0
            remaining -= alloc

策略：
  室外温度 < 库内温度 - 2℃ 且室外湿度在舒适区 → 增大新风比
  室外温度 > 库内温度 + 2℃ → 减小新风，回风循环
  室外湿度 > 设定上限 + 5% → 限制新风，防湿负荷
  室外湿度 < 设定下限 - 5% → 限制新风，防干燥

偏差标定记录表：
  测点ID | 距回风口距离 | 平均偏差 | 最大偏差 | 备注
  s01    | 2m           | +0.3℃   | +0.8℃   | 送风口下
  s02    | 8m           | +1.8℃   | +3.2℃   | 远角
  s03    | 靠西墙       | +2.1℃   | +4.5℃   | 西晒

步骤1：手动模式，机组固定设定运行，观察库内分布
步骤2：开启感知融合，但不下发控制，对比融合值与机组回风
步骤3：开启设定值修正，小增益（0.2）起步，观察24h
步骤4：逐步增大增益至系统稳定无振荡
步骤5：注入扰动（开门、模拟越限），验证响应
步骤6：故障注入（通信中断、单点失效），验证兜底