import os import sys import time import re import logging import threading import queue import subprocess from collections import defaultdict from PySide6.QtWidgets import (QMainWindow, QWidget, QVBoxLayout, QHBoxLayout, QPushButton, QLabel, QGridLayout, QScrollArea, QToolButton, QFrame, QApplication, QCheckBox, QMessageBox, QSlider) 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 logger = logging.getLogger(__name__) class MainWindow(QMainWindow): # 防抖阈值（连续相同状态次数） STATE_THRESHOLD = 3 # 放包判断位移阈值（像素） RELEASE_DELTA_THRESHOLD = 5 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) 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.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 # 语音引擎（在工作线程内初始化） 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.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.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.timer = QTimer() self.timer.timeout.connect(self.check_idle) self.timer.start(1000) 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) # 摄像头信号指示条（左右两段） bar = QWidget() bar.setFixedHeight(8) bar_layout = QHBoxLayout(bar) bar_layout.setContentsMargins(0, 0, 0, 0) bar_layout.setSpacing(2) left_bar = QFrame() right_bar = QFrame() for fb in (left_bar, right_bar): fb.setFixedHeight(8) fb.setStyleSheet("background-color: #dcdfe6; border-radius: 2px;") bar_layout.addWidget(left_bar) bar_layout.addWidget(right_bar) cell_layout.addWidget(bar) self._bucket_signal_labels[bucket.id] = (left_bar, right_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) 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() event.accept() # ---------- 语音控制 ---------- 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.update_ui_for_state(HookState.IDLE) 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) 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"} 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): try: import pythoncom pythoncom.CoInitialize() except Exception: pass speaker = None try: import win32com.client as _win32com speaker = _win32com.Dispatch("SAPI.SpVoice") speaker.Rate = 2 except Exception: logging.warning("SAPI 不可用，仅可使用 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) 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 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 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) 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 / self.video_aspect_ratio) 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 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)) return lat is not None and 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离线") # ---------- 硬件初始化 ---------- def init_hardware(self): if self.config.module_type == ModuleType.RFID and self.config.rfid: 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 ) 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.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.update_ui_for_state(HookState.IDLE) # 先切 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 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) else: self._unsubscribe_all_topics() self.left_label.clear() self.right_label.clear() QApplication.processEvents() 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({ "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 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) if cam.position == CameraPosition.LEFT: self.left_objects = all_objects else: self.right_objects = all_objects self.evaluate_hook_state() draw_objects = [o for o in all_objects if 'ready' not in o['label'].lower()] for obj in draw_objects: if not obj.get('paired'): obj['dimmed'] = True 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 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()] def find_nearest_pair(hooks, lugs): 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 left_pair = find_nearest_pair(left_hooks, left_lugs) right_pair = find_nearest_pair(right_hooks, right_lugs) 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: any_ready = any('ready' in o['label'].lower() for o in self.left_objects + self.right_objects) has_intersecting_pair = self._has_intersecting_pair(left_pair, right_pair) if any_ready and has_intersecting_pair: new_state = HookState.READY elif has_intersecting_pair: new_state = HookState.HOOKING else: new_state = HookState.IDLE if self.hook_state == HookState.READY and new_state != HookState.READY: threshold = 15 else: threshold = 1 if new_state == HookState.IDLE else 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) 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 # ---------- UI 更新与语音 ---------- 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 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() 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 ping_one(ip: str) -> tuple: try: if sys.platform == "win32": r = subprocess.run(["ping", "-n", "1", ip], capture_output=True, timeout=5, text=True, creationflags=subprocess.CREATE_NO_WINDOW) else: r = subprocess.run(["ping", "-c", "1", ip], capture_output=True, timeout=5, text=True) if r.returncode != 0: return (ip, -1) m = re.search(r'time[=<]\s*(\d+(?:\.\d+)?)\s*ms', r.stdout, re.IGNORECASE) return (ip, float(m.group(1)) if m else 0) except subprocess.TimeoutExpired: return (ip, -1) except Exception: return (ip, -1) with ThreadPoolExecutor(max_workers=len(ips)) as pool: results = dict(pool.map(ping_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 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: 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 _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) with open(self._config_path, "w", encoding="utf-8") as f: cfg.write(f) logger.info(f"TTS 设置已保存: {self._config_path} → tts_enabled={val}") @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/ 目录权限。" )