中小型档案室轻量化方案：低成本温湿度监控与八防基础搭建指南

感知层：网口/RS485温湿度变送器 + 水浸（选配） + 门磁/PIR（选配）
        │ Modbus TCP / RTU转TCP
边缘层：PoE交换机 + 边缘网关/工控机/树莓派类节点，跑采集容器
        │ MQTT/HTTP →
平台层：本地Grafana+InfluxDB，或轻量SaaS/内网Web；告警经短信/钉钉/邮件/Webhook

"""
light_collector.py - 中小型档案室轻量采集服务
- Modbus TCP 轮询温湿度，写 InfluxDB，经 MQTT 通知
- 含迟滞启停联动示例（控制智能继电器/插座）
"""

import asyncio, json, logging
from pymodbus.client import AsyncModbusTcpClient
from influxdb_client import InfluxDBClient, Point
from influxdb_client.client.write_api import SYNCHRONOUS

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")

CFG = {
    "devices": {
        "room-a-ret": {"host": "10.20.1.11", "unit": 1, "scale_t": 0.1, "scale_h": 0.1},
        "room-a-ws":  {"host": "10.20.1.12", "unit": 1, "scale_t": 0.1, "scale_h": 0.1},
    },
    "poll": 10.0,
    "influx": {"url": "http://localhost:8086", "token": "", "org": "arch", "bucket": "env"},
    "mqtt": {"host": "localhost", "topic": "archive/env"},
    "ctrl": {  # 可选联动
        "relay_host": "10.20.1.30", "coil": 0,
        "t_lo": 19.0, "t_hi": 21.0, "h_lo": 45.0, "h_hi": 55.0,
        "min_run": 300, "min_off": 60
    }
}

class LightCollector:
    def __init__(self):
        self.clients = {}
        self.influx = InfluxDBClient(url=CFG["influx"]["url"], token=CFG["influx"]["token"], org=CFG["influx"]["org"])
        self.wapi = self.influx.write_api(write_options=SYNCHRONOUS)
        self.state = {"dev": None, "since": 0}  # 简单迟滞状态机
        self._stop = False

    async def client(self, host):
        c = self.clients.get(host)
        if c is None or not c.connected:
            c = AsyncModbusTcpClient(host)
            await c.connect()
            self.clients[host] = c
        return c

    async def read_dev(self, name, cfg):
        try:
            c = await self.client(cfg["host"])
            rr = await c.read_holding_registers(0, 2, slave=cfg["unit"])
            if rr.isError():
                return None, None, False
            raw_t, raw_h = rr.registers
            # 假设16bit有符号定点，0.1℃/0.1%RH；按厂家表调整
            t = raw_t * cfg["scale_t"] if False else raw_t/10.0
            h = raw_h / 10.0
            return t, h, True
        except Exception as e:
            logging.warning("read %s failed: %s", name, e)
            return None, None, False

    async def ctrl_loop(self, env):
        # env: 聚合后的库区温湿度
        c = CFG["ctrl"]
        now = asyncio.get_event_loop().time()
        on = self.state.get("on", False)
        since = self.state.get("since", 0)
        # 迟滞启停：越上限开制冷/除湿，越下限关；简化示例
        if env["t"] > c["t_hi"] and not on:
            await self.set_relay(True)
            self.state = {"on": True, "since": now}
        elif env["t"] < c["t_lo"] and on:
            await self.set_relay(False)
            self.state = {"on": False, "since": now}
        # 运行/停机最小时间保护
        if on and now - since < c.get("min_run", 300):
            return
        if not on and now - since < c.get("min_off", 60):
            return

    async def set_relay(self, on):
        try:
            c = await self.client(CFG["ctrl"]["relay_host"])
            await c.write_coil(CFG["ctrl"]["coil"], on, slave=1)
            logging.info("relay set %s", on)
        except Exception as e:
            logging.warning("relay ctrl failed: %s", e)

    async def run(self):
        while not self._stop:
            agg_t, agg_h, n = [], [], 0
            for name, cfg in CFG["devices"].items():
                t, h, ok = await self.read_dev(name, cfg)
                if ok:
                    p = Point("env").tag("device", name).field("temperature", t).field("humidity", h).field("quality",1)
                    self.wapi.write(bucket=CFG["influx"]["bucket"], org=CFG["influx"]["org"], record=p)
                    agg_t.append(t); agg_h.append(h); n+=1
            if n:
                env = {"t": sum(agg_t)/n, "h": sum(agg_h)/n}
                await self.ctrl_loop(env)
            await asyncio.sleep(CFG["poll"])

if __name__ == "__main__":
    svc = LightCollector()
    try:
        asyncio.run(svc.run())
    except KeyboardInterrupt:
        svc._stop = True