可运行

sri-pysdk/xclient/xmqtt.py

@@ -8,6 +8,7 @@ class Client(mqtt.Client):
     def __init__(self, host='127.0.0.1', port=41883):
         super().__init__()
         self.connect(host=host, port=port, keepalive=60)
+        self.casper_info = {}
         # if subscribe_topic:
         #     def on_message(client, userdata, message):
         #         # print(f"#userdata: {userdata}")

sri-server-bg02/lib/TopicListener.py

@@ -8,19 +8,21 @@ import os
 
 from lib.UserWorkRecord import UserWorkRecord
 
-class LocalClass(object):
-    last_log_key = str()
+# class LocalClass(object):
+#     last_log_key = str()
+
+topic_info_dict = {}  # {<topic_name>: <last_log_key>}
 
 def method_v001(_client, _userdata, message):
     """消息处理方法"""
     # --- debug ---
-    methods.debug_log(f"TopicListener|17", f"#message.payload: {message.payload}")
+    # methods.debug_log(f"TopicListener|17", f"#message.payload: {message.payload}")
     # return
 
     # --- check ---
     # if not methods.is_dir(Global.save_dir):
     #     out = methods.run_command(f'mkdir -p {Global.save_dir}', callback=True)
-    #     methods.debug_log('MessageListener46', f"#out: {out}")
+    #     methods.debug_log('TopicListener|25', f"#out: {out}")
 
     # --- check ---
     try:
@@ -30,15 +32,15 @@ def method_v001(_client, _userdata, message):
         vehicle_id = log_dict.get('VehicleID')
         cockpit_id = log_dict.get('cockpitID')
     except Exception as exception:
-        methods.debug_log("method_v001|41", f"#message.payload: {message.payload}")
-        methods.debug_log("method_v001|41", f"#exception: {exception.__class__.__name__}")
-        methods.debug_log("method_v001|41", f"#traceback: {traceback.format_exc()}")
+        methods.debug_log("TopicListener|35", f"#message.payload: {message.payload}")
+        methods.debug_log("TopicListener|35", f"#exception: {exception.__class__.__name__}")
+        methods.debug_log("TopicListener|35", f"#traceback: {traceback.format_exc()}")
         return
 
     # --- check ---
     log_uuid = UserWorkRecord.get_log_uuid(user_uuid, log_time, vehicle_id, cockpit_id)
     if not log_uuid:
-        methods.debug_log(f"MessageListenerV001|error50", f"user_uuid: {user_uuid}, log_time: {log_time}")
+        methods.debug_log(f"TopicListener|error50", f"user_uuid: {user_uuid}, log_time: {log_time}")
         return
 
     # --- fill log_list ---
@@ -94,16 +96,21 @@ def method_v001(_client, _userdata, message):
 
     ]
 
+    # --- get ---
+    topic = _client.casper_info.get('topic')
+    last_log_key = topic_info_dict.get(topic)
+
     # --- check ---
     log_key = '-'.join(log_list[1:])
-    if LocalClass.last_log_key and LocalClass.last_log_key == log_key:
+    if last_log_key and last_log_key == log_key:
         return
 
     # --- update --
-    LocalClass.last_log_key = log_key
+    # last_log_key = log_key
+    topic_info_dict[topic] = log_key
 
     # --- save ---
-    # methods.debug_log(f"MessageListenerV001|113", f"#log_list: {log_list}")
+    # methods.debug_log(f"TopicListener|113", f"#log_list: {log_list}")
     log_file_path = os.path.join(Global.save_dir, f'{log_uuid}.log')
     methods.write_text(log_file_path, ' | '.join(log_list) + '\n', 'a')
 
@@ -165,6 +172,8 @@ def method_c001(_client, _userdata, message):
     }
     """
     # --- debug ---
+    # methods.debug_log(f"TopicListener|168", f"#_client: {_client.casper_info} | {type(_client.casper_info)}")
+    # methods.debug_log(f"TopicListener|168", f"#_client: {_userdata} | {type(_userdata)}")
     # methods.debug_log(f"TopicListener|168", f"#message.payload: {message.payload}")
     # return
 
@@ -172,9 +181,9 @@ def method_c001(_client, _userdata, message):
     try:
         log_dict = json.loads(message.payload)
     except Exception as exception:
-        methods.debug_log("method_c001|79", f"#message.payload: {message.payload}")
-        methods.debug_log("method_c001|79", f"#exception: {exception.__class__.__name__}")
-        methods.debug_log("method_c001|79", f"#traceback: {traceback.format_exc()}")
+        methods.debug_log("TopicListener|79", f"#message.payload: {message.payload}")
+        methods.debug_log("TopicListener|79", f"#exception: {exception.__class__.__name__}")
+        methods.debug_log("TopicListener|79", f"#traceback: {traceback.format_exc()}")
         return
 
     # --- check ---
@@ -185,7 +194,7 @@ def method_c001(_client, _userdata, message):
     cockpit_id = log_dict.get('cockpitID')
     log_uuid = UserWorkRecord.get_log_uuid(user_uuid, log_time, vehicle_id, cockpit_id)
     if not log_uuid:
-        methods.debug_log(f"MessageListenerC001|error92", f"user_uuid: {user_uuid}, log_time: {log_time}")
+        methods.debug_log(f"TopicListener|error92", f"user_uuid: {user_uuid}, log_time: {log_time}")
         return
 
     # --- fill log_list ---
@@ -239,16 +248,21 @@ def method_c001(_client, _userdata, message):
 
     ]
 
+    # --- get ---
+    topic = _client.casper_info.get('topic')
+    last_log_key = topic_info_dict.get(topic)
+
     # --- check ---
     log_key = '-'.join(log_list[1:])
-    if LocalClass.last_log_key and LocalClass.last_log_key == log_key:
+    if last_log_key and last_log_key == log_key:
         return
 
     # --- update --
-    LocalClass.last_log_key = log_key
+    # last_log_key = log_key
+    topic_info_dict[topic] = log_key
 
     # --- save ---
-    # methods.debug_log(f"MessageListenerC001|158", f"#log_list: {log_list}")
+    # methods.debug_log(f"TopicListener|158", f"#log_list: {log_list}")
     log_file_path = os.path.join(Global.save_dir, f'{log_uuid}.log')
     methods.write_text(log_file_path, ' | '.join(log_list) + '\n', 'a')
 
@@ -265,7 +279,9 @@ class TopicListener:
     def subscribe(self):
 
 
+        topic_info_dict[self.subscribe_topic] = str()
         
+
         if self.method_name == 'method_v001':
             method = method_v001
         elif self.method_name == 'method_c001':
@@ -275,6 +291,9 @@ class TopicListener:
 
 
         client = Global.emqx_factory.Client(host=Global.emqx_host, port=Global.emqx_port)
+        client.casper_info = {
+            'topic': self.subscribe_topic
+        }
         client.start_subscribe_loop(
             decorate_method=method,
             subscribe_topic=self.subscribe_topic