LAS.py

@@ -14,7 +14,7 @@ from common.sys_comm import (
     get_utc_time_ms
 )
 
-from mqtt import mqtt_process as mp
+import mqtt.mqtt_process as mp
 from mqtt.mqtt_process import (
     MQTTClientThread, RobustMQTTClient, MQTTConsumerThread, g_mqtt_client, g_mqtt_consumer
 )

mqtt/mqtt_process.py

@@ -113,6 +113,7 @@ class MQTTClientThread(threading.Thread):
             LOGINFO("MQTT Connected successfully!")
             client.subscribe(TOPICS.dev_tracker_targets)
             client.subscribe(TOPICS.las_alarm_plan_update, qos=2)
+            client.subscribe(TOPICS.las_test, qos=2)
         else:
             LOGERR(f"MQTT failed to connect, return code {rc}")
 
@@ -208,7 +209,8 @@ class MQTTClientThread(threading.Thread):
 class RobustMQTTClient(threading.Thread):
     def __init__(self, max_workers=4):
         super().__init__(name="RobustMQTTClient")
-        self.client: mqtt.Client = mqtt.Client()
+        client_id: str = "LAS_" + str(sys_comm.g_sys_conf["sp_id"])
+        self.client: mqtt.Client = mqtt.Client(client_id= client_id)
         self.client.username_pw_set(MQTT_USERNAME, MQTT_PASSWD)
         self.client.on_connect = self.on_connect
         self.client.on_disconnect = self.on_disconnect
@@ -230,6 +232,7 @@ class RobustMQTTClient(threading.Thread):
             LOGINFO("MQTT connected successfully!")
             client.subscribe(TOPICS.dev_tracker_targets)
             client.subscribe(TOPICS.las_alarm_plan_update, qos=2)
+            client.subscribe(TOPICS.las_test, qos=2)
         else:
             LOGERR(f"MQTT failed to connect, rc={rc}")
 

mqtt/mqtt_recv.py

@@ -197,6 +197,22 @@ def deal_alarm_plan_update(msg:mqtt.MQTTMessage):
             LOGERR(f"[{frame.filename}:{frame.lineno}] @{frame.name}(), error: {e}")
 
 
+
+# 告警计划变更
+def deal_las_test(msg:mqtt.MQTTMessage):
+    try:
+        mqtt_send.las_test_resp()
+
+    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}")
+
+
 # 设备消息分发处理：/dev/#
 def deal_dev_msg(client:mqtt.Client, userdaata, msg:mqtt.MQTTMessage):
     try:
@@ -271,6 +287,9 @@ def deal_las_msg(client:mqtt.Client, userdaata, msg:mqtt.MQTTMessage):
         if (check_topic(topic,Topic_Pattern.las_alarm_plan_update)):
             deal_alarm_plan_update(msg)
 
+        elif (check_topic(topic,Topic_Pattern.las_test)):
+            deal_las_test(msg)
+
     except json.JSONDecodeError as e:
         tb_info = traceback.extract_tb(e.__traceback__)
         for frame in tb_info:
@@ -297,7 +316,7 @@ def process_message(client:mqtt.Client, userdata, msg:mqtt.MQTTMessage):
     elif (check_topic(topic,Topic_Pattern.das_all)):     # 设备代理
         return
         deal_das_msg(client, userdata, msg)
-    elif (check_topic(topic,Topic_Pattern.las_all)):     # 只接受告警配置相关
+    elif (check_topic(topic,Topic_Pattern.las_all)):     # las相关
         deal_las_msg(client, userdata, msg)
     else:
         # LOGDBG(f"recv invalid topic: {msg.topic}")

mqtt/mqtt_send.py

@@ -73,6 +73,24 @@ def alarm_event(
     send_msg(TOPICS.las_alarm_event, format_json, 2)
 
 
+# 测试
+def las_test_resp():
+    module_name     = sys_comm.g_sys_conf["module_name"]
+    platform        = sys_comm.g_sys_conf["platform"]
+    host_ip         = sys_comm.g_sys_conf["host_ip"]
+    max_log_files   = sys_comm.g_sys_conf["max_log_files"]
+    max_log_size    = sys_comm.g_sys_conf["max_log_size"]
+    log_lvl         = sys_comm.g_sys_conf["log_lvl"]
+    sp_id           = sys_comm.g_sys_conf["sp_id"]
+    format_json = dict()
+    format_json["module_name"]      = module_name
+    format_json["platform"]         = platform
+    format_json["host_ip"]          = host_ip
+    format_json["max_log_files"]    = max_log_files
+    format_json["max_log_size"]     = max_log_size
+    format_json["log_lvl"]          = log_lvl
+    format_json["sp_id"]            = sp_id
+    send_msg(TOPICS.las_test_resp, format_json, 2)
 
 
 

mqtt/mqtt_topics.py

@@ -50,6 +50,8 @@ class TOPICS(str):
     las_all                 = "/las/#"
     las_alarm_plan_update   = "/las/alarm_plan_update"      # 告警计划变更
     las_alarm_event         = "/las/alarm_event"            # 上报告警事件
+    las_test                = "/las/test"                   # 测试
+    las_test_resp           = "/las/test_resp"              # 测试响应
 
 
 # topic匹配规则
@@ -108,4 +110,6 @@ class Topic_Pattern(str):
     # 告警联动服务
     las_all                 = r"^/las/.*$"
     las_alarm_plan_update   = r"^/las/alarm_plan_update$"       # 告警计划变更
-    las_alarm_event         = r"^/las/alarm_eventr$"            # 上报告警事件
+    las_alarm_event         = r"^/las/alarm_eventr$"            # 上报告警事件
+    las_test                = r"^/las/test$"                    # 测试
+    las_test_resp           = r"^/las/test_resp$"               # 测试响应