feat: 数采做成单独服务

apps/em/cd.py

@@ -4,10 +4,10 @@ import json
 import time
 from django.core.cache import cache
 from apps.utils.thread import MyThread
-import struct
 import uuid
 import logging
 import threading
+import requests
 
 myLogger = logging.getLogger('log')
 
@@ -136,7 +136,7 @@ def get_tyy_data_2(*args, retry=1):
 sc_all = {}
 sc_lock = threading.Lock()
 
-def get_tyy_data(*args, retry=1):
+def get_tyy_data_1(*args, retry=1):
     host, port = args[0], int(args[1])
     global sc_all
     sc = None
@@ -240,4 +240,13 @@ def get_tyy_data_3(*args, retry=2):
         except ParseError:
             raise
         except Exception as e:
-            raise ParseError(f"未知错误: {str(e)}")
+            raise ParseError(f"未知错误: {str(e)}")
+        
+
+def get_tyy_data(*args):
+    host, port = args[0], int(args[1])
+    r = requests.get(f"http://127.0.0.1:2300?host={host}&port={port}")
+    res = r.json()
+    if "err_msg" in res:
+        raise ParseError(res["err_msg"])
+    return res

out_service/cd.py

@@ -0,0 +1,104 @@
+from http.server import BaseHTTPRequestHandler, HTTPServer
+import json
+from urllib.parse import urlparse, parse_qs
+import socket
+
+sc_all = {
+    "192.168.1.220_6000": None,
+    "192.168.1.235_6000": None,
+    "192.168.1.225_6000": None,
+    "192.168.1.230_6000": None,
+}
+
+def get_checksum(body_msg):
+    return sum(body_msg) & 0xFF
+
+def handle_bytes(arr):
+    if len(arr) < 8:
+        return f"返回数据长度错误-{arr}"
+    
+    if arr[0] != 0xEB or arr[1] != 0x90:
+        return "数据头不正确"
+
+
+    # 读取长度信息
+    length_arr = arr[2:4][::-1]  # 反转字节
+    length = int.from_bytes(length_arr, byteorder='little', signed=True)  # 小端格式
+    
+    # 提取内容
+    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]
+
+class JSONRequestHandler(BaseHTTPRequestHandler):
+    def ok(self, data):
+        self.send_response(200)
+        self.send_header('Content-Type', 'application/json; charset=utf-8')
+        self.end_headers()
+        self.wfile.write(json.dumps(data, ensure_ascii=False).encode('utf-8'))
+
+    def error(self, err_msg):
+        self.send_response(400)
+        self.send_header('Content-Type', 'application/json; charset=utf-8')
+        self.end_headers()
+        data = {
+            "err_msg": err_msg
+        }
+        self.wfile.write(json.dumps(data, ensure_ascii=False).encode('utf-8'))
+
+    def do_GET(self):
+        parsed_url = urlparse(self.path)
+        query_params = parse_qs(parsed_url.query)
+        host = query_params.get('host', ['127.0.0.1'])[0]
+        port = query_params.get('port', ['6000'])[0]
+        addr = f'{host}_{port}'
+        if addr not in sc_all:
+            self.error(f'{addr} 未找到')
+            return
+        sc = None
+        def connect_and_send():
+            nonlocal sc
+            sc = sc_all[addr]
+            if sc is None:
+                sc = socket.socket()
+                sc.settimeout(5)  # 设置超时
+                sc.connect((host, int(port)))
+                sc_all[f"{host}_{port}"] = sc
+            try:
+                sc.sendall(b"R")
+            except Exception as e:
+                try:
+                    sc.close()
+                except Exception:
+                    pass
+                self.error(f'采集器连接失败: {e}')
+
+        connect_and_send()
+        resp = sc.recv(1024)
+        res = handle_bytes(resp)
+        if isinstance(res, str):
+            self.error(res)
+        else:
+            self.ok(res)
+def run(server_class=HTTPServer, handler_class=JSONRequestHandler, port=2300):
+    server_address = ('', port)
+    httpd = server_class(server_address, handler_class)
+    print(f'Starting httpd server on port {port}...')
+    httpd.serve_forever()
+
+if __name__ == '__main__':
+    run()