import os import sys import time import re import logging import configparser import threading import queue import subprocess from concurrent.futures import ThreadPoolExecutor from collections import defaultdict, deque from PySide6.QtWidgets import (QMainWindow, QWidget, QVBoxLayout, QHBoxLayout, QPushButton, QLabel, QGridLayout, QScrollArea, QToolButton, QFrame, QApplication, QCheckBox, QMessageBox, QSlider, QDialog, QDialogButtonBox) from PySide6.QtCore import Qt, QTimer, QSize, Signal from PySide6.QtGui import QIcon, QPixmap, QResizeEvent, QKeyEvent, QCloseEvent from .models import * from .rtsp_player import RTSPPlayer from .rfid_client import RFIDClient from .modbus_client import ModbusClient from .mqtt_client import MQTTClient from .auto_restart import AutoRestartScheduler logger = logging.getLogger(__name__) class MainWindow(QMainWindow): # 防抖阈值（连续相同状态次数） STATE_THRESHOLD = 3 # 放包判断位移阈值（像素） RELEASE_DELTA_THRESHOLD = 5 # 轨迹阶段判断阈值（像素） RELEASE_TRAJ_Y = 8 SEPARATE_TRAJ_DIST = 15 ENGAGE_DIST = 40 # 命中放包/分离后屏蔽语音的顺延秒数 SUPPRESS_SECONDS = 8 # 轨迹窗口：保留最近 2 秒 / 最多 30 帧 TRAJ_WINDOW_SECS = 2.0 TRAJ_MAX_FRAMES = 30 # MQTT 消息超时（秒）后清除残留画框 MQTT_CLEAR_TIMEOUT = 1 # 标定采样时长（秒），期间对每帧有效配对取平均 CALIB_DURATION_SECS = 1.0 # 自动标定参数见 [AI] 配置：auto_calib_ready_frames / auto_calib_duration_secs / auto_calib_min_samples APPTITLE = "HookVision" # 跨线程信号 rfid_tag_signal = Signal(object) rfid_state_signal = Signal(object) modbus_data_signal = Signal(object) mqtt_message_signal = Signal(str, object) voice_signal = Signal(str) camera_signal_signal = Signal(dict) restart_trigger_signal = Signal() def __init__(self, config: Config, config_path: str = ""): super().__init__() self.config = config self._config_path = config_path self.buckets = config.buckets self.active_bucket: BucketModel = None self.switch_mode = SwitchMode.AUTO self._subscribed_topics: set = set() self._current_ai_module_id: Optional[int] = None self.left_player: RTSPPlayer = None self.right_player: RTSPPlayer = None self.rfid_client = None self.modbus_client = None self.mqtt_client = None self._last_rfid_tag: Optional[str] = None self.video_aspect_ratio = 16 / 9 self._ratio_16_9 = config.video.ratio_16_9 self.left_label = None self.right_label = None # 挂钩状态相关 self.hook_state = HookState.IDLE self.prev_state = HookState.IDLE self.state_counter = 0 # 存储左右画面最近检测对象 self.left_objects = [] self.right_objects = [] # 前一帧 hook/lug 中心位置（用于轨迹判断） self.prev_left_hook = None self.prev_right_hook = None self.prev_left_lug = None self.prev_right_lug = None # 配对轨迹（每侧：{t, hook_cx, hook_cy, lug_cx, lug_cy, dist}），用于放包/分离检测与画线 self._pair_traj: dict = {"left": [], "right": []} # 放包/分离期间屏蔽 挂包中/挂包成功 语音的截止时间 self._voice_suppress_until: float = 0.0 # 当前叠加显示的阶段（放包/分离/空） self._trace_phase: str = "" # 未标定侧告警去重（side → 已告警） self._calib_warned: set = set() self.debug_trace = getattr(config.ai, 'debug_trace', True) # MQTT 消息超时后是否已清除画框（防重复清空） self._overlay_cleared = False # 标定采集状态：点击标定后连续采样约 CALIB_DURATION_SECS 秒取平均 self._calib_active = False self._calib_until: float = 0.0 self._calib_samples: dict = {} # side → [sum_xr, sum_yr, count] self._calib_badge_labels: dict = {} # bucket.id → QLabel # 自动标定（窗口内 ready 帧数达标）：未标定侧自动标定，已标定侧不覆盖 self._auto_calib_state: dict = { # side → 窗口计数状态 side: {"ready_times": deque(), "samples": deque()} for side in ("left", "right") } self._side_ready: dict = {"left": False, "right": False} # 语音引擎（在工作线程内初始化） self.last_voice_state = None self._last_hooking_voice_time = 0.0 self.voice_enabled = True self._voice_volume = 100 self.ai_enabled = True self._last_left_draw: list = [] self._last_right_draw: list = [] self._voice_queue = queue.Queue(maxsize=32) self._voice_thread = threading.Thread(target=self._voice_worker, daemon=True) self._voice_thread.start() self._voices_dir = self._get_voices_dir() self._use_tts = self.config.ai.tts_enabled # RTSP 退避重连 self._rtsp_retry_left: Optional[QTimer] = None self._rtsp_retry_right: Optional[QTimer] = None self._rtsp_retry_delay_left: int = 1 self._rtsp_retry_delay_right: int = 1 self._last_mqtt_connect_attempt: float = 0.0 self._last_mqtt_process: float = 0.0 # 摄像头信号监控 self._camera_signal: dict = {} # ip → 延迟(ms), -1=不可达 self._bucket_signal_labels: dict = {} # bucket.id → QLabel self._last_signal_check: float = 0.0 self._signal_check_thread: threading.Thread = None self._offline_reported: set = set() # 已播报离线的 IP # 连接跨线程信号到主线程槽 self.rfid_tag_signal.connect(self._on_rfid_tag_impl) self.rfid_state_signal.connect(self._on_rfid_state_impl) self.modbus_data_signal.connect(self._on_modbus_data_impl) self.mqtt_message_signal.connect(self._on_mqtt_message_impl) self.voice_signal.connect(self._speak_impl) self.camera_signal_signal.connect(self._on_camera_signal) self.restart_trigger_signal.connect(self._on_auto_restart_impl) self.setWindowTitle(config.app_name) logo_path = os.path.join(os.path.dirname(__file__), "..", "assets", "logo.png") if os.path.exists(logo_path): self.setWindowIcon(QIcon(logo_path)) screen = QApplication.primaryScreen() screen_geometry = screen.availableGeometry() screen_width = screen_geometry.width() screen_height = screen_geometry.height() window_width = int(screen_width * 0.9) window_height = int(screen_height * 0.85) x = (screen_width - window_width) // 2 y = (screen_height - window_height) // 2 self.setGeometry(x, y, window_width, window_height) self.setMinimumSize(800, 600) self.setup_ui() self.showFullScreen() self.update_video_size() self.init_hardware() self._started = True self.timer = QTimer() self.timer.timeout.connect(self.check_idle) self.timer.start(1000) self.auto_restart_scheduler = AutoRestartScheduler(config.auto_restart) self.auto_restart_scheduler.on_restart_triggered = self._on_auto_restart_triggered self.auto_restart_scheduler.start() def setup_ui(self): central = QWidget() self.setCentralWidget(central) main_layout = QVBoxLayout(central) main_layout.setContentsMargins(10, 10, 10, 10) main_layout.setSpacing(10) # 标题栏 title_widget = QWidget() title_layout = QHBoxLayout(title_widget) title_layout.setContentsMargins(0, 0, 0, 0) logo_path = os.path.join(os.path.dirname(__file__), "..", "assets", "logo.png") if os.path.exists(logo_path): logo_label = QLabel() logo_pixmap = QPixmap(logo_path).scaled(28, 28, Qt.KeepAspectRatio, Qt.SmoothTransformation) logo_label.setPixmap(logo_pixmap) logo_label.setFixedSize(28, 28) title_layout.addWidget(logo_label) self.title_label = QLabel(self.config.app_name) self.title_label.setStyleSheet("font-size: 22px; font-weight: bold; padding: 5px;") title_layout.addWidget(self.title_label) title_layout.addStretch() btn_style = """ QPushButton { background-color: transparent; border: none; font-size: 18px; padding: 6px 12px; border-radius: 5px; } QPushButton:hover { background-color: rgba(255, 255, 255, 0.1); } """ self.min_btn = QPushButton("—") self.min_btn.setStyleSheet(btn_style) self.min_btn.clicked.connect(self.showMinimized) self.fullscreen_btn = QPushButton("⛶") self.fullscreen_btn.setStyleSheet(btn_style) self.fullscreen_btn.setToolTip("全屏 / 还原") self.fullscreen_btn.clicked.connect(self._toggle_fullscreen) self.close_btn = QPushButton("✕") self.close_btn.setStyleSheet(btn_style) self.close_btn.clicked.connect(self.close) title_layout.addWidget(self.min_btn) title_layout.addWidget(self.fullscreen_btn) title_layout.addWidget(self.close_btn) main_layout.addWidget(title_widget) # 工位区域 bucket_group = QFrame() bucket_group.setStyleSheet("QFrame { border-radius: 8px; padding: 8px; }") bucket_layout = QVBoxLayout(bucket_group) scroll_area = QScrollArea() scroll_area.setWidgetResizable(True) scroll_area.setFrameShape(QFrame.NoFrame) scroll_area.setStyleSheet("QScrollArea { background: transparent; border: none; }") self.bucket_container = QWidget() self.bucket_grid = QGridLayout(self.bucket_container) self.bucket_grid.setSpacing(15) self.bucket_grid.setContentsMargins(10, 10, 10, 10) self.bucket_grid.setAlignment(Qt.AlignLeft | Qt.AlignTop) # 按 code_x 分组纵向排列 groups = defaultdict(list) for bucket in self.buckets: groups[bucket.code_x].append(bucket) sorted_keys = sorted(groups.keys()) self.bucket_buttons = [] col = 0 for key in sorted_keys: buckets_in_group = groups[key] for row, bucket in enumerate(buckets_in_group): cell = QWidget() cell_layout = QVBoxLayout(cell) cell_layout.setContentsMargins(0, 0, 0, 0) cell_layout.setSpacing(2) btn = QToolButton() btn.setFixedSize(100, 100) btn.setToolButtonStyle(Qt.ToolButtonTextUnderIcon) btn.setText(bucket.description) btn.setIcon(self.get_bucket_icon(bucket.is_active)) btn.setIconSize(QSize(64, 64)) btn.setStyleSheet(""" QToolButton { border-radius: 12px; font-size: 14px; font-weight: bold; } """) btn.clicked.connect(lambda checked, b=bucket: self.on_bucket_clicked(b)) cell_layout.addWidget(btn, 0, Qt.AlignHCenter) # 底部状态条三等分：左摄像头信号 | 右摄像头信号 | 标定状态（与图标同宽） bar = QWidget() bar.setFixedSize(100, 8) bar_layout = QHBoxLayout(bar) bar_layout.setContentsMargins(0, 0, 0, 0) bar_layout.setSpacing(2) left_bar = QFrame() right_bar = QFrame() calib_bar = QFrame() for fb in (left_bar, right_bar, calib_bar): fb.setFixedHeight(8) fb.setStyleSheet("background-color: #dcdfe6; border-radius: 2px;") bar_layout.addWidget(left_bar, 1) bar_layout.addWidget(right_bar, 1) bar_layout.addWidget(calib_bar, 1) cell_layout.addWidget(bar, 0, Qt.AlignHCenter) self._bucket_signal_labels[bucket.id] = (left_bar, right_bar) self._calib_badge_labels[bucket.id] = calib_bar self.bucket_grid.addWidget(cell, row, col) self.bucket_buttons.append((btn, bucket)) col += 1 self.bucket_container.setLayout(self.bucket_grid) scroll_area.setWidget(self.bucket_container) bucket_layout.addWidget(scroll_area) main_layout.addWidget(bucket_group) # 视频区域（仅保留边框） video_container = QWidget() video_layout = QHBoxLayout(video_container) video_layout.setSpacing(15) video_layout.setContentsMargins(0, 0, 0, 0) self.left_label = QLabel("左侧画面") self.left_label.setScaledContents(False) self.left_label.setAlignment(Qt.AlignCenter) self.left_label.setStyleSheet(""" QLabel { border: 2px solid #dcdfe6; border-radius: 8px; background-color: black; } """) self.right_label = QLabel("右侧画面") self.right_label.setScaledContents(False) self.right_label.setAlignment(Qt.AlignCenter) self.right_label.setStyleSheet(""" QLabel { border: 2px solid #dcdfe6; border-radius: 8px; background-color: black; } """) video_layout.addWidget(self.left_label) video_layout.addWidget(self.right_label) main_layout.addWidget(video_container, 1) # 状态栏 status_widget = QWidget() status_layout = QHBoxLayout(status_widget) status_layout.setContentsMargins(5, 5, 15, 5) self.module_status = QLabel("通讯状态: 初始化") self.module_status.setStyleSheet("color: #606266; font-weight: bold;") self.rfid_tag_label = QLabel("当前卡号: 无") self.rfid_tag_label.setStyleSheet("color: #409eff; font-weight: bold; margin-left: 20px;") self.hook_status_label = QLabel("挂钩状态: 空闲") self.hook_status_label.setStyleSheet("color: #ffa500; font-weight: bold; margin-left: 20px;") self.voice_checkbox = QCheckBox("语音播报") self.voice_checkbox.setChecked(True) self.voice_checkbox.setStyleSheet("color: #606266; font-weight: bold;") self.voice_checkbox.stateChanged.connect(self.on_voice_toggle) self.tts_checkbox = QCheckBox("TTS语音") self.tts_checkbox.setChecked(self._use_tts) self.tts_checkbox.setStyleSheet("color: #606266; font-weight: bold;") self.tts_checkbox.toggled.connect(self._on_tts_toggle) self.volume_btn = QPushButton(f"音量 {self._voice_volume}%") self.volume_btn.setFixedWidth(80) self.volume_btn.setStyleSheet(""" QPushButton { color: #606266; font-weight: bold; font-size: 12px; border: 1px solid #dcdfe6; border-radius: 4px; padding: 2px 6px; background: transparent; } QPushButton:hover { border-color: #409eff; color: #409eff; } """) self.volume_btn.clicked.connect(self._show_volume_popup) self.ai_checkbox = QCheckBox("AI启用") self.ai_checkbox.setChecked(True) self.ai_checkbox.setStyleSheet("color: #606266; font-weight: bold;") self.ai_checkbox.stateChanged.connect(self.on_ai_toggle) self.auto_switch_btn = QCheckBox("自动模式") self.auto_switch_btn.setChecked(True) self.auto_switch_btn.setStyleSheet("color: #606266; font-weight: bold;") self.auto_switch_btn.clicked.connect(self.toggle_switch_mode) self.autostart_checkbox = QCheckBox("开机自启") self.autostart_checkbox.setChecked(self._read_autostart_flag()) self.autostart_checkbox.setStyleSheet("color: #606266; font-weight: bold; margin-left: 10px;") self.autostart_checkbox.stateChanged.connect(self._on_autostart_toggle) status_layout.addWidget(self.module_status) status_layout.addWidget(self.rfid_tag_label) status_layout.addWidget(self.hook_status_label) status_layout.addStretch() status_layout.addWidget(self.voice_checkbox) status_layout.addWidget(self.tts_checkbox) status_layout.addWidget(self.volume_btn) status_layout.addWidget(self.ai_checkbox) status_layout.addWidget(self.auto_switch_btn) status_layout.addWidget(self.autostart_checkbox) self.ratio_btn = QPushButton("16:9" if self._ratio_16_9 else "4:3") self.ratio_btn.setStyleSheet("background:transparent; border:none; color:#606266; font-weight:bold; padding:4px 8px;") self.ratio_btn.setToolTip("切换画面比例 16:9 / 4:3") self.ratio_btn.clicked.connect(self._toggle_aspect_ratio) status_layout.addWidget(self.ratio_btn) self.calib_btn = QPushButton("标定") self.calib_btn.setFixedWidth(72) self.calib_btn.setStyleSheet(""" QPushButton { color: #606266; font-weight: bold; font-size: 12px; border: 1px solid #dcdfe6; border-radius: 4px; padding: 2px 6px; background: transparent; } QPushButton:hover { border-color: #409eff; color: #409eff; } """) self.calib_btn.setToolTip("标定可以起吊的 hook/lug 偏移（当前工位，左右各采 1 秒取平均）") self.calib_btn.clicked.connect(self.on_calibrate_clicked) status_layout.addWidget(self.calib_btn) self.reset_calib_btn = QPushButton("重置标定") self.reset_calib_btn.setFixedWidth(80) self.reset_calib_btn.setStyleSheet(""" QPushButton { color: #f56c6c; font-weight: bold; font-size: 12px; border: 1px solid #f56c6c; border-radius: 4px; padding: 2px 6px; background: transparent; } QPushButton:hover { background-color: rgba(245, 108, 108, 0.1); } """) self.reset_calib_btn.setToolTip("清除当前工位的标定数据，清除后该工位会在识别到 ready 时自动重新标定") self.reset_calib_btn.clicked.connect(self.on_reset_calibrate_clicked) status_layout.addWidget(self.reset_calib_btn) main_layout.addWidget(status_widget) self.update_video_size() # ---------- 事件与资源管理 ---------- def closeEvent(self, event: QCloseEvent): self._cancel_rtsp_retry("left") self._cancel_rtsp_retry("right") if self.rfid_client: self.rfid_client.disconnect() if self.modbus_client: self.modbus_client.disconnect() if self.mqtt_client: self.mqtt_client.disconnect() if self.left_player: self.left_player.stop() if self.right_player: self.right_player.stop() if hasattr(self, 'auto_restart_scheduler'): self.auto_restart_scheduler.stop() event.accept() # ---------- 自动重启 ---------- def _on_auto_restart_triggered(self): self.restart_trigger_signal.emit() def _on_auto_restart_impl(self): ar = self.config.auto_restart target_text = "重启电脑" if ar.target.value == "device" else "重启软件" dialog = QDialog(self) dialog.setWindowTitle("自动重启") dialog.setWindowFlag(Qt.WindowStaysOnTopHint, True) dialog.setModal(True) dialog.setMinimumWidth(300) layout = QVBoxLayout(dialog) label = QLabel(f"{ar.countdown} 秒后将自动{target_text}", dialog) label.setAlignment(Qt.AlignCenter) label.setStyleSheet("font-size: 14px; padding: 10px;") layout.addWidget(label) btn_layout = QHBoxLayout() btn_now = QPushButton("立即重启") btn_cancel = QPushButton("取消") btn_layout.addWidget(btn_now) btn_layout.addWidget(btn_cancel) layout.addLayout(btn_layout) remaining = ar.countdown def update_label(): nonlocal remaining remaining -= 1 if remaining <= 0: timer.stop() dialog.accept() else: label.setText(f"{remaining} 秒后将自动{target_text}") timer = QTimer(dialog) timer.timeout.connect(update_label) timer.start(1000) btn_now.clicked.connect(dialog.accept) btn_cancel.clicked.connect(dialog.reject) result = dialog.exec() timer.stop() if result == QDialog.Accepted: self.auto_restart_scheduler.perform_restart() else: logger.info("自动重启已取消") # ---------- 语音控制 ---------- def on_voice_toggle(self, state): self.voice_enabled = (state == Qt.Checked) def _on_tts_toggle(self): self._use_tts = self.tts_checkbox.isChecked() self.config.ai.tts_enabled = self._use_tts try: self._save_config() except Exception as e: logger.warning(f"保存 TTS 设置失败: {e}") def _on_volume_changed(self, value: int): self._voice_volume = value self.volume_btn.setText(f"音量 {value}%") def _preview_volume(self, vol: int): self._voice_volume = vol self.volume_btn.setText(f"音量 {vol}%") if not hasattr(self, '_preview_timer'): self._preview_timer = QTimer() self._preview_timer.setSingleShot(True) self._preview_timer.timeout.connect(lambda: self._speak("挂包中")) self._preview_timer.start(300) def _show_volume_popup(self): from PySide6.QtWidgets import QDialog, QVBoxLayout from PySide6.QtCore import Qt dlg = QDialog(self) dlg.setWindowFlags(Qt.Popup | Qt.FramelessWindowHint) dlg.setStyleSheet("background: #2c2c2c; border: 1px solid #555; border-radius: 6px;") layout = QVBoxLayout(dlg) layout.setContentsMargins(12, 8, 12, 8) slider = QSlider(Qt.Horizontal) slider.setRange(0, 100) slider.setValue(self._voice_volume) slider.setFixedWidth(160) slider.setStyleSheet(""" QSlider::groove:horizontal { height: 4px; background: #555; border-radius: 2px; } QSlider::handle:horizontal { width: 14px; margin: -5px 0; border-radius: 7px; background: #409eff; } QSlider::sub-page:horizontal { background: #409eff; border-radius: 2px; } """) slider.valueChanged.connect(self._preview_volume) layout.addWidget(slider) dlg.adjustSize() btn_global = self.volume_btn.mapToGlobal(self.volume_btn.rect().topLeft()) dlg.move(btn_global.x(), btn_global.y() - dlg.height() - 4) dlg.exec() def on_ai_toggle(self, state): enabled = (state == Qt.Checked) if enabled == self.ai_enabled: return self.ai_enabled = enabled if enabled and not self.mqtt_client: self.mqtt_client = MQTTClient(self.on_mqtt_message) self._connect_current_ai() elif not enabled and self.mqtt_client: self.mqtt_client.disconnect() self.mqtt_client = None self._subscribed_topics.clear() self._current_ai_module_id = None if not enabled: self.hook_state = HookState.IDLE self.prev_state = HookState.IDLE self.state_counter = 0 self.left_objects = [] self.right_objects = [] self.prev_left_hook = None self.prev_right_hook = None self.prev_left_lug = None self.prev_right_lug = None self._pair_traj = {"left": [], "right": []} self._voice_suppress_until = 0.0 self._trace_phase = "" self.update_ui_for_state(HookState.IDLE) self._last_left_draw = [] self._last_right_draw = [] if self.left_player: self.left_player.set_ai_objects([]) if self.right_player: self.right_player.set_ai_objects([]) else: if self.left_player and self._last_left_draw: self.left_player.set_ai_objects(self._last_left_draw) if self.right_player and self._last_right_draw: self.right_player.set_ai_objects(self._last_right_draw) self._refresh_calib_ui() def _speak(self, msg: str): self.voice_signal.emit(msg) def _speak_impl(self, msg: str): try: self._voice_queue.put_nowait(msg) except queue.Full: pass @staticmethod def _get_voices_dir() -> str: if getattr(sys, 'frozen', False): base = sys._MEIPASS else: base = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) return os.path.join(base, "assets", "voices") def _voice_msg_to_path(self, msg: str) -> str: mapping = {"挂包中": "hooking", "挂包成功": "ready", "放包中": "releasing", "读卡器离线": "offline_reader", "左侧摄像头离线": "offline_cam_left", "右侧摄像头离线": "offline_cam_right"} if msg in mapping: return os.path.join(self._voices_dir, f"{mapping[msg]}.wav") m = re.search(r'到达(\d+)号工位', msg) if m: return os.path.join(self._voices_dir, f"arrive_{m.group(1)}.wav") return "" def _voice_worker(self): speaker = None tts_engine = None if sys.platform == "win32": try: import pythoncom pythoncom.CoInitialize() except Exception: pass try: import win32com.client as _win32com speaker = _win32com.Dispatch("SAPI.SpVoice") speaker.Rate = 2 except Exception: logging.warning("SAPI 不可用，仅可使用 WAV 语音") else: try: import pyttsx3 tts_engine = pyttsx3.init() tts_engine.setProperty('rate', 150) except Exception: logging.warning("pyttsx3 不可用，将回退到 WAV 播放") while True: msg = self._voice_queue.get() if msg is None: break try: if self._use_tts and speaker is not None: speaker.Volume = self._voice_volume speaker.Speak(msg) elif self._use_tts and tts_engine is not None: tts_engine.setProperty('volume', self._voice_volume / 100.0) tts_engine.say(msg) tts_engine.runAndWait() else: self._play_wav(msg) except Exception as e: logging.warning(f"语音播报失败: {e}") self._voice_queue.task_done() def _play_wav(self, msg: str): path = self._voice_msg_to_path(msg) if not path: return if sys.platform == "win32": self._play_wav_windows(path) else: self._play_wav_linux(path) def _play_wav_windows(self, path: str): import struct, wave, io, winsound try: with wave.open(path, 'rb') as wf: params = wf.getparams() raw = bytearray(wf.readframes(wf.getnframes())) gain = self._voice_volume / 100.0 if params.sampwidth == 2: fmt = '<' + 'h' * (len(raw) // 2) samples = list(struct.unpack_from(fmt, bytes(raw))) samples = [max(-32768, min(32767, int(s * gain))) for s in samples] raw = bytearray(struct.pack(fmt, *samples)) elif params.sampwidth == 1: samples = list(raw) samples = [max(0, min(255, int((s - 128) * gain) + 128)) for s in samples] raw = bytearray(samples) buf = io.BytesIO() with wave.open(buf, 'wb') as wf: wf.setparams(params) wf.writeframes(bytes(raw)) winsound.PlaySound(buf.getvalue(), winsound.SND_MEMORY) except Exception: winsound.PlaySound(path, winsound.SND_FILENAME | winsound.SND_ASYNC) def _play_wav_linux(self, path: str): import struct, wave, io, numpy as np try: with wave.open(path, 'rb') as wf: params = wf.getparams() raw = bytearray(wf.readframes(wf.getnframes())) gain = self._voice_volume / 100.0 if params.sampwidth == 2: fmt = '<' + 'h' * (len(raw) // 2) samples = list(struct.unpack_from(fmt, bytes(raw))) samples = [max(-32768, min(32767, int(s * gain))) for s in samples] raw = bytearray(struct.pack(fmt, *samples)) elif params.sampwidth == 1: samples = list(raw) samples = [max(0, min(255, int((s - 128) * gain) + 128)) for s in samples] raw = bytearray(samples) import sounddevice as sd if params.sampwidth == 2: dtype = np.int16 elif params.sampwidth == 4: dtype = np.int32 else: dtype = np.float32 audio_data = np.frombuffer(bytes(raw), dtype=dtype) sd.play(audio_data, samplerate=params.framerate, blocking=True) except Exception as e: logging.warning(f"sounddevice 播放失败: {e}，回退到 aplay") try: import subprocess subprocess.run(['aplay', '-q', path], timeout=10, check=False) except Exception: pass # ---------- 工位图标 ---------- def get_bucket_icon(self, is_active): icon_dir = os.path.join(os.path.dirname(__file__), "..", "assets") if not os.path.exists(icon_dir): os.makedirs(icon_dir, exist_ok=True) return QIcon() if is_active: icon_path = os.path.join(icon_dir, "bucket_active.svg") else: icon_path = os.path.join(icon_dir, "bucket_normal.svg") if os.path.exists(icon_path): return QIcon(icon_path) else: pixmap = QPixmap(64, 64) pixmap.fill(Qt.green if is_active else Qt.gray) return QIcon(pixmap) def refresh_bucket_icons(self): for btn, bucket in self.bucket_buttons: icon = self.get_bucket_icon(bucket.is_active) btn.setIcon(icon) badge = self._calib_badge_labels.get(bucket.id) if badge is not None: if self._is_bucket_calibrated(bucket): badge.setStyleSheet("background-color: #67c23a; border-radius: 2px;") tip = "标定状态：已标定" for side, side_name in (("left", "左"), ("right", "右")): x = getattr(bucket, f"hook_lug_x_offset_ratio_{side}", None) y = getattr(bucket, f"hook_lug_y_offset_ratio_{side}", None) if x is not None and y is not None: tip += f"\n{side_name}侧 X={x:.3f} Y={y:.3f}" else: badge.setStyleSheet("background-color: #dcdfe6; border-radius: 2px;") tip = "标定状态：未标定" badge.setToolTip(tip) if bucket.is_active: btn.setStyleSheet(""" QToolButton { background-color: #e1f3d8; border: 2px solid #67c23a; border-radius: 12px; font-size: 14px; font-weight: bold; color: #1f2d3d; } """) else: btn.setStyleSheet(""" QToolButton { background-color: transparent; border: 2px solid #d0d7de; border-radius: 12px; font-size: 14px; font-weight: bold; color: #1f2d3d; } QToolButton:hover { background-color: rgba(0, 0, 0, 0.05); border-color: #409eff; } """) # ---------- 视频尺寸自适应 ---------- def update_video_size(self): if not self.left_label or not self.right_label: return container = self.left_label.parentWidget() if not container: return available_width = container.width() - 15 single_width = available_width // 2 target_height = int(single_width / (16/9 if self._ratio_16_9 else 4/3)) if target_height > container.height(): target_height = container.height() self.left_label.setFixedSize(single_width, target_height) self.right_label.setFixedSize(single_width, target_height) def update_players_target_size(self): if self.left_player: self.left_player.set_target_size(self.left_label.width(), self.left_label.height()) if self.right_player: self.right_player.set_target_size(self.right_label.width(), self.right_label.height()) def resizeEvent(self, event: QResizeEvent): self.update_video_size() self.update_players_target_size() super().resizeEvent(event) def keyPressEvent(self, event: QKeyEvent): if event.key() == Qt.Key_Escape and self.isFullScreen(): self.showNormal() elif event.key() == Qt.Key_F11: if self.isFullScreen(): self.showNormal() else: self.showFullScreen() else: super().keyPressEvent(event) def _toggle_fullscreen(self): if self.isFullScreen(): self.showNormal() else: self.showFullScreen() def _toggle_aspect_ratio(self): self._ratio_16_9 = not self._ratio_16_9 self.ratio_btn.setText("16:9" if self._ratio_16_9 else "4:3") self.update_video_size() self.update_players_target_size() self._save_config() # ---------- 加载动画和超时 ---------- def show_loading_on_label(self, label: QLabel): label.setText("加载中...") label.setAlignment(Qt.AlignCenter) label.setStyleSheet(label.styleSheet() + "color: white; font-size: 20px;") def hide_loading_on_label(self, label: QLabel): label.setText("") label.setStyleSheet(""" QLabel { border: 2px solid #dcdfe6; border-radius: 8px; background-color: black; } """) def show_timeout_on_label(self, label: QLabel): label.setText("连接超时") label.setAlignment(Qt.AlignCenter) label.setStyleSheet(label.styleSheet() + "color: red; font-size: 20px;") def on_player_timeout(self, label: QLabel, player_ref, side: str): self.show_timeout_on_label(label) # 只停止当前活动的播放器（防止旧信号误杀新播放器） if side == "left" and self.left_player is player_ref: self.left_player.stop() self.left_player = None elif side == "right" and self.right_player is player_ref: self.right_player.stop() self.right_player = None self._start_rtsp_retry(side) def _cancel_rtsp_retry(self, side: str): timer = self._rtsp_retry_left if side == "left" else self._rtsp_retry_right if timer: timer.stop() timer.deleteLater() if side == "left": self._rtsp_retry_left = None self._rtsp_retry_delay_left = 1 else: self._rtsp_retry_right = None self._rtsp_retry_delay_right = 1 def _start_rtsp_retry(self, side: str): if not self.active_bucket: return cam = next((c for c in self.active_bucket.cameras if c.position == (CameraPosition.LEFT if side == "left" else CameraPosition.RIGHT)), None) if not cam: return if not self._is_camera_online(cam): return label = self.left_label if side == "left" else self.right_label delay = self._rtsp_retry_delay_left if side == "left" else self._rtsp_retry_delay_right label.setText(f"正在重连({delay}s)...") label.setAlignment(Qt.AlignCenter) label.setStyleSheet(label.styleSheet() + "color: orange; font-size: 20px;") timer = QTimer(self) timer.setSingleShot(True) timer.timeout.connect(lambda: self._retry_rtsp(side)) timer.start(delay * 1000) if side == "left": self._rtsp_retry_left = timer else: self._rtsp_retry_right = timer def _retry_rtsp(self, side: str): if not self.active_bucket: return cam = next((c for c in self.active_bucket.cameras if c.position == (CameraPosition.LEFT if side == "left" else CameraPosition.RIGHT)), None) if not cam: return label = self.left_label if side == "left" else self.right_label if not self._is_camera_online(cam): self.show_timeout_on_label(label) self._cancel_rtsp_retry(side) return self.show_loading_on_label(label) w, h = label.width(), label.height() timeout = self.config.video.rtsp_timeout player = RTSPPlayer(cam.rtsp_url, w, h, timeout_sec=timeout) player.frame_ready.connect(label.setPixmap) player.first_frame_received.connect(lambda s=side: self._on_rtsp_retry_success(s)) player.timeout.connect(lambda l=label, p=player, s=side: self.on_player_timeout(l, p, s)) old = self.left_player if side == "left" else self.right_player if old: try: old.frame_ready.disconnect() except TypeError: pass try: old.first_frame_received.disconnect() except TypeError: pass try: old.timeout.disconnect() except TypeError: pass old.stop() if side == "left": self.left_player = player else: self.right_player = player player.start() def _on_rtsp_retry_success(self, side: str): self.hide_loading_on_label(self.left_label if side == "left" else self.right_label) self._cancel_rtsp_retry(side) # 重连成功后恢复 AI 对象标注 if self.active_bucket and self.ai_enabled: objects = self._last_left_draw if side == "left" else self._last_right_draw player = self.left_player if side == "left" else self.right_player if player: player.set_ai_objects(objects) def _is_camera_online(self, cam) -> bool: m = re.search(r'rtsp://(?:[^@]+@)?([^:/]+)', cam.rtsp_url) if not m: return False lat = self._camera_signal.get(m.group(1)) # 尚未检测过该 IP 时，默认认为在线，先尝试连接 if lat is None: return True return lat >= 0 def on_bucket_clicked(self, bucket: BucketModel): if self.switch_mode != SwitchMode.HAND: self.switch_mode = SwitchMode.HAND self.auto_switch_btn.setChecked(False) self.auto_switch_btn.setText("手动模式") self.activate_bucket(bucket) def on_rfid_state_changed(self, online: bool): self.rfid_state_signal.emit(online) def _on_rfid_state_impl(self, online: bool): if online: self.module_status.setText("通讯状态: RFID正常") else: self.module_status.setText("通讯状态: RFID离线") logger.warning("读卡器离线") if self.voice_enabled and getattr(self, '_started', False): self._speak("读卡器离线") # ---------- 硬件初始化 ---------- def init_hardware(self): if self.config.module_type == ModuleType.RFID and self.config.rfid: # 收集所有配置的工位卡号，过滤掉读卡器同时读到的无关卡 valid_codes = {b.code_x for b in self.buckets if b.code_x} self.rfid_client = RFIDClient( self.config.rfid, self.on_rfid_tag, on_state_changed=self.on_rfid_state_changed, tag_offset=18, tag_length=4, hex_output=True, valid_codes=valid_codes ) if self.rfid_client.connect(): self.module_status.setText("通讯状态: RFID正常") else: self.module_status.setText("通讯状态: RFID离线") if self.config.module_type == ModuleType.HD3200 and self.config.modbus: self.modbus_client = ModbusClient(self.config.modbus, self.on_modbus_data) if self.modbus_client.connect(): self.module_status.setText("通讯状态: Modbus正常") else: self.module_status.setText("通讯状态: Modbus离线") if self.config.ai.enabled: self.mqtt_client = MQTTClient(self.on_mqtt_message) self._connect_current_ai() def _connect_current_ai(self, bucket=None): if bucket is None: bucket = self.active_bucket idx = None if bucket and bucket.ai_module_id is not None: idx = bucket.ai_module_id elif self.config.ai_modules: idx = 0 if idx is not None and idx < len(self.config.ai_modules): target = self.config.ai_modules[idx] self.mqtt_client.disconnect() self.mqtt_client.connect(target.broker, target.port) self._current_ai_module_id = idx if bucket: topics = [c.topic for c in bucket.cameras if c.topic] if topics: self.mqtt_client.subscribe(topics) self._subscribed_topics = set(topics) # ---------- 激活工位 ---------- def activate_bucket(self, bucket: BucketModel): if self.active_bucket == bucket: return self._cancel_rtsp_retry("left") self._cancel_rtsp_retry("right") if bucket and self.voice_enabled: self._speak(f"到达{bucket.description}") for player in (self.left_player, self.right_player): if player is None: continue try: player.frame_ready.disconnect() except TypeError: pass try: player.first_frame_received.disconnect() except TypeError: pass try: player.timeout.disconnect() except TypeError: pass player.stop() self.left_player = None self.right_player = None if self.active_bucket: self.active_bucket.is_active = False self.active_bucket = bucket if bucket: bucket.is_active = True self.refresh_bucket_icons() self._refresh_calib_ui() self.show_loading_on_label(self.left_label) self.show_loading_on_label(self.right_label) # 重置挂钩状态 self.hook_state = HookState.IDLE self.prev_state = HookState.IDLE self.state_counter = 0 self.left_objects = [] self.right_objects = [] self.prev_left_hook = None self.prev_right_hook = None self.prev_left_lug = None self.prev_right_lug = None self.last_voice_state = None self._last_hooking_voice_time = 0.0 self._pair_traj = {"left": [], "right": []} self._voice_suppress_until = 0.0 self._trace_phase = "" self._side_ready = {"left": False, "right": False} self._auto_calib_state = { side: {"ready_times": deque(), "samples": deque()} for side in ("left", "right") } self.update_ui_for_state(HookState.IDLE) self._last_left_draw = [] self._last_right_draw = [] # 先切 AI 模块 + 订阅 MQTT topic（非阻塞），再启 RTSP if bucket and self.config.ai.enabled and self.mqtt_client: if bucket.ai_module_id != self._current_ai_module_id: self._unsubscribe_all_topics() self._connect_current_ai(bucket) if bucket: if self.mqtt_client: topics = [c.topic for c in bucket.cameras if c.topic] stale = self._subscribed_topics - set(topics) for t in stale: self.mqtt_client.unsubscribe(t) new_t = [t for t in topics if t not in self._subscribed_topics] if new_t: self.mqtt_client.subscribe(new_t) self._subscribed_topics = set(topics) # 再启 RTSP left_cam = next((c for c in bucket.cameras if c.position == CameraPosition.LEFT), None) right_cam = next((c for c in bucket.cameras if c.position == CameraPosition.RIGHT), None) if left_cam and self._is_camera_online(left_cam): w = self.left_label.width() h = self.left_label.height() timeout = self.config.video.rtsp_timeout player = RTSPPlayer(left_cam.rtsp_url, w, h, timeout_sec=timeout) player.frame_ready.connect(self.left_label.setPixmap) player.first_frame_received.connect(lambda p=player: self.hide_loading_on_label(self.left_label)) player.timeout.connect(lambda p=player: self.on_player_timeout(self.left_label, p, "left")) self.left_player = player player.start() elif left_cam: self.show_timeout_on_label(self.left_label) if self.voice_enabled: self._speak("左侧摄像头离线") if right_cam and self._is_camera_online(right_cam): w = self.right_label.width() h = self.right_label.height() timeout = self.config.video.rtsp_timeout player = RTSPPlayer(right_cam.rtsp_url, w, h, timeout_sec=timeout) player.frame_ready.connect(self.right_label.setPixmap) player.first_frame_received.connect(lambda p=player: self.hide_loading_on_label(self.right_label)) player.timeout.connect(lambda p=player: self.on_player_timeout(self.right_label, p, "right")) self.right_player = player player.start() elif right_cam: self.show_timeout_on_label(self.right_label) if self.voice_enabled: self._speak("右侧摄像头离线") else: self._unsubscribe_all_topics() self.left_label.clear() self.right_label.clear() QApplication.processEvents() self.left_label.clear() self.right_label.clear() def _unsubscribe_all_topics(self): if self.mqtt_client and self._subscribed_topics: for t in self._subscribed_topics: self.mqtt_client.unsubscribe(t) self._subscribed_topics.clear() # ---------- RFID 标签回调 ---------- def on_rfid_tag(self, tag: str): self.rfid_tag_signal.emit(tag) def _on_rfid_tag_impl(self, tag: str): self._last_rfid_tag = tag if tag: self.rfid_tag_label.setText(f"当前卡号: {tag}") else: self.rfid_tag_label.setText("当前卡号: 无") if tag is None: if self.switch_mode == SwitchMode.AUTO and self.active_bucket: self.activate_bucket(None) return for bucket in self.buckets: if bucket.code_x == tag: if self.switch_mode == SwitchMode.AUTO: self.activate_bucket(bucket) break # ---------- Modbus 数据回调 ---------- def on_modbus_data(self, bools): self.modbus_data_signal.emit(bools) def _on_modbus_data_impl(self, bools): for bucket in self.buckets: try: addr_x = int(bucket.code_x) except (ValueError, TypeError): continue if addr_x < 0 or addr_x >= len(bools): continue if bucket.code_y and bucket.code_y.strip(): try: addr_y = int(bucket.code_y) except (ValueError, TypeError): continue if addr_y < 0 or addr_y >= len(bools): continue if bools[addr_x] and bools[addr_y]: if self.switch_mode == SwitchMode.AUTO and self.active_bucket != bucket: self.activate_bucket(bucket) break else: if bools[addr_x]: if self.switch_mode == SwitchMode.AUTO and self.active_bucket != bucket: self.activate_bucket(bucket) break # ---------- MQTT 消息处理 ---------- def parse_ai_objects(self, payload: dict, display_width: int, display_height: int, include_ready: bool = True) -> list: if payload is None or not isinstance(payload, dict): return [] orig_width = payload.get("width", 640) orig_height = payload.get("height", 360) scale_x = display_width / orig_width if orig_width > 0 else 1.0 scale_y = display_height / orig_height if orig_height > 0 else 1.0 objects = [] for obj_str in payload.get("objects", []): parts = obj_str.split('|') if len(parts) >= 6: try: x = int(float(parts[1]) * scale_x) y = int(float(parts[2]) * scale_y) w = int(float(parts[3]) * scale_x) h = int(float(parts[4]) * scale_y) raw_label = parts[5] label = raw_label if 'hoist' in raw_label: label = 'hook' elif 'cup' in raw_label: label = 'lug' if 'hook' in label: color = (0, 0, 255) elif 'lug' in label: color = (255, 0, 0) elif 'ready' in label: if not include_ready: continue color = (0, 255, 255) else: color = (128, 128, 128) objects.append({ "track_id": parts[0], "x": x, "y": y, "w": w, "h": h, "label": label, "raw_label": raw_label, "color": color }) except (ValueError, IndexError): continue return objects def on_mqtt_message(self, topic: str, payload: dict): self.mqtt_message_signal.emit(topic, payload) def _on_mqtt_message_impl(self, topic: str, payload: dict): if payload is None or not isinstance(payload, dict): return if not self.active_bucket: return if not self.ai_enabled: return now = time.time() if now - self._last_mqtt_process < 0.03: return self._last_mqtt_process = now self._overlay_cleared = False for cam in self.active_bucket.cameras: if cam.topic == topic: self.update_video_size() if cam.position == CameraPosition.LEFT: display_width = self.left_label.width() display_height = self.left_label.height() else: display_width = self.right_label.width() display_height = self.right_label.height() if display_width == 0 or display_height == 0: display_width = 640 display_height = 360 all_objects = self.parse_ai_objects(payload, display_width, display_height, include_ready=True) # ready 标签仅用于自动标定触发，不进检测对象/不参与状态判定 side = "left" if cam.position == CameraPosition.LEFT else "right" self._side_ready[side] = any('ready' in o['label'].lower() for o in all_objects) objects = [o for o in all_objects if 'ready' not in o['label'].lower()] if cam.position == CameraPosition.LEFT: self.left_objects = objects else: self.right_objects = objects self.evaluate_hook_state() if self._calib_active: self._sample_calibration() self._update_auto_calibration(side, self._side_ready[side]) draw_objects = [] for obj in all_objects: if 'ready' in obj['label'].lower(): continue obj['alpha'] = 0.4 draw_objects.append(obj) draw_objects.extend(self._trace_objects(side)) status = self._trace_status_text() if status: color = (0, 0, 255) if self._trace_phase else (0, 200, 255) draw_objects.append({"text": status, "pos": (10, 20), "anchor": "tr", "color": color}) if cam.position == CameraPosition.LEFT: self._last_left_draw = draw_objects if self.left_player: self.left_player.set_ai_objects(draw_objects) else: self._last_right_draw = draw_objects if self.right_player: self.right_player.set_ai_objects(draw_objects) break # ---------- 挂钩状态评估 ---------- def _nearest_pair(self, hooks, lugs): """在 hook/lug 列表中选中心距最近的配对""" if not hooks or not lugs: return None best_pair = None best_dist = float('inf') for h in hooks: hx = h['x'] + h['w'] / 2 hy = h['y'] + h['h'] / 2 for l in lugs: lx = l['x'] + l['w'] / 2 ly = l['y'] + l['h'] / 2 dist = (hx - lx) ** 2 + (hy - ly) ** 2 if dist < best_dist: best_dist = dist best_pair = (h, l) return best_pair def _best_intersecting_pair(self, side: str): """该侧最近且框相交的 hook/lug 配对；无则返回 None""" if side == "left": objects = self.left_objects else: objects = self.right_objects hooks = [o for o in objects if 'hook' in o['label'].lower()] lugs = [o for o in objects if 'lug' in o['label'].lower()] pairs = [] for h in hooks: hx = h['x'] + h['w'] / 2 hy = h['y'] + h['h'] / 2 for l in lugs: lx = l['x'] + l['w'] / 2 ly = l['y'] + l['h'] / 2 pairs.append(((hx - lx) ** 2 + (hy - ly) ** 2, h, l)) pairs.sort(key=lambda t: t[0]) for _, h, l in pairs: if self.rects_intersect(h, l): return (h, l) return None def _compute_offset_ratios(self, side: str): """该侧最近相交配对的归一化偏移 (xr, yr)；无有效配对返回 None""" pair = self._best_intersecting_pair(side) if not pair: return None h, l = pair if self._lug_touches_top(l): return None lw, lh = l['w'], l['h'] if lw <= 0 or lh <= 0: return None xr = (h['x'] + h['w'] / 2 - (l['x'] + l['w'] / 2)) / lw yr = ((h['y'] + h['h']) - (l['y'] + l['h'])) / lh return (xr, yr) # ---------- 标定（可以起吊偏移） ---------- @staticmethod def _is_bucket_calibrated(bucket) -> bool: for side in ("left", "right"): x = getattr(bucket, f"hook_lug_x_offset_ratio_{side}", None) y = getattr(bucket, f"hook_lug_y_offset_ratio_{side}", None) if x is not None and y is not None: return True return False def on_calibrate_clicked(self): if self._calib_active: return bucket = self.active_bucket if not bucket: QMessageBox.information(self, "标定", "请先选择一个工位") return if not self.ai_enabled: QMessageBox.information(self, "标定", "请先启用 AI") return if self._is_bucket_calibrated(bucket): ret = QMessageBox.question( self, "标定", f"{bucket.description} 已标定，要重新标定吗？", QMessageBox.Yes | QMessageBox.No | QMessageBox.Cancel, QMessageBox.No, ) if ret == QMessageBox.Cancel: return if ret == QMessageBox.No: self._reset_calibration() return self._start_calibration() def _start_calibration(self): self._calib_active = True self._calib_until = time.time() + self.CALIB_DURATION_SECS self._calib_samples = {} self._calib_warned.clear() self.calib_btn.setEnabled(False) self.calib_btn.setText("标定中...") def _sample_calibration(self): now = time.time() if now >= self._calib_until: self._finish_calibration() return for side in ("left", "right"): ratios = self._compute_offset_ratios(side) if ratios is None: continue buf = self._calib_samples.setdefault(side, [0.0, 0.0, 0]) buf[0] += ratios[0] buf[1] += ratios[1] buf[2] += 1 def _finish_calibration(self): self._calib_active = False if self.calib_btn: self.calib_btn.setEnabled(True) bucket = self.active_bucket if not bucket: self._refresh_calib_ui() return sides = {"left": None, "right": None} for side in ("left", "right"): buf = self._calib_samples.get(side) if buf and buf[2] > 0: xr = buf[0] / buf[2] yr = buf[1] / buf[2] setattr(bucket, f"hook_lug_x_offset_ratio_{side}", xr) setattr(bucket, f"hook_lug_y_offset_ratio_{side}", yr) sides[side] = (xr, yr) self._save_config() self.refresh_bucket_icons() self._refresh_calib_ui() lines = [f"{bucket.description} 标定完成"] for side in ("left", "right"): v = sides[side] if v is None: lines.append(f" {side}侧: 未采到有效配对（未标定）") else: lines.append(f" {side}侧: x={v[0]:.3f} y={v[1]:.3f}") QMessageBox.information(self, "标定", "\n".join(lines)) def _reset_calibration(self): bucket = self.active_bucket if not bucket: return for side in ("left", "right"): setattr(bucket, f"hook_lug_x_offset_ratio_{side}", None) setattr(bucket, f"hook_lug_y_offset_ratio_{side}", None) self._save_config() self.refresh_bucket_icons() self._refresh_calib_ui() # ---------- 自动标定（连续 ready） ---------- def on_reset_calibrate_clicked(self): bucket = self.active_bucket if not bucket: return if not self._is_bucket_calibrated(bucket): QMessageBox.information(self, "重置标定", "当前工位尚未标定，无需重置") return ret = QMessageBox.question( self, "重置标定", f"确认清除 {bucket.description} 的标定数据？\n清除后该工位会在识别到 ready 时自动重新标定。", QMessageBox.Yes | QMessageBox.No, QMessageBox.No, ) if ret != QMessageBox.Yes: return for side in ("left", "right"): setattr(bucket, f"hook_lug_x_offset_ratio_{side}", None) setattr(bucket, f"hook_lug_y_offset_ratio_{side}", None) self._save_config() self.refresh_bucket_icons() self._refresh_calib_ui() logger.info(f"标定已重置: {bucket.description}") def _update_auto_calibration(self, side: str, ready: bool): """该侧窗口内 ready 帧数达标后采样取平均，自动写入未标定侧的标定值。 已标定侧不覆盖；切换到未标定工位后识别到 ready 即自动标定，无需手动操作。 DeepStream 跳帧不可靠，采用窗口计数而非连续帧： 1 秒（auto_calib_duration_secs）内出现 auto_calib_ready_frames 帧 ready 即触发； 非 ready 消息不清零，天然容忍跳帧与偶发闪烁。 """ if self._calib_active: return bucket = self.active_bucket if not bucket: return has_cam = any(c.position.value == side for c in bucket.cameras) if not has_cam: return x_attr = f"hook_lug_x_offset_ratio_{side}" y_attr = f"hook_lug_y_offset_ratio_{side}" if getattr(bucket, x_attr) is not None and getattr(bucket, y_attr) is not None: return if not ready: return st = self._auto_calib_state[side] now = time.time() window = self.config.ai.auto_calib_duration_secs st["ready_times"].append(now) while st["ready_times"] and now - st["ready_times"][0] > window: st["ready_times"].popleft() ratios = self._compute_offset_ratios(side) if ratios is not None: st["samples"].append((now, ratios[0], ratios[1])) while st["samples"] and now - st["samples"][0][0] > window: st["samples"].popleft() if len(st["ready_times"]) < self.config.ai.auto_calib_ready_frames: return if len(st["samples"]) < self.config.ai.auto_calib_min_samples: return x = sum(s[1] for s in st["samples"]) / len(st["samples"]) y = sum(s[2] for s in st["samples"]) / len(st["samples"]) setattr(bucket, x_attr, x) setattr(bucket, y_attr, y) st["ready_times"].clear() st["samples"].clear() self._save_config() self.refresh_bucket_icons() self._refresh_calib_ui() logger.info(f"自动标定完成 {side}侧: x={x:.3f}, y={y:.3f} (工位{bucket.id})") def _refresh_calib_ui(self): if not self.calib_btn: return bucket = self.active_bucket if not bucket or not self.ai_enabled: self.calib_btn.setEnabled(False) self.calib_btn.setText("标定") self.calib_btn.setToolTip("标定可以起吊的 hook/lug 偏移（当前工位）") if self.reset_calib_btn: self.reset_calib_btn.setEnabled(False) return self.calib_btn.setEnabled(True) if self.reset_calib_btn: self.reset_calib_btn.setEnabled(self._is_bucket_calibrated(bucket)) parts = [] for side in ("left", "right"): x = getattr(bucket, f"hook_lug_x_offset_ratio_{side}", None) y = getattr(bucket, f"hook_lug_y_offset_ratio_{side}", None) if x is not None and y is not None: parts.append(f"{side}侧 已标定 (x={x:.3f}, y={y:.3f})") else: parts.append(f"{side}侧 未标定") self.calib_btn.setToolTip(bucket.description + "\n" + "\n".join(parts)) if self._is_bucket_calibrated(bucket): self.calib_btn.setText("已标定") self.calib_btn.setStyleSheet(""" QPushButton { color: #67c23a; font-weight: bold; font-size: 12px; border: 1px solid #67c23a; border-radius: 4px; padding: 2px 6px; background: transparent; } """) else: self.calib_btn.setText("标定") self.calib_btn.setStyleSheet(""" QPushButton { color: #606266; font-weight: bold; font-size: 12px; border: 1px solid #dcdfe6; border-radius: 4px; padding: 2px 6px; background: transparent; } QPushButton:hover { border-color: #409eff; color: #409eff; } """) def evaluate_hook_state(self): left_hooks = [o for o in self.left_objects if 'hook' in o['label'].lower()] left_lugs = [o for o in self.left_objects if 'lug' in o['label'].lower()] right_hooks = [o for o in self.right_objects if 'hook' in o['label'].lower()] right_lugs = [o for o in self.right_objects if 'lug' in o['label'].lower()] left_pair = self._nearest_pair(left_hooks, left_lugs) right_pair = self._nearest_pair(right_hooks, right_lugs) self._push_pair_traj("left", left_pair) self._push_pair_traj("right", right_pair) self._update_suppress_window() # 吊耳贴着画面上沿（框顶被裁）：不再判定状态，状态框不显示 if (left_pair and self._lug_touches_top(left_pair[1])) or \ (right_pair and self._lug_touches_top(right_pair[1])): self.hook_state = HookState.IDLE self.prev_state = HookState.IDLE self.state_counter = 0 self._clear_state_display() return is_releasing = False if left_pair: h, l = left_pair if self.prev_left_hook and self.prev_left_lug: dy_h = (h['y'] + h['h'] / 2) - self.prev_left_hook[1] dy_l = (l['y'] + l['h'] / 2) - self.prev_left_lug[1] if dy_h > self.RELEASE_DELTA_THRESHOLD and dy_l > self.RELEASE_DELTA_THRESHOLD: is_releasing = True self.prev_left_hook = (h['x'] + h['w'] / 2, h['y'] + h['h'] / 2) self.prev_left_lug = (l['x'] + l['w'] / 2, l['y'] + l['h'] / 2) else: self.prev_left_hook = None self.prev_left_lug = None if right_pair: h, l = right_pair if self.prev_right_hook and self.prev_right_lug: dy_h = (h['y'] + h['h'] / 2) - self.prev_right_hook[1] dy_l = (l['y'] + l['h'] / 2) - self.prev_right_lug[1] if dy_h > self.RELEASE_DELTA_THRESHOLD and dy_l > self.RELEASE_DELTA_THRESHOLD: is_releasing = True self.prev_right_hook = (h['x'] + h['w'] / 2, h['y'] + h['h'] / 2) self.prev_right_lug = (l['x'] + l['w'] / 2, l['y'] + l['h'] / 2) else: self.prev_right_hook = None self.prev_right_lug = None if is_releasing: new_state = HookState.RELEASING else: # 可以起吊：hook/lug 双轴归一化偏移落在标定容差内且框相交（语义≠挂牢） can_lift = False if self.active_bucket: can_lift = (self._side_match("left", left_pair) or self._side_match("right", right_pair)) if can_lift: new_state = HookState.READY elif self._has_intersecting_pair(left_pair, right_pair): new_state = HookState.HOOKING else: new_state = HookState.IDLE # 防抖：进出 READY 对称用 ready_debounce（防单帧丢配对闪红绿）， # 非 READY → IDLE 即时，其余过渡用 STATE_THRESHOLD if new_state == HookState.IDLE and self.hook_state != HookState.READY: threshold = 1 elif new_state == HookState.READY or self.hook_state == HookState.READY: threshold = self.config.ai.ready_debounce else: threshold = self.STATE_THRESHOLD if new_state == self.prev_state: self.state_counter += 1 else: self.prev_state = new_state self.state_counter = 1 if self.state_counter >= threshold: if new_state != self.hook_state: self.hook_state = new_state self.update_ui_for_state(new_state) self.state_counter = 0 for obj in self.left_objects + self.right_objects: obj.pop('paired', None) if left_pair: left_pair[0]['paired'] = True left_pair[1]['paired'] = True if right_pair: right_pair[0]['paired'] = True right_pair[1]['paired'] = True def rects_intersect(self, a, b): ax1, ay1, ax2, ay2 = a['x'], a['y'], a['x']+a['w'], a['y']+a['h'] bx1, by1, bx2, by2 = b['x'], b['y'], b['x']+b['w'], b['y']+b['h'] return not (ax2 < bx1 or bx2 < ax1 or ay2 < by1 or by2 < ay1) @staticmethod def _lug_touches_top(l) -> bool: """吊耳检测框是否贴着画面上沿（框顶被画面裁切）""" return l['y'] <= 0 def _has_intersecting_pair(self, left_pair, right_pair): if left_pair: h, l = left_pair if self.rects_intersect(h, l): return True if right_pair: h, l = right_pair if self.rects_intersect(h, l): return True return False def _side_match(self, side: str, pair) -> bool: """该侧是否满足"可以起吊"：hook/lug 框相交，且双轴归一化偏移都在标定容差内。 X 用中心点、Y 用框底边：升到一定高度后检测框顶部会被裁掉，中心 Y 不可靠，底边贴近吊耳坐实位置不受裁切影响。""" bucket = self.active_bucket if not bucket or not pair: return False h, l = pair if not self.rects_intersect(h, l): return False if self._lug_touches_top(l): return False lw, lh = l['w'], l['h'] if lw <= 0 or lh <= 0: return False xr = (h['x'] + h['w'] / 2 - (l['x'] + l['w'] / 2)) / lw yr = ((h['y'] + h['h']) - (l['y'] + l['h'])) / lh cal_x = getattr(bucket, f"hook_lug_x_offset_ratio_{side}", None) cal_y = getattr(bucket, f"hook_lug_y_offset_ratio_{side}", None) if cal_x is None or cal_y is None: if side not in self._calib_warned: self._calib_warned.add(side) logger.warning(f"工位 {bucket.id} 未标定 {side} 侧偏移，跳过可以起吊判定") return False tol = self.config.ai.offset_ratio_tolerance return abs(xr - cal_x) <= tol and abs(yr - cal_y) <= tol def _push_pair_traj(self, side: str, pair): """把当前帧配对中心与距离追加进轨迹队列""" if not pair: self._pair_traj[side] = [] return h, l = pair hx, hy = h['x'] + h['w'] / 2, h['y'] + h['h'] / 2 lx, ly = l['x'] + l['w'] / 2, l['y'] + l['h'] / 2 dist = ((hx - lx) ** 2 + (hy - ly) ** 2) ** 0.5 now = time.time() traj = self._pair_traj[side] traj.append({"t": now, "hook_cx": hx, "hook_cy": hy, "lug_cx": lx, "lug_cy": ly, "lug_w": l['w'], "lug_h": l['h'], "dist": dist}) while len(traj) > self.TRAJ_MAX_FRAMES: traj.pop(0) if traj[-1]['t'] - traj[0]['t'] > self.TRAJ_WINDOW_SECS: traj.pop(0) def _window_traj(self, side: str): """返回该侧轨迹窗口（最近 TRAJ_WINDOW_SECS）""" traj = self._pair_traj[side] if not traj: return traj now = traj[-1]['t'] return [p for p in traj if now - p['t'] <= self.TRAJ_WINDOW_SECS] def _update_suppress_window(self): """基于轨迹判定放包/分离阶段，命中则顺延屏蔽语音""" now = time.time() hit = "" for side in ("left", "right"): win = self._window_traj(side) if len(win) < 5: continue first, last = win[0], win[-1] dy_h = last['hook_cy'] - first['hook_cy'] dy_l = last['lug_cy'] - first['lug_cy'] min_dist = min(p['dist'] for p in win) max_dist = max(p['dist'] for p in win) # 放包：hook 与 lug 同步下移，且距离保持贴合 if dy_h >= self.RELEASE_TRAJ_Y and dy_l >= self.RELEASE_TRAJ_Y and min_dist < self.ENGAGE_DIST: hit = "放包" # 分离：距离相离（任意方向），且之前曾贴合 elif max_dist - min_dist >= self.SEPARATE_TRAJ_DIST and min_dist < self.ENGAGE_DIST: hit = "分离" if hit: self._voice_suppress_until = max(now, self._voice_suppress_until) + self.SUPPRESS_SECONDS self._trace_phase = hit else: self._trace_phase = "" def _is_voice_suppressed(self) -> bool: return time.time() < self._voice_suppress_until def _maybe_clear_stale_overlay(self): """MQTT 消息超时后清除残留画框/轨迹，避免画面卡住旧框""" now = time.time() if not self.ai_enabled or self._overlay_cleared: return if self._last_mqtt_process > 0 and now - self._last_mqtt_process > self.MQTT_CLEAR_TIMEOUT: self._last_left_draw = [] self._last_right_draw = [] self._pair_traj = {"left": [], "right": []} self._trace_phase = "" self._voice_suppress_until = 0.0 if self.left_player: self.left_player.set_ai_objects([]) if self.right_player: self.right_player.set_ai_objects([]) self._overlay_cleared = True def _trace_objects(self, side: str) -> list: """构造叠加对象：挂钩轨迹、吊耳轨迹、挂钩-吊耳连线、阶段文本""" if not self.debug_trace: return [] win = self._window_traj(side) if len(win) < 2: return [] hook_pts = [(p['hook_cx'], p['hook_cy']) for p in win] lug_pts = [(p['lug_cx'], p['lug_cy']) for p in win] objs = [ {"points": hook_pts, "color": (0, 255, 0), "alpha": 0.4, "thickness": 1}, {"points": lug_pts, "color": (0, 255, 255), "alpha": 0.4, "thickness": 1}, ] hx, hy = win[-1]['hook_cx'], win[-1]['hook_cy'] lx, ly = win[-1]['lug_cx'], win[-1]['lug_cy'] color = (0, 0, 255) if self._trace_phase == "分离" else (255, 165, 0) objs.append({"points": [(hx, hy), (lx, ly)], "color": color, "alpha": 0.4, "thickness": 1}) return objs def _trace_status_text(self) -> str: if not self.debug_trace: return "" if self._trace_phase: remain = max(0.0, self._voice_suppress_until - time.time()) tag = "RELEASE" if self._trace_phase == "放包" else "SEP" return f"{tag} {remain:.1f}s" return "" # ---------- UI 更新与语音 ---------- def _clear_state_display(self): """吊耳被画面上沿裁切时不判定：状态文字清空，边框恢复默认""" self.hook_status_label.setText("") style = """ QLabel { border: 4px solid #dcdfe6; border-radius: 8px; background-color: black; } """ self.left_label.setStyleSheet(style) self.right_label.setStyleSheet(style) def update_ui_for_state(self, state): if state == HookState.IDLE: border_color = "#dcdfe6" text = "空闲" elif state == HookState.RELEASING: border_color = "#ffa500" text = "放包中" elif state == HookState.HOOKING: border_color = "#ff0000" text = "挂包中" elif state == HookState.READY: border_color = "#00ff00" text = "挂包成功" else: border_color = "#dcdfe6" text = "" self.hook_status_label.setText(f"挂钩状态: {text}") style = f""" QLabel {{ border: 4px solid {border_color}; border-radius: 8px; background-color: black; }} """ self.left_label.setStyleSheet(style) self.right_label.setStyleSheet(style) if self.voice_enabled and state in (HookState.HOOKING, HookState.READY): now = time.time() if self._is_voice_suppressed(): return if state == HookState.HOOKING: if state != self.last_voice_state or now - self._last_hooking_voice_time > 15: self._speak("挂包中") self.last_voice_state = state self._last_hooking_voice_time = now elif state != self.last_voice_state: self._speak("挂包成功") self.last_voice_state = state # ---------- 定时任务 ---------- def check_idle(self): if self.rfid_client: self.rfid_client.check_idle() self._maybe_clear_stale_overlay() if self._calib_active: self._sample_calibration() now = time.time() if self.config.ai.enabled and self.mqtt_client and not self.mqtt_client.connected: if now - self._last_mqtt_connect_attempt > 30: self._last_mqtt_connect_attempt = now self._connect_current_ai() # 每 5 秒检测一次摄像头信号 if now - self._last_signal_check > 5: self._last_signal_check = now if not getattr(self, '_signal_check_thread', None) or not self._signal_check_thread.is_alive(): self._signal_check_thread = threading.Thread(target=self._check_all_cameras, daemon=True) self._signal_check_thread.start() def _collect_camera_ips(self) -> list: ips = set() for bucket in self.buckets: for cam in bucket.cameras: m = re.search(r'rtsp://(?:[^@]+@)?([^:/]+)', cam.rtsp_url) if m: ips.add(m.group(1)) return sorted(ips) def _check_all_cameras(self): ips = self._collect_camera_ips() if not ips: return def check_one(ip: str) -> tuple: try: import socket s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) s.settimeout(2) r = s.connect_ex((ip, 554)) s.close() return (ip, 0 if r == 0 else -1) except Exception: return (ip, -1) with ThreadPoolExecutor(max_workers=len(ips)) as pool: results = dict(pool.map(check_one, ips)) self.camera_signal_signal.emit(results) def _on_camera_signal(self, results: dict): was_offline = {ip for ip, lat in self._camera_signal.items() if lat is not None and lat < 0} self._camera_signal.update(results) now_online = {ip for ip, lat in results.items() if lat >= 0} revived = now_online & was_offline went_offline = {ip for ip, lat in results.items() if lat < 0} if went_offline and self.active_bucket: for cam in self.active_bucket.cameras: m = re.search(r'rtsp://(?:[^@]+@)?([^:/]+)', cam.rtsp_url) if m and m.group(1) in went_offline: ip = m.group(1) if ip not in self._offline_reported and self.voice_enabled: self._offline_reported.add(ip) side_name = "左" if cam.position == CameraPosition.LEFT else "右" self._speak(f"{side_name}侧摄像头离线") logger.warning(f"摄像头离线: {ip}") side = "left" if cam.position == CameraPosition.LEFT else "right" label = self.left_label if side == "left" else self.right_label cur = self.left_player if side == "left" else self.right_player if cur is not None: cur.stop() if side == "left": self.left_player = None else: self.right_player = None self._cancel_rtsp_retry(side) self.show_timeout_on_label(label) if revived and self.active_bucket: for cam in self.active_bucket.cameras: m = re.search(r'rtsp://(?:[^@]+@)?([^:/]+)', cam.rtsp_url) if m and m.group(1) in revived: self._offline_reported.discard(m.group(1)) side = "left" if cam.position == CameraPosition.LEFT else "right" cur = self.left_player if side == "left" else self.right_player if cur is None: self._cancel_rtsp_retry(side) self._retry_rtsp(side) for bucket in self.buckets: bars = self._bucket_signal_labels.get(bucket.id) if not bars: continue left_bar, right_bar = bars for bar, cam in zip((left_bar, right_bar), bucket.cameras): m = re.search(r'rtsp://(?:[^@]+@)?([^:/]+)', cam.rtsp_url) if not m: bar.setStyleSheet("background-color: #dcdfe6; border-radius: 2px;") continue ip = m.group(1) lat = self._camera_signal.get(ip) if lat is None: bar.setStyleSheet("background-color: #dcdfe6; border-radius: 2px;") elif lat < 0: bar.setStyleSheet("background-color: #f56c6c; border-radius: 2px;") elif lat < 50: bar.setStyleSheet("background-color: #67c23a; border-radius: 2px;") elif lat < 200: bar.setStyleSheet("background-color: #e6a23c; border-radius: 2px;") else: bar.setStyleSheet("background-color: #f56c6c; border-radius: 2px;") def toggle_switch_mode(self, checked: bool): if checked: self.switch_mode = SwitchMode.AUTO self.auto_switch_btn.setText("自动模式") if self._last_rfid_tag: self._on_rfid_tag_impl(self._last_rfid_tag) elif self.active_bucket: self.activate_bucket(None) else: self.switch_mode = SwitchMode.HAND self.auto_switch_btn.setText("手动模式") # ---------- 配置保存 ---------- def _find_bucket_section(self, cfg, bucket_id): for section in cfg.sections(): if not re.match(r'^Bucket:\d+$', section): continue try: if cfg.getint(section, "id") == bucket_id: return section except (ValueError, configparser.Error): continue return None def _save_config(self): import configparser cfg = configparser.ConfigParser() cfg.read(self._config_path, encoding="utf-8") if not cfg.has_section("AI"): cfg.add_section("AI") val = "true" if self.config.ai.tts_enabled else "false" cfg.set("AI", "tts_enabled", val) cfg.set("AI", "offset_ratio_tolerance", str(self.config.ai.offset_ratio_tolerance)) cfg.set("AI", "ready_debounce", str(self.config.ai.ready_debounce)) cfg.set("AI", "auto_calib_ready_frames", str(self.config.ai.auto_calib_ready_frames)) cfg.set("AI", "auto_calib_duration_secs", str(self.config.ai.auto_calib_duration_secs)) cfg.set("AI", "auto_calib_min_samples", str(self.config.ai.auto_calib_min_samples)) for bucket in self.buckets: section = self._find_bucket_section(cfg, bucket.id) if not section: continue for side in ("left", "right"): x = getattr(bucket, f"hook_lug_x_offset_ratio_{side}", None) y = getattr(bucket, f"hook_lug_y_offset_ratio_{side}", None) key_x = f"hook_lug_x_offset_ratio_{side}" key_y = f"hook_lug_y_offset_ratio_{side}" if x is not None and y is not None: cfg.set(section, key_x, f"{x:.6f}") cfg.set(section, key_y, f"{y:.6f}") else: cfg.remove_option(section, key_x) cfg.remove_option(section, key_y) if not cfg.has_section("Video"): cfg.add_section("Video") cfg.set("Video", "ratio_16_9", "true" if self._ratio_16_9 else "false") with open(self._config_path, "w", encoding="utf-8") as f: cfg.write(f) logger.info(f"配置已保存: {self._config_path}") @staticmethod def _autostart_config_dir(): if sys.platform == "win32": return os.path.join(os.environ.get("APPDATA", os.path.expanduser("~")), "HookVision") return os.path.join(os.path.expanduser("~"), ".config", "HookVision") @staticmethod def _autostart_config_path(): return os.path.join(MainWindow._autostart_config_dir(), "autostart.txt") # 改为 txt @staticmethod def _autostart_desktop_path(): return os.path.expanduser("~/.config/autostart/HookVision.desktop") @staticmethod def _read_autostart_flag(): path = MainWindow._autostart_config_path() try: with open(path, "r", encoding="utf-8") as f: content = f.read().strip().lower() return content == "true" except: return False @staticmethod def _write_autostart_flag(enabled: bool): path = MainWindow._autostart_config_path() os.makedirs(os.path.dirname(path), exist_ok=True) with open(path, "w", encoding="utf-8") as f: f.write("true" if enabled else "false") @staticmethod def _autostart_command(): if getattr(sys, 'frozen', False): return sys.executable else: main_script = os.path.abspath(sys.argv[0]) python_exe = sys.executable if ' ' in python_exe: python_exe = f'"{python_exe}"' if ' ' in main_script: main_script = f'"{main_script}"' if sys.platform == "win32": pythonw = python_exe.replace('python.exe', 'pythonw.exe') if pythonw != python_exe and os.path.exists(pythonw): python_exe = pythonw return f'{python_exe} {main_script}' @staticmethod def _apply_system_autostart(enabled: bool) -> bool: cmd = MainWindow._autostart_command() try: if sys.platform == "win32": import winreg key = r"Software\Microsoft\Windows\CurrentVersion\Run" with winreg.OpenKey(winreg.HKEY_CURRENT_USER, key, 0, winreg.KEY_SET_VALUE) as rk: if enabled: winreg.SetValueEx(rk, MainWindow.APPTITLE, 0, winreg.REG_SZ, cmd) else: try: winreg.DeleteValue(rk, MainWindow.APPTITLE) except OSError: pass return True else: desktop_path = MainWindow._autostart_desktop_path() if enabled: os.makedirs(os.path.dirname(desktop_path), exist_ok=True) with open(desktop_path, "w", encoding="utf-8") as f: f.write(f"""[Desktop Entry] Type=Application Name={MainWindow.APPTITLE} Exec={cmd} Terminal=false X-GNOME-Autostart-enabled=true """) else: try: os.remove(desktop_path) except OSError: pass return True except Exception as e: logger.error(f"设置开机自启失败: {e}") return False def _on_autostart_toggle(self, state): enabled = (state == Qt.Checked) self._write_autostart_flag(enabled) ok = self._apply_system_autostart(enabled) if not ok: QMessageBox.warning( self, "设置失败", "开机自启动设置失败，请检查权限或手动添加。\n" "Windows: 将本程序快捷方式放入 '启动' 文件夹。\n" "Linux: 检查 ~/.config/autostart/ 目录权限。" )