基于腾讯云 IoT Explorer 的 POE 网口温湿度传感器接入：物模型与属性上报实践

PoE交换机 ── Modbus TCP ──> 边缘网关(机房内)
                              │
                  MQTT over TLS (8883)
                              ▼
                  腾讯云 IoT Explorer
                      │ 规则引擎
                      ▼
             云函数/CKafka → CTSDB/InfluxDB → 云图/Grafana

{
  "version": "1.0",
  "properties": [
    { "id": "temperature", "name": "温度", "define": { "type": "float", "min": "-40", "max": "85", "step": "0.1", "unit": "°C" } },
    { "id": "humidity", "name": "湿度", "define": { "type": "float", "min": "0", "max": "100", "step": "0.1", "unit": "%RH" } },
    { "id": "sensor_status", "name": "传感器状态", "define": { "type": "enum", "mapping": { "0": "正常", "1": "故障", "2": "离线", "3": "越限" } } },
    { "id": "power_voltage", "name": "供电电压", "define": { "type": "float", "min": "9", "max": "36", "step": "0.1", "unit": "V" } },
    { "id": "room_id", "name": "机房编号", "define": { "type": "string", "len": 64 } },
    { "id": "last_collect_ts", "name": "采集时间戳", "define": { "type": "int64", "unit": "ms" } }
  ],
  "events": [
    { "id": "temp_high_alarm", "name": "温度高告警", "type": "alert", "params": [
        { "id": "value", "define": { "type": "float" } },
        { "id": "threshold", "define": { "type": "float" } }
    ]},
    { "id": "humidity_high_alarm", "name": "湿度高告警", "type": "alert", "params": [
        { "id": "value", "define": { "type": "float" } }
    ]},
    { "id": "sensor_offline", "name": "传感器离线", "type": "alert", "params": [
        { "id": "sensor_id", "define": { "type": "string" } }
    ]}
  ],
  "actions": []
}

import time, json, logging
from pymodbus.client import ModbusTcpClient
import paho.mqtt.client as mqtt
from hashlib import sha256
import hmac, base64, os

PRODUCT_ID = os.environ["PRODUCT_ID"]
DEVICE_NAME = os.environ["DEVICE_NAME"]
DEVICE_SECRET = os.environ["DEVICE_SECRET"]
BROKER = f"{PRODUCT_ID}.iotcloud.tencentdevices.com"
PORT = 8883
CA_FILE = "root.ca.pem"   # 腾讯云根证书或Amazon CA1可用，按文档
SENSORS = [{"id": "s1", "ip": "192.168.10.10", "port": 502, "unit": 1}]
ROOM_ID = os.environ.get("ROOM_ID", "room-default")
REPORT_INT = 15.0

def sign_password():
    # 简化：使用腾讯云推荐的密钥认证方式，paho_tls_set + username/password
    # 实际部署建议用SDK；此处给手动构造示例
    conn_id = "edgegw"
    expiry = 0
    nonce = str(int(time.time()))
    sign_content = f"deviceName={DEVICE_NAME}&productId={PRODUCT_ID}&expiry={expiry}&nonce={nonce}"
    raw = hmac.new(DEVICE_SECRET.encode(), sign_content.encode(), sha256).digest()
    sig = base64.b64encode(raw).decode()
    username = f"{DEVICE_NAME};hmacsha256"
    password = f"{sig}"
    return username, password

client = mqtt.Client(client_id=f"{PRODUCT_ID}{DEVICE_NAME}", transport="tcp")
client.tls_set(ca_certs=CA_FILE, cert_reqs=2)
uname, pwd = sign_password()
client.username_pw_set(uname, pwd)
client.connect_async(BROKER, PORT, keepalive=60)

# Explorer 属性上报主题
UP_TOPIC = f"$thing/up/property/{PRODUCT_ID}/{DEVICE_NAME}"
DOWN_TOPIC = f"$thing/down/property/{PRODUCT_ID}/{DEVICE_NAME}"

def on_connect(c, u, f, rc):
    logging.info("connected %s", rc)
    c.subscribe(DOWN_TOPIC)
    # 上线后上报一次
    report()

def read_sensor(s):
    c = ModbusTcpClient(host=s["ip"], port=s["port"], timeout=1.0)
    try:
        if not c.connect():
            return None
        r = c.read_holding_registers(0, 3, unit=s["unit"])
        if r.is_error():
            return None
        temp = r.registers[0]/10.0
        humi = r.registers[1]/10.0
        status_word = r.registers[2]
        return temp, humi, status_word
    finally:
        c.close()

def build_payload():
    temps, humis, st = [], [], 0
    sensor_status = 0
    last_ts = int(time.time()*1000)
    detail = []
    for s in SENSORS:
        d = read_sensor(s)
        if d is None:
            sensor_status = 2  # 离线
            detail.append({"sensor_id": s["id"], "status": "offline"})
            continue
        t, h, sw = d
        temps.append(t); humis.append(h)
        if sw & 0x1: sensor_status = 1
        detail.append({"sensor_id": s["id"], "temperature": t, "humidity": h, "sw": sw})
    payload = {
        "method": "report",
        "clientToken": f"{DEVICE_NAME}-{int(time.time())}",
        "params": {
            "temperature": round(sum(temps)/len(temps), 1) if temps else 0.0,
            "humidity": round(sum(humis)/len(humis), 1) if humis else 0.0,
            "sensor_status": sensor_status,
            "room_id": ROOM_ID,
            "last_collect_ts": last_ts
        }
    }
    # 扩展明细走自定义topic或事件，此处附在property下行调试用event上报
    return payload, detail

def report():
    payload, detail = build_payload()
    client.publish(UP_TOPIC, json.dumps(payload), qos=1)
    # 可选：上报事件
    evt = {"method": "event_post", "clientToken": payload["clientToken"],
           "params": {"sensors_detail": detail}}
    client.publish(f"$thing/up/event/{PRODUCT_ID}/{DEVICE_NAME}", json.dumps(evt), qos=1)

def loop():
    while True:
        try:
            if client.is_connected():
                report()
        except Exception as e:
            logging.warning("report err %s", e)
        client.loop_start()
        time.sleep(REPORT_INT)

if __name__ == "__main__":
    client.on_connect = on_connect
    client.loop_start()
    while True:
        time.sleep(1)

def main(event, context):
    for msg in parse_kafka_or_direct(event):
        dev = msg["deviceName"]
        props = msg["payload"]["params"]
        point = {
          "measurement": "env",
          "tags": {"room": props.get("room_id"), "dev": dev},
          "fields": {
             "temp": props["temperature"],
             "humi": props["humidity"],
             "status": props["sensor_status"]
          },
          "time": props.get("last_collect_ts", int(time.time()*1000))
        }
        write_influx(point)   # influxdb-client v2
    return "ok"