socket

python/api.py

@@ -1,20 +1,53 @@
 from models import *
+import requests
+import json
 from logger import logger
 from serial.tools import list_ports
+from utils.hlm_http_request import forward_request
+from sockets.socket_client import socket_manager
+
 @app.get("/")
 async def index():
-    logger.info("Starting")
+    await socket_manager.send_message(msg="测试")
     return {"message": "Hello World"}
 
 
-@app.get("/api/hello")
-async def hello():
-    return {"app_name": "FastAPI框架学习", "app_version": "v0.0.1"}
-
-
-@app.get("/scan_serials")
+@app.get("/scan_serials", description="扫描可用的设备端口")
 async def scanSerials():
-    '''扫描串口'''
+    """扫描串口"""
     ports = list_ports.comports()
-    print("Scanning",ports)
-    return {"message": "Hello World"}
+    print("Scanning", ports)
+    return {"message": "Hello World"}
+
+
+@app.api_route(
+    "/forward_request", methods=["GET", "POST"], description="代理转发hlm项目得请求"
+)
+async def forwardRequest(request: HlmForwardRequest):
+    """
+    转发HTTP请求到目标URL
+
+    :param request: FastAPI Request对象
+    :return: 目标接口的响应
+    """
+    try:
+        if request.method == "GET":
+            params = request.query_params
+        elif request.method == "POST":
+            params = json.dump(request.query_params)
+        else:
+            raise UnicornException("仅支持GET和POST方法")
+        target_url = request.target_url
+        method = request.method.upper()
+        headers = request.headers
+        if not target_url:
+            raise UnicornException("目标url地址不能为空")
+        # 调用 hlm_http_request 中的 forward_request 函数
+        response = forward_request(
+            target_url, params=params, method=method, headers=headers
+        )
+        return response
+    except requests.RequestException as e:
+        raise UnicornException(e)
+    except Exception as e:
+        raise UnicornException(e)

python/config.ini

@@ -9,7 +9,7 @@ host=127.0.0.1
 app_run=api:app
 # 端口号
 port=7074
-debug=false
+debug=true
 env=prod
 # 线程数
 works=1

python/mcu/DeviceHandler.py

@@ -0,0 +1,136 @@
+import time
+from mcu.SerialIns import SerialIns
+
+class DeviceHandler:
+
+    def __init__(self, windows=None):
+        super().__init__()
+        self.windows = windows
+        self.serial_ins = None
+        self.port_name = ""
+        self.connect_state = False
+        self.is_running = False
+
+    def to_connect_com(self, port_name, is_test=False):
+        self.close_connect()
+        time.sleep(0.5)
+        try:
+            # 原值为9600
+            self.serial_ins = SerialIns(port_name=port_name, baud=115200)
+            # self.serial_ins = SerialIns(port_name=port_name, baud=9600)
+            if self.serial_ins.serial_handle:
+                self.sign_data.emit(
+                    {
+                        "_type": "show_info",
+                        "plugins_mode": "remote_control",
+                        "data": "遥控设备 打开串口成功",
+                    }
+                )
+
+                self.connect_state = True
+                self.sign_data.emit(
+                    {
+                        "_type": "remote_control_connect",
+                        "plugins_mode": "remote_control",
+                        "data": port_name,
+                    }
+                )
+                self.port_name = port_name
+                if is_test is False:
+                    self.start()
+                return True
+
+            else:
+                self.sign_data.emit(
+                    {
+                        "_type": "show_info",
+                        "plugins_mode": "remote_control",
+                        "data": "遥控设备 打开串口失败",
+                    }
+                )
+
+                self.serial_ins = None
+                self.connect_state = False
+        except:
+            self.sign_data.emit(
+                {
+                    "_type": "show_info",
+                    "plugins_mode": "remote_control",
+                    "data": "遥控设备 打开串口失败",
+                }
+            )
+            self.serial_ins = None
+            self.connect_state = False
+            return False
+
+    def close_connect(self):
+        self.port_name = ""
+        if self.connect_state:
+            self.serial_ins.close_serial_port()
+            self.connect_state = False
+
+    def __del__(self):
+        self.close_connect()
+
+    def analysis_received_data(self):
+        if not self.connect_state:
+            return
+        receive_data = self.serial_ins.read_cmd(out_time=1, check=None)
+        if receive_data is False:
+            self.connect_state = False
+            return False
+        if not receive_data:
+            return
+        else:
+            print(
+                "read receive_data {}".format(
+                    self.serial_ins.change_hex_to_int(receive_data)
+                )
+            )
+            pass
+        # 数据 结构 command，按命令解析
+        if receive_data[0] == 1:
+            # 扫码数据
+            bar_code = receive_data[1:].decode()
+            bar_code = bar_code.replace("\r", "")
+            bar_code = bar_code.replace("\n", "")
+
+            self.sign_data.emit(
+                {"_type": 0, "plugins_mode": "remote_control", "data": bar_code}
+            )
+            return
+        if receive_data[0] == 9:
+            button_value = receive_data[1]
+            data = {"button_value": button_value}
+            self.sign_data.emit(
+                {"_type": 9, "plugins_mode": "remote_control", "data": data}
+            )
+            return
+        pass
+
+    def run(self):
+        # self.show_info.emit("未连接")
+        # self.data_command_sign.emit(data)
+        self.is_running = True
+        while 1:
+            time.sleep(0.05)
+            if not self.connect_state:
+                self.sign_data.emit(
+                    {
+                        "_type": "show_info",
+                        "plugins_mode": "remote_control",
+                        "data": "遥控设备 未连接",
+                    }
+                )
+                break
+            self.analysis_received_data()
+
+        self.is_running = False
+        if not self.connect_state:
+            self.sign_data.emit(
+                {
+                    "_type": "show_info",
+                    "plugins_mode": "remote_control",
+                    "data": "遥控设备 未连接",
+                }
+            )

python/mcu/SerialIns.py

@@ -0,0 +1,322 @@
+#!usr/bin/python3
+import serial.tools.list_ports
+import time
+
+
+# https://blog.csdn.net/weixin_44289254/article/details/121583562
+
+
+class SerialIns(object):
+    def __init__(self, port_name=None, baud=9600, timeout=0.01):
+        self.port_name = port_name
+        self.baud = baud
+        self.serial_handle = None
+        # self.serial_handle = self.check_connect()
+        try:
+            self.serial_handle = serial.Serial(
+                port=self.port_name, baudrate=self.baud, timeout=timeout
+            )
+            print("{}打开成功".format(self.port_name))
+        except:
+            self.serial_handle = None
+            print("{}打开失败".format(self.port_name))
+
+        self.receive_data = b""
+        # self.receive_data = bytearray(b'\x00\UU\x02d\x00\x9b\x9b')
+
+    def read_line(self):
+        c = self.serial_handle.inWaiting()
+        if c:
+            time.sleep(0.02)
+            # 确保缓存写入完成
+            if c == self.serial_handle.inWaiting():
+                return self.serial_handle.readline()
+        return None
+
+    def read_all(self):
+        return self.serial_handle.read_all()
+
+    def check_connect(self):
+        if not self.port_name:
+            self.scan_serial_port()
+            if self.port_name:
+                self.serial_handle = serial.Serial(self.port_name, self.baud)
+        else:
+            try:
+                # if self.scan_serial_port(self.port_name):
+                self.serial_handle = serial.Serial(self.port_name, self.baud)
+                print("{}打开成功".format(self.port_name))
+            except:
+                print("{}打开失败".format(self.port_name))
+
+        return self.serial_handle
+
+    def clearn_flush(self):  # 清空接收缓存
+        if self.serial_handle:
+            self.serial_handle.flushInput()
+            self.receive_data = b""
+
+    def write_cmd(self, data: list):
+        if self.serial_handle:
+            # data = [(0xff & par1), (0xff & (par1 >> 8))]
+            # self.clearn_flush()
+            buf = bytearray(b"")
+            buf.extend([0x55, 0x55, (0xFF & len(data))])
+            buf.extend(data)
+            buf.extend([0xFF & ~sum(data)])
+            # 55 55 02 5a 01 a4
+            # print("send buf  {}".format(self.change_hex_to_int(buf)))
+            try:
+                self.serial_handle.write(buf)
+                return True
+            except:
+                self.serial_handle = None
+                _recv_data = b""
+                return False
+
+    def read_cmd_out_time(self, scan_interval=0.1, out_time=1):
+        if self.serial_handle:
+            """
+            获取数据，设计超时机制
+            :param scan_interval:
+            :param out_time:
+            :return:
+            """
+            n = 0
+            while 1:
+                n += 1
+                time.sleep(scan_interval)
+                if out_time <= n * scan_interval:
+                    return False
+                receive_data = self.read_cmd()
+                if receive_data:
+                    return receive_data
+        return False
+
+    def read_cmd1(self, out_time=1, check=None):
+        s = time.time()
+        while 1:
+            try:
+                _recv_data = self.serial_handle.read_all()  # 读取接收到的数据
+            except:
+                self.serial_handle = None
+                _recv_data = b""
+                return False
+
+            self.receive_data += _recv_data  # 写入所有缓存数据
+            # print(self.receive_data)
+            if not self.receive_data or len(self.receive_data) < 4:
+                return
+            if self.receive_data[0] == 0x55 and self.receive_data[1] == 0x55:
+                # print("read ori ", self.change_hex_to_int(self.receive_data))
+                data_len = self.receive_data[2]
+                if len(self.receive_data) < data_len + 4:
+                    # 此处需要超时机制
+                    print("1数据长度不够，等待下次读取")
+                    # 超时退出
+                    if time.time() - s > out_time:
+                        time.sleep(0.01)
+                        return
+                _data = self.receive_data[3 : data_len + 4]
+                # 更新缓存区
+                self.receive_data = self.receive_data[data_len + 4 :]
+                # 校验数据
+                if check is None:
+                    if 0xFF & ~sum(_data[:-1]) == _data[-1]:
+                        # print("data:", self.change_hex_to_int(data[:-1]))
+                        return _data[:-1]
+                    else:
+                        print("数据异常，丢弃")
+                        return
+                else:
+                    if _data[-1] == check:
+                        return _data[:-1]
+                    else:
+                        print("数据异常，丢弃")
+                        return
+            else:
+                # print("起始位不是 55 55 进行移除", self.receive_data[0])
+                # 起始位不是 55 55 进行移除
+                index = 0
+                for index, r_data in enumerate(self.receive_data):
+                    if r_data == 0x55:
+                        break
+
+                index += 1
+                if index >= len(self.receive_data):
+                    self.receive_data = b""
+                else:
+                    self.receive_data = self.receive_data[index - 1 :]
+
+    def read_cmd111(self, out_time=1, check=None):
+        while 1:
+            try:
+                _recv_data = self.serial_handle.read_all()  # 读取接收到的数据
+            except BaseException as e:
+                print("串口接收报错", e)
+                self.serial_handle = None
+                _recv_data = b""
+                return False
+
+            # print("read_cmd", _recv_data)
+            if not _recv_data or len(_recv_data) < 4:
+                # print("数据长度不够")
+                return
+
+            print("2   _recv_data", self.change_hex_to_int(_recv_data))
+            if _recv_data[0] == 0x55 and _recv_data[1] == 0x55:
+                # print("read ori ", self.change_hex_to_int(self.receive_data))
+                data_len = _recv_data[2]
+                if len(_recv_data) < data_len + 4:
+                    # 此处需要超时机制
+                    print("2数据长度不够，等待下次读取")
+                    # if time.time() - s > out_time:
+                    #     time.sleep(0.01)
+                    #     return
+                    return
+
+                _data = _recv_data[3 : data_len + 4]
+                # 校验数据
+                if check is None:
+                    if 0xFF & ~sum(_data[:-1]) == _data[-1]:
+                        # print("data:", self.change_hex_to_int(data[:-1]))
+                        return _data[:-1]
+                    else:
+                        print("数据异常，丢弃")
+                        return
+                else:
+                    if _data[-1] == check:
+                        return _data[:-1]
+                    else:
+                        print("数据异常，丢弃")
+                        return
+            else:
+                # 起始位不是 55 55 进行移除
+                while self.receive_data:
+                    if len(self.receive_data) == 1:
+                        if self.receive_data[0] == 0x55:
+                            break
+                        else:
+                            self.receive_data = b""
+                    else:
+                        if (
+                            self.receive_data[0] == 0x55
+                            and self.receive_data[1] == 0x55
+                        ):
+                            break
+                        else:
+                            self.receive_data = self.receive_data[1:]
+
+    def read_cmd(self, out_time=1, check=None, out_time_n=5):
+        n = 0
+        while 1:
+            try:
+                read_d = self.serial_handle.read_all()  # 读取接收到的数据
+                self.receive_data += read_d
+            except BaseException as e:
+                print("串口接收报错", e)
+                self.serial_handle = None
+                return False
+
+            if len(self.receive_data) < 4:
+                break
+
+            if self.receive_data[0] == 0x55 and self.receive_data[1] == 0x55:
+                # print("read ori ", self.change_hex_to_int(self.receive_data))
+                data_len = self.receive_data[2]
+                if len(self.receive_data) < data_len + 4:
+                    # 此处需要超时机制
+                    # print("数据长度不够，等待下次读取")
+                    # 超时退出
+                    # if not self.serial_handle.txdone():
+                    #     return None
+                    # n += 1
+                    # if n > out_time_n:
+                    #     return None
+                    # time.sleep(0.01)
+                    continue
+                _data = self.receive_data[3 : data_len + 4]
+                # 更新缓存区
+                self.receive_data = self.receive_data[data_len + 4 :]
+                # 校验数据
+                if 0xFF & ~sum(_data[:-1]) == _data[-1]:
+                    # print("receive_data:", self.change_hex_to_int(self.receive_data[:-1]))
+                    return _data[:-1]
+                else:
+                    return None
+            else:
+                # print("起始位不是 55 55 进行移除", self.receive_data[0])
+                # 起始位不是 55 55 进行移除
+                while self.receive_data:
+                    if len(self.receive_data) == 1:
+                        if self.receive_data[0] == 0x55:
+                            break
+                        else:
+                            self.receive_data = b""
+                    else:
+                        if (
+                            self.receive_data[0] == 0x55
+                            and self.receive_data[1] == 0x55
+                        ):
+                            break
+                        else:
+                            self.receive_data = self.receive_data[1:]
+
+    def change_hex_to_int(self, _bytearray):
+        return " ".join([hex(x)[2:].zfill(2) for x in _bytearray])
+
+    def scan_serial_port(self, port_name=None):
+        plist = list(serial.tools.list_ports.comports())
+        if len(plist) <= 0:
+            return None
+
+        if port_name:
+            for i in plist:
+                # print("port", i)
+                if i.name == port_name:
+                    return i
+            return None
+        else:
+            for i in plist:
+                if "CH340" in i.description:
+                    self.port_name = i.name
+                    print("----------", i)
+            return plist
+
+    def open_serial_port(self):
+        if self.serial_handle:
+            if not s.check_connect():
+                self.serial_handle.open()
+        return True
+
+    def close_serial_port(self):
+        if self.serial_handle:
+            self.serial_handle.close()
+            print("{}串口已关闭".format(self.port_name))
+            self.serial_handle = None
+
+    def __del__(self):
+        self.close_serial_port()
+
+
+if __name__ == "__main__":
+    s = SerialIns(port_name="COM11", baud=115200, timeout=0.1)
+    s.scan_serial_port()
+    s.clearn_flush()
+    print("-" * 30)
+    for i in range(2):
+        data = [0x55, 0x55, 0x2, 0x5A, 0x1, 0xA4]
+        print("command data {}".format(s.change_hex_to_int(data)))
+        # s.write_cmd(data=data)
+        buf = bytearray(b"")
+        buf.extend(data)
+        s.serial_handle.write(buf)
+
+        time.sleep(0.1)
+        data = s.read_cmd()
+        # print("data-->", data)
+        if data:
+            print("data--> {}".format(s.change_hex_to_int(data)))
+        time.sleep(1)
+    print("-" * 30)
+    s.close_serial_port()

python/models.py

@@ -1 +1,6 @@
 from middleware import *
+class HlmForwardRequest(BaseModel):
+    method: str = Field(default="GET", description="请求方法")
+    headers: dict = Field(default={}, description="请求头")
+    target_url: str = Field(default="", description="目标地址")
+    query_params:str = Field(default="", description="请求参数")

python/sockets/connect_manager.py

@@ -1,21 +1,32 @@
 from models import WebSocket
 from logger import logger
+import json
+
 class ConnectionManager:
     def __init__(self):
         self.active_connections: list[WebSocket] = []
 
+    def jsonMessage(self, code=0, msg="", data: object = None):
+        """json字符串数据"""
+        jsonData = {"code": code, "msg": msg, "data": data}
+        return json.dumps(jsonData)
+
     async def connect(self, websocket: WebSocket):
+        '''连接事件'''
         await websocket.accept()
         self.active_connections.append(websocket)
         logger.info("socket 已连接")
 
     def disconnect(self, websocket: WebSocket):
+        '''断开连接事件'''
         self.active_connections.remove(websocket)
         logger.info("socket 连接断开")
 
     async def send_personal_message(self, message: str, websocket: WebSocket):
+        '''向用户发送消息'''
         await websocket.send_json(message)
 
     async def broadcast(self, message: str):
+        """广播消息"""
         for connection in self.active_connections:
-            await connection.send_json(message)
+            await connection.send_json(message)

python/sockets/message_handler.py

@@ -0,0 +1,19 @@
+from .connect_manager import ConnectionManager
+from models import WebSocket
+
+
+# socket消息发送逻辑处理方法
+async def handlerSend(
+    manager: ConnectionManager, receiveData: str, websocket: WebSocket
+):
+    # 处理消息发送逻辑
+    jsonType = receiveData.get("type")
+    print("receiveData", receiveData)
+    match jsonType:
+        case "ping":
+            """发送心跳"""
+            data = manager.jsonMessage("pong")
+            await manager.send_personal_message(data, websocket)
+        case "pong":
+            """发送心跳"""
+            pass

python/sockets/socket_client.py

@@ -0,0 +1,37 @@
+# socket_manager.py
+import socket, json
+import websockets
+from settings import APP_HOST,PORT
+from middleware import UnicornException
+class SocketClient:
+
+    def __init__(self, uri="ws://127.0.0.1:7074"):
+        self.uri = uri
+        self.websocket = None
+
+    async def connect(self):
+        if self.websocket == None:
+            self.websocket = await websockets.connect(self.uri)
+            print(f"Local Socket Connected to {self.uri}")
+
+    async def send_message(self, code=0, msg="", data: object = None):
+        if self.websocket:
+            json_data = json.dumps({"code": code, "msg": msg, "data": data})
+            await self.websocket.send(json_data)
+            print("Message sent:", json_data)
+        else:
+            print("WebSocket client is not connected")
+
+    async def close(self):
+        if self.websocket:
+            await self.websocket.close()
+            print("WebSocket connection closed")
+
+    def jsonMessage(self, code=0, msg="", data: object = None):
+        """json字符串数据"""
+        jsonData = {"code": code, "msg": msg, "data": data}
+        return json.dumps(jsonData)
+
+
+# 创建全局 SocketManager 实例
+socket_manager = SocketClient(f"ws://{APP_HOST}:{PORT}/ws")

python/sockets/socket_server.py

@@ -2,51 +2,40 @@ import json
 import asyncio
 from models import *
 from .connect_manager import ConnectionManager
-
-def jsonMessage(code=0, msg="", data: object = None):
-    """json字符串数据"""
-    jsonData = {"code": code, "msg": msg, "data": data}
-    return json.dumps(jsonData)
+from .message_handler import *
+import time
+from .socket_client import socket_manager
 manager = ConnectionManager()
 active_connections = set()
 
-
-
 @app.websocket("/ws")
 async def websocket_endpoint(websocket: WebSocket):
     await manager.connect(websocket)
     active_connections.add(websocket)
     try:
+        await socket_manager.connect()
         async def handler_messages():
             while True:
-                data = await websocket.receive_json()
-                await handlerSend(data,websocket)
-        await asyncio.gather( handler_messages(),)
+                byteDats = await websocket.receive()
+                # print("byteDats", byteDats)
+                await handlerSend(manager, json.loads(byteDats), websocket)
+        await asyncio.gather(handler_messages())
     except WebSocketDisconnect:
+        socket_manager.close()
         print("Client disconnected")
     finally:
         active_connections.discard(websocket)
         # if websocket:
         #     await websocket.close()
+
+
 @app.on_event("shutdown")
 async def shutdown_event():
     print("Shutting down...")
+    socket_manager.close()
     # 清理操作
     for connection in list(active_connections):
         try:
             await connection.close()
         except Exception as e:
             print(f"Error closing connection: {e}")
-async def handlerSend(receiveData,websocket):
-    # 处理消息发送逻辑
-    jsonType = receiveData.get("type")
-    match jsonType:
-        case 'ping':
-            '''发送心跳'''
-            data = jsonMessage("pong")
-            await manager.send_personal_message(data,websocket)
-        case 'pong':
-            '''发送心跳'''
-            pass
-
-

python/utils/hlm_http_request.py

@@ -0,0 +1,22 @@
+import requests
+from middleware import UnicornException
+def forward_request(target_url, params=None, method="GET", headers=None):
+    """
+    转发HTTP请求到目标URL
+
+    :param target_url: 目标接口的URL
+    :param params: 请求参数（字典格式）
+    :param method: 请求方法（GET或POST）
+    :param headers: 请求头（字典格式）
+    :return: 目标接口的响应
+    """
+    try:
+        if method.upper() == "GET":
+            response = requests.get(target_url, params=params, headers=headers)
+        elif method.upper() == "POST":
+            response = requests.post(target_url, data=params, headers=headers)
+        else:
+            raise UnicornException("仅支持GET和POST方法")
+        return response.json()  # 假设目标接口返回JSON格式的数据
+    except requests.RequestException as e:
+        raise UnicornException(e)