档案库房物联网搭建：温湿度、漏水、防盗多传感器联动调试实战

┌──────────────────────────────────────────────────────────────────┐
│  决策层（联动策略引擎）                                           │
│  ├── 规则引擎（IF-THEN + 优先级 + 抑制窗）                      │
│  ├── 设备状态机（模式管理 + 安全联锁）                           │
│  └── 审计日志（操作 trace + 环境曲线 + 处置记录）                │
└──────────────────────┬───────────────────────────────────────────┘
                       │ MQTT / HTTP API
                       │ 策略下发 + 状态回传
┌──────────────────────▼───────────────────────────────────────────┐
│  控制层（DDC / 边缘网关）                                        │
│  ├── 温湿度 PID 回路（暖通联动）                                 │
│  ├── 漏水定位 + 关阀逻辑（水浸联动）                             │
│  ├── 门禁 + 红外 + 震动（防盗联动）                              │
│  └── 本地自治（网络断开时独立运行）                              │
└──────────────────────┬───────────────────────────────────────────┘
                       │ Modbus TCP / BACnet / 干接点
┌──────────────────────▼───────────────────────────────────────────┐
│  感知层（多模态传感器网络）                                       │
│  ├── 温湿度：高精度变送器（回风口 + 工作区 + 角落）              │
│  ├── 漏水：定位式漏水绳 + 区域式漏水传感器                       │
│  ├── 防盗：门磁 + 红外双鉴 + 震动传感器 + 视频联动              │
│  └── 辅助：烟感、CO₂、压差、表面温度                            │
└──────────────────────────────────────────────────────────────────┘

档案库房 A（120㎡，层高 3m）

┌─────────────────────────────────────────────────────────────────┐
│  空调区（北侧）                                                  │
│  📍 漏水绳沿冷凝水管敷设（6m）                                    │
│  📍 区域式漏水块 ×2（空调底部）                                   │
│  📍 回风温湿度 ×1（风管）                                        │
│  📍 工作区温湿度 ×2（1.5m 高）                                   │
├─────────────────────────────────────────────────────────────────┤
│  主通道                                                          │
│  📍 红外双鉴 ×2（对角覆盖）                                      │
│  📍 温湿度 ×1（通道中央）                                        │
├─────────────────────────────────────────────────────────────────┤
│  密集架区（东侧）                                                │
│  📍 温湿度 ×4（每排端部）                                        │
│  📍 震动传感器 ×2（主通道侧密集架）                               │
│  📍 温湿度 ×2（角落，气流死角）                                   │
├─────────────────────────────────────────────────────────────────┤
│  入口区                                                          │
│  📍 门磁 ×1（主门）                                              │
│  📍 温湿度 ×1（缓冲区）                                          │
│  📍 视频摄像头 ×1（联动触发）                                     │
├─────────────────────────────────────────────────────────────────┤
│  管井/外墙（南侧）                                               │
│  📍 区域式漏水块 ×2（管井底部）                                   │
│  📍 温湿度 ×1（墙角）                                            │
└─────────────────────────────────────────────────────────────────┘

总计：温湿度 12 + 漏水绳 6m + 区域漏水 4 + 红外双鉴 2 + 门磁 1 + 震动 2 + 视频 1

"""
multi_sensor_rules.py - 多传感器联动规则引擎（核心逻辑）
"""

from enum import Enum, auto
from dataclasses import dataclass, field
from typing import List, Callable, Dict
import time

class Priority(Enum):
    CRITICAL = 1   # 立即处置：消防、水浸、入侵
    HIGH = 2        # 快速处置：温湿度越限超阈值、门禁异常
    MEDIUM = 3      # 值班处置：单点失效、通信中断
    LOW = 4         # 记录巡检：校准到期、滤网提醒

class ActionType(Enum):
    ALARM = auto()          # 告警通知
    CONTROL = auto()        # 设备控制
    VIDEO = auto()          # 视频联动
    LOCK = auto()           # 门禁控制
    SUPPRESS = auto()       # 告警抑制（防风暴）

@dataclass
class SensorEvent:
    sensor_id: str
    sensor_type: str        # temp_hum / water / door / pir / vibration
    zone: str
    value: float
    threshold: float
    timestamp: float
    quality: bool = True

@dataclass
class Rule:
    rule_id: str
    name: str
    priority: Priority
    condition: Callable[[SensorEvent, Dict], bool]
    actions: List[tuple]    # [(ActionType, target, params)]
    suppress_window: int = 300  # 抑制窗口（秒），防止重复触发
    last_triggered: float = 0.0

class MultiSensorRuleEngine:
    def __init__(self):
        self.rules: List[Rule] = []
        self.event_history: List[SensorEvent] = []
        self.suppressed_count = 0

    def add_rule(self, rule: Rule):
        self.rules.append(rule)

    def evaluate(self, event: SensorEvent, context: Dict) -> List[tuple]:
        """评估所有规则，返回触发的动作列表"""
        triggered_actions = []
        now = time.time()

        for rule in sorted(self.rules, key=lambda r: r.priority.value):
            # 抑制窗口检查
            if now - rule.last_triggered < rule.suppress_window:
                continue

            try:
                if rule.condition(event, context):
                    rule.last_triggered = now
                    for action in rule.actions:
                        triggered_actions.append((rule.priority, action))
                        print(f"[{rule.priority.name}] {rule.name}: {action}")
            except Exception as e:
                print(f"规则 {rule.rule_id} 评估异常: {e}")

        self.event_history.append(event)
        return triggered_actions

    def get_active_alarms(self) -> List[Dict]:
        """返回当前活跃告警（未恢复）"""
        # 简化：实际需维护告警状态机
        return []


# ============ 规则定义 ============

def create_rules() -> MultiSensorRuleEngine:
    engine = MultiSensorRuleEngine()

    # ---- 水浸联动（CRITICAL）----
    engine.add_rule(Rule(
        rule_id="WATER-001",
        name="漏水检测-关阀+告警",
        priority=Priority.CRITICAL,
        condition=lambda e, ctx: e.sensor_type == "water" and e.value > 0,
        actions=[
            (ActionType.CONTROL, "water_valve_main", {"state": "close"}),
            (ActionType.ALARM, "sms", {"level": "critical", "msg": "漏水检测触发，已关闭主水阀"}),
            (ActionType.VIDEO, "camera_entrance", {"preset": "water_valve"}),
        ],
        suppress_window=60
    ))

    # ---- 防盗联动（CRITICAL）----
    engine.add_rule(Rule(
        rule_id="SEC-001",
        name="非工作时间入侵检测",
        priority=Priority.CRITICAL,
        condition=lambda e, ctx: (
            e.sensor_type in ("pir", "vibration") and
            not ctx.get("working_hours", True)
        ),
        actions=[
            (ActionType.ALARM, "sms", {"level": "critical", "msg": "非工作时间检测到入侵"}),
            (ActionType.VIDEO, "camera_all", {"action": "record"}),
            (ActionType.CONTROL, "siren", {"state": "on"}),
        ],
        suppress_window=30
    ))

    # ---- 门禁异常（HIGH）----
    engine.add_rule(Rule(
        rule_id="DOOR-001",
        name="门禁异常-超时未关",
        priority=Priority.HIGH,
        condition=lambda e, ctx: (
            e.sensor_type == "door" and
            e.value == 1 and  # 门开
            time.time() - ctx.get("door_open_time", 0) > 300  # 开门超5min
        ),
        actions=[
            (ActionType.ALARM, "sms", {"level": "high", "msg": "库房门开启超时"}),
            (ActionType.VIDEO, "camera_entrance", {"action": "snapshot"}),
        ],
        suppress_window=120
    ))

    # ---- 温湿度越限（HIGH）----
    engine.add_rule(Rule(
        rule_id="ENV-001",
        name="温度越限-启动空调",
        priority=Priority.HIGH,
        condition=lambda e, ctx: (
            e.sensor_type == "temp_hum" and
            e.value > ctx.get("temp_max", 24.0)
        ),
        actions=[
            (ActionType.CONTROL, "ac_main", {"state": "on", "setpoint": 20.0}),
            (ActionType.ALARM, "platform", {"level": "high", "msg": f"温度越限: {e.value}℃"}),
        ],
        suppress_window=600
    ))

    # ---- 温湿度持续越限（CRITICAL 升级）----
    engine.add_rule(Rule(
        rule_id="ENV-002",
        name="温湿度越限超30min-升级告警",
        priority=Priority.CRITICAL,
        condition=lambda e, ctx: (
            e.sensor_type == "temp_hum" and
            ctx.get("env_out_of_range_duration", 0) > 1800  # 30min
        ),
        actions=[
            (ActionType.ALARM, "sms", {"level": "critical", "msg": "环境越限超30min，启动备用机组"}),
            (ActionType.CONTROL, "ac_backup", {"state": "on"}),
        ],
        suppress_window=300
    ))

    # ---- 安防布防/撤防（时间驱动）----
    engine.add_rule(Rule(
        rule_id="SEC-002",
        name="工作时间自动撤防",
        priority=Priority.LOW,
        condition=lambda e, ctx: (
            e.sensor_type == "time" and
            ctx.get("is_work_start", False)
        ),
        actions=[
            (ActionType.CONTROL, "security_system", {"state": "disarm"}),
        ],
        suppress_window=3600
    ))

    return engine


# ---- 使用示例 ----
if __name__ == "__main__":
    engine = create_rules()

    # 模拟漏水事件
    water_event = SensorEvent(
        sensor_id="water-pipe-01",
        sensor_type="water",
        zone="A-空调区",
        value=1,  # 检测到水
        threshold=0,
        timestamp=time.time(),
        quality=True
    )

    context = {
        "working_hours": False,
        "temp_max": 24.0,
        "door_open_time": time.time() - 400,
        "env_out_of_range_duration": 0,
    }

    actions = engine.evaluate(water_event, context)
    print(f"\n触发 {len(actions)} 条动作")

时间轴（秒）                    事件
─────────────────────────────────────────────────────────────────
T+0     漏水绳检测到导通
T+0.5   控制器确认（去抖滤波，排除误报）
T+1     边缘网关收到告警，上报平台
T+2     平台联动引擎触发 → 关闭主水阀 + 声光告警 + 短信通知
T+3     视频摄像头转向漏水点，开始录像
T+5     平台确认水阀状态回读（已关闭）
T+10    运维人员收到通知，确认收到
T+30    系统自动巡检：漏水是否持续？阀位是否保持？
T+60    若漏水消失，进入恢复流程：手动确认 → 开阀 → 记录

# Modbus TCP 漏水控制器
# 假设寄存器表：0x0000 = 状态字，0x0001~0x0006 = 6 段漏水绳定位值
pymodbus read_holding_registers 0 10 --host 10.20.1.50 --port 502 --unit 1

# BACnet 门禁控制器
bacwi read 10.20.1.51 1001 analog-input 1 present-value

# SNMP 交换机 PoE 端口（确认传感器在线）
snmpwalk -v2c -c public 10.20.1.254 1.3.6.1.2.1.105.1.1.1.3  # poePortPower

"""
debug_open_loop.py - 开环调试辅助脚本
"""

import requests
import time

class OpenLoopDebugger:
    def __init__(self, api_base):
        self.api_base = api_base

    def inject_event(self, sensor_id, value):
        """模拟传感器事件注入"""
        payload = {
            "sensor_id": sensor_id,
            "value": value,
            "timestamp": int(time.time()),
            "quality": True,
            "debug": True  # 标记为调试事件，不触发真实联动
        }
        resp = requests.post(f"{self.api_base}/api/v1/debug/inject", json=payload)
        return resp.json()

    def check_received(self, sensor_id, timeout=5):
        """确认平台是否收到事件"""
        start = time.time()
        while time.time() - start < timeout:
            resp = requests.get(f"{self.api_base}/api/v1/sensor/{sensor_id}/latest")
            if resp.json().get("received"):
                return True
            time.sleep(0.5)
        return False

    def verify_action(self, action_id, expected_state):
        """确认执行器是否动作"""
        resp = requests.get(f"{self.api_base}/api/v1/action/{action_id}/status")
        return resp.json().get("state") == expected_state


# 使用示例
debugger = OpenLoopDebugger("http://collector.local:8080")

# 模拟漏水事件
result = debugger.inject_event("water-pipe-01", 1)
print(f"事件注入: {result}")

# 确认平台收到
received = debugger.check_received("water-pipe-01")
print(f"平台收到: {received}")

# 确认联动动作（需手动触发联动或等待规则引擎）
# time.sleep(3)
# action_ok = debugger.verify_action("water_valve_main", "closed")
# print(f"阀门已关: {action_ok}")

# 告警抑制策略
class AlarmSuppressor:
    def __init__(self):
        self.window = {}  # {rule_id: last_trigger_time}
        self.same_sensor_window = 60  # 同传感器 60s 内只报一次
        self.same_zone_window = 120   # 同区域 120s 内合并

    def should_suppress(self, rule_id, sensor_id, zone):
        now = time.time()
        # 规则级抑制
        if rule_id in self.window:
            if now - self.window[rule_id] < 60:
                return True
        # 传感器级抑制
        key = f"sensor:{sensor_id}"
        if key in self.window:
            if now - self.window[key] < self.same_sensor_window:
                return True
        # 区域级合并
        key = f"zone:{zone}"
        if key in self.window:
            if now - self.window[key] < self.same_zone_window:
                return True
        self.window[rule_id] = now
        self.window[f"sensor:{sensor_id}"] = now
        self.window[f"zone:{zone}"] = now
        return False