import cv2 import tkinter as tk from tkinter import Canvas from PIL import Image, ImageTk import threading import numpy as np from config import config_manager class PreviewWindow: """采集卡预览窗口""" def __init__(self): self.config = config_manager.config self.caps = {} self.frames = {} self.large_window = None self.running = True def init_cameras(self): """初始化所有相机""" print("🔧 开始初始化采集卡...") loaded_count = 0 # 如果没有活动配置，加载所有配置 active_groups = [g for g in self.config['groups'] if g.get('active', True)] if not active_groups: print("⚠️ 没有活动的配置组，将尝试加载所有配置组") active_groups = self.config['groups'] for i, group in enumerate(active_groups): try: cam_idx = group['camera_index'] print(f" 尝试打开采集卡 {cam_idx} ({group['name']})...") cap = cv2.VideoCapture(int(cam_idx), cv2.CAP_DSHOW) if not cap.isOpened(): print(f" DSHOW模式失败，尝试默认模式...") cap = cv2.VideoCapture(int(cam_idx)) if cap.isOpened(): cap.set(cv2.CAP_PROP_FRAME_WIDTH, group['camera_width']) cap.set(cv2.CAP_PROP_FRAME_HEIGHT, group['camera_height']) # 测试读取一帧 ret, test_frame = cap.read() if ret: self.caps[i] = { 'cap': cap, 'group': group, 'name': group['name'] } loaded_count += 1 print(f" ✅ 采集卡 {cam_idx} 初始化成功") else: cap.release() print(f" ❌ 采集卡 {cam_idx} 无法读取帧") else: print(f" ❌ 采集卡 {cam_idx} 无法打开") except Exception as e: print(f" ❌ 采集卡 {group.get('camera_index', '?')} 初始化失败: {e}") if loaded_count == 0: print("⚠️ 警告：没有成功加载任何采集卡！") print("请检查：") print("1. 采集卡是否正确连接") print("2. 采集卡索引是否正确") print("3. 是否有活动的配置组") else: print(f"✅ 成功加载 {loaded_count} 个采集卡") def capture_frames(self): """捕获帧""" while self.running: for idx, data in self.caps.items(): try: ret, frame = data['cap'].read() if ret and frame is not None: # 裁剪到配置的区域 height, width = frame.shape[:2] crop_top = 30 crop_bottom = min(crop_top + 720, height) crop_width = min(1280, width) # 确保裁剪范围有效 if crop_bottom > crop_top and crop_width > 0: frame = frame[crop_top:crop_bottom, 0:crop_width] # 转换颜色空间 frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB) self.frames[idx] = frame_rgb else: print(f"⚠️ 采集卡 {idx} 裁剪参数无效") else: # 读取失败，清除旧帧 if idx in self.frames: self.frames[idx] = None except Exception as e: print(f"捕获帧 {idx} 错误: {e}") import traceback traceback.print_exc() import time time.sleep(0.01) # 避免CPU占用过高 def create_grid_window(self): """创建网格窗口""" root = tk.Tk() root.title("采集卡预览 - 点击放大") root.geometry("1000x700") # 获取显示配置 display = self.config.get('display', {}) preview_width = display.get('preview_width', 640) preview_height = display.get('preview_height', 360) columns = display.get('preview_columns', 2) rows = display.get('preview_rows', 2) canvas = Canvas(root, bg='black') canvas.pack(fill=tk.BOTH, expand=True) # 存储图像对象 self.photo_objects = {} def update_frames_once(): """在主线程中更新一帧（使用after循环）""" if not self.running: return try: canvas.delete("all") # 如果没有加载任何采集卡，显示提示 if not self.caps: canvas.create_text( root.winfo_width() // 2, root.winfo_height() // 2, text="未找到可用的采集卡\n\n请检查：\n1. 采集卡是否正确连接\n2. 配置组是否有活动的采集卡\n3. 采集卡索引是否正确", fill='yellow', font=('Arial', 14), justify=tk.CENTER ) root.after(33, update_frames_once) return # 计算每个预览窗口的位置和大小 window_width = root.winfo_width() if root.winfo_width() > 1 else 1000 window_height = root.winfo_height() if root.winfo_height() > 1 else 700 cell_width = window_width // columns cell_height = window_height // rows frame_idx = 0 for idx in self.caps.keys(): if idx in self.frames and self.frames[idx] is not None: row = frame_idx // columns col = frame_idx % columns x = col * cell_width y = row * cell_height # 调整图像大小 try: frame = self.frames[idx] h, w = frame.shape[:2] scale = min(cell_width / w, cell_height / h) * 0.9 new_w = int(w * scale) new_h = int(h * scale) resized_frame = cv2.resize(frame, (new_w, new_h)) # 转换为PIL图像 pil_image = Image.fromarray(resized_frame) photo = ImageTk.PhotoImage(image=pil_image) # 保持引用，避免被GC self.photo_objects[idx] = photo # 绘制图像 center_x = x + cell_width // 2 center_y = y + cell_height // 2 canvas.create_image(center_x, center_y, image=photo, anchor='center') # 绘制标签 name = self.caps[idx]['name'] canvas.create_text(center_x, y + 20, text=name, fill='white', font=('Arial', 12, 'bold')) frame_idx += 1 except Exception as e: print(f"处理帧 {idx} 错误: {e}") else: # 显示等待提示 row = frame_idx // columns col = frame_idx % columns x = col * cell_width y = row * cell_height center_x = x + cell_width // 2 center_y = y + cell_height // 2 name = self.caps[idx]['name'] canvas.create_text(center_x, center_y, text=f"{name}\n等待画面...", fill='gray', font=('Arial', 12)) frame_idx += 1 except Exception as e: print(f"更新帧错误: {e}") import traceback traceback.print_exc() # 约30fps root.after(33, update_frames_once) def on_canvas_click(event): """点击画布事件""" window_width = root.winfo_width() window_height = root.winfo_height() cell_width = window_width // columns cell_height = window_height // rows col = int(event.x // cell_width) row = int(event.y // cell_height) index = row * columns + col # 找到对应的配置 if index < len(self.caps): idx = list(self.caps.keys())[index] self.show_large_window(idx) canvas.bind('', on_canvas_click) # 使用after在主线程中循环刷新 root.after(33, update_frames_once) def on_closing(): """关闭窗口""" self.running = False for data in self.caps.values(): data['cap'].release() root.destroy() root.protocol("WM_DELETE_WINDOW", on_closing) root.mainloop() def show_large_window(self, idx): """显示大窗口""" if self.large_window is not None and self.large_window.winfo_exists(): self.large_window.destroy() self.large_window = tk.Toplevel() self.large_window.title(f"放大视图 - {self.caps[idx]['name']}") self.large_window.geometry("1280x720") canvas = Canvas(self.large_window, bg='black') canvas.pack(fill=tk.BOTH, expand=True) photo_obj = {} def update_large_once(): if not self.running or not self.large_window.winfo_exists(): return try: # 获取窗口大小 window_width = self.large_window.winfo_width() if self.large_window.winfo_width() > 1 else 1280 window_height = self.large_window.winfo_height() if self.large_window.winfo_height() > 1 else 720 if idx in self.frames and self.frames[idx] is not None: canvas.delete("all") frame = self.frames[idx] h, w = frame.shape[:2] # 调整到窗口大小 scale = min(window_width / w, window_height / h) new_w = int(w * scale) new_h = int(h * scale) resized_frame = cv2.resize(frame, (new_w, new_h)) pil_image = Image.fromarray(resized_frame) photo = ImageTk.PhotoImage(image=pil_image) photo_obj['img'] = photo canvas.create_image(window_width // 2, window_height // 2, image=photo, anchor='center') else: # 显示等待提示 canvas.delete("all") canvas.create_text( window_width // 2, window_height // 2, text="等待画面...", fill='gray', font=('Arial', 16) ) except Exception as e: print(f"更新大窗口错误: {e}") import traceback traceback.print_exc() self.large_window.after(33, update_large_once) self.large_window.after(33, update_large_once) def run(self): """运行预览""" self.init_cameras() if not self.caps: # 如果没有加载任何采集卡，仍然显示窗口并显示错误信息 print("⚠️ 没有可用的采集卡，将显示错误提示窗口") # 启动捕获线程 if self.caps: capture_thread = threading.Thread(target=self.capture_frames, daemon=True) capture_thread.start() # 创建并显示网格窗口 import time time.sleep(0.5) # 等待几帧 self.create_grid_window() if __name__ == "__main__": preview = PreviewWindow() preview.run()