CameraMachine/python/mcu/DeviceControl.py

import asyncio
import serial.tools.list_ports
import time
from .SerialIns import SerialIns
from sockets import socket_manager
# mcu命令
class DeviceControl:
    def __init__(self):
        self.serial_ins = None
        self.connected_ports_dict = {}  # 已连接的ports
    def scan_serial_port(self) -> dict:
        # 获取所有可用串口列表
        ports_dict = {}
        ports = serial.tools.list_ports.comports()
        # 遍历所有端口并打印信息
        for port in ports:
            if "CH340" in port.description:
                ports_dict[port.name] = {
                    "name": port.name,
                    "device": port.device,
                    "description": port.description,
                    "hwid": port.hwid,
                    "manufacturer": port.manufacturer,
                    "product": port.product,
                    "serial_number": port.serial_number,
                }
        if len(ports_dict) <= 0:
            return {}
        return ports_dict

    def remove_port(self, port_name):
        '''移除串口'''
        print("remove", port_name)
        self.sign_data.emit(
            {
                "_type": "remove_port",
                "plugins_mode": "auto_select_com",
                "data": {"port_name": port_name},
            }
        )
    def sendMsg(self,code=0,msg="",data=None):
        asyncio.run(socket_manager.send_message(code=code, msg=msg, data=data))
        # print(code,msg,data)
    def add_port_by_linkage(self, port_name):
        # port_value :串口基础信息
        # todo 根据prot_value 信息自动进行连接
        print("add", port_name)
        # 对没有连接的设备进行尝试连接
        message_data = {
            "_type": "show_info",
            "plugins_mode": "auto_select_com",
            "data": {"text": "开始识别接口：{}".format(port_name)},
        }

        self.sendMsg(msg="开始识别接口：{}".format(port_name), data=message_data)
        time.sleep(1)
        """
        步骤：
        1、进行临时连接，并发送命令，成功后，自动连接对应设备
        """
        try:
            # 尝试使用115200波特率链接
            serial_handle =  serial.Serial(
                port=port_name, baudrate=115200, timeout=0.5
            )
        except:
            message_data = {
                "_type": "show_info",
                "plugins_mode": "auto_select_com",
                "data": {"text": "串口：{} 被占用,或无法识别".format(port_name)},
            }
            self.sendMsg(
                msg="串口：{} 被占用,或无法识别".format(port_name).format(port_name),
                data=message_data,
            )
            print("串口：{} 被占用".format(port_name))
            return

        time.sleep(2)
        print("开始发送命令")
        data = [90, 1]
        try:
            serial_handle.flushInput()  # 尝试重置输入缓冲区
        except serial.SerialTimeoutException:
            print("超时错误：无法在规定时间内重置输入缓冲区。")
            serial_handle.close()
            return
        print("尝试写入数据")

        buf = bytearray(b"")
        buf.extend([0x55, 0x55, (0xFF & len(data))])
        buf.extend(data)
        buf.extend([0xFF & ~sum(data)])
        try:
            self.receive_data = b""
            serial_handle.write(buf)
        except serial.SerialTimeoutException:
            print("写入数据错误")
            serial_handle.close()
            return

        time.sleep(0.3)
        print("尝试接收命令")
        receive_data = self.read_cmd(serial_handle)
        device_id = 0

        if receive_data:
            print("receive_data", receive_data)
            if receive_data[0] == 90:
                connect_flag = receive_data[1]
                device_id = receive_data[2]

        print("关闭串口：{}".format(port_name))
        serial_handle.close()

        if device_id > 0:
            if device_id == 1:
                self.to_connect_com(port_name)
                message_data = {
                    "_type": "show_info",
                    "plugins_mode": "auto_select_com",
                    "data": {"text": "MCU开始连接"},
                }
                self.sendMsg(
                    msg="MCU开始连接",
                    data=message_data,
                )
                self.connected_ports_dict[port_name] = "MCU"
            message_data = {
                "_type": "select_port_name",
                "plugins_mode": "auto_select_com",
                "data": {
                    "device_name": "mcu" if device_id == 1 else "remote_control",
                    "port_name": port_name,
                },
            }
            self.sendMsg(
                msg="MCU连接成功",
                data=message_data,
            )
        else:
            print("串口无法识别")
            # 走其他途径处理

        # 检查当前MCU链接是否正常
        # 正常跳过；记录为其他列表
        # 不正常进行尝试连接
        # 连接不上，记录为其他列表

    def read_cmd(self, serial_handle, check=None):
        n = 0
        while 1:
            try:
                read_d = serial_handle.read_all()  # 读取接收到的数据
                self.receive_data += read_d
            except BaseException as e:
                print("171串口接收报错", 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 to_connect_com(self, port_name):
        self.close_connect()
        time.sleep(0.5)
        try:
            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.sendMsg(msg="条码识别 打开串口成功")
                self.connect_state = True
                self.start()
                return True
            else:
                self.sendMsg(code=1, msg="条码识别 打开串口失败")
                self.serial_ins = None
                self.connect_state = False
        except:
            self.sendMsg(code=1, msg="条码识别 打开串口失败")
            # print("条码识别 打开串口失败")
            self.serial_ins = None
            self.connect_state = False
            return False

    def close_connect(self):
        self.port_name = ""
        if self.serial_ins:
            self.serial_ins.close_serial_port()
            self.connect_state = False


if __name__ == "__main__":
    device_control = DeviceControl()
    p_list = []
    temp_ports_dict = {}
    while 1:
        time.sleep(1)
        ports_dict = device_control.scan_serial_port()
        temp_ports_dict = ports_dict

        if not ports_dict:
            # 全部清空 移除所有串口
            if p_list:
                _p = p_list.pop()
                device_control.remove_port(_p)
            continue

        if ports_dict:
            # print(plist)
            for index, _i in enumerate(p_list):
                if _i not in ports_dict:
                    _p = p_list.pop(index)
                    device_control.remove_port(_p)

            for _port_name, _port_value in ports_dict.items():
                if _port_name not in p_list:
                    # try:
                        p_list.append(_port_name)
                        device_control.add_port_by_linkage(_port_name)
                    # except BaseException as e:
                    #     print(e.__traceback__.tb_frame.f_globals["__file__"])   # 发生异常所在的文件
                    #     print(e.__traceback__.tb_lineno)                        # 发生异常所在的行数
                    #     print("串口不存在{} {}".format(_port_name, e))

                # threading.Thread(target=self.add_port, args=(_port_name, _port_value)).start()
                # self.add_port(_p)