from config.settings import app_set_config from hardware.relay import RelayController import time import threading from datetime import datetime class VisualCallback: # 类变量，用于存储实例引用，实现单例检测 _instance = None _lock = threading.Lock() def __new__(cls): """检测实例是否存在，实现单例模式""" with cls._lock: if cls._instance is None: cls._instance = super().__new__(cls) return cls._instance def __init__(self): """初始化视觉回调处理器""" # 避免重复初始化 if hasattr(self, '_initialized') and self._initialized: return self.angle_mode = "normal" self.relay_controller = RelayController() self.init_weight = 100 self.mould_finish_weight = 0 self.mould_need_weight = 100 self.finish_count = 0 self.overflow = False # 线程安全的参数传递 self._current_angle = None self._overflow_detected = None self._new_data_available = threading.Event() self._is_processing = threading.Lock() self._stop_event = threading.Event() # 创建并启动单个持续运行的线程 self.callback_thread = threading.Thread( target=self._run_thread_loop, daemon=True ) self.callback_thread.start() self._initialized = True def angle_visual_callback(self, current_angle, overflow_detected): """ 视觉控制主逻辑，供外部推送数据 使用单个持续运行的线程，通过参数设置传递数据 如果线程正在处理数据，则丢弃此次推送 """ print(f"{datetime.now().strftime('%H:%M:%S.%f')[:-3]} 回调:{current_angle}") # 尝试获取处理锁，若失败则说明正在处理，丢弃数据 if not self._is_processing.acquire(blocking=False): print("回调线程仍在执行，丢弃此次推送数据") return try: # 更新参数 self._current_angle = current_angle if overflow_detected is not None: self._overflow_detected = overflow_detected # 通知线程有新数据可用 self._new_data_available.set() finally: # 释放处理锁 self._is_processing.release() def _run_thread_loop(self): """ 线程主循环，持续运行 等待新数据，然后调用处理方法 """ while not self._stop_event.is_set(): # 等待新数据可用 self._new_data_available.wait() # 重置事件 self._new_data_available.clear() # 获取当前参数（使用临时变量避免被其他线程修改） current_angle = self._current_angle overflow_detected = self._overflow_detected # 处理数据 self._process_angle_callback(current_angle, overflow_detected) time.sleep(0.5) def __del__(self): self.relay_controller.close_all() def _process_angle_callback(self, current_angle, overflow_detected): """ 内部方法，实际处理视觉回调逻辑 在异步线程中执行 """ try: # print('current_angle:', current_angle, 'overflow_detected:', overflow_detected) # return # 检测溢出状态 print(f"{self.angle_mode}") self.overflow = overflow_detected in ["大堆料", "小堆料"] if current_angle is None: print("无法获取当前角度，跳过本次调整") return print(f"{datetime.now().strftime('%H:%M:%S.%f')[:-3]} 当前角度: {current_angle:.2f}°") if True: # 状态机控制逻辑 if self.angle_mode == "normal": # 正常模式大于app_set_config.angle_threshold=60度 if self.overflow: self.angle_mode = "reducing" else: # 保持正常开门 # print(f'当前重量:{self.mould_finish_weight:.2f}kg, 目标重量:{self.mould_need_weight:.2f}kg') if self.mould_need_weight > 0: if self.mould_finish_weight / self.mould_need_weight >= 0.8: print(f"完成重量占比{self.mould_finish_weight/self.mould_need_weight:.2f}，半开出砼门") # 半开出砼门 if current_angle > app_set_config.target_angle: # 角度已降至目标范围，关闭出砼门 self.relay_controller.control(self.relay_controller.DOOR_LOWER_OPEN, 'close') self.relay_controller.control(self.relay_controller.DOOR_LOWER_CLOSE, 'open') time.sleep(0.3) self.relay_controller.control(self.relay_controller.DOOR_LOWER_CLOSE, 'close') else: self.relay_controller.control(self.relay_controller.DOOR_LOWER_CLOSE, 'close') self.relay_controller.control(self.relay_controller.DOOR_LOWER_OPEN, 'open') time.sleep(0.32) self.relay_controller.control(self.relay_controller.DOOR_LOWER_OPEN, 'close') else: # 全开砼门 if current_angle > app_set_config.angle_threshold: #self.relay_controller.control(self.relay_controller.DOOR_LOWER_CLOSE,'close') self.relay_controller.control(self.relay_controller.DOOR_LOWER_OPEN, 'close') else: self.relay_controller.control(self.relay_controller.DOOR_LOWER_CLOSE,'close') self.relay_controller.control(self.relay_controller.DOOR_LOWER_OPEN,'open') else: # 全开砼门 if current_angle > app_set_config.angle_threshold: self.relay_controller.control(self.relay_controller.DOOR_LOWER_OPEN, 'close') else: self.relay_controller.control(self.relay_controller.DOOR_LOWER_OPEN, 'open') elif self.angle_mode == "reducing": # 角度减小模式 if self.overflow: if current_angle <= app_set_config.target_angle: # 角度已达到目标范围，仍有堆料，进入维持模式 print(f"角度已降至{current_angle:.2f}°，仍有堆料，进入维持模式") if current_angle <= app_set_config.min_angle: self.relay_controller.control(self.relay_controller.DOOR_LOWER_CLOSE, 'close') self.relay_controller.control(self.relay_controller.DOOR_LOWER_OPEN, 'open') time.sleep(0.1) self.angle_mode = "maintaining" else: print(f"角度大于30，继续关闭") self.relay_controller.control(self.relay_controller.DOOR_LOWER_OPEN, 'close') self.relay_controller.control(self.relay_controller.DOOR_LOWER_CLOSE, 'open') else: # 无堆料，恢复正常模式 print(f"角度已降至{current_angle:.2f}°，无堆料，恢复正常模式") self.angle_mode = "normal" elif self.angle_mode == "maintaining": # 维持模式 - 使用脉冲控制 if not self.overflow: # 堆料已消除，恢复正常模式 print("堆料已消除，恢复正常模式") self.angle_mode = "normal" else: # 继续维持角度控制 print("进入维持模式") # 关门时间 if current_angle > app_set_config.target_angle: # 角度已降至目标范围，关闭出砼门 self.relay_controller.control(self.relay_controller.DOOR_LOWER_OPEN, 'close') self.relay_controller.control(self.relay_controller.DOOR_LOWER_CLOSE, 'open') #time.sleep(0.3) # self.relay_controller.control(self.relay_controller.DOOR_LOWER_CLOSE, 'close') else: self.relay_controller.control(self.relay_controller.DOOR_LOWER_CLOSE, 'close') self.relay_controller.control(self.relay_controller.DOOR_LOWER_OPEN, 'open') #time.sleep(0.32) # self.relay_controller.control(self.relay_controller.DOOR_LOWER_OPEN, 'close') elif self.angle_mode == "recovery": # 恢复模式 - 逐步打开门 if self.overflow: # 又出现堆料，回到角度减小模式 print("恢复过程中又检测到堆料，回到角度减小模式") self.angle_mode = "maintaining" else: # 堆料已消除，恢复正常模式 print("堆料已消除，恢复正常模式") self.relay_controller.control(self.relay_controller.DOOR_LOWER_OPEN, 'open') self.relay_controller.control(self.relay_controller.DOOR_LOWER_CLOSE, 'close') self.angle_mode = "normal" # else: # 浇筑满，关闭下料门 # self.relay_controller.control_lower_close() except Exception as e: print(f"处理视觉回调时发生异常: {e}") @classmethod def instance_exists(cls): """检测实例是否存在""" return cls._instance is not None def shutdown(self): """关闭线程，清理资源""" # 设置停止事件 self._stop_event.set() # 唤醒线程以便它能检测到停止事件 self._new_data_available.set() # 等待线程结束 if self.callback_thread.is_alive(): self.callback_thread.join(timeout=1.0) # 创建默认实例 visual_callback_instance = VisualCallback() # 兼容层，保持原来的函数调用方式可用 def angle_visual_callback(current_angle, overflow_detected): """ 兼容旧版本的函数调用方式 将调用转发到默认实例的angle_visual_callback方法 """ visual_callback_instance.angle_visual_callback(current_angle, overflow_detected)