wangfq
a2cfe4602c
基于《DLD960_IoT_MQTT协议.md》V1.00 实现 功能： - MQTT Broker 连接管理 - 设备自动发现（订阅 dld960/+/dev/# 通配符） - 设备信息查询（dev_info_query） - 网络配置（SSC/IoT TCP）+ Topic 配置 - 实时线圈数据监控（loop_data）+ 事件上报（event_report） - 心跳监控（heartbeat） - 控制命令：密码验证/设置、出厂初始化、设备复位 项目结构： - main.py 主窗口 (tkinter GUI) - dbn_mqtt_tool/ - protocol.py DLD960 IoT MQTT 协议定义 - mqtt_client.py MQTT 客户端封装 (paho-mqtt) - device_manager.py 设备发现与状态管理
112 lines
3.5 KiB
Python
112 lines
3.5 KiB
Python
"""
|
|
设备管理器 — 管理已发现的设备及其状态
|
|
"""
|
|
|
|
import threading
|
|
from typing import Callable, Optional
|
|
from dataclasses import dataclass, field
|
|
from datetime import datetime
|
|
|
|
|
|
@dataclass
|
|
class DeviceInfo:
|
|
dev_serial: str
|
|
model: str = ""
|
|
hard_ver: str = ""
|
|
soft_ver: str = ""
|
|
product_code: str = ""
|
|
net_enabled: bool = False
|
|
iot_enabled: bool = False
|
|
bus: dict = field(default_factory=dict)
|
|
last_seen: Optional[datetime] = None
|
|
online: bool = False
|
|
loop_data: Optional[dict] = None
|
|
events: list = field(default_factory=list)
|
|
|
|
|
|
class DeviceManager:
|
|
"""管理设备列表和状态"""
|
|
|
|
def __init__(self):
|
|
self._devices: dict[str, DeviceInfo] = {}
|
|
self._lock = threading.Lock()
|
|
self._callbacks: list[Callable] = []
|
|
|
|
def on_change(self, callback: Callable):
|
|
self._callbacks.append(callback)
|
|
|
|
def _notify(self):
|
|
for cb in self._callbacks:
|
|
try:
|
|
cb()
|
|
except Exception:
|
|
pass
|
|
|
|
def get_or_create(self, dev_serial: str) -> DeviceInfo:
|
|
with self._lock:
|
|
if dev_serial not in self._devices:
|
|
self._devices[dev_serial] = DeviceInfo(dev_serial=dev_serial)
|
|
return self._devices[dev_serial]
|
|
|
|
def update_from_dev_info(self, dev_serial: str, data: dict):
|
|
with self._lock:
|
|
dev = self.get_or_create(dev_serial)
|
|
dev.dev_serial = data.get("dev_serial", dev_serial)
|
|
dev.model = data.get("model", dev.model)
|
|
dev.hard_ver = data.get("hard_ver", dev.hard_ver)
|
|
dev.soft_ver = data.get("soft_ver", dev.soft_ver)
|
|
dev.product_code = data.get("product_code", dev.product_code)
|
|
sc = data.get("sub_code", {})
|
|
dev.net_enabled = sc.get("net", dev.net_enabled)
|
|
dev.iot_enabled = sc.get("iot", dev.iot_enabled)
|
|
dev.bus = data.get("bus", dev.bus)
|
|
dev.last_seen = datetime.now()
|
|
dev.online = True
|
|
self._notify()
|
|
|
|
def update_loop_data(self, dev_serial: str, data: dict):
|
|
with self._lock:
|
|
dev = self.get_or_create(dev_serial)
|
|
dev.loop_data = data
|
|
dev.last_seen = datetime.now()
|
|
dev.online = True
|
|
self._notify()
|
|
|
|
def update_event(self, dev_serial: str, data: dict):
|
|
with self._lock:
|
|
dev = self.get_or_create(dev_serial)
|
|
events = data.get("events", [])
|
|
for evt in events:
|
|
evt["_ts"] = datetime.now()
|
|
dev.events.insert(0, evt)
|
|
dev.events = dev.events[:100] # 保留最近 100 条
|
|
dev.last_seen = datetime.now()
|
|
dev.online = True
|
|
self._notify()
|
|
|
|
def update_heartbeat(self, dev_serial: str, data: dict):
|
|
with self._lock:
|
|
dev = self.get_or_create(dev_serial)
|
|
dev.last_seen = datetime.now()
|
|
dev.online = True
|
|
self._notify()
|
|
|
|
def set_offline(self):
|
|
"""标记所有设备离线（断开连接时调用）"""
|
|
with self._lock:
|
|
now = datetime.now()
|
|
for dev in self._devices.values():
|
|
if dev.last_seen and (now - dev.last_seen).seconds > 120:
|
|
dev.online = False
|
|
self._notify()
|
|
|
|
@property
|
|
def devices(self) -> dict[str, DeviceInfo]:
|
|
with self._lock:
|
|
return dict(self._devices)
|
|
|
|
@property
|
|
def online_devices(self) -> list[DeviceInfo]:
|
|
with self._lock:
|
|
return [d for d in self._devices.values() if d.online]
|