机关单位档案室智能化改造：从单点监测到十二防一体化系统演进

阶段一              阶段二              阶段三              阶段四
单点监测    ──▶   联网汇聚    ──▶   联动控制    ──▶   一体化决策
（看得见）          （看得到）          （动得了）          （判得准）

传感器层 → RS485/网口/无线 → 边缘网关 → TCP/IP → 监控平台
                                     │
                              本地缓存+断网续传
                                     │
                              就地联动逻辑

# linkage_engine.py
from enum import Enum
from dataclasses import dataclass

class EventType(Enum):
    TEMP_HIGH = "temp_high"
    TEMP_LOW = "temp_low"
    HUMI_HIGH = "humi_high"
    HUMI_LOW = "humi_low"
    SMOKE = "smoke"
    WATER_LEAK = "water_leak"
    PM25_HIGH = "pm25_high"
    DOOR_ALARM = "door_alarm"

@dataclass
class Action:
    device: str
    command: str   # start/stop/set
    param: dict = None

# 联动规则表：优先级高的先执行
LINKAGE_RULES = [
    # 消防最高优先级
    {"event": EventType.SMOKE, "priority": 0,
     "actions": [Action("ac", "stop"), Action("fresh_air", "stop"),
                 Action("exhaust", "start"), Action("siren", "start")]},
    # 水浸次高
    {"event": EventType.WATER_LEAK, "priority": 1,
     "actions": [Action("humidifier", "stop"), Action("dehumidifier", "stop"),
                 Action("alarm", "start")]},
    # 温湿度
    {"event": EventType.TEMP_HIGH, "priority": 2,
     "actions": [Action("ac", "start"), Action("ac", "set", {"mode": "cool"})]},
    {"event": EventType.HUMI_HIGH, "priority": 2,
     "actions": [Action("dehumidifier", "start")]},
    {"event": EventType.HUMI_LOW, "priority": 2,
     "actions": [Action("humidifier", "start")]},
    # 安防
    {"event": EventType.DOOR_ALARM, "priority": 1,
     "actions": [Action("camera", "start_rec"), Action("siren", "start")]},
    # 空气质量
    {"event": EventType.PM25_HIGH, "priority": 3,
     "actions": [Action("purifier", "start"), Action("purifier", "set", {"level": "high"})]},
]

class LinkageEngine:
    def __init__(self):
        self.interlocks = {
            ("dehumidifier", "humidifier"): "mutex",  # 互斥
            ("ac", "heater"): "mutex",
        }

    def handle_event(self, event: EventType, zone: str) -> list:
        """处理事件，返回执行动作列表"""
        rules = sorted(
            [r for r in LINKAGE_RULES if r["event"] == event],
            key=lambda x: x["priority"]
        )
        executed = []
        for rule in rules:
            for action in rule["actions"]:
                if self._check_interlock(action, executed):
                    executed.append(action)
        return executed

    def _check_interlock(self, action: Action, executed: list) -> bool:
        """检查互锁，防止冲突动作同时执行"""
        for (dev_a, dev_b), kind in self.interlocks.items():
            if kind == "mutex":
                if action.device == dev_a:
                    for e in executed:
                        if e.device == dev_b:
                            return False  # 互锁设备已在运行
                if action.device == dev_b:
                    for e in executed:
                        if e.device == dev_a:
                            return False
        return True

┌────────────────────────────────────────────────────┐
│               一体化决策层                         │
│  · 多源数据融合  · 风险等级评估  · 预案自动匹配  │
└───────────────────┬────────────────────────────────┘
                    │
┌───────────────────┼────────────────────────────────┐
│ 联动执行层        │       │ 感知接入层             │
│ 空调/除湿/新风    │       │ 温湿度/烟感/门磁/PM2.5│
│ 消防/门禁/净化    │       │ 照度/静电/水浸/VOC    │
└────────────────────────────────────────────────────┘

# risk_assessor.py
from dataclasses import dataclass
from typing import Dict

@dataclass
class RiskFactor:
    name: str
    weight: float      # 权重
    value: float       # 当前值
    threshold: float   # 阈值

# 十二防风险权重
RISK_WEIGHTS = {
    "fire":      0.25,
    "water":     0.15,
    "theft":     0.15,
    "temp":      0.12,
    "humi":      0.12,
    "air":       0.08,
    "light":     0.05,
    "static":    0.04,
    "mold":      0.02,
    "pest":      0.02,
}

class RiskAssessor:
    def evaluate(self, readings: Dict[str, float]) -> dict:
        """综合评估当前风险等级"""
        score = 0.0
        details = {}

        # 防火
        smoke = readings.get("smoke", 0)
        if smoke > 0:
            score += RISK_WEIGHTS["fire"] * min(smoke / 5.0, 1.0)
            details["fire"] = "高风险" if smoke > 3 else "关注"

        # 防潮/高温
        temp = readings.get("temperature", 22)
        if temp > 24 or temp < 14:
            deviation = max(abs(temp - 24), abs(temp - 14)) / 10
            score += RISK_WEIGHTS["temp"] * min(deviation, 1.0)
            details["temp"] = "超标"

        humi = readings.get("humidity", 50)
        if humi > 60 or humi < 45:
            deviation = max(abs(humi - 60), abs(humi - 45)) / 15
            score += RISK_WEIGHTS["humi"] * min(deviation, 1.0)
            details["humi"] = "超标"

        # 防水
        water = readings.get("water_leak", 0)
        if water > 0:
            score += RISK_WEIGHTS["water"]
            details["water"] = "漏水"

        # 防有害气体
        voc = readings.get("voc", 0)
        if voc > 0.6:
            score += RISK_WEIGHTS["air"] * min(voc / 1.0, 1.0)
            details["air"] = "有害气体超标"

        # 防光
        lux = readings.get("illuminance", 0)
        if lux > 300:
            score += RISK_WEIGHTS["light"] * min(lux / 500, 1.0)
            details["light"] = "光照过强"

        # 防静电
        static = readings.get("static", 0)
        if static > 2000:
            score += RISK_WEIGHTS["static"] * min(static / 5000, 1.0)
            details["static"] = "静电偏高"

        # 确定风险等级
        if score >= 0.5:
            level = "高风险"
        elif score >= 0.25:
            level = "中风险"
        elif score >= 0.1:
            level = "低风险"
        else:
            level = "正常"

        return {"score": round(score, 3), "level": level, "details": details}