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
import settings,os
import logging
logger = logging.getLogger(__name__)
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,
                )
                # 循环监听消息
                loop = asyncio.get_event_loop()
                loop.create_task(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.saveScanData(self.goods_art_no)
            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.saveScanData(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
    def saveScanData(self, scan_data):
        """保存扫码数据"""
        if scan_data == None or scan_data == "":
            return
        scan_path = settings.SCAN_DIR
        if scan_path ==None or scan_path == "":
            return
        try:
            # 验证文件名的有效性
            # 替换可能导致问题的字符
            invalid_chars = '<>:"/\\|?*'
            safe_scan_data = scan_data
            for char in invalid_chars:
                safe_scan_data = safe_scan_data.replace(char, '_')
            # 确保文件名不为空
            if not safe_scan_data.strip():
                logger.info(f"扫码文件,无效的文件名")
                return
            
            # 获取目录下所有txt文件
            txt_files = [f for f in os.listdir(scan_path) if f.lower().endswith('.txt')]
            txt_filename = safe_scan_data + ".txt"
            new_txt_path = os.path.join(scan_path, txt_filename)
            # 如果只有一个txt文件，则重命名并覆盖内容
            if len(txt_files) == 1:
                try:
                    old_txt_path = os.path.join(scan_path, txt_files[0])
                    os.rename(old_txt_path, new_txt_path)
                    
                    # 覆盖文件内容
                    with open(new_txt_path, 'w', encoding='utf-8') as f:
                        f.write(scan_data)
                        
                except Exception as e:
                    logger.info(f"扫码文件,重命名或写入文件失败: {e}")
                    
                    # 如果重命名失败，创建新文件
                    with open(new_txt_path, 'w', encoding='utf-8') as f:
                        f.write(scan_data)
            
            # 如果有多个txt文件，删除它们并创建新文件
            elif len(txt_files) > 1:
                for txt_file in txt_files:
                    try:
                        txt_file_path = os.path.join(scan_path, txt_file)
                        os.remove(txt_file_path)
                    except Exception as e:
                        logger.info(f"扫码文件,删除文件失败 {txt_file}: {e}")
                
                # 创建新的txt文件
                with open(new_txt_path, 'w', encoding='utf-8') as f:
                    f.write(scan_data)
            
            # 如果没有txt文件，直接创建新文件
            else:
                with open(new_txt_path, 'w', encoding='utf-8') as f:
                    f.write(scan_data)
                
        except Exception as e:
            logger.info(f"扫码文件,保存文件时发生错误: {e}")
    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
        self.close_connect()