CameraMachine/python/mcu/ProgramItem.py

import asyncio
import json
import os

from .BaseClass import BaseClass
import settings
import time
from .capture.module_digicam import DigiCam
from .capture.module_watch_dog import FileEventHandler


class ProgramItem(BaseClass):
    # program_sign = Signal(dict)
    # program_refresh_photo_list_sign = Signal()

    def __init__(
        self,
        websocket_manager,
        action_data: any,
        mcu,
        goods_art_no: str = None,
        image_index: int = -1,
        smart_shooter=None,
    ):
        super().__init__(BaseClass)
        # 1 表示等待中，2表示没有等待
        self.wait_state = 2
        self.msg_type = "mcu"
        print("action_data====>", action_data)
        self.data = action_data
        print("ProgramItem smart_shooter", smart_shooter)
        self.smart_shooter = smart_shooter
        if smart_shooter == None:
            self.capture_one = DigiCam()
            captrure_folder_path = self.capture_one.getCaptureFolderPath()
            self.watch_dog = FileEventHandler()
            self.watch_dog.goods_art_no = goods_art_no
            self.watch_dog.image_index = image_index
            self.watch_dog.mcu = mcu
            self.watch_dog.start_observer(captrure_folder_path)
        else:
            print("使用smart_shooter")
        print("21 =========ProgramItem=======action_data=====")
        print(action_data)
        self.action_id = self.get_value(action_data, "id")
        self.mode_type = self.get_value(action_data, "mode_type")
        # self.action_type = self.get_value(action_data, "execution_type", "程序1")
        self.action_name = self.get_value(action_data, "action_name", "")
        self.is_wait = self.get_value(action_data, "is_wait", False)
        self.is_need_confirm = self.get_value(action_data, "is_need_confirm", False)
        self.image_index = self.get_value(action_data, "picture_index", 99)
        self.camera_height = self.get_value(action_data, "camera_height", 0)
        self.camera_angle = float(self.get_value(action_data, "camera_angle", 0.0))
        self.af_times = self.get_value(action_data, "number_focus", 0)
        self.shoe_overturn = self.get_value(action_data, "shoe_upturn", False)
        self.is_photograph = self.get_value(action_data, "take_picture", True)
        self.turntable_position = float(
            self.get_value(action_data, "turntable_position", 0.0)
        )
        self.turntable_angle = float(
            self.get_value(action_data, "turntable_angle", 0.0)
        )
        self.delay_time = self.get_value(action_data, "pre_delay", 0.0)
        self.after_delay_time = self.get_value(action_data, "after_delay", 0.0)
        self.is_led = self.get_value(action_data, "led_switch", False)

        self.last_photograph_time = None  # 最近一次拍照时间
        self.goods_art_no = goods_art_no  # 货号

        self.set_other()
        self.error_info_text = ""  # 错误提示信息

        # self.setParent(parent)
        self.mcu = mcu

        # if is_show:
        #     self.parent = parent
        #     self.setupUi(self)
        #     self.init()
        #     self.show()

    def set_other(self):
        if self.mode_type == "其他配置":
            self.turntable_position = None
            self.turntable_angle = None
            self.delay_time = 0
            self.after_delay_time = 0
            self.is_led = False

    def get_value(self, data, key, default=None):
        if key not in data:
            data[key] = default
            print("取值模式==>获取到key:{}的值为:{}".format(key, data[key]))
            return default
        print("默认值模式==>获取到key:{}的值为:{}".format(key, data[key]))
        return data[key]

    def reset(self):
        self.set_state(state_value=0)
        self.error_info_text = ""  # 错误提示信息

    def init(self):
        self.icon_dict = {
            0: "resources/other_icon/0weikaishi.png",
            1: "resources/other_icon/1jinxingzhong.png",
            2: "resources/other_icon/2yijieshu.png",
            99: "resources/other_icon/11yichang.png",
        }
        if self.is_wait:
            msg = "{}--等待".format(self.action_name)
            self.sendSocketMessage(msg=msg, device_status=0)
        else:
            msg = "{}".format(self.action_name)
            self.sendSocketMessage(msg=msg, device_status=2)
        tips_text = "程序：{}，对焦：{}".format(1, 1)
        self.sendSocketMessage(msg=tips_text, device_status=2)
        self.set_state(state_value=0)

    # def next_step_clicked(self, *args, **kwargs):
    #     self.windows.event.set()

    def set_state(self, state_value):
        self.state = state_value
        # icon = QPixmap(self.icon_dict[self.state])
        # icon = icon.scaled(
        #     self.ui_icon.size(), Qt.KeepAspectRatio, Qt.SmoothTransformation
        # )
        # self.ui_icon.setPixmap(icon)
        # if self.state == 0:
        #     self.ui_retry.setText("")
        #     self.ui_retry.setEnabled(False)
        # if self.state == 1:
        #     self.ui_retry.setText("")
        #     self.ui_retry.setEnabled(False)
        # if self.state == 2:
        #     self.ui_retry.setText("单步")
        #     self.ui_retry.setEnabled(True)
        # if self.state == 99:
        #     self.ui_retry.setText("重试")
        #     self.ui_retry.setEnabled(True)
        #     self.ui_action_name.setToolTip(self.error_info_text)

    def check_mcu_move_is_stop(self, re_check=False):
        self.error_info_text = ""
        # 发送基础数据信息
        # self.mcu.to_get_mcu_base_info()
        _s = time.time()
        last_num_1 = self.mcu.last_mcu_info_data["num"]
        self.mcu.cleanAllReceiveData()
        while 1:
            if self.mcu.action_state != 1:
                # 外部终止，停止运行
                return False
            cr_time = time.time()
            print(cr_time - _s, cr_time, _s)
            if cr_time - _s > 8:
                self.error_info_text = "MCU检测运动未停止，自动退出"
                self.set_state(state_value=99)  # 标记异常
                print("MCU检测运动未停止，自动退出")
                self.sendSocketMessage(msg=self.error_info_text, device_status=-1)
                return False
                # return True
            # 存在时间间隙，导致误认为所有设备已完成运动
            if self.mcu.mcu_move_state == 2:
                return True
            else:
                self.mcu.to_get_mcu_base_info()
                self.mcu.send_all_cmd()
                time.sleep(0.5)
                self.mcu.get_basic_info_mcu()
                # return True

            time.sleep(0.1)
            # self.mcu.to_get_mcu_base_info()

    def run(self, total_len=5, *args):
        if total_len == 1:
            self.mode_type = "其他配置"
            self.set_other()

        print("1{}  - is run".format(self.action_name))
        self.set_state(state_value=1)
        if settings.IS_TEST:
            self.do_run()
        else:
            try:
                self.do_run()
            except BaseException as e:
                self.sendSocketMessage(
                    msg="p_item 错误：{}".format(e), device_status=-1
                )
                self.set_state(state_value=99)

        self.set_state(state_value=2)
        return True

    def run_only_mcu(self, *args):
        # ============连接MCU 处理步进电机与舵机等
        if settings.IS_MCU:
            if self.shoe_overturn:
                self.mcu.to_deal_device(device_name="overturn_steering")
                time.sleep(0.1)
            if self.camera_height is not None:
                self.mcu.to_device_move(
                    device_name="camera_high_motor", value=self.camera_height
                )
                time.sleep(0.1)
            if self.camera_angle is not None:
                self.mcu.to_device_move(
                    device_name="camera_steering", value=self.camera_angle
                )
                time.sleep(0.1)
            if self.turntable_position is not None:
                self.mcu.to_device_move(
                    device_name="turntable_position_motor",
                    value=self.turntable_position,
                )
                time.sleep(0.1)
            if self.turntable_angle is not None:
                self.mcu.to_device_move(
                    device_name="turntable_steering", value=self.turntable_angle
                )
                time.sleep(0.1)
            self.mcu.send_all_cmd()

    def do_run(self, *args):
        # if not self.goods_art_no:  # and self.action_name != "初始化位置"
        #     return False
        start_time = time.time()
        # ============连接MCU 处理步进电机与舵机等
        if settings.IS_MCU:
            if self.mode_type != "其他配置" and self.check_mcu_move_is_stop() is False:
                # MCU运动是否有停止检查，设定超时时间
                return
            print(
                "{} 检查停止时间1：{}".format(
                    self.action_name, time.time() - start_time
                )
            )
            if self.is_led:
                self.mcu.to_deal_device(device_name="laser_position", value=1)
            else:
                self.mcu.to_deal_device(device_name="laser_position", value=0)

            if self.shoe_overturn:
                self.mcu.to_deal_device(device_name="overturn_steering")
                # time.sleep(0.1)
            if self.camera_height is not None:
                self.mcu.to_device_move(
                    device_name="camera_high_motor", value=self.camera_height
                )
                # time.sleep(0.1)
            if self.camera_angle is not None:
                self.mcu.to_device_move(
                    device_name="camera_steering", value=self.camera_angle
                )
                # time.sleep(0.1)

            if self.turntable_position is not None:
                self.mcu.to_device_move(
                    device_name="turntable_position_motor",
                    value=self.turntable_position,
                )
                # time.sleep(0.1)

            if self.turntable_angle is not None:
                self.mcu.to_device_move(
                    device_name="turntable_steering", value=self.turntable_angle
                )
                # time.sleep(0.1)

            # MCU运动是否有停止检查，设定超时时间
            self.mcu.send_all_cmd()
            if self.mode_type != "其他配置":
                time.sleep(1.2)
                print("二次检查")
                if self.check_mcu_move_is_stop(re_check=True) is False:
                    print("MCU检测运动未停止，自动退出,   提前退出")
                    return

        if self.delay_time:
            time.sleep(self.delay_time)
        if self.is_photograph:
            # print("photograph==================")
            self.mcu.to_deal_device(device_name="buzzer", times=1)
            # 用于临时拍照计数
            is_af = True if self.af_times > 0 else False
            loop = asyncio.get_event_loop()
            if self.smart_shooter != None:
                if is_af:
                    # 指定自动对焦
                    loop.create_task(
                        self.smart_shooter.CameraAutofocus(),
                        name="CameraAutofocus",
                    )
                # 拍照
                print("smart shooter 拍照")
                loop.create_task(
                    self.smart_shooter.CameraShooter(msg_type="run_mcu"),
                    name="CameraShooter",
                )
            else:
                # 指定自动对焦
                self.capture_one.photograph(is_af=is_af)
            self.last_photograph_time = time.time()  # 记录最近一次拍照时间
        # print("{} 拍照时间：{}".format(self.action_name, time.time() - start_time))
        print("{}-{}已完成".format(self.mode_type, self.action_name))

        if self.after_delay_time:
            print("拍照后延时：{}".format(self.after_delay_time))
            time.sleep(self.after_delay_time)
        return True

    def rephotograph_one_pic(self, *args):
        """
        1、获取最近一张照片
        2、判断拍照时间距离当前节点的最近拍照时间是否小于3秒
        3、删除该照片
        4、重新触发进行拍照，并更新最近拍照时间
        """
        if settings.RUNNING_MODE == "普通模式":
            return

        print("-----100-2--self", self)
        # 需要删除最近一张照片
        # if self.last_photograph_time is not None:
        #     record = self.windows.image_process_data.photo_todo_list
        #     if record:
        #         last_record = record[-1]
        #         if self.goods_art_no == last_record["goods_art_no"]:
        #             if last_record["is_photo"]:
        #                 photo_create_time_formate = last_record[
        #                     "photo_create_time_formate"
        #                 ]
        #                 image_path = last_record["image_path"]
        #                 print(
        #                     "photo_create_time_formate,self.last_photograph_time",
        #                     photo_create_time_formate,
        #                     self.last_photograph_time,
        #                 )
        #                 # 如果照片的时间大于触发时间，且小于3秒，则认定是当前触发
        #                 if (
        #                     self.last_photograph_time
        #                     < photo_create_time_formate
        #                     < self.last_photograph_time
        #                     + settings.PHOTO_TRANSFER_TIME_INTERVAL
        #                 ):
        #                     if os.path.exists(image_path):
        #                         os.remove(image_path)
        #                         self.windows.add_goods_images_count(
        #                             self.goods_art_no, flag=False
        #                         )
        #                         if (
        #                             self.windows.image_process_data.goods_art_no_times_record
        #                         ):
        #                             goods_art_no = self.windows.image_process_data.goods_art_no_times_record[
        #                                 0
        #                             ][
        #                                 "goods_art_no"
        #                             ]
        #                             if goods_art_no == self.goods_art_no:
        #                                 self.windows.image_process_data.goods_art_no_times_record[
        #                                     0
        #                                 ][
        #                                     "number"
        #                                 ] -= 1
        #                         self.program_refresh_photo_list_sign.emit()

        # if self.is_photograph:
        #     # 重新拍照
        #     self.last_photograph_time = time.time()  # 记录最近一次拍照时间
        #     print("last_photograph_time:", self.last_photograph_time)
        #     print("photograph==================")
        #     self.mcu.to_deal_device(device_name="buzzer", times=1)
        #     # 用于临时拍照计数
        #     self.windows.add_goods_images_count(self.goods_art_no)
        #     if self.af_times > 0:
        #         self.windows.capture_one.photograph(is_af=True)
        #     else:
        #         self.windows.capture_one.photograph(is_af=False)

    def do_retry(self, *args, **kwargs):
        """
        重试操作，注意事项：
        1、需要根据当前系统的角度位置，重新进行返回并拍摄
        :param args:
        :param kwargs:
        :return:
        """
        # self.program_sign.emit({})
        # print("do_retry")
        self.reset()
        self.mcu.action_state = 1
        self.run()
        self.mcu.action_state = 2

    def get_photo_node_name_and_sound(self, photo_take_time):
        data = {
            "action_name": self.action_name,
            "flag": False,
            "is_sound_play": False,
        }
        if self.state == 0:
            return data
        if self.last_photograph_time is None:
            return data
        # if (
        #     self.last_photograph_time
        #     < photo_take_time
        #     < self.last_photograph_time + settings.PHOTO_TRANSFER_TIME_INTERVAL
        # ):
        #     # 认定为当节点拍摄
        #     # 如果为待用户确认则播放声音
        #     data["flag"] = True
        #     if settings.RUNNING_MODE == "待用户确认模式":
        #         _f = False
        #         if settings.RUNNING_MODE_DETAIL == "所有节点待确认":
        #             _f = True
        #         else:
        #             print("self.is_need_confirm", self.is_need_confirm)
        #             if self.is_need_confirm:
        #                 _f = True
        #             else:
        #                 _f = False
        #         if _f:
        #             data["is_sound_play"] = True
        #             print("========is_sound_play===========")
        #             self.windows.playsound.tips_type = "photo_confirm"
        #             self.windows.playsound.start()
        #     return data
        return data