TCP 与 SNMP 双协议：RJ45 温湿度记录仪对接机房服务器监控大屏实战

RJ45记录仪 ──Modbus TCP──▶ 边缘采集服务 ──本地状态库──▶ SNMP子代理 ──SNMP GET──▶ NMS/Grafana+Telegraf
                                              │
                                              └──HTTP/InfluxDB直推（并行）

记录仪 ──SNMP agent内置──▶ NMS直接轮询
       └─Modbus TCP保留给配置/补传

# 轮询单台验证
mbpoll -a -t 3 -r 1 -c 4 -P 502 -u 1 10.20.1.51
# -t 3 = holding registers, -r 1 注意mbpoll是1-based，映射到0-based偏移

"""
env_bridge.py - 边缘采集 + SNMP子代理状态供给
"""

import asyncio, logging, time
from pymodbus.client import AsyncModbusTcpClient
from dataclasses import dataclass, asdict
import json

DEVICES = {
    "A01": {"host": "10.20.1.51", "unit": 1},
    "A02": {"host": "10.20.1.52", "unit": 1},
}
POLL_PERIOD = 5.0
STATE_TTL = 30.0   # 状态库老化阈值

@dataclass
class EnvSample:
    dev: str
    temp: float
    hum: float
    dewpoint: float
    status: int
    ts: float
    reachable: bool = True

class Bridge:
    def __init__(self):
        self.clients = {}
        self.state = {}        # dev -> EnvSample
        self._stop = False

    async def ensure_client(self, dev):
        cfg = DEVICES[dev]
        c = self.clients.get(dev)
        if c is None or not c.connected:
            c = AsyncModbusTcpClient(cfg["host"], port=502)
            await c.connect()
            self.clients[dev] = c
        return c

    async def poll_one(self, dev):
        cfg = DEVICES[dev]
        try:
            c = await self.ensure_client(dev)
            rr = await c.read_holding_registers(0, 5, slave=cfg["unit"])
            if rr.isError():
                raise RuntimeError(str(rr))
            v = rr.registers
            temp = v[0]/10.0 if v[0] < 0x8000 else (v[0]-0x10000)/10.0
            hum = v[1]/10.0
            dp = v[2]/10.0 if v[2] < 0x8000 else (v[2]-0x10000)/10.0
            status = v[3]
            ts = time.time()
            self.state[dev] = EnvSample(dev, temp, hum, dp, status, ts)
        except Exception as e:
            logging.warning("poll %s failed: %s", dev, e)
            old = self.state.get(dev)
            if old:
                old.reachable = False
            else:
                self.state[dev] = EnvSample(dev, float('nan'), float('nan'), float('nan'), 0, time.time(), False)

    async def poll_loop(self):
        while not self.stop:
            await asyncio.gather(*(self.poll_one(d) for d in DEVICES))
            await asyncio.sleep(POLL_PERIOD)

    # ---- SNMP子代理侧拉取接口 ----
    def snapshot_json(self) -> str:
        out = {}
        now = time.time()
        for dev, s in self.state.items():
            stale = (now - s.ts) > STATE_TTL
            out[dev] = {
                "temp": s.temp, "hum": s.hum, "dewpoint": s.dewpoint,
                "status": s.status, "reachable": s.reachable and not stale,
                "age_s": round(now - s.ts, 1)
            }
        return json.dumps(out)

if __name__ == "__main__":
    b = Bridge()
    loop = asyncio.get_event_loop()
    loop.create_task(b.poll_loop())
    try:
        loop.run_forever()
    except KeyboardInterrupt:
        b.stop = True

envKitEntry ::= SEQUENCE {
  envKitIndex      INTEGER,      -- x.1.<idx>
  envKitId         OCTET STRING, -- x.2.<idx>
  envKitTemp       Gauge32,      -- x.3.<idx>  ×10 后返回，或返回原生
  envKitHum        Gauge32,      -- x.4.<idx>
  envKitDewpoint   INTEGER,      -- x.5.<idx>
  envKitStatus     INTEGER,      -- x.6.<idx>
  envKitReachable  INTEGER,      -- x.7.<idx>  1=up 2=down
  envKitAge        Gauge32       -- x.8.<idx>  s
}

#!/usr/bin/env python3
"""
snmp_pass_persist.py - net-snmp pass_persist handler
从env_bridge拉取JSON状态，映射为SNMP叶子
"""
import sys, json, time

BRIDGE_STATE_PATH = "/run/env_bridge/state.json"   # 或由本地HTTP拉取

OID_BASE = ".1.3.6.1.4.1.4711.1.1"
INDEX_MAP = {}   # dev -> idx

def load_state():
    try:
        with open(BRDIGE_STATE_PATH) as f:
            return json.load(f)
    except Exception:
        return {}

def build_rows(state):
    rows = []
    idx = 1
    INDEX_MAP.clear()
    for dev, m in state.items():
        INDEX_MAP[dev] = idx
        prefix = f"{OID_BASE}.3.{idx}"
        rows.append((f"{OID_BASE}.1.{idx}", "INTEGER", str(idx)))
        rows.append((f"{OID_BASE}.2.{idx}", "OCTETSTR", dev.encode().hex()))
        rows.append((f"{OID_BASE}.3.{idx}", "GAUGE", str(int(round(m.get('temp',0)*10)))))
        rows.append((f"{OID_BASE}.4.{idx}", "GAUGE", str(int(round(m.get('hum',0)*10)))))
        rows.append((f"{OID_BASE}.5.{idx}", "INTEGER", str(int(round(m.get('dewpoint',0)*10)))))
        rows.append((f"{OID_BASE}.6.{idx}", "INTEGER", str(m.get('status',0))))
        rows.append((f"{OID_BASE}.7.{idx}", "INTEGER", "1" if m.get('reachable') else "2"))
        rows.append((f"{OID_BASE}.8.{idx}", "GAUGE", str(int(m.get('age_s',0)))))
        idx += 1
    return rows

def handle_pdu():
    # pass_persist协议：snmpd发送 "ping\nPONG\n" 或 "get\n<oid>\n" / "getnext\n<oid>\n"
    while True:
        line = sys.stdin.readline()
        if not line:
            break
        cmd = line.strip()
        if cmd == "ping":
            sys.stdout.write("PONG\n")
            sys.stdout.flush()
            continue
        if cmd in ("get", "getnext", "set"):
            oid = sys.stdin.readline().strip()
            state = load_state()
            rows = build_rows(state)
            keymap = {o:(t,v) for (o,t,v) in rows}
            if cmd == "getnext":
                # 简化：返回下一个>=oid的叶子；生产用完备字典序
                cand = sorted(keymap.keys())
                hit = next((k for k in cand if k >= oid), None)
                if hit:
                    t,v = keymap[hit]
                    sys.stdout.write(f"{t}\n{hit}\n{v}\n")
                else:
                    sys.stdout.write("NONE\n\n")
            else:
                if oid in keymap:
                    t,v = keymap[oid]
                    sys.stdout.write(f"{t}\n{oid}\n{v}\n")
                else:
                    sys.stdout.write("NONE\n\n")
            sys.stdout.flush()
            continue

if __name__ == "__main__":
    handle_pdu()

pass_persist .1.3.6.1.4.1.4711.1.1 /usr/local/bin/snmp_pass_persist.py

snmpwalk -v2c -c public localhost .1.3.6.1.4.1.4711.1.1
snmpget -v2c -c public localhost .1.3.6.1.4.1.4711.1.1.3.1

[[inputs.snmp.table]]
  oid = ".1.3.6.1.4.1.4711.1.1"
  name = "envkit"
  [[inputs.snmp.table.field]]
    oid = ".1.3.6.1.4.1.4711.1.1.2.1"
    name = "dev_id"
    is_tag = true
  [[inputs.snmp.table.field]]
    oid = ".1.3.6.1.4.1.4711.1.1.3.1"
    name = "temp_x10"
  [[inputs.snmp.table.field]]
    oid = ".1.3.6.1.4.1.4711.1.1.4.1"
    name = "hum_x10"
  [[inputs.snmp.table.field]]
    oid = ".1.3.6.1.4.1.4711.1.1.7.1"
    name = "reachable"

$room, $rack_row, $sensor
FROM env WHERE room=$room

async def set_setpoint(dev, t_sp, h_sp):
    cfg = DEVICES[dev]
    c = await ensure_client(dev)
    await c.write_registers(0x10, [int(t_sp*10), int(h_sp*10)], slave=cfg["unit"])