楼宇自控IoT接入：基于TCP/IP的以太网温湿度传感器批量组态实操

管理网/楼控BAS网 (VLAN 100, 10.10.10.0/24)
   └── 边缘网关 x2 (HA)
         ├── Modbus TCP 南向 (VLAN 200, 192.168.200.0/24)
         │      ├── 接入交换机/PoE
         │      └── 以太网温湿度传感器 N台, port 502
         └── BACnet/IP 北向 (VLAN 100, UDP/47808=BIP)

# sensor_template.yaml
template_id: eth-th-01
transport: modbus-tcp
port: 502
unit_id: 1            # 多数网口变送器从站地址可设，批量时统一或按IP映射
poll_period_ms: 2000
timeout_ms: 1500
retry: 2
registers:
  - {name: temperature, block: holding, offset: 0, type: int16, scale: 0.1, signed: true}
  - {name: humidity,    block: holding, offset: 1, type: int16, scale: 0.1, signed: false}
  - {name: dewpoint,    block: holding, offset: 2, type: int16, scale: 0.1, signed: true}
  - {name: status_word, block: holding, offset: 3, type: uint16, scale: 1.0}
  - {name: uptime_min,  block: holding, offset: 4, type: uint32, wordorder: hilo}
alarms:
  - {point: temperature, hi: 28.0, lo: 16.0, deadband: 0.5}
  - {point: humidity, hi: 65.0, lo: 35.0, deadband: 2.0}
meta:
  location_attr: floor_zone   # 组态时填充

alias,ip,unit_id,floor,zone,ddc_loop,enable
AHU1-RA-T,192.168.200.11,1,L2,Z1,DL1,1
AHU1-SA-T,192.168.200.12,1,L2,Z2,DL1,1
CRAC-1-T,192.168.200.21,1,L3,CRAC,DL2,1
...

"""gen_config.py - 由清单+模板生成组态"""
import yaml, csv, json
from jinja2 import Template  # 或直接拼dict

tpl = yaml.safe_load(open("sensor_template.yaml"))
devs = []
with open("devices.csv") as f:
    for row in csv.DictReader(f):
        if row["enable"] != "1":
            continue
        devs.append({
            "alias": row["alias"],
            "ip": row["ip"],
            "unit_id": int(row["unit_id"]),
            "floor": row["floor"],
            "zone": row["zone"],
            "ddc_loop": row["ddc_loop"],
            "poll_period_ms": tpl["poll_period_ms"],
            "points": tpl["registers"],
            "alarms": tpl["alarms"],
        })

config = {
    "gateway": {"southbound": {"modbus_tcp": devs}},
    "northbound": {"bacnet": {"device_instance_base": 10000, "increment": 1}},
}
json.dump(config, open("gateway_config.json","w"), indent=2)
print(f"generated {len(devs)} devices")

"""gw_modbus_poller.py"""
import asyncio, logging
from pymodbus.client import AsyncModbusTcpClient
from dataclasses import dataclass, asdict

@dataclass
class DevCfg:
    alias: str; ip: str; unit: int; period: float
    points: list
    floor: str; zone: str; ddc_loop: str

class Southbound:
    def __init__(self, devs, cache):
        self.devs = devs
        self.cache = cache          # dict alias->point dict, 也推北向
        self.clients = {}
    async def client(self, d):
        c = self.clients.get(d.ip)
        if c is None or not c.connected:
            c = AsyncModbusTcpClient(d.ip, port=502)
            await c.connect()
            self.clients[d.ip] = c
        return c
    async def poll(self, d):
        try:
            c = await self.client(d)
            # 单次读连续块，offset0..max_off
            max_off = max(p["offset"] for p in d.points) + (2 if any(p["type"]=="uint32" for p in d.points) else 1)
            rr = await asyncio.wait_for(
                c.read_holding_registers(0, max_off, slave=d.unit), timeout=2.0)
            if rr.isError():
                raise RuntimeError(str(rr))
            regs = rr.registers
            vals = {}
            for p in d.points:
                raw = regs[p["offset"]]
                if p["type"] == "int16":
                    v = raw/10.0 if not p.get("signed",True) else (raw-0x10000)/10.0 if raw>=0x8000 else raw/10.0
                elif p["type"] == "uint16":
                    v = raw*p["scale"]
                elif p["type"] == "uint32":
                    hi = regs[p["offset"]]; lo = regs[p["offset"]+1]
                    v = ((hi<<16)|lo)*p["scale"]
                vals[p["name"]] = v
            vals["_online"] = True
            vals["_ts"] = asyncio.get_event_loop().time()
            self.cache[d.alias] = vals
        except Exception as e:
            logging.warning("poll %s fail: %s", d.alias, e)
            cur = self.cache.setdefault(d.alias, {})
            cur["_online"] = False
    async def run(self):
        while True:
            for d in self.devs:
                asyncio.create_task(self.poll(d))
            await asyncio.sleep(min(d.period for d in self.devs)/1000.0)

"""gw_bacnet.py - 简化示例，勿直接上产线"""
from bacpypes3.ipv4.app import Application
from bacpypes3.local.device import DeviceObject
from bacpypes3.local.object import AnalogInputObject
from bacpypes3.primitivedata import CharacterString, Real
import asyncio, json

def build_app(cache, devs, device_instance=10101, addr="10.10.10.50/24"):
    d = DeviceObject(objectIdentifier=("device", device_instance),
                     objectName="ETH-TH-GW",
                     servicesSupported=[...])
    app = Application(d, addr)
    # 建对象
    inst = 0
    for dev in devs:
        for pt in ("temperature","humidity"):
            inst += 1
            ai = AnalogInputObject(
                objectIdentifier=("analogInput", inst),
                objectName=f"{dev['alias']}.{pt}",
                units="degreesCelsius" if pt=="temperature" else "percentRelativeHumidity",
                presentValue=Real(0.0),
                covIncrement=0.2,
            )
            app.add_object(ai)
            dev[f"_obj_{pt}"] = ai
    return app, d

async def push_loop(app, cache, devs):
    while True:
        for dev in devs:
            vals = cache.get(dev["alias"], {})
            for pt in ("temperature","humidity"):
                obj = dev.get(f"_obj_{pt}")
                if obj and pt in vals:
                    obj.presentValue = Real(float(vals[pt]))
        await asyncio.sleep(1.0)

"""discover.py - Modbus探测+SNMP可选，产出实际清单"""
import asyncio
from pymodbus.client import AsyncModbusTcpClient

async def probe(ip):
    c = AsyncModbusTcpClient(ip, port=502)
    try:
        await asyncio.wait_for(c.connect(), timeout=1.0)
        rr = await asyncio.wait_for(c.read_holding_registers(0,4,slave=1), timeout=1.0)
        if not rr.isError():
            return ip, rr.registers
    except Exception:
        pass
    finally:
        c.close()
    return ip, None

async def scan(subnet_ips):
    res = {}
    for ip in subnet_ips:
        r = await probe(ip)
        if r[1]: res[ip]=r[1]
    return res

"""validate.py - 组态与现场一致性"""
import json, asyncio
from pymodbus.client import AsyncModbusTcpClient

async def check(config, cache):
    errs=[]
    for d in config["gateway"]["southbound"]["modbus_tcp"]:
        c = AsyncModbusTcpClient(d["ip"], port=502)
        ok=False
        try:
            await c.connect(); 
            rr=await c.read_holding_registers(0,1,slave=d["unit_id"])
            ok=not rr.isError()
        except Exception as e:
            errs.append((d["alias"],"conn",str(e)))
        finally:
            c.close()
        cached=cache.get(d["alias"],{})
        if cached and not cached.get("_online"):
            errs.append((d["alias"],"offline",""))
        # 质量检查
        v=cached.get("temperature")
        if v is not None and (v<-40 or v>85):
            errs.append((d["alias"],"range",v))
    return errs