CameraMachine/python/mcu/LineControl.py

import time

# from import_qt_mode import *
import asyncio
from .SerialIns import SerialIns
from .BaseClass import BaseClass
from utils.SingletonType import SingletonType
from databases import SqlQuery, PhotoRecord, DeviceConfig, CRUD, insert_photo_records
from .capture.module_digicam import DigiCam
from .capture.module_watch_dog import FileEventHandler
from sockets.connect_manager import ConnectionManager

class LineControl(BaseClass):
    # sign_data = Signal(dict)

    def __init__(self, websocket_manager: ConnectionManager):
        super().__init__(websocket_manager)
        self.serial_ins = None
        self.connect_state = False
        self.is_running = False
        self.msg_type = "blue_tooth"
        self.port_name = None
        self.goods_art_no = None
        # 0 闲置;1进行中;2已完成;
        self.photo_take_state = 0

    async def to_connect_com(self, port_name):
        self.port_name = port_name
        self.close_connect()
        await asyncio.sleep(0.5)
        try:
            # 原值为9600
            self.serial_ins = SerialIns(port_name=port_name, baud=115200)
            if self.serial_ins.serial_handle:
                self.connect_state = True
                message = {
                    "_type": "show_info",
                    "plugins_mode": "remote_control",
                    "data": {"msg": "有线遥控器 打开串口成功", "port_name": port_name},
                }
                self.sendSocketMessage(
                    code=0,
                    msg="有线遥控器 打开串口成功",
                    data=message,
                    device_status=2,
                )
                # 循环监听消息
                await self.run()
                return True
            else:
                message = {
                    "_type": "show_info",
                    "plugins_mode": "remote_control",
                    "data": {"msg": "有线遥控器 打开串口失败"},
                }
                self.sendSocketMessage(
                    code=1,
                    msg="有线遥控器 打开串口失败",
                    data=message,
                    device_status=-1,
                )
                self.serial_ins = None
                self.connect_state = False
                return False
        except:
            message = {
                "_type": "show_info",
                "plugins_mode": "remote_control",
                "data": {"msg": "有线遥控器 打开串口失败"},
            }
            self.sendSocketMessage(
                code=1,
                msg="有线遥控器 打开串口失败",
                data=message,
                device_status=-1,
            )
            self.serial_ins = None
            self.connect_state = False
            return False

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

    def __del__(self):
        self.close_connect()

    async def to_connect_linecontrol(self):
        """连接有线控制器"""
        print("to_connect_linecontrol 连接有线控制器,连接状态", self.connect_state)
        if self.connect_state:
            return
        message = {
            "_type": "show_info",
            "plugins_mode": "remote_control",
            "data": "有线遥控器 打开成功",
        }
        self.close_connect()
        print(message)
        self.sendSocketMessage(
            code=0, msg="有线遥控器 打开蓝牙成功", data=message, device_status=2
        )
        self.connect_state = True
        self.is_running = True
        await self.run()

    def handlerAction(self, button_value):
        """处理拍照动作按键[左 右]"""
        control_program = "执行左脚程序" if button_value == 1 else "执行右脚程序"
        match button_value:
            case 1:
                control_program = "执行左脚程序"
            case 2:
                control_program = "执行右脚程序"
        if self.goods_art_no == None or self.goods_art_no == "":
            input_data = {
                "data": {
                    "action": control_program,
                    "goods_art_no": "",
                },
                "type": "run_mcu",
            }
            self.msg_type = "blue_tooth_scan"
            self.sendSocketMessage(
                code=0,
                msg=f"准备执行[{control_program}]",
                data=input_data,
                device_status=2,
            )
            self.msg_type = "blue_tooth"
            return
        self.photo_take_state = 1
        input_data = {
            "data": {
                "action": control_program,
                "goods_art_no": self.goods_art_no,
            },
            "type": "run_mcu",
        }
        self.msg_type = "blue_tooth_scan"
        self.sendSocketMessage(
            code=0,
            msg=f"准备执行[{control_program}]",
            data=input_data,
            device_status=2,
        )
        self.goods_art_no = None
        self.msg_type = "blue_tooth"
        self.photo_take_state = 2

    async def analysis_received_data(self,):
        await asyncio.sleep(0.01)
        if not self.connect_state:
            return
        receive_data = self.serial_ins.read_cmd(out_time=1, check=0x6B)
        if receive_data is False:
            self.connect_state = False
            return False

        if not receive_data:
            return
        else:
            print("有线控制器receive_data", receive_data)
            # 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}
            # )
            message = {"_type": 0, "plugins_mode": "remote_control", "data": bar_code}
            print("有线控制器 扫码数据1", message)
            self.goods_art_no = bar_code
            self.sendSocketMessage(code=0, msg="", data=message, device_status=2)
            return
        if receive_data[0] == 9:
            button_value = receive_data[1]
            data = {"button_value": button_value}
            message = {"_type": 9, "plugins_mode": "remote_control", "data": data}
            if button_value in [1, 2]:
                # 扫描货号
                if self.photo_take_state != 0:
                    self.sendSocketMessage(1, "前置拍照未完成,请稍后", device_status=-1)
                    return
                print("收到货号信息", self.goods_art_no)
                self.handlerAction(button_value)
                self.photo_take_state = 0
            if button_value in [3]:
                # 处理遥控器单拍
                self.msg_type = "handler_take_picture"
                # 0 闲置;1进行中;2已完成;
                _data = {"type": self.msg_type, "data": None}
                self.sendSocketMessage(0, "处理单拍消息", data=_data, device_status=-1)
                self.msg_type = "blue_tooth"
            if button_value in [9]:
                # 处理停止
                self.msg_type = "stop_action"
                # 0 闲置;1进行中;2已完成;
                _data = {"type": self.msg_type, "data": None}
                self.sendSocketMessage(
                    0, "停止执行组合动作", data=_data, device_status=-1
                )
                self.msg_type = "blue_tooth"
            self.sendSocketMessage(code=0, msg="", data=message, device_status=2)
            return
        pass

    async def run(self):
        self.is_running = True
        while True:
            await asyncio.sleep(0.06)
            if not self.connect_state:
                message = {
                    "_type": "show_info",
                    "plugins_mode": "remote_control",
                    "data": {"msg": "有线遥控器 未连接"},
                }
                self.sendSocketMessage(
                    code=1,
                    msg="有线遥控器 未连接",
                    data=message,
                    device_status=-1,
                )
                break
            await self.analysis_received_data()

        self.is_running = False
        if not self.connect_state:
            message = {
                "_type": "show_info",
                "plugins_mode": "remote_control",
                "data": {"msg": "有线遥控器 未连接"},
            }
            self.sendSocketMessage(
                code=1,
                msg="有线遥控器 未连接",
                data=message,
                device_status=-1,
            )