CameraMachine/python/mcu/DeviceControl.py

import asyncio
import serial.tools.list_ports
import time, json
from .SerialIns import SerialIns
from utils.SingletonType import SingletonType
from .BaseClass import BaseClass
from sockets import ConnectionManager

# mcu命令
class DeviceControl(BaseClass,metaclass=SingletonType):

    def __init__(self, websocket_manager: ConnectionManager):
        super().__init__(websocket_manager)
        self.msg_type = "mcu"
        self.serial_ins = None
        self.connected_ports_dict = {}  # 已连接的ports
        self.p_list = []
        self.temp_ports_dict = {}
        self.is_running = False
        self.connect_state = False

    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)
        data = {
            "_type": "remove_port",
            "plugins_mode": "auto_select_com",
            "data": {"port_name": port_name},
        }
        self.sendSocketMessage(1, "串口被移除", 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.sendSocketMessage(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.sendSocketMessage(
                1,
                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("超时错误：无法在规定时间内重置输入缓冲区。")
            self.sendSocketMessage(
                1,
                msg="超时错误：无法在规定时间内重置输入缓冲区。",
                data=None,
            )
            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.sendSocketMessage(
                    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.sendSocketMessage(
                msg="MCU连接成功",
                data=message_data,
            )
        else:
            print("串口无法识别")
            # 走其他途径处理

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

    def to_connect_com(self, port_name):
        # 关闭串口
        print("to_connect_com", port_name)
        self.close_connect()
        time.sleep(0.3)
        self.connect_state = False
        try:
            self.serial_ins = SerialIns(port_name=port_name, baud=115200, timeout=0.1)
            if not self.serial_ins.serial_handle:
                message_data= {
                        "_type": "show_info",
                        "plugins_mode": "mcu",
                        "data": "MCU 打开串口失败",
                    }
                self.sendSocketMessage(
                    msg="MCU 打开串口失败",
                    data=message_data,
                )
                self.serial_ins = None
                self.connect_state = False
                return False

        except:
            message_data={
                "_type": "show_info",
                "plugins_mode": "mcu",
                "data": "MCU 打开串口失败",
            }
            self.sendSocketMessage(
                msg="MCU 打开串口失败",
                data=message_data,
            )
            self.serial_ins = None
            self.connect_state = False
            return False

        message_data={"_type": "show_info", "plugins_mode": "mcu", "data": "MCU 开始连接"}
        self.sendSocketMessage(
            msg="MCU 开始连接",
            data=message_data,
        )
        # =======================发送连接请求=================================
        cmd = 90
        data = [cmd, 1]
        print("405  发送 连接请求  -----------------------------------------")
        print(self.serial_ins)
        # self.serial_ins.clearn_flush()
        self.serial_ins.write_cmd(data)

        # 延迟接收数据
        time.sleep(0.3)
        receive_data = self.serial_ins.read_cmd(out_time=1)
        if receive_data:
            print(
                "409  receive_data--90:{}".format(self.change_hex_to_int(receive_data))
            )
        if receive_data:
            # receive_data[2]=1 表示为MCU设备编号
            if receive_data[0] == 90 and receive_data[2] == 1:
                connect_flag = receive_data[1]
                # 是否有初始化
                try:
                    mcu_has_been_set = receive_data[
                        6
                    ]  # 设备是否有初始化 ，1 表示已初始化
                except:
                    mcu_has_been_set = 99  # 未知状态
                print("MCU初始化信息{}".format(mcu_has_been_set))
                message_data = {
                    "_type": "show_info",
                    "plugins_mode": "mcu",
                    "data": "MCU 已连接",
                }
                self.sendSocketMessage(
                    msg="MCU 已连接",
                    data=message_data,
                )
                self.connect_state = True
                self.is_running = True
                print("MCU 已连接")
                self.port_name = port_name
                return

        print("MCU 连接失败")
        message_data = {
            "_type": "show_info",
            "plugins_mode": "mcu",
            "data": "MCU 连接失败",
        }
        self.sendSocketMessage(
            msg="MCU 连接失败",
            data=message_data,
        )
        self.close_connect()

    def close_connect(self):
        self.port_name = ""
        if self.serial_ins:
            self.serial_ins.close_serial_port()
            self.is_running = False
            self.connect_state = False
            self.connected_ports_dict = {}  # 已连接的ports
            self.p_list = []
            self.temp_ports_dict = {}
            print("关闭MCU")


async def checkMcuConnection(device_ctrl: DeviceControl):
    if device_ctrl.is_running == True:
        message = {
            "_type": "select_port_name",
            "plugins_mode": "auto_select_com",
            "data": device_ctrl.temp_ports_dict,
        }
        device_ctrl.sendSocketMessage(code=0, msg="MCU连接成功", data=message)
        return
    """实时检测串口是否连接"""
    while True:
        await asyncio.sleep(0.5)
        ports_dict = device_ctrl.scan_serial_port()
        device_ctrl.temp_ports_dict = ports_dict
        if not ports_dict:
            # 全部清空 移除所有串口
            if device_ctrl.p_list:
                _p = device_ctrl.p_list.pop()
                device_ctrl.remove_port(_p)
            print("串口未连接,请检查")
            device_ctrl.sendSocketMessage(code=1, msg="串口未连接,请检查")
            continue
        if ports_dict:
            for index, _i in enumerate(device_ctrl.p_list):
                if _i not in ports_dict:
                    _p = device_ctrl.p_list.pop(index)
                    device_ctrl.remove_port(_p)
            for _port_name, _port_value in ports_dict.items():
                if _port_name not in device_ctrl.p_list:
                    try:
                        device_ctrl.p_list.append(_port_name)
                        device_ctrl.add_port_by_linkage(_port_name)
                    except BaseException as e:
                        print("串口不存在{} {}".format(_port_name, e))