feat: 优化设备采集cd

apps/em/cd.py

@@ -4,6 +4,41 @@ import json
 import time
 from django.core.cache import cache
 from apps.utils.thread import MyThread
+import struct
+
+def get_checksum(body_msg):
+    return sum(body_msg) & 0xFF
+
+def handle_bytes(arr):
+    if len(arr) < 8:
+        return "返回数据长度错误"
+    
+    if arr[0] != 0xEB or arr[1] != 0x90:
+        return "数据头不正确"
+
+
+    # 读取长度信息
+    length_arr = arr[2:4][::-1]  # 反转字节
+    length = struct.unpack('<H', bytes(length_arr))[0]  # 小端格式
+    
+    # 提取内容
+    body = arr[4:4 + length - 3]
+
+    # 校验和检查
+    check_sum = get_checksum(body)
+    if check_sum != arr[length + 1]:
+        return "校验错误"
+        
+
+    # 尾部标识检查
+    if arr[length + 2] != 0xFF or arr[length + 3] != 0xFE:
+        return "尾错误"
+
+    content = body.decode('utf-8')
+
+    res = json.loads(content)
+    
+    return res[0]
 
 def get_tyy_data_t(host, port):
     cd_thread_key = f"cd_thread_{host}_{port}"
@@ -21,18 +56,17 @@ def get_tyy_data_t(host, port):
                 connect_and_send(retry-1)
         except OSError as e:
             sc = None
-            if retry > 0:
-                connect_and_send(retry-1)
-            else:
-                cache.set(cd_thread_key, {"err_msg": f"采集器连接失败-{str(e)}"})
+            cache.set(cd_thread_key, {"err_msg": f"采集器连接失败-{str(e)}"})
         except ConnectionResetError:
             sc = None
-            if retry > 0:
-                connect_and_send(retry-1)
-            else:
-                cache.set(cd_thread_key, {"err_msg": "采集器重置了连接"})
+            cache.set(cd_thread_key, {"err_msg": "采集器重置了连接"})
         except socket.timeout:
+            sc = None
             cache.set(cd_thread_key, {"err_msg": "采集器连接超时"})
+        except Exception as e:
+            sc = None
+            cache.set(cd_thread_key, {"err_msg": f"采集器连接失败-{str(e)}"})
+
     while True:
         cd_thread_val = cache.get(cd_thread_key, default=None)
         if cd_thread_val is None:
@@ -43,23 +77,21 @@ def get_tyy_data_t(host, port):
                     pass
             break
         elif cd_thread_val == "get":
+            cache.set(cd_thread_key, "working")
             connect_and_send()
             if sc is None:
-                cache.set(cd_thread_key, {"err_msg": "采集器连接失败"})
                 continue
             resp = sc.recv(1024)
-            if not resp:
-                cache.set(cd_thread_key, {"err_msg": f"采集器返回空数据-{str(resp)}"})
-            elif len(resp) < 8:
-                cache.set(cd_thread_key, {"err_msg": f"设备未启动-{str(resp)}"})
+            res = handle_bytes(resp)
+            if isinstance(res, str):
+                cache.set(cd_thread_key, {"err_msg": f'采集器返回数据错误-{res}'})
             else:
-                json_data = resp[5:-4]
-                json_str = json_data.decode('utf-8')
-                res = json.loads(json_str)
                 cache.set(cd_thread_key, res)
         time.sleep(0.3)
 
-def get_tyy_data(*args):
+def get_tyy_data(*args, sleep=0):
+    if sleep > 0:
+        time.sleep(sleep)
     host, port = args[0], int(args[1])
     cd_thread_key = f"cd_thread_{host}_{port}"
     cd_thread_val = cache.get(cd_thread_key, default=None)
@@ -69,17 +101,23 @@ def get_tyy_data(*args):
         cd_thread.start()
     cache.set(cd_thread_key, "get")
     num = 0
+    get_val = False
+
     while True:
         num += 1
         if num > 8:
             break
         val = cache.get(cd_thread_key)
         if isinstance(val, dict):
+            get_val = True
             if "err_msg" in val:
                 raise ParseError(val["err_msg"])
             return val
         time.sleep(0.3)
-    raise ParseError("获取数据超时")
+
+    if not get_val:
+        cache.set(cd_thread_key, None)
+        get_tyy_data(*args, sleep=2)
 
 
 if __name__ == '__main__':