#!/usr/bin/env python3 """ AI监控系统 PyQt6图形界面 连接WebSocket服务，实时显示监控画面和告警信息 """ import sys import json import asyncio import threading import base64 from datetime import datetime from PyQt6.QtWidgets import * from PyQt6.QtCore import * from PyQt6.QtGui import * import websockets class WebSocketWorker(QThread): """WebSocket连接工作线程""" message_received = pyqtSignal(dict) connection_status = pyqtSignal(bool, str) def __init__(self, url): super().__init__() self.url = url self.running = False self.websocket = None def run(self): """运行WebSocket连接""" loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) try: self.running = True loop.run_until_complete(self.connect_websocket()) except Exception as e: self.connection_status.emit(False, f"连接失败: {str(e)}") finally: loop.close() async def connect_websocket(self): """连接WebSocket服务器""" try: async with websockets.connect(self.url) as websocket: self.websocket = websocket self.connection_status.emit(True, "连接成功") while self.running: try: message = await websocket.recv() data = json.loads(message) self.message_received.emit(data) except websockets.exceptions.ConnectionClosed: break except Exception as e: print(f"接收消息错误: {e}") break except Exception as e: self.connection_status.emit(False, f"连接错误: {str(e)}") def stop(self): """停止WebSocket连接""" self.running = False if self.websocket: self.websocket.close() class VideoWidget(QLabel): """视频显示控件""" def __init__(self): super().__init__() self.setMinimumSize(640, 480) self.setAlignment(Qt.AlignmentFlag.AlignCenter) self.setStyleSheet(""" QLabel { background-color: #2b2b2b; border: 2px solid #555; border-radius: 8px; color: #888; font-size: 14px; } """) self.setText("等待视频流...") def update_image(self, base64_data): """更新显示的图像""" try: # 解码base64图像数据 image_data = base64.b64decode(base64_data) pixmap = QPixmap() pixmap.loadFromData(image_data) # 缩放图像以适应控件大小 scaled_pixmap = pixmap.scaled( self.size(), Qt.AspectRatioMode.KeepAspectRatio, Qt.TransformationMode.SmoothTransformation ) self.setPixmap(scaled_pixmap) except Exception as e: print(f"图像显示错误: {e}") self.setText("图像解码失败") class AlertListWidget(QListWidget): """告警列表控件""" def __init__(self): super().__init__() self.setMaximumHeight(200) self.setStyleSheet(""" QListWidget { background-color: #1e1e1e; border: 1px solid #444; border-radius: 4px; color: white; font-family: 'Courier New', monospace; } QListWidget::item { padding: 8px; border-bottom: 1px solid #333; } QListWidget::item:selected { background-color: #0078d4; } """) def add_alert(self, alert_data): """添加告警信息""" timestamp = datetime.fromtimestamp(alert_data['timestamp']).strftime("%H:%M:%S") camera_id = alert_data['camera_id'] event_type = alert_data['event_type'] video_file = alert_data.get('video_file', '') # 创建告警项目 item_text = f"[{timestamp}] 摄像头{camera_id} - 事件类型{event_type}" item = QListWidgetItem(item_text) item.setData(Qt.ItemDataRole.UserRole, alert_data) # 根据事件类型设置颜色 if event_type == 0: item.setForeground(QColor("#4CAF50")) # 绿色 - 正常 elif event_type == 1: item.setForeground(QColor("#FF9800")) # 橙色 - 警告 else: item.setForeground(QColor("#F44336")) # 红色 - 严重告警 self.insertItem(0, item) # 插入到顶部 # 限制列表长度 if self.count() > 100: self.takeItem(self.count() - 1) class AIMonitorGUI(QMainWindow): """AI监控系统主界面""" def __init__(self): super().__init__() self.websocket_worker = None self.camera_widgets = {} # 存储各摄像头的视频控件 self.init_ui() def init_ui(self): """初始化用户界面""" self.setWindowTitle("AI监控系统 v1.0 (PyQt6)") self.setGeometry(100, 100, 1400, 900) # 设置深色主题 self.setStyleSheet(""" QMainWindow { background-color: #1e1e1e; color: white; } QMenuBar { background-color: #2d2d2d; border-bottom: 1px solid #444; } QMenu { background-color: #2d2d2d; border: 1px solid #444; } QStatusBar { background-color: #2d2d2d; border-top: 1px solid #444; color: #ccc; } QPushButton { background-color: #0078d4; border: none; color: white; padding: 8px 16px; border-radius: 4px; font-weight: bold; } QPushButton:hover { background-color: #106ebe; } QPushButton:pressed { background-color: #005a9e; } QPushButton:disabled { background-color: #666; color: #999; } QLabel { color: white; } """) # 创建菜单栏 self.create_menu_bar() # 创建中央控件 central_widget = QWidget() self.setCentralWidget(central_widget) # 主布局 main_layout = QVBoxLayout(central_widget) # 顶部控制栏 control_layout = self.create_control_bar() main_layout.addLayout(control_layout) # 中间内容区域 content_layout = QHBoxLayout() # 左侧：摄像头视频区域 self.video_area = self.create_video_area() content_layout.addWidget(self.video_area, 2) # 右侧：信息面板 info_panel = self.create_info_panel() content_layout.addWidget(info_panel, 1) main_layout.addLayout(content_layout) # 底部：告警列表 alert_layout = self.create_alert_section() main_layout.addLayout(alert_layout) # 创建状态栏 self.statusBar().showMessage("就绪") # 自动连接WebSocket QTimer.singleShot(1000, self.connect_websocket) def create_menu_bar(self): """创建菜单栏""" menubar = self.menuBar() # 文件菜单 file_menu = menubar.addMenu('文件') connect_action = QAction('连接服务器', self) connect_action.setShortcut(QKeySequence.StandardKey.Open) connect_action.triggered.connect(self.connect_websocket) file_menu.addAction(connect_action) disconnect_action = QAction('断开连接', self) disconnect_action.setShortcut(QKeySequence.StandardKey.Close) disconnect_action.triggered.connect(self.disconnect_websocket) file_menu.addAction(disconnect_action) file_menu.addSeparator() exit_action = QAction('退出', self) exit_action.setShortcut(QKeySequence.StandardKey.Quit) exit_action.triggered.connect(self.close) file_menu.addAction(exit_action) # 帮助菜单 help_menu = menubar.addMenu('帮助') about_action = QAction('关于', self) about_action.triggered.connect(self.show_about) help_menu.addAction(about_action) def create_control_bar(self): """创建顶部控制栏""" layout = QHBoxLayout() # 连接状态指示器 self.status_label = QLabel("未连接") self.status_label.setStyleSheet(""" QLabel { padding: 6px 12px; background-color: #666; border-radius: 4px; font-weight: bold; } """) layout.addWidget(self.status_label) # 连接按钮 self.connect_btn = QPushButton("连接服务器") self.connect_btn.clicked.connect(self.toggle_connection) layout.addWidget(self.connect_btn) # 刷新按钮 refresh_btn = QPushButton("刷新") refresh_btn.clicked.connect(self.refresh_videos) layout.addWidget(refresh_btn) # 间隔 layout.addStretch() # 当前时间 self.time_label = QLabel() self.time_label.setStyleSheet("color: #ccc;") layout.addWidget(self.time_label) # 更新时间定时器 self.timer = QTimer() self.timer.timeout.connect(self.update_time) self.timer.start(1000) self.update_time() return layout def create_video_area(self): """创建视频显示区域""" scroll_area = QScrollArea() scroll_area.setWidgetResizable(True) scroll_area.setStyleSheet(""" QScrollArea { border: 1px solid #444; background-color: #2d2d2d; border-radius: 8px; } """) # 视频网格容器 self.video_container = QWidget() self.video_grid = QGridLayout(self.video_container) scroll_area.setWidget(self.video_container) return scroll_area def create_info_panel(self): """创建右侧信息面板""" panel = QWidget() layout = QVBoxLayout(panel) # 系统信息组 info_group = QGroupBox("系统信息") info_group.setStyleSheet(""" QGroupBox { font-weight: bold; border: 2px solid #444; border-radius: 8px; margin-top: 1ex; padding-top: 10px; } QGroupBox::title { subcontrol-origin: margin; left: 10px; padding: 0 5px 0 5px; } """) info_layout = QVBoxLayout() self.camera_count_label = QLabel("摄像头数量: 0") self.frame_count_label = QLabel("处理帧数: 0") self.alert_count_label = QLabel("告警数量: 0") info_layout.addWidget(self.camera_count_label) info_layout.addWidget(self.frame_count_label) info_layout.addWidget(self.alert_count_label) info_group.setLayout(info_layout) layout.addWidget(info_group) # 统计图表区域 stats_group = QGroupBox("告警统计") stats_group.setStyleSheet(""" QGroupBox { font-weight: bold; border: 2px solid #444; border-radius: 8px; margin-top: 1ex; padding-top: 10px; } QGroupBox::title { subcontrol-origin: margin; left: 10px; padding: 0 5px 0 5px; } """) self.stats_label = QLabel("暂无数据") self.stats_label.setAlignment(Qt.AlignmentFlag.AlignCenter) self.stats_label.setStyleSheet(""" QLabel { padding: 20px; background-color: #2d2d2d; border-radius: 4px; color: #888; } """) stats_layout = QVBoxLayout() stats_layout.addWidget(self.stats_label) stats_group.setLayout(stats_layout) layout.addWidget(stats_group) layout.addStretch() return panel def create_alert_section(self): """创建告警列表区域""" layout = QVBoxLayout() # 标题 title_label = QLabel("🚨 实时告警") title_label.setStyleSheet(""" QLabel { font-size: 16px; font-weight: bold; color: #FF6B6B; margin-bottom: 8px; } """) layout.addWidget(title_label) # 告警列表 self.alert_list = AlertListWidget() layout.addWidget(self.alert_list) return layout def connect_websocket(self): """连接WebSocket服务器""" if self.websocket_worker and self.websocket_worker.isRunning(): return self.statusBar().showMessage("正在连接服务器...") # 创建WebSocket工作线程 self.websocket_worker = WebSocketWorker("ws://localhost:8765") self.websocket_worker.message_received.connect(self.handle_message) self.websocket_worker.connection_status.connect(self.handle_connection_status) self.websocket_worker.start() def disconnect_websocket(self): """断开WebSocket连接""" if self.websocket_worker: self.websocket_worker.stop() self.websocket_worker.wait() self.websocket_worker = None self.status_label.setText("未连接") self.status_label.setStyleSheet(""" QLabel { padding: 6px 12px; background-color: #666; border-radius: 4px; font-weight: bold; } """) self.connect_btn.setText("连接服务器") self.statusBar().showMessage("已断开连接") def toggle_connection(self): """切换连接状态""" if self.websocket_worker and self.websocket_worker.isRunning(): self.disconnect_websocket() else: self.connect_websocket() def handle_connection_status(self, connected, message): """处理连接状态变化""" if connected: self.status_label.setText("已连接") self.status_label.setStyleSheet(""" QLabel { padding: 6px 12px; background-color: #4CAF50; border-radius: 4px; font-weight: bold; } """) self.connect_btn.setText("断开连接") self.statusBar().showMessage("连接成功") else: self.status_label.setText("连接失败") self.status_label.setStyleSheet(""" QLabel { padding: 6px 12px; background-color: #F44336; border-radius: 4px; font-weight: bold; } """) self.connect_btn.setText("连接服务器") self.statusBar().showMessage(message) def handle_message(self, data): """处理接收到的WebSocket消息""" msg_type = data.get('msg_type') if msg_type == 'frame': self.handle_frame_message(data) elif msg_type == 'alert': self.handle_alert_message(data) def handle_frame_message(self, data): """处理帧消息""" camera_id = data.get('camera_id') image_base64 = data.get('image_base64') if not image_base64: return # 获取或创建摄像头控件 if camera_id not in self.camera_widgets: self.create_camera_widget(camera_id) # 更新视频显示 self.camera_widgets[camera_id]['video'].update_image(image_base64) # 更新统计信息 current_count = int(self.frame_count_label.text().split(': ')[1]) self.frame_count_label.setText(f"处理帧数: {current_count + 1}") def handle_alert_message(self, data): """处理告警消息""" # 添加到告警列表 self.alert_list.add_alert(data) # 更新统计信息 current_count = int(self.alert_count_label.text().split(': ')[1]) self.alert_count_label.setText(f"告警数量: {current_count + 1}") # 状态栏提示 camera_id = data.get('camera_id') event_type = data.get('event_type') self.statusBar().showMessage(f"摄像头{camera_id}触发告警 - 事件类型{event_type}") def create_camera_widget(self, camera_id): """创建摄像头视频控件""" # 摄像头容器 camera_widget = QWidget() camera_layout = QVBoxLayout(camera_widget) # 标题 title_label = QLabel(f"摄像头 {camera_id}") title_label.setAlignment(Qt.AlignmentFlag.AlignCenter) title_label.setStyleSheet(""" QLabel { font-weight: bold; font-size: 14px; padding: 8px; background-color: #0078d4; border-radius: 4px 4px 0 0; } """) camera_layout.addWidget(title_label) # 视频显示 video_widget = VideoWidget() camera_layout.addWidget(video_widget) # 状态标签 status_label = QLabel("🔴 离线") status_label.setAlignment(Qt.AlignmentFlag.AlignCenter) status_label.setStyleSheet(""" QLabel { padding: 4px; font-size: 12px; background-color: #2d2d2d; border-top: 1px solid #444; color: #F44336; } """) camera_layout.addWidget(status_label) # 添加到网格布局 row = (camera_id - 1) // 2 col = (camera_id - 1) % 2 self.video_grid.addWidget(camera_widget, row, col) # 保存控件引用 self.camera_widgets[camera_id] = { 'video': video_widget, 'status': status_label } # 更新摄像头数量 self.camera_count_label.setText(f"摄像头数量: {len(self.camera_widgets)}") def refresh_videos(self): """刷新视频显示""" # 清空现有视频控件 for widget_info in self.camera_widgets.values(): if hasattr(widget_info['video'], 'clear'): widget_info['video'].clear() def update_time(self): """更新当前时间显示""" current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S") self.time_label.setText(current_time) def show_about(self): """显示关于对话框""" QMessageBox.about(self, "关于", "AI监控系统 v1.0 (PyQt6)\n\n" "基于Python + PyQt6的实时视频监控解决方案\n" "支持RTSP视频流接入和AI智能检测\n\n" "功能特性：\n" "• 多路RTSP视频流监控\n" "• 实时AI目标检测\n" "• 智能告警推送\n" "• 历史视频回放\n" "• 现代化图形界面\n\n" "技术栈：Python, PyQt6, OpenCV, YOLO, WebSocket") def closeEvent(self, event): """窗口关闭事件""" self.disconnect_websocket() event.accept() def main(): """主函数""" app = QApplication(sys.argv) # 设置应用信息 app.setApplicationName("AI监控系统") app.setApplicationVersion("1.0") app.setOrganizationName("AI Monitor") # 创建主窗口 window = AIMonitorGUI() window.show() # 运行应用 sys.exit(app.exec()) if __name__ == "__main__": main()