智慧档案馆信创适配：环境监控平台与恒温恒湿设备对接方案

┌──────────────────────────────────────────────────────────────────┐
│ 北向集成层（信创平台内）                                          │
│  Web控制台 / Grafana替代（国产BI或自研前端） / 告警通知 / 工单     │
│  REST API + WebSocket + MQ                                        │
└──────────────────────┬───────────────────────────────────────────┘
                       │ 内部数据总线（Kafka/NATS/Redis Stream，信创验证）
┌──────────────────────▼───────────────────────────────────────────┐
│ 平台服务层（国产OS+国产库）                                       │
│  采集网关管理 · 设备模型 · 点表映射 · 联动规则引擎 · 审计日志     │
└──────────────────────┬───────────────────────────────────────────┘
                       │ 北向协议代理 / EdgeAgent
┌──────────────────────▼───────────────────────────────────────────┐
│ 边缘层（信创边缘节点或工控机）                                     │
│  协议转换 · 缓存 · 本地自治 · 断网续传                            │
└──────────────────────┬───────────────────────────────────────────┘
                       │ Modbus TCP / BACnet/IP / MQTT
┌──────────────────────▼───────────────────────────────────────────┐
│ 设备层                                                          │
│  网口温湿度变送器 · DDC/PLC · 恒温恒湿机组控制器 · 水阀/风阀执行器 │
└──────────────────────────────────────────────────────────────────┘

# store_backend.py - 存储适配层参考
from abc import ABC, abstractmethod

class MetricSink(ABC):
    @abstractmethod
    async def write_point(self, measurement, tags, fields, ts):
        ...

class InfluxSink(MetricSink):
    def __init__(self, client): self.c = client
    async def write_point(self, m, tags, fields, ts):
        from influxdb_client import Point
        p = Point(m).time(ts)
        for k,v in tags.items(): p = p.tag(k,v)
        for k,v in fields.items(): p = p.field(k,v)
        self.c.write_api().write(record=p)

class TDEngineSink(MetricSink):
    def __init__(self, conn): self.conn = conn  # taosrest 或 taospy
    async def write_point(self, m, tags, fields, ts):
        cols = ",".join(list(tags.keys())+list(fields.keys()))
        vals = ",".join([f"'{v}'" if isinstance(v,str) else str(v)
                        for v in list(tags.values())+list(fields.values())])
        sql = f"INSERT INTO {m} ({cols}) VALUES ({vals})"
        await self.conn.execute(sql)

"""
edge_agent.py - 信创边缘侧采集与北向上行（精简参考）
- 读配置，连南向设备，归一化后写本地缓冲+北向MQTT
- 接收北向下发，写回设备；失败退避，状态回报
"""

import asyncio, json, logging, signal
from typing import Dict, Any
from pymodbus.client import AsyncModbusTcpClient

logging.basicConfig(level=logging.INFO)

class EdgeAgent:
    def __init__(self, cfg: Dict[str, Any]):
        self.cfg = cfg
        self.clients: Dict[str, AsyncModbusTcpClient] = {}
        self.local_buf = []          # 断网续传缓冲
        self.uplink_ok = False
        self._stop = False

    async def modbus_read(self, dev):
        host = dev["host"]; unit = dev["unit"]
        c = self.clients.get(host)
        if c is None or not c.connected:
            c = AsyncModbusTcpClient(host, port=dev.get("port",502))
            await c.connect()
            self.clients[host] = c
        rr = await c.read_holding_registers(0, 4, slave=unit)
        if rr.isError():
            return None
        regs = rr.registers
        # 示例：ret_temp, ret_hum, setpoint_temp, setpoint_hum
        return {
            "ret_temp": regs[0]/10.0, "ret_hum": regs[1]/10.0,
            "setpoint_temp": regs[2]/10.0, "setpoint_hum": regs[3]/10.0,
        }

    async def modbus_write_setpoint(self, dev, sp_t, sp_h):
        c = self.clients.get(dev["host"])
        if c is None or not c.connected:
            return False
        # 写保持寄存器 2,3（0-based offset），缩放回原值
        await c.write_registers(2, [int(sp_t*10), int(sp_h*10)], slave=dev["unit"])
        # 回读确认
        rr = await c.read_holding_registers(2, 2, slave=dev["unit"])
        if rr.isError():
            return False
        got_t, got_h = rr.registers[0]/10.0, rr.registers[1]/10.0
        return abs(got_t-sp_t) < 0.2 and abs(got_h-sp_h) < 0.5

    async def poll_loop(self):
        while not self._stop:
            for dev in self.cfg["devices"]:
                try:
                    state = await self.modbus_read(dev)
                    if state:
                        msg = {"device": dev["id"], "ts": asyncio.get_event_loop().time(), "state": state}
                        # 北向上行（MQTT示意，实际接信创消息总线）
                        await self.publish(msg)
                        self.local_buf.append(msg)
                except Exception as e:
                    logging.warning("poll %s: %s", dev["id"], e)
            await asyncio.sleep(self.cfg.get("poll", 5.0))

    async def publish(self, msg):
        # 占位：NATS/Kafka/MQTT over TLS/国密网关
        self.uplink_ok = True

    async def control_handler(self, cmd):
        dev = next(d for d in self.cfg["devices"] if d["id"]==cmd["device"])
        ok = await self.modbus_write_setpoint(dev, cmd["sp_t"], cmd["sp_h"])
        return {"ack": ok, "device": cmd["device"]}

    def stop(self):
        self._stop = True

async def main():
    cfg = json.load(open("/etc/edge_agent/config.json"))
    agent = EdgeAgent(cfg)
    loop = asyncio.get_event_loop()
    for s in (signal.SIGTERM, signal.SIGINT):
        loop.add_signal_handler(s, agent.stop)
    await agent.poll_loop()

if __name__ == "__main__":
    asyncio.run(main())

# Modbus TCP 抓包
tcpdump -i any -nn 'host 10.20.1.11 and port 502' -w /tmp/mb.pcap
# BACnet/IP
tcpdump -i any -nn 'port 47808' -w /tmp/bac.pcap
# 国产OS上确认监听与连通
ss -tlnp | grep 502
nc -zv 10.20.1.11 502