import sys import os import yaml import requests import cv2 import numpy as np from PyQt6.QtWidgets import (QApplication, QMainWindow, QWidget, QVBoxLayout, QHBoxLayout, QPushButton, QLabel, QMessageBox, QGroupBox, QTextEdit, QStatusBar, QSizePolicy, QSystemTrayIcon, QMenu) from PyQt6.QtCore import QThread, pyqtSignal, Qt, pyqtSlot, QPoint from PyQt6.QtGui import QImage, QPixmap, QFont, QPainter, QColor, QBrush, QPen, QAction, QCursor, QIcon class ConfigManager: DEFAULT_CONFIG = { "video_stream_url": "http://222.243.138.146:9002/live/cam890134962b.live.mp4", "door_ip": "192.168.188.205", "door_port": 14460, "door_password": "", "door_id": 1 } @staticmethod def load_config(config_path="config.yaml"): if not os.path.exists(config_path): return ConfigManager.DEFAULT_CONFIG try: with open(config_path, 'r', encoding='utf-8') as f: config = yaml.safe_load(f) if not config: return ConfigManager.DEFAULT_CONFIG # Merge with defaults to ensure all keys exist final_config = ConfigManager.DEFAULT_CONFIG.copy() final_config.update(config) return final_config except Exception as e: print(f"Error loading config: {e}") return ConfigManager.DEFAULT_CONFIG class VideoThread(QThread): change_pixmap_signal = pyqtSignal(QImage) connection_status_signal = pyqtSignal(bool) def __init__(self, url): super().__init__() self.url = url self.running = True def run(self): # Attempt to open the video stream # Note: ws:// streams with .mp4 extension often require specific handling # or an ffmpeg backend. We try standard cv2 first. cap = cv2.VideoCapture(self.url) if not cap.isOpened(): # Fallback/Retry logic could go here pass while self.running: ret, frame = cap.read() if ret: self.connection_status_signal.emit(True) rgb_image = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB) h, w, ch = rgb_image.shape bytes_per_line = ch * w convert_to_qt_format = QImage(rgb_image.data, w, h, bytes_per_line, QImage.Format.Format_RGB888) p = convert_to_qt_format.scaled(480, 360, Qt.AspectRatioMode.KeepAspectRatio) self.change_pixmap_signal.emit(p) else: self.connection_status_signal.emit(False) # Try to reconnect if stream drops cap.release() cv2.waitKey(1000) # Wait a bit before reconnecting cap = cv2.VideoCapture(self.url) cap.release() def stop(self): self.running = False self.wait() class RemoteOpenWorker(QThread): result_signal = pyqtSignal(str, bool) def __init__(self, ip, port, password, door_id): super().__init__() self.ip = ip self.port = port self.password = password self.door_id = door_id def run(self): url = f"http://{self.ip}:{self.port}/openDoorControl" payload = { "pass": self.password, "doorId": self.door_id } try: response = requests.post(url, json=payload, timeout=5) if response.status_code == 200: try: data = response.json() if data.get("result") is True: self.result_signal.emit(f"Remote Open Success: {data.get('message', 'OK')}", True) else: self.result_signal.emit(f"Remote Open Failed: {data.get('message', 'Unknown error')}", False) except ValueError: self.result_signal.emit(f"Invalid JSON response: {response.text}", False) else: self.result_signal.emit(f"HTTP Error: {response.status_code}", False) except Exception as e: self.result_signal.emit(f"Connection Error: {str(e)}", False) class DoorControlWorker(QThread): result_signal = pyqtSignal(str, bool) # message, success def __init__(self, ip, port, password, control_way): super().__init__() self.ip = ip self.port = port self.password = password self.control_way = control_way def run(self): url = f"http://{self.ip}:{self.port}/setEmergencyControl" payload = { "pass": self.password, "controlWay": self.control_way } try: # Image implies form-data or json? # The postman screenshot shows 'Body' -> 'raw' -> 'JSON'. response = requests.post(url, json=payload, timeout=5) if response.status_code == 200: try: data = response.json() # Check 'result' field based on screenshot if data.get("result") is True: self.result_signal.emit(f"Success: {data.get('message', 'OK')}", True) else: self.result_signal.emit(f"Failed: {data.get('message', 'Unknown error')}", False) except ValueError: self.result_signal.emit(f"Invalid JSON response: {response.text}", False) else: self.result_signal.emit(f"HTTP Error: {response.status_code}", False) except Exception as e: self.result_signal.emit(f"Connection Error: {str(e)}", False) class FloatingButton(QWidget): clicked = pyqtSignal() def __init__(self, parent=None): super().__init__(parent) self.setWindowFlags(Qt.WindowType.FramelessWindowHint | Qt.WindowType.WindowStaysOnTopHint | Qt.WindowType.Tool) self.setAttribute(Qt.WidgetAttribute.WA_TranslucentBackground) self.resize(60, 60) self.dragging = False self.offset = QPoint() # Initial position (e.g., right-center of screen) screen = QApplication.primaryScreen().geometry() self.move(screen.width() - 80, screen.height() // 2) def paintEvent(self, event): painter = QPainter(self) painter.setRenderHint(QPainter.RenderHint.Antialiasing) # Draw circle painter.setBrush(QBrush(QColor("#2196F3"))) # Blue like the "Always Open" button painter.setPen(QPen(Qt.GlobalColor.white, 2)) painter.drawEllipse(5, 5, 50, 50) # Draw "Open" text or icon painter.setPen(Qt.GlobalColor.white) font = painter.font() font.setBold(True) font.setPointSize(10) painter.setFont(font) painter.drawText(self.rect(), Qt.AlignmentFlag.AlignCenter, "Open") def mousePressEvent(self, event): if event.button() == Qt.MouseButton.LeftButton: self.dragging = True self.offset = event.globalPosition().toPoint() - self.pos() self.start_pos = event.globalPosition().toPoint() def mouseMoveEvent(self, event): if self.dragging: self.move(event.globalPosition().toPoint() - self.offset) def mouseReleaseEvent(self, event): if event.button() == Qt.MouseButton.LeftButton: self.dragging = False end_pos = event.globalPosition().toPoint() dist = (end_pos - self.start_pos).manhattanLength() if dist < 5: # Threshold self.clicked.emit() class MainWindow(QMainWindow): def __init__(self): super().__init__() self.config = ConfigManager.load_config() self.setWindowTitle("大门控制工具 (Door Control)") self.resize(450, 800) # Phone-like aspect ratio # Set Application Icon icon_path = "icon.png" if os.path.exists(icon_path): self.app_icon = QIcon(icon_path) else: # Fallback to generated door icon self.app_icon = self.create_door_icon() self.setWindowIcon(self.app_icon) # Also set the taskbar icon explicitely for some Windows versions if needed, # but usually setWindowIcon handles it if the window is shown. QApplication.setWindowIcon(self.app_icon) # Main Layout central_widget = QWidget() self.setCentralWidget(central_widget) main_layout = QVBoxLayout(central_widget) main_layout.setSpacing(20) main_layout.setContentsMargins(20, 20, 20, 20) # 1. Video Section (Top) # video_group = QGroupBox("实时监控") # Simplified: Just a label, maybe with a frame video_container = QWidget() video_layout = QVBoxLayout(video_container) video_layout.setContentsMargins(0,0,0,0) self.video_label = QLabel("正在连接视频流...\nConnecting...") self.video_label.setAlignment(Qt.AlignmentFlag.AlignCenter) self.video_label.setStyleSheet("background-color: black; color: white; font-size: 14px; border-radius: 10px;") self.video_label.setMinimumSize(400, 300) self.video_label.setMaximumHeight(350) video_layout.addWidget(self.video_label) main_layout.addWidget(video_container) # 2. Controls Section (Bottom) controls_container = QWidget() controls_layout = QVBoxLayout(controls_container) controls_layout.setSpacing(15) # Common button style btn_font = QFont() btn_font.setPointSize(16) # Bigger font btn_font.setBold(True) def create_btn(text, color): btn = QPushButton(text) btn.setFont(btn_font) btn.setMinimumHeight(60) # Taller buttons btn.setStyleSheet(f""" QPushButton {{ background-color: {color}; color: white; border-radius: 8px; }} QPushButton:pressed {{ background-color: {color}dd; # Slightly darker on press }} """) return btn # Remote Open Button self.btn_remote_open = create_btn("远程开门 (Remote Open)", "#FF9800") # Orange self.btn_remote_open.clicked.connect(self.send_remote_open) controls_layout.addWidget(self.btn_remote_open) # Normal Mode self.btn_online = create_btn("正常模式 (Normal - 0)", "#4CAF50") # Green self.btn_online.clicked.connect(lambda: self.send_command(0)) controls_layout.addWidget(self.btn_online) # Always Open self.btn_open = create_btn("常开模式 (Always Open - 1)", "#2196F3") # Blue self.btn_open.clicked.connect(lambda: self.send_command(1)) controls_layout.addWidget(self.btn_open) # Always Closed self.btn_close = create_btn("常闭模式 (Always Closed - 2)", "#f44336") # Red self.btn_close.clicked.connect(lambda: self.send_command(2)) controls_layout.addWidget(self.btn_close) main_layout.addWidget(controls_container) # 3. Log Section (Very bottom, smaller) self.log_output = QTextEdit() self.log_output.setReadOnly(True) self.log_output.setMaximumHeight(100) # Smaller log area self.log_output.setStyleSheet("background-color: #f0f0f0; border: 1px solid #ccc; border-radius: 5px;") main_layout.addWidget(self.log_output) # Start Video self.start_video() # --- System Tray & Floating Button --- self.tray_icon = QSystemTrayIcon(self) self.tray_icon.setIcon(self.app_icon) # Create Tray Menu tray_menu = QMenu() show_action = QAction("显示 (Show)", self) show_action.triggered.connect(self.show_window) quit_action = QAction("退出 (Quit)", self) quit_action.triggered.connect(self.quit_app) tray_menu.addAction(show_action) tray_menu.addAction(quit_action) self.tray_icon.setContextMenu(tray_menu) self.tray_icon.activated.connect(self.on_tray_icon_activated) self.tray_icon.show() # Floating Button self.floating_button = FloatingButton() self.floating_button.clicked.connect(self.show_window) # Initially hidden if main window is shown self.floating_button.hide() def create_door_icon(self): pixmap = QPixmap(64, 64) pixmap.fill(Qt.GlobalColor.transparent) painter = QPainter(pixmap) painter.setRenderHint(QPainter.RenderHint.Antialiasing) # Draw Door Frame painter.setBrush(QBrush(QColor("#4CAF50"))) # Green door painter.setPen(QPen(QColor("#388E3C"), 2)) painter.drawRoundedRect(12, 4, 40, 56, 4, 4) # Draw Door Knob painter.setBrush(QBrush(QColor("#FFC107"))) # Gold knob painter.setPen(Qt.PenStyle.NoPen) painter.drawEllipse(42, 32, 6, 6) painter.end() return QIcon(pixmap) def show_window(self): self.show() self.setWindowState(Qt.WindowState.WindowNoState) self.activateWindow() self.floating_button.hide() def quit_app(self): if hasattr(self, 'video_thread'): self.video_thread.stop() QApplication.quit() def on_tray_icon_activated(self, reason): if reason == QSystemTrayIcon.ActivationReason.Trigger: if self.isVisible(): self.hide() self.floating_button.show() else: self.show_window() def closeEvent(self, event): if self.tray_icon.isVisible(): self.hide() self.floating_button.show() event.ignore() else: self.quit_app() def start_video(self): url = self.config.get("video_stream_url") self.log(f"Attempting to connect to stream: {url}") self.video_thread = VideoThread(url) self.video_thread.change_pixmap_signal.connect(self.update_image) self.video_thread.connection_status_signal.connect(self.update_video_status) self.video_thread.start() @pyqtSlot(QImage) def update_image(self, qt_img): self.video_label.setPixmap(QPixmap.fromImage(qt_img)) self.video_label.setText("") # Clear loading text @pyqtSlot(bool) def update_video_status(self, connected): if not connected and not self.video_label.pixmap(): self.video_label.setText("视频流连接失败或已断开\nStream Disconnected") def send_remote_open(self): self.log("Sending remote open command...") self.set_buttons_enabled(False) self.remote_worker = RemoteOpenWorker( self.config['door_ip'], self.config['door_port'], self.config['door_password'], self.config['door_id'] ) self.remote_worker.result_signal.connect(self.handle_command_result) self.remote_worker.finished.connect(lambda: self.set_buttons_enabled(True)) self.remote_worker.start() def send_command(self, mode): mode_text = {0: "正常 (Normal)", 1: "常开 (Always Open)", 2: "常闭 (Always Closed)"}[mode] self.log(f"Sending command: {mode_text}...") # Disable buttons temporarily self.set_buttons_enabled(False) self.worker = DoorControlWorker( self.config['door_ip'], self.config['door_port'], self.config['door_password'], mode ) self.worker.result_signal.connect(self.handle_command_result) self.worker.finished.connect(lambda: self.set_buttons_enabled(True)) self.worker.start() def handle_command_result(self, message, success): if success: self.log(f"SUCCESS: {message}") # QMessageBox.information(self, "成功", f"操作成功\n{message}") else: self.log(f"ERROR: {message}") # QMessageBox.warning(self, "错误", f"操作失败\n{message}") def set_buttons_enabled(self, enabled): self.btn_remote_open.setEnabled(enabled) self.btn_online.setEnabled(enabled) self.btn_open.setEnabled(enabled) self.btn_close.setEnabled(enabled) def log(self, message): self.log_output.append(message) # Scroll to bottom cursor = self.log_output.textCursor() cursor.movePosition(cursor.MoveOperation.End) self.log_output.setTextCursor(cursor) if __name__ == "__main__": app = QApplication(sys.argv) window = MainWindow() window.show() sys.exit(app.exec())