修改适配新版设备通信协议

mqtt/mqtt_process.py

@@ -87,7 +87,7 @@ class MQTTConsumerThread(threading.Thread):
 def on_connect(client:mqtt.Client, userdata, flags, rc):
     if rc == 0:
         LOGINFO("MQTT Connected successfully!")
-        client.subscribe(TOPICS.dev_dsp_data)
+        client.subscribe(TOPICS.dev_tracker_targets)
         # client.subscribe(TOPICS.das_all)
         # client.subscribe(TOPICS.mps_all)
         client.subscribe(TOPICS.las_alarm_plan_update, qos=2)

mqtt/mqtt_recv.py

@@ -63,6 +63,8 @@ def check_topic(topic:str, pattern:str) -> bool:
     ]
 }
 """
+
+
 # 处理来自设备的实时消息： /dev/{device_id}/dsp_data
 def deal_dsp_data(msg:mqtt.MQTTMessage):
     try:
@@ -100,6 +102,43 @@ def deal_dsp_data(msg:mqtt.MQTTMessage):
             LOGERR(f"[{frame.filename}:{frame.lineno}] @{frame.name}(), error: {e}")
 
 
+# 处理来自设备的实时数据： /dev/{device_id}/tracker_targets
+def deal_tracker_targets(msg:mqtt.MQTTMessage):
+    try:
+        parts = msg.topic.split('/')
+        dev_id = parts[2]
+
+        device: Device = g_Dev.g_dev_mgr.find_dev_map(dev_id)
+        if not g_Dev.g_dev_mgr.find_dev_map(dev_id):
+            device = Device(dev_id)
+            g_Dev.g_dev_mgr.push_dev_map(dev_id, device)
+
+        payload = json.loads(msg.payload.decode('utf-8'))
+
+        # 处理 target
+        if ("tracker_targets" in payload):
+            tracker_targets = payload["tracker_targets"]
+
+            timestamp = get_utc_time_s()
+            pose = POSE_E.POSE_4.value
+            rtd_unit = {
+                "timestamp": timestamp,
+                "pose": pose,
+                "target_point": tracker_targets
+            }
+            device.put_rtd_unit(rtd_unit)
+            device.update_keepalive(timestamp)
+
+    except json.JSONDecodeError as e:
+        tb_info = traceback.extract_tb(e.__traceback__)
+        for frame in tb_info:
+            LOGERR(f"[{frame.filename}:{frame.lineno}] @{frame.name}(), error:{e}, {e.doc}")
+    except Exception as e:
+        tb_info = traceback.extract_tb(e.__traceback__)
+        for frame in tb_info:
+            LOGERR(f"[{frame.filename}:{frame.lineno}] @{frame.name}(), error: {e}")
+
+
 # 实时位置姿态
 def deal_realtime_pos(msg:mqtt.MQTTMessage):
     try:
@@ -175,8 +214,13 @@ def deal_dev_msg(client:mqtt.Client, userdaata, msg:mqtt.MQTTMessage):
 
         # 设备实时数据
         elif (check_topic(topic,Topic_Pattern.dev_dsp_data)):
+            return
             deal_dsp_data(msg)
 
+        # 设备实时数据
+        elif (check_topic(topic,Topic_Pattern.dev_tracker_targets)):
+            deal_tracker_targets(msg)
+
         # 点云数据
         elif (check_topic(topic,Topic_Pattern.dev_cloudpoint)):
             return

mqtt/mqtt_topics.py

@@ -7,6 +7,7 @@ class TOPICS(str):
     dev_rep_dev_info    = "/dev/+/report_device_info"       # 设备信息
     dev_rep_dev_param   = "/dev/+/report_device_param"      # 设备参数
     dev_dsp_data        = "/dev/+/dsp_data"                 # 设备实时数据
+    dev_tracker_targets = "/dev/+/tracker_targets"          # 设备实时数据（新版V2.1.1, 2025-09-08）
     dev_cloudpoint      = "/dev/+/cloudpoint"               # 点云数据
     dev_rep_fall_event  = "/dev/+/report_falling_event"     # 跌倒事件
     dev_rep_pres_event  = "/dev/+/report_presence_event"    # 存在事件
@@ -59,7 +60,8 @@ class Topic_Pattern(str):
     dev_keepalive       = r"^/dev/[^/]+/keepalive$"             # 心跳保活（请求）
     dev_rep_dev_info    = r"^/dev/[^/]+/report_device_info$"    # 设备信息
     dev_rep_dev_param   = r"^/dev/[^/]+/report_device_param$"   # 上报设备参数
-    dev_dsp_data        = r"^/dev/[^/]+/dsp_data$"              # 设备实时数据
+    dev_dsp_data        = r"^/dev/[^/]+/dsp_data$"              # 设备实时数据（旧版）
+    dev_tracker_targets = r"^/dev/[^/]+/tracker_targets$"       # 设备实时数据（新版V2.1.1, 2025-09-08）
     dev_cloudpoint      = r"^/dev/[^/]+/cloudpoint$"            # 点云数据
     dev_rep_fall_event  = r"^/dev/[^/]+/report_falling_event$"  # 跌倒事件
     dev_rep_pres_event  = r"^/dev/[^/]+/report_presence_event$" # 存在事件