档案库房环境监测系统的事件驱动设计：异常温湿度自动联动策略

┌────────────────────────────────────────────────────────────────────┐
│  事件等级           │ 触发条件                    │ 响应要求        │
├─────────────────────┼─────────────────────────────┼────────────────┤
│  P0 紧急            │ 温度>32℃ 或 湿度>70%RH     │ <10s 自动联动  │
│                     │ 或 消防报警                  │ 短信+电话通知  │
│  P1 严重            │ 温度>28℃ 或 湿度>60%RH     │ <30s 自动联动  │
│                     │ 持续>15min 或 设备离线       │ 平台弹窗+短信  │
│  P2 警告            │ 偏离目标带±3%RH/±2℃        │ <60s 记录+预警 │
│                     │ 持续>30min                  │ 平台弹窗       │
│  P3 提示            │ 偏离预警带±1.5%RH/±1℃      │ <120s 记录     │
│                     │ 设备维护提醒                 │ 工单生成       │
└────────────────────────────────────────────────────────────────────┘

from dataclasses import dataclass, field
from enum import Enum
from time import time
from typing import Optional

class EventLevel(Enum):
    P0_CRITICAL = 0
    P1_MAJOR = 1
    P2_WARNING = 2
    P3_INFO = 3

class EventType(Enum):
    TEMP_HIGH = "temp_high"
    TEMP_LOW = "temp_low"
    HUM_HIGH = "hum_high"
    HUM_LOW = "hum_low"
    DEVICE_OFFLINE = "device_offline"
    DEVICE_FAULT = "device_fault"
    LINKAGE_SUCCESS = "linkage_success"
    LINKAGE_FAILED = "linkage_failed"

@dataclass
class EnvEvent:
    event_id: str
    event_type: EventType
    level: EventLevel
    zone_id: str
    sensor_id: str
    trigger_value: float
    threshold: float
    timestamp: float = field(default_factory=time)
    linked_actions: list = field(default_factory=list)
    acknowledged: bool = False
    resolved_at: Optional[float] = None

@dataclass
class LinkageRule:
    rule_id: str
    event_type: EventType
    level: EventLevel
    condition: str          # 附加条件表达式
    actions: list           # 动作列表
    cooldown: int = 300     # 同一规则冷却时间（秒）
    max_retries: int = 3
    escalation_after: int = 600  # 未恢复则升级（秒）

class LinkageEngine:
    def __init__(self):
        self.rules: dict = {}
        self.event_history: list = []
        self._last_fire: dict = {}   # rule_id -> timestamp

    def register_rule(self, rule: LinkageRule):
        self.rules[rule.rule_id] = rule

    def on_event(self, event: EnvEvent) -> list:
        results = []
        now = time()

        for rule in self.rules.values():
            if rule.event_type != event.event_type:
                continue
            if rule.level.value > event.level.value:
                continue

            # 冷却检查
            last = self._last_fire.get(rule.rule_id, 0)
            if now - last < rule.cooldown:
                continue

            # 附加条件评估
            if rule.condition and not self._eval_condition(rule.condition, event):
                continue

            # 执行联动动作
            for action in rule.actions:
                ok = self._execute_action(action, event)
                results.append({
                    'rule_id': rule.rule_id,
                    'action': action,
                    'success': ok,
                    'timestamp': now
                })

            self._last_fire[rule.rule_id] = now
            event.linked_actions.extend(results)

        self.event_history.append(event)
        return results

    def _eval_condition(self, condition: str, event: EnvEvent) -> bool:
        # 简单条件表达式解析：如 "duration>900" "gradient>5"
        # 实际工程中可接入规则引擎（Drools/Rete）
        return True

    def _execute_action(self, action: dict, event: EnvEvent) -> bool:
        atype = action.get('type')
        if atype == 'start_device':
            return modbus_write(action['device_id'], action['register'], action['value'])
        elif atype == 'stop_device':
            return modbus_write(action['device_id'], action['register'], 0)
        elif atype == 'close_valve':
            return gpio_write(action['valve_id'], 0)
        elif atype == 'alarm':
            return send_alarm(action.get('channel'), event)
        elif atype == 'sms':
            return send_sms(action['to'], f"档案库房告警:{event.event_type.value} 值:{event.trigger_value}")
        return False

场景：特藏库温度持续上升

T+0s   传感器上报 28.5℃ → 触发 R003（P1）
       → 启动备用空调
       → 平台弹窗 + 短信通知

T+60s  温度 29.8℃ → 事件升级
       → 关闭该区照明和多余设备
       → 打开相邻区补风阀

T+180s 温度 31.2℃ → 升级为 P0
       → 启动全部可用制冷设备
       → 电话通知值班主管
       → 触发消防预检（检查是否有火情）

T+300s 温度 32.5℃ → 触发熔断规则
       → 停止非关键区空调，集中冷量到特藏库
       → 启动应急排风
       → 生成紧急事件报告

class CascadeBreaker:
    def __init__(self, max_events_per_min=20, max_linkage_per_event=5):
        self._event_count = []
        self._linkage_count = {}
        self.max_events = max_events_per_min
        self.max_linkage = max_linkage_per_event

    def allow_event(self) -> bool:
        now = time()
        self._event_count = [t for t in self._event_count if now - t < 60]
        if len(self._event_count) >= self.max_events:
            return False
        self._event_count.append(now)
        return True

    def allow_linkage(self, event_id: str) -> bool:
        count = self._linkage_count.get(event_id, 0)
        if count >= self.max_linkage:
            return False
        self._linkage_count[event_id] = count + 1
        return True

def audit_trail(zone_id: str, start: float, end: float) -> list:
    """按时间范围查询某分区的完整事件溯源链"""
    return [e for e in event_store
            if e.zone_id == zone_id
            and start <= e.timestamp <= end]