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): super().__init__(BaseClass) # 1 表示等待中，2表示没有等待 self.wait_state = 2 self.data = action_data self.capture_one = DigiCam() captrure_folder_path = self.capture_one.getCaptureFolderPath() self.watch_dog = FileEventHandler() self.watch_dog.start_observer(captrure_folder_path) 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 return default 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: # print("p_item 错误：{}".format(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) 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: # print("拍照前延时：{}".format(self.delay_time)) time.sleep(self.delay_time) if self.is_photograph: print("拍照==>", time.time()) # print("photograph==================") self.mcu.to_deal_device(device_name="buzzer", times=1) # 用于临时拍照计数 is_af = True if self.af_times > 0 else False 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.action_type, self.action_name)) if True: 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