档案库房环境异常预警系统：温湿度趋势预测与设备预调节方案

┌─────────────────────────────────────────────────────────────────────┐
│ 档案库房环境异常预警系统架构                                       │
│                                                                     │
│  ┌───────────────────────────────────────────────────────────────┐ │
│  │ 数据采集层                                                     │ │
│  │  ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐            │ │
│  │  │温湿度   │ │室外气象 │ │空调状态 │ │电力参数 │            │ │
│  │  │传感器   │ │站数据   │ │反馈信号 │ │监测     │            │ │
│  │  └────┬────┘ └────┬────┘ └────┬────┘ └────┬────┘            │ │
│  └───────┼───────────┼───────────┼───────────┼──────────────────┘ │
│          │           │           │           │                     │
│  ┌───────▼───────────▼───────────▼───────────▼──────────────────┐ │
│  │ 特征工程层                                                     │ │
│  │  · 时间对齐（30秒窗口）                                        │ │
│  │  · 异常值过滤（3σ + 物理范围）                                 │ │
│  │  · 衍生特征计算（温变速率、湿度变化率、焓值、露点温度）       │ │
│  │  · 空间聚合（区域均值/极差/标准差）                            │ │
│  └───────────────────────┬───────────────────────────────────────┘ │
│                          │                                         │
│  ┌───────────────────────▼───────────────────────────────────────┐ │
│  │ 预测与预警引擎                                                 │ │
│  │  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐        │ │
│  │  │短期预测      │  │中期预测      │  │异常检测      │        │ │
│  │  │（30min~2h） │  │（2h~24h）   │  │（实时）      │        │ │
│  │  │线性外推     │  │周期模型      │  │偏离基线     │        │ │
│  │  └──────┬───────┘  └──────┬───────┘  └──────┬───────┘        │ │
│  └─────────┼──────────────────┼──────────────────┼───────────────┘ │
│            │                  │                  │                   │
│  ┌─────────▼──────────────────▼──────────────────▼───────────────┐ │
│  │ 预警决策层                                                     │ │
│  │  · 预警等级判定（提示/关注/警告/紧急）                         │ │
│  │  · 预警聚合（多传感器交叉确认，避免误报）                      │ │
│  │  · 根因分析（是设备故障？室外天气？还是设定值不合理？）        │ │
│  └───────────────────────┬───────────────────────────────────────┘ │
│                          │                                         │
│  ┌───────────────────────▼───────────────────────────────────────┐ │
│  │ 预调节执行层                                                   │ │
│  │  · 自动生成调控策略                                            │ │
│  │  · 设备联动控制（空调/除湿/加湿/新风）                        │ │
│  │  · 执行确认与回退机制                                          │ │
│  └───────────────────────────────────────────────────────────────┘ │
│                                                                     │
│  ┌───────────────────────────────────────────────────────────────┐ │
│  │ 人机交互层                                                     │ │
│  │  · 预警推送（APP/短信/大屏）                                   │ │
│  │  · 调控建议展示                                                │ │
│  │  · 历史回溯与模型自优化                                        │ │
│  └───────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────────────┐
│ 短期趋势预测示意图                                                 │
│                                                                     │
│  温度(℃)                                                            │
│  26┤                                           × 预测超标点       │
│    │                                      ×                       │
│  25┤                                ×     阈值上限(24℃)           │
│    │                          ×                                   │
│  24┤                    ×──────────────────────── 阈值线            │
│    │              ×                                             │
│  23┤        ×                                                   │
│    │    ×  当前值(22.8℃)                                        │
│  22┤  ×                                                         │
│    │×                                                            │
│  21┼───────────────────────────────────────────────────────────── │
│    现在         30min      1h        1.5h       2h                 │
│                                                                     │
│  温变速率: +0.3℃/10min = +1.8℃/h                                  │
│  距阈值(24℃)差距: 1.2℃                                             │
│  预计超标时间: 1.2 / 1.8 × 60 = 40分钟                            │
│  → 触发"关注"级预警，建议提前启动空调                              │
└─────────────────────────────────────────────────────────────────────┘

# trend_predictor.py
import time
import numpy as np
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Tuple
from enum import Enum

class AlertLevel(Enum):
    INFO = "info"           # 提示：趋势偏离但短期内不会超标
    ATTENTION = "attention" # 关注：预计30~60分钟内可能超标
    WARNING = "warning"     # 警告：预计10~30分钟内可能超标
    CRITICAL = "critical"   # 紧急：预计10分钟内超标或已超标

@dataclass
class TrendPoint:
    """趋势数据点"""
    timestamp: float
    temperature: float
    humidity: float

@dataclass
class PredictionResult:
    """预测结果"""
    zone_id: str
    sensor_id: str
    current_temp: float
    current_humi: float
    temp_slope: float        # ℃/h
    humi_slope: float        # %RH/h
    predicted_temp: float    # 预测值（指定时间窗口后）
    predicted_humi: float
    time_to_temp_limit: Optional[float] = None  # 距离温度超标的时间(秒)
    time_to_humi_limit: Optional[float] = None  # 距离湿度超标的时间(秒)
    alert_level: AlertLevel = AlertLevel.INFO
    alert_reason: str = ""

class TrendPredictor:
    """温湿度趋势预测器"""

    def __init__(self,
                 temp_upper: float = 24.0,
                 temp_lower: float = 14.0,
                 humi_upper: float = 60.0,
                 humi_lower: float = 45.0,
                 min_data_points: int = 6,
                 prediction_window: float = 1800):  # 30分钟
        self.temp_upper = temp_upper
        self.temp_lower = temp_lower
        self.humi_upper = humi_upper
        self.humi_lower = humi_lower
        self.min_data_points = min_data_points
        self.prediction_window = prediction_window
        self.history: Dict[str, List[TrendPoint]] = {}
        self.lock = threading.RLock()

    def add_reading(self, zone_id: str, sensor_id: str,
                   temperature: float, humidity: float):
        """添加传感器读数"""
        key = f"{zone_id}:{sensor_id}"
        with self.lock:
            if key not in self.history:
                self.history[key] = []
            self.history[key].append(TrendPoint(
                timestamp=time.time(),
                temperature=temperature,
                humidity=humidity
            ))
            # 保留最近2小时数据
            cutoff = time.time() - 7200
            self.history[key] = [
                p for p in self.history[key]
                if p.timestamp > cutoff
            ]

    def predict(self, zone_id: str, sensor_id: str) -> Optional[PredictionResult]:
        """执行趋势预测"""
        key = f"{zone_id}:{sensor_id}"
        with self.lock:
            if key not in self.history:
                return None
            points = self.history[key]

        if len(points) < self.min_data_points:
            return None

        # 取最近N个点进行拟合
        recent = points[-self.min_data_points:]
        timestamps = np.array([p.timestamp for p in recent])
        temperatures = np.array([p.temperature for p in recent])
        humidities = np.array([p.humidity for p in recent])

        # 线性拟合
        temp_coeffs = np.polyfit(timestamps, temperatures, 1)
        humi_coeffs = np.polyfit(timestamps, humidities, 1)

        temp_slope = temp_coeffs[0] * 3600  # ℃/h
        humi_slope = humi_coeffs[0] * 3600  # %RH/h

        current_time = timestamps[-1]
        current_temp = temperatures[-1]
        current_humi = humidities[-1]

        # 预测未来值
        future_time = current_time + self.prediction_window
        predicted_temp = np.polyval(temp_coeffs, future_time)
        predicted_humi = np.polyval(humi_coeffs, future_time)

        # 计算距离超标的时间
        time_to_temp_limit = self._time_to_limit(
            current_temp, temp_slope, self.temp_upper, self.temp_lower
        )
        time_to_humi_limit = self._time_to_limit(
            current_humi, humi_slope, self.humi_upper, self.humi_lower
        )

        # 判定预警等级
        alert_level, alert_reason = self._determine_alert_level(
            current_temp, current_humi,
            predicted_temp, predicted_humi,
            time_to_temp_limit, time_to_humi_limit
        )

        return PredictionResult(
            zone_id=zone_id,
            sensor_id=sensor_id,
            current_temp=current_temp,
            current_humi=current_humi,
            temp_slope=temp_slope,
            humi_slope=humi_slope,
            predicted_temp=predicted_temp,
            predicted_humi=predicted_humi,
            time_to_temp_limit=time_to_temp_limit,
            time_to_humi_limit=time_to_humi_limit,
            alert_level=alert_level,
            alert_reason=alert_reason
        )

    def _time_to_limit(self, current: float, slope: float,
                      upper: float, lower: float) -> Optional[float]:
        """计算距离超标的时间（秒）"""
        if abs(slope) < 0.01:  # 变化极小
            return None

        if slope > 0 and current < upper:
            # 上升趋势，检查上限
            return (upper - current) / (slope / 3600)
        elif slope < 0 and current > lower:
            # 下降趋势，检查下限
            return (current - lower) / (abs(slope) / 3600)

        return None

    def _determine_alert_level(self, current_temp: float, current_humi: float,
                              predicted_temp: float, predicted_humi: float,
                              time_to_temp: Optional[float],
                              time_to_humi: Optional[float]) -> Tuple[AlertLevel, str]:
        """判定预警等级"""
        reasons = []

        # 已超标
        if current_temp > self.temp_upper or current_temp < self.temp_lower:
            reasons.append(f"当前温度已超标({current_temp:.1f}℃)")
            return AlertLevel.CRITICAL, "; ".join(reasons)
        if current_humi > self.humi_upper or current_humi < self.humi_lower:
            reasons.append(f"当前湿度已超标({current_humi:.1f}%RH)")
            return AlertLevel.CRITICAL, "; ".join(reasons)

        # 预测将超标
        min_time = None
        if time_to_temp is not None:
            min_time = time_to_temp if min_time is None else min(min_time, time_to_temp)
            if time_to_temp <= 600:  # 10分钟内
                reasons.append(f"温度预计{time_to_temp/60:.0f}分钟后超标")
            elif time_to_temp <= 1800:  # 30分钟内
                reasons.append(f"温度预计{time_to_temp/60:.0f}分钟后超标")

        if time_to_humi is not None:
            min_time = time_to_humi if min_time is None else min(min_time, time_to_humi)
            if time_to_humi <= 600:
                reasons.append(f"湿度预计{humi_time/60:.0f}分钟后超标")
            elif time_to_humi <= 1800:
                reasons.append(f"湿度预计{humi_time/60:.0f}分钟后超标")

        if not reasons:
            return AlertLevel.INFO, "趋势正常"

        # 根据最短超标时间判定等级
        if min_time is not None:
            if min_time <= 600:
                return AlertLevel.CRITICAL, "; ".join(reasons)
            elif min_time <= 1800:
                return AlertLevel.WARNING, "; ".join(reasons)
            elif min_time <= 3600:
                return AlertLevel.ATTENTION, "; ".join(reasons)

        return AlertLevel.INFO, "; ".join(reasons)

    def predict_all_zones(self, zone_sensors: Dict[str, List[str]]) -> List[PredictionResult]:
        """对所有区域的所有传感器进行预测"""
        results = []
        for zone_id, sensors in zone_sensors.items():
            for sensor_id in sensors:
                result = self.predict(zone_id, sensor_id)
                if result:
                    results.append(result)
        return results

# medium_term_predictor.py
import numpy as np
from dataclasses import dataclass
from typing import Dict, List, Optional
import math

@dataclass
class WeatherForecast:
    """室外天气预报数据"""
    timestamp: float
    outdoor_temp: float
    outdoor_humi: float
    solar_radiation: float  # W/m²
    wind_speed: float       # m/s

class PeriodicPredictor:
    """基于周期模型的中期预测"""

    def __init__(self, history_days: int = 7):
        self.history_days = history_days
        self.hourly_baselines: Dict[int, dict] = {}  # hour -> {temp_mean, humi_mean, ...}
        self.lock = threading.RLock()

    def build_baseline(self, historical_data: List[TrendPoint]):
        """基于历史数据建立逐时基线"""
        with self.lock:
            hourly_values = {h: {'temps': [], 'humis': []} for h in range(24)}

            for point in historical_data:
                hour = int((point.timestamp % 86400) / 3600)
                hourly_values[hour]['temps'].append(point.temperature)
                hourly_values[hour]['humis'].append(point.humidity)

            for hour, values in hourly_values.items():
                if values['temps']:
                    self.hourly_baselines[hour] = {
                        'temp_mean': np.mean(values['temps']),
                        'temp_std': np.std(values['temps']),
                        'humi_mean': np.mean(values['humis']),
                        'humi_std': np.std(values['humis'])
                    }

    def predict_hourly(self, hour: int,
                      weather: Optional[WeatherForecast] = None) -> dict:
        """预测指定小时的环境参数"""
        if hour not in self.hourly_baselines:
            return {"status": "no_baseline"}

        baseline = self.hourly_baselines[hour]
        predicted_temp = baseline['temp_mean']
        predicted_humi = baseline['humi_mean']

        # 如果有天气预报，进行修正
        if weather:
            # 室外温度对库房的影响（简化模型）
            # 影响系数取决于外墙传热、遮阳、隔热等
            temp_influence_factor = 0.15  # 室外温度变化1℃，库房变化0.15℃
            predicted_temp += (weather.outdoor_temp - 25.0) * temp_influence_factor

            # 太阳辐射影响
            solar_factor = 0.002  # 每增加100W/m²，温度上升约0.2℃
            predicted_temp += weather.solar_radiation * solar_factor

            # 湿度影响（室外湿度对库房的影响较小，主要是新风引入时）
            if weather.outdoor_humi > 70:
                humi_influence_factor = 0.05
                predicted_humi += (weather.outdoor_humi - 50) * humi_influence_factor

        return {
            'hour': hour,
            'predicted_temp': predicted_temp,
            'predicted_humi': predicted_humi,
            'temp_std': baseline['temp_std'],
            'humi_std': baseline['humi_std'],
            'confidence': self._calculate_confidence(baseline)
        }

    def _calculate_confidence(self, baseline: dict) -> float:
        """计算预测置信度"""
        # 基于历史数据的波动程度
        temp_cv = baseline['temp_std'] / baseline['temp_mean'] if baseline['temp_mean'] > 0 else 1
        humi_cv = baseline['humi_std'] / baseline['humi_mean'] if baseline['humi_mean'] > 0 else 1

        # 波动越小，置信度越高
        confidence = max(0, 100 - (temp_cv * 200 + humi_cv * 200))
        return min(100, confidence)

    def generate_daily_forecast(self,
                               weather_forecast: List[WeatherForecast]) -> List[dict]:
        """生成全天预测"""
        forecast = []
        for weather in weather_forecast:
            hour = int((weather.timestamp % 86400) / 3600)
            prediction = self.predict_hourly(hour, weather)
            forecast.append(prediction)
        return forecast

# anomaly_detector.py
import numpy as np
from typing import Dict, List, Optional
from dataclasses import dataclass

@dataclass
class AnomalyEvent:
    """异常事件"""
    sensor_id: str
    zone_id: str
    anomaly_type: str
    severity: str          # low / medium / high / critical
    description: str
    detected_at: float
    metric_value: float
    expected_value: Optional[float] = None
    threshold: Optional[float] = None

class AnomalyDetector:
    """多维度异常检测器"""

    def __init__(self,
                 temp_rate_threshold: float = 2.0,    # ℃/h
                 humi_rate_threshold: float = 10.0,   # %RH/h
                 spatial_deviation_threshold: float = 1.5,  # ℃
                 history_comparison_window: int = 7):  # 与最近N天的同期对比
        self.temp_rate_threshold = temp_rate_threshold
        self.humi_rate_threshold = humi_rate_threshold
        self.spatial_deviation_threshold = spatial_deviation_threshold
        self.history_comparison_window = history_comparison_window
        self.predictor = None  # 引用TrendPredictor

    def set_predictor(self, predictor):
        self.predictor = predictor

    def detect_rate_anomaly(self, zone_id: str, sensor_id: str) -> Optional[AnomalyEvent]:
        """检测变化速率异常"""
        if not self.predictor:
            return None

        prediction = self.predictor.predict(zone_id, sensor_id)
        if not prediction:
            return None

        anomalies = []

        # 温度变化速率异常
        if abs(prediction.temp_slope) > self.temp_rate_threshold:
            severity = "critical" if abs(prediction.temp_slope) > self.temp_rate_threshold * 2 else "high"
            anomalies.append(AnomalyEvent(
                sensor_id=sensor_id,
                zone_id=zone_id,
                anomaly_type="temperature_rate",
                severity=severity,
                description=f"温度变化速率异常: {prediction.temp_slope:+.1f}℃/h",
                detected_at=time.time(),
                metric_value=prediction.temp_slope,
                threshold=self.temp_rate_threshold
            ))

        # 湿度变化速率异常
        if abs(prediction.humi_slope) > self.humi_rate_threshold:
            severity = "critical" if abs(prediction.humi_slope) > self.humi_rate_threshold * 2 else "high"
            anomalies.append(AnomalyEvent(
                sensor_id=sensor_id,
                zone_id=zone_id,
                anomaly_type="humidity_rate",
                severity=severity,
                description=f"湿度变化速率异常: {prediction.humi_slope:+.1f}%RH/h",
                detected_at=time.time(),
                metric_value=prediction.humi_slope,
                threshold=self.humi_rate_threshold
            ))

        return anomalies if anomalies else None

    def detect_spatial_anomaly(self, zone_readings: Dict[str, float]) -> List[AnomalyEvent]:
        """检测空间一致性异常"""
        if len(zone_readings) < 3:
            return []

        values = list(zone_readings.values())
        mean_val = np.mean(values)
        anomalies = []

        for sensor_id, value in zone_readings.items():
            deviation = abs(value - mean_val)
            if deviation > self.spatial_deviation_threshold:
                severity = "high" if deviation > self.spatial_deviation_threshold * 2 else "medium"
                anomalies.append(AnomalyEvent(
                    sensor_id=sensor_id,
                    zone_id="unknown",
                    anomaly_type="spatial_deviation",
                    severity=severity,
                    description=f"传感器读数与区域均值偏差过大: {deviation:.1f}℃",
                    detected_at=time.time(),
                    metric_value=value,
                    expected_value=mean_val,
                    threshold=self.spatial_deviation_threshold
                ))

        return anomalies

    def detect_historical_deviation(self, current_value: float,
                                   historical_baseline: float,
                                   baseline_std: float) -> Optional[AnomalyEvent]:
        """检测与历史基线的偏离"""
        deviation = abs(current_value - historical_baseline)
        z_score = deviation / baseline_std if baseline_std > 0 else 0

        if z_score > 3:  # 超过3个标准差
            return AnomalyEvent(
                sensor_id="aggregate",
                zone_id="all",
                anomaly_type="historical_deviation",
                severity="high" if z_score > 4 else "medium",
                description=f"与历史同期偏离{z_score:.1f}个标准差",
                detected_at=time.time(),
                metric_value=current_value,
                expected_value=historical_baseline,
                threshold=baseline_std * 3
            )
        return None

# alert_aggregator.py
from typing import Dict, List, Optional
from dataclasses import dataclass
from datetime import datetime

@dataclass
class AggregatedAlert:
    """聚合后的预警"""
    alert_id: str
    zone_id: str
    alert_level: AlertLevel
    triggered_sensors: List[str]
    primary_reason: str
    prediction_results: List[PredictionResult]
    anomaly_events: List[AnomalyEvent]
    created_at: float
    acknowledged: bool = False
    auto_resolved: bool = False

class AlertAggregator:
    """预警聚合器"""

    def __init__(self, aggregation_window: float = 300):  # 5分钟聚合窗口
        self.aggregation_window = aggregation_window
        self.pending_alerts: Dict[str, List[PredictionResult]] = {}
        self.active_alerts: Dict[str, AggregatedAlert] = {}
        self.lock = threading.RLock()

    def add_prediction(self, prediction: PredictionResult):
        """添加预测结果到聚合缓存"""
        if prediction.alert_level == AlertLevel.INFO:
            return

        key = f"{prediction.zone_id}"
        with self.lock:
            if key not in self.pending_alerts:
                self.pending_alerts[key] = []
            self.pending_alerts[key].append(prediction)

    def aggregate(self) -> List[AggregatedAlert]:
        """执行聚合，生成聚合预警"""
        aggregated = []
        current_time = time.time()

        with self.lock:
            for zone_id, predictions in list(self.pending_alerts.items()):
                # 过滤出在聚合窗口内的预测
                recent = [p for p in predictions
                         if current_time - current_time <= self.aggregation_window]

                if not recent:
                    continue

                # 取最高预警等级
                max_level = max(p.alert_level for p in recent)

                # 确认预警（需要至少2个传感器或持续2个周期确认）
                confirmed = self._confirm_alert(zone_id, recent)

                if confirmed:
                    alert_id = f"{zone_id}_{int(current_time)}"
                    aggregated_alert = AggregatedAlert(
                        alert_id=alert_id,
                        zone_id=zone_id,
                        alert_level=max_level,
                        triggered_sensors=[p.sensor_id for p in recent],
                        primary_reason=self._summarize_reason(recent),
                        prediction_results=recent,
                        anomaly_events=[],
                        created_at=current_time
                    )
                    self.active_alerts[alert_id] = aggregated_alert
                    aggregated.append(aggregated_alert)

                # 清空已处理的预测
                self.pending_alerts[zone_id] = []

        return aggregated

    def _confirm_alert(self, zone_id: str, predictions: List[PredictionResult]) -> bool:
        """确认预警是否有效（防误报）"""
        # 规则1: 至少2个传感器同时预警
        if len(predictions) >= 2:
            return True

        # 规则2: 单个传感器但预警等级为CRITICAL
        if len(predictions) == 1 and predictions[0].alert_level == AlertLevel.CRITICAL:
            return True

        # 规则3: 连续2次预测都预警（在下次聚合时检查）
        # 这里简化处理，实际应检查历史记录
        return False

    def _summarize_reason(self, predictions: List[PredictionResult]) -> str:
        """汇总预警原因"""
        reasons = []
        for p in predictions:
            if p.alert_reason:
                reasons.append(f"{p.sensor_id}: {p.alert_reason}")
        return "; ".join(reasons)

    def resolve_alert(self, alert_id: str):
        """手动确认解决预警"""
        with self.lock:
            if alert_id in self.active_alerts:
                self.active_alerts[alert_id].acknowledged = True

    def auto_resolve_expired(self, predictor: TrendPredictor,
                           zone_sensors: Dict[str, List[str]]):
        """自动解除已恢复的预警"""
        current_time = time.time()
        with self.lock:
            for alert_id, alert in list(self.active_alerts.items()):
                if alert.acknowledged:
                    continue

                # 重新检查该区域的所有传感器
                all_clear = True
                for sensor_id in alert.triggered_sensors:
                    prediction = predictor.predict(alert.zone_id, sensor_id)
                    if prediction and prediction.alert_level != AlertLevel.INFO:
                        all_clear = False
                        break

                if all_clear:
                    alert.auto_resolved = True
                    del self.active_alerts[alert_id]

┌─────────────────────────────────────────────────────────────────────┐
│ 预调节决策流程                                                     │
│                                                                     │
│  预警触发                                                          │
│     │                                                               │
│     ▼                                                               │
│  分析超标方向                                                       │
│  ┌────┴────┐                                                        │
│  │温度偏高│温度偏低│湿度偏高│湿度偏低│                              │
│  └────┬────┴────┬───┴────┬───┴────┬──┘                            │
│       │         │        │        │                                 │
│       ▼         ▼        ▼        ▼                                 │
│  ┌────────┐ ┌────────┐ ┌────────┐ ┌────────┐                      │
│  │预冷策略│ │预热策略│ │除湿策略│ │加湿策略│                      │
│  └───┬────┘ └───┬────┘ └───┬────┘ └───┬────┘                      │
│      │          │          │          │                             │
│      ▼          ▼          ▼          ▼                             │
│  ┌─────────────────────────────────────────────┐                   │
│  │ 设备选择（基于分区、优先级、互锁）          │                   │
│  └──────────────────────┬──────────────────────┘                   │
│                         │                                          │
│                         ▼                                          │
│  ┌─────────────────────────────────────────────┐                   │
│  │ 执行力度计算（轻度/中度/全力）              │                   │
│  │ · 轻度：预计刚好在阈值前拉回 → 单设备低功率│                   │
│  │ · 中度：趋势较陡 → 单设备全功率            │                   │
│  │ · 全力：多个区域同时预警 → 多设备联动      │                   │
│  └──────────────────────┬──────────────────────┘                   │
│                         │                                          │
│                         ▼                                          │
│  ┌─────────────────────────────────────────────┐                   │
│  │ 下发控制指令 → 设备执行                     │                   │
│  └──────────────────────┬──────────────────────┘                   │
│                         │                                          │
│                         ▼                                          │
│  ┌─────────────────────────────────────────────┐                   │
│  │ 效果跟踪（5分钟后检查趋势是否改善）         │                   │
│  │ · 改善 → 维持当前策略                      │                   │
│  │ · 未改善 → 升级力度或增加设备              │                   │
│  │ · 恶化 → 紧急模式（所有可用设备全力）      │                   │
│  └─────────────────────────────────────────────┘                   │
└─────────────────────────────────────────────────────────────────────┘

# pre_adjustment.py
from enum import Enum
from dataclasses import dataclass
from typing import Dict, List, Optional

class AdjustmentAction(Enum):
    PRECOOL = "precool"           # 预冷
    PREHEAT = "preheat"           # 预热
    DEHUMIDIFY = "dehumidify"     # 预除湿
    HUMIDIFY = "humidify"         # 预加湿
    INCREASE_VENTILATION = "increase_ventilation"  # 增大新风
    REDUCE_VENTILATION = "reduce_ventilation"      # 减少新风
    NONE = "none"

class AdjustmentIntensity(Enum):
    LIGHT = "light"       # 轻度：单设备低功率
    MODERATE = "moderate" # 中度：单设备全功率
    FULL = "full"         # 全力：多设备联动

@dataclass
class PreAdjustmentCommand:
    """预调节指令"""
    zone_id: str
    action: AdjustmentAction
    intensity: AdjustmentIntensity
    target_devices: List[str]
    estimated_duration: float  # 预计持续时间（秒）
    reason: str
    predicted_exceed_time: Optional[float] = None

class PreAdjustmentEngine:
    """预调节决策引擎"""

    def __init__(self, zone_controller):
        self.zone_controller = zone_controller
        self.active_commands: Dict[str, PreAdjustmentCommand] = {}
        self.lock = threading.RLock()

    def generate_command(self, prediction: PredictionResult) -> Optional[PreAdjustmentCommand]:
        """根据预测结果生成预调节指令"""
        zone_id = prediction.zone_id

        # 判断需要什么方向的调节
        action = self._determine_action(prediction)
        if action == AdjustmentAction.NONE:
            return None

        # 计算调节力度
        intensity = self._calculate_intensity(prediction)

        # 选择设备
        target_devices = self._select_devices(zone_id, action)

        if not target_devices:
            return None

        # 计算预计持续时间
        estimated_duration = self._estimate_duration(prediction, action, intensity)

        command = PreAdjustmentCommand(
            zone_id=zone_id,
            action=action,
            intensity=intensity,
            target_devices=target_devices,
            estimated_duration=estimated_duration,
            reason=prediction.alert_reason,
            predicted_exceed_time=prediction.time_to_temp_limit or prediction.time_to_humi_limit
        )

        with self.lock:
            self.active_commands[zone_id] = command

        return command

    def _determine_action(self, prediction: PredictionResult) -> AdjustmentAction:
        """确定调节方向"""
        # 温度偏高
        if prediction.time_to_temp_limit and prediction.temp_slope > 0:
            if prediction.predicted_temp > prediction.temp_upper:
                return AdjustmentAction.PRECOOL

        # 温度偏低
        if prediction.time_to_temp_limit and prediction.temp_slope < 0:
            if prediction.predicted_temp < prediction.temp_lower:
                return AdjustmentAction.PREHEAT

        # 湿度偏高
        if prediction.time_to_humi_limit and prediction.humi_slope > 0:
            if prediction.predicted_humi > prediction.humi_upper:
                return AdjustmentAction.DEHUMIDIFY

        # 湿度偏低
        if prediction.time_to_humi_limit and prediction.humi_slope < 0:
            if prediction.predicted_humi < prediction.humi_lower:
                return AdjustmentAction.HUMIDIFY

        return AdjustmentAction.NONE

    def _calculate_intensity(self, prediction: PredictionResult) -> AdjustmentIntensity:
        """计算调节力度"""
        # 根据距离超标的时间和变化速率判断
        min_time = None
        if prediction.time_to_temp_limit:
            min_time = prediction.time_to_temp_limit
        if prediction.time_to_humi_limit:
            if min_time is None:
                min_time = prediction.time_to_humi_limit
            else:
                min_time = min(min_time, prediction.time_to_humi_limit)

        if min_time is None:
            return AdjustmentIntensity.LIGHT

        if min_time <= 600:  # 10分钟内超标
            return AdjustmentIntensity.FULL
        elif min_time <= 1800:  # 30分钟内超标
            return AdjustmentIntensity.MODERATE
        else:
            return AdjustmentIntensity.LIGHT

    def _select_devices(self, zone_id: str, action: AdjustmentAction) -> List[str]:
        """选择目标设备"""
        if zone_id not in self.zone_controller.zones:
            return []

        zone_config = self.zone_controller.zones[zone_id]
        devices = zone_config.devices

        selected = []
        for device in devices:
            if action == AdjustmentAction.PRECOOL and 'ac' in device.lower():
                selected.append(device)
            elif action == AdjustmentAction.PREHEAT and 'heater' in device.lower():
                selected.append(device)
            elif action == AdjustmentAction.DEHUMIDIFY and 'dehumidifier' in device.lower():
                selected.append(device)
            elif action == AdjustmentAction.HUMIDIFY and 'humidifier' in device.lower():
                selected.append(device)

        return selected

    def _estimate_duration(self, prediction: PredictionResult,
                          action: AdjustmentAction,
                          intensity: AdjustmentIntensity) -> float:
        """估计调节持续时间"""
        # 简化模型：根据超标幅度和设备能力估算
        base_duration = 1800  # 基础30分钟

        if prediction.time_to_temp_limit:
            exceed_magnitude = abs(prediction.predicted_temp - prediction.temp_upper)
        else:
            exceed_magnitude = abs(prediction.predicted_humi - prediction.humi_upper)

        # 幅度越大，需要的时间越长
        duration = base_duration + exceed_magnitude * 300  # 每超标0.1℃/RH增加5分钟

        # 力度系数
        intensity_factor = {
            AdjustmentIntensity.LIGHT: 1.5,
            AdjustmentIntensity.MODERATE: 1.0,
            AdjustmentIntensity.FULL: 0.7
        }

        return duration * intensity_factor.get(intensity, 1.0)

    def check_effectiveness(self, zone_id: str,
                          current_prediction: PredictionResult) -> str:
        """检查预调节效果"""
        with self.lock:
            if zone_id not in self.active_commands:
                return "no_active_command"

            command = self.active_commands[zone_id]

            # 检查趋势是否改善
            if current_prediction.alert_level == AlertLevel.INFO:
                # 预警已解除
                del self.active_commands[zone_id]
                return "resolved"

            if current_prediction.time_to_temp_limit and current_prediction.time_to_temp_limit > 3600:
                # 超标时间推迟到1小时以上，说明有效
                return "improving"

            if current_prediction.time_to_temp_limit and current_prediction.time_to_temp_limit < 600:
                # 超标时间仍在10分钟以内，需要升级
                return "needs_escalation"

            return "monitoring"

┌─────────────────────────────────────────────────────────────────────┐
│ 部署架构图                                                         │
│                                                                     │
│  ┌───────────────────────────────────────────────────────────────┐ │
│  │ 边缘层（库房本地）                                             │ │
│  │  · 传感器数据采集网关（每库房1台）                             │ │
│  │  · 本地预测引擎（轻量级，运行短期预测）                        │ │
│  │  · 本地设备控制器（执行预调节指令）                            │ │
│  │  · 断网续传缓存（本地存储7天数据）                             │ │
│  └───────────────────────────┬───────────────────────────────────┘ │
│                              │ 4G/光纤                             │
│  ┌───────────────────────────▼───────────────────────────────────┐ │
│  │ 平台层（中心服务器）                                           │ │
│  │  · 中期/长期预测引擎                                           │ │
│  │  · 异常检测与预警聚合                                          │ │
│  │  · 历史数据分析与模型自优化                                     │ │
│  │  · 预警推送服务（短信/APP/邮件）                              │ │
│  └───────────────────────────────────────────────────────────────┘ │
│                                                                     │
│  关键设计：                                                         │
│  - 边缘层独立运行：即使与平台失联，本地预测和预调节仍能工作        │
│  - 平台层负责全局优化：跨库房对比、季节性模型更新                  │
│  - 数据同步：边缘层定时上报数据，平台层下发模型参数更新            │
└─────────────────────────────────────────────────────────────────────┘