档案库房环境控制优化：温湿度波动抑制与净化设备调度策略解析

def median_filter(readings, window=5):
    """滑动中值滤波，剔除脉冲噪声"""
    if len(readings) < window:
        return readings[-1]
    return median(readings[-window:])

def iqr_outlier_detect(points, k=1.5):
    """IQR法标记异常点"""
    vals = [p["value"] for p in points]
    q1, q3 = percentile(vals, 25), percentile(vals, 75)
    iqr = q3 - q1
    lower, upper = q1 - k * iqr, q3 + k * iqr
    return [p for p in points if lower <= p["value"] <= upper]

def weighted_fusion(points, weights):
    """
    weights: 基于位置/响应时间的先验权重
    回风口: 0.15（响应快但局部）
    远角:   0.35（最差工况代表）
    外墙侧: 0.25（传热影响）
    门附近: 0.10（开门扰动）
    通道内: 0.15（架体间微环境）
    """
    valid = iqr_outlier_detect(points)
    total_w = sum(weights[p["id"]] for p in valid)
    fused_temp = sum(p["temp"] * weights[p["id"]] for p in valid) / total_w
    fused_humi = sum(p["humi"] * weights[p["id"]] for p in valid) / total_w
    return fused_temp, fused_humi

MAX_RATE_TEMP = 0.5  # ℃/min
MAX_RATE_HUMI = 2.0  # %RH/min

def rate_limit(current, previous, interval_sec):
    rate_temp = abs(current["temp"] - previous["temp"]) / (interval_sec / 60)
    rate_humi = abs(current["humi"] - previous["humi"]) / (interval_sec / 60)
    if rate_temp > MAX_RATE_TEMP or rate_humi > MAX_RATE_HUMI:
        return previous, "suspect"
    return current, "valid"

温度死区：±1.5℃（设定22℃，控制范围20.5–23.5℃）
湿度死区：±5%RH（设定50%，控制范围45–55%RH）

def anti_oscillation(state, command, history, cooldown=300):
    """
    cooldown: 设备动作冷却时间（秒）
    同一设备短时间内不允许反向动作
    """
    now = time.time()
    last_action = history.get(command["device_id"], {}).get("last_action_time", 0)
    
    # 冷却期内禁止反向动作
    if now - last_action < cooldown:
        last_mode = history[command["device_id"]]["mode"]
        if command["mode"] != last_mode:
            return None, "cooldown"
    
    # 相邻设备互斥检查
    if command["type"] == "cool" and state.get("dehumid_running"):
        # 制冷与除湿同时运行需确认负荷确实叠加
        if state["temp_spread"] < 1.0:  # 温差小，不需要同时制冷
            return None, "mutual_exclusion"
    
    return command, "allowed"

前馈输入：
  - 室外温度/湿度变化率（气象站数据）
  - 开门事件（门磁信号）
  - 新风阀开度变化

补偿动作：
  - 室外温度快速上升 → 提前降低设定温度0.5℃
  - 门磁触发 → 临时提高风机转速，加速气流混合
  - 新风阀开大 → 提前启动再热/再湿

def fresh_air_schedule(outdoor, indoor, cfg):
    """
    基于室外气象和室内状态的新风调度
    """
    actions = []
    
    # 免费制冷：室外温度适宜且湿度在舒适区
    if (outdoor["temp"] < indoor["temp"] - 2.0 and
        cfg["humi_low"] <= outdoor["humi"] <= cfg["humi_high"]):
        damper_pos = min(100, int((indoor["temp"] - outdoor["temp"]) * 20))
        actions.append(("damper", damper_pos))
    
    # 高湿限制：室外湿度过高，关小新风防湿负荷
    elif outdoor["humi"] > cfg["humi_high"] + 5:
        actions.append(("damper", 10))  # 最小新风量维持正压
    
    # 低温限制：冬季室外过冷，限制新风防冻
    elif outdoor["temp"] < cfg["temp_low"] - 2:
        actions.append(("damper", 10))
    
    # 正常工况：按CO₂浓度调节
    else:
        if indoor.get("co2", 800) > 1000:
            actions.append(("damper", 50))
        elif indoor.get("co2", 800) < 600:
            actions.append(("damper", 20))
        else:
            actions.append(("damper", 30))
    
    return actions

def humidity_coordination(fused, unit_state, dehumid_state, cfg):
    """
    优先级：机组除湿 > 独立除湿机
    机组除湿能力不足时才启动独立除湿机
    """
    if fused["avg_humi"] > cfg["humi_high"]:
        # 机组已在除湿模式且满负荷（阀门100%）
        if unit_state["mode"] == "dehumid" and unit_state["valve_pos"] >= 95:
            if not dehumid_state["running"]:
                return [("dehumid", "start")]
        # 机组未除湿，先切机组
        elif unit_state["mode"] != "dehumid":
            return [("unit_mode", "dehumid")]
    
    elif fused["avg_humi"] < cfg["humi_low"]:
        # 类似逻辑，机组优先
        if unit_state["mode"] != "humid":
            return [("unit_mode", "humid")]
    
    else:
        # 回到死区内，逐步退出辅助设备
        if dehumid_state["running"]:
            return [("dehumid", "stop")]
    
    return []

优化前（仅机组PID，无协同）：
  日温度波动：4.2℃
  日湿度波动：12%RH
  压缩机启停：28次/天
  均匀性（温差）：3.8℃

优化后（感知融合+协同控制+调度）：
  日温度波动：1.6℃
  日湿度波动：4.1%RH
  压缩机启停：8次/天
  均匀性（温差）：1.2℃