# vision/camera.py import cv2 import threading import queue import time import numpy as np from datetime import datetime from typing import Optional, Tuple, Dict, Any class DualCameraController: """双摄像头控制器 - 支持多线程捕获和同步帧获取""" def __init__(self, camera_configs: Dict[str, Dict[str, Any]], max_queue_size: int = 10, sync_threshold_ms: float = 50.0): # 摄像头配置 self.camera_configs = camera_configs # 摄像头对象和队列 self.cameras: Dict[str, cv2.VideoCapture] = {} self.frame_queues: Dict[str, queue.Queue] = {} self.capture_threads: Dict[str, threading.Thread] = {} # 线程控制 self.stop_event = threading.Event() self.max_queue_size = max_queue_size self.sync_threshold_ms = sync_threshold_ms self.last_sync_pair: Tuple[Optional[np.ndarray], Optional[np.ndarray]] = (None, None) # 摄像头状态 self.is_running = False def set_camera_config(self, camera_id: str, ip: str, username: str = "admin", password: str = "XJ123456", port: int = 554, channel: int = 1): """设置指定摄像头的配置""" if camera_id in ['cam1', 'cam2']: self.camera_configs[camera_id].update({ 'ip': ip, 'username': username, 'password': password, 'port': port, 'channel': channel }) print(f"摄像头 {camera_id} 配置已更新: IP={ip}") else: raise ValueError(f"无效的摄像头ID: {camera_id}") def _build_rtsp_url(self, camera_id: str) -> str: """构建RTSP URL""" config = self.camera_configs[camera_id] return f"rtsp://{config['username']}:{config['password']}@{config['ip']}:{config['port']}/Streaming/Channels/{config['channel']}01" def _capture_thread(self, camera_id: str): """摄像头捕获线程""" cap = self.cameras[camera_id] q = self.frame_queues[camera_id] rtsp_url = self._build_rtsp_url(camera_id) print(f"启动 {camera_id} 捕获线程") while not self.stop_event.is_set(): try: # print('aaaaa') ret, frame = cap.read() if ret and frame is not None: # 在帧右上角添加时间戳 current_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3] # 获取帧尺寸 height, width = frame.shape[:2] # 设置文字参数 font = cv2.FONT_HERSHEY_SIMPLEX font_scale = 0.6 thickness = 2 color = (0, 255, 0) # 绿色 # 计算文字位置（右上角） text_size = cv2.getTextSize(current_time, font, font_scale, thickness)[0] text_x = width - text_size[0] - 10 # 距离右边10像素 text_y = 30 # 距离顶部30像素 # 添加文字背景（半透明） overlay = frame.copy() cv2.rectangle(overlay, (text_x - 5, text_y - text_size[1] - 5), (text_x + text_size[0] + 5, text_y + 5), (0, 0, 0), -1) cv2.addWeighted(overlay, 0.7, frame, 0.3, 0, frame) # 添加时间戳文字 cv2.putText(frame, current_time, (text_x, text_y), font, font_scale, color, thickness) # 使用高精度时间戳 timestamp = time.time() # 检查队列是否已满 if q.qsize() >= self.max_queue_size: # 队列已满，丢弃最旧帧（FIFO） try: q.get_nowait() # 移除最旧帧 q.put_nowait((timestamp, frame)) except queue.Empty: # 理论上不会发生，但安全处理 pass else: # 队列未满，直接添加 q.put_nowait((timestamp, frame)) else: print(f"{camera_id} 读取失败，重连中...") time.sleep(1) cap.open(rtsp_url) except Exception as e: print(f"{camera_id} 捕获异常: {e}") time.sleep(1) print(f"{camera_id} 捕获线程已停止") def start_cameras(self) -> bool: """启动双摄像头""" if self.is_running: print("摄像头已在运行中") return True try: # 初始化摄像头和队列 for camera_id in ['cam2']: rtsp_url = self._build_rtsp_url(camera_id) cap = cv2.VideoCapture(rtsp_url) if not cap.isOpened(): print(f"无法打开摄像头 {camera_id}: {rtsp_url}") # 清理已打开的摄像头 self.release() return False self.cameras[camera_id] = cap self.frame_queues[camera_id] = queue.Queue(maxsize=self.max_queue_size) print(f"摄像头 {camera_id} 初始化成功: {rtsp_url}") # 启动捕获线程 self.stop_event.clear() for camera_id in ['cam2']: thread = threading.Thread( target=self._capture_thread, args=(camera_id,), daemon=True ) self.capture_threads[camera_id] = thread thread.start() self.is_running = True print("双摄像头系统启动成功") return True except Exception as e: print(f"启动摄像头失败: {e}") self.release() return False def get_latest_frames(self, sync_threshold_ms: Optional[float] = None) -> Optional[Tuple[np.ndarray, np.ndarray]]: """获取最新的同步帧对""" return if not self.is_running: print("摄像头未运行") return None sync_threshold = sync_threshold_ms or self.sync_threshold_ms sync_threshold_sec = sync_threshold / 1000.0 # 检查队列是否有数据 if (self.frame_queues['cam1'].empty() or self.frame_queues['cam2'].empty()): return None try: # 获取最新帧 ts1, f1 = self.frame_queues['cam1'].queue[-1] ts2, f2 = self.frame_queues['cam2'].queue[-1] dt = abs(ts1 - ts2) if dt < sync_threshold_sec: # 时间差在阈值内，认为是同步的 frame1, frame2 = f1.copy(), f2.copy() self.last_sync_pair = (frame1, frame2) return (frame1, frame2) else: # 搜索最近5帧找最小时间差 min_dt = float('inf') best_pair = None # 获取最近5帧 cam1_frames = list(self.frame_queues['cam1'].queue)[-5:] cam2_frames = list(self.frame_queues['cam2'].queue)[-5:] for t1_local, f1_local in cam1_frames: for t2_local, f2_local in cam2_frames: dt_local = abs(t1_local - t2_local) if dt_local < min_dt and dt_local < sync_threshold_sec * 2: # 更宽松的阈值 min_dt = dt_local best_pair = (f1_local.copy(), f2_local.copy()) if best_pair: self.last_sync_pair = best_pair return best_pair else: # 没找到同步帧，返回最新非同步帧 return (f1.copy(), f2.copy()) except Exception as e: print(f"获取帧对失败: {e}") return None def get_single_frame(self, camera_id: str) -> Optional[np.ndarray]: """获取单个摄像头的最新帧""" if not self.is_running: print("摄像头未运行") return None if camera_id not in self.frame_queues: print(f"无效的摄像头ID: {camera_id}") return None try: if not self.frame_queues[camera_id].empty(): _, frame = self.frame_queues[camera_id].queue[-1] return frame.copy() return None except Exception as e: print(f"获取单帧失败: {e}") return None def get_single_latest_frame(self) -> Optional[np.ndarray]: """获取单个摄像头的最新帧""" if not self.is_running: print("摄像头未运行") return None try: frame_latest = None dt_t1 = None # 获取cam1的最新帧 if not self.frame_queues['cam2'].empty(): dt_t1, frame_latest = self.frame_queues['cam2'].queue[-1] # 获取cam2的最新帧，选择时间戳更新的那个 # if frame_latest is None: # if not self.frame_queues['cam2'].empty(): # dt_t2, frame2 = self.frame_queues['cam2'].queue[-1] # if dt_t1 is None or dt_t2 > dt_t1: # frame_latest = frame2 # 返回最新帧的副本（如果找到） return frame_latest.copy() if frame_latest is not None else None except Exception as e: print(f"获取单帧失败: {e}") return None def get_single_latest_frame2(self) -> Optional[np.ndarray]: """获取单个摄像头的最新帧""" if not self.is_running: print("摄像头未运行") return None try: frame_latest = None dt_t1 = None # 获取cam1的最新帧 if not self.frame_queues['cam2'].empty(): dt_t1, frame_latest = self.frame_queues['cam2'].queue[-1] # 获取cam2的最新帧，选择时间戳更新的那个 if frame_latest is None: if not self.frame_queues['cam1'].empty(): dt_t2, frame2 = self.frame_queues['cam1'].queue[-1] if dt_t1 is None or dt_t2 > dt_t1: frame_latest = frame2 # 返回最新帧的副本（如果找到） return frame_latest.copy() if frame_latest is not None else None except Exception as e: print(f"获取单帧失败: {e}") return None def get_notification_frame(self, camera_id: str = None, use_sync: bool = True) -> Optional[np.ndarray]: """根据通知参数获取最近的帧 Args: camera_id: 摄像头ID ('cam1', 'cam2')，如果为None则根据use_sync决定 use_sync: 是否使用同步帧对，如果为True则返回同步帧对，否则返回指定摄像头的单帧 Returns: 单帧图像或同步帧对 """ if not self.is_running: print("摄像头未运行") return None if use_sync: # 获取同步帧对，返回拼接后的图像 frames = self.get_latest_frames() if frames: frame1, frame2 = frames # 调整大小并拼接 h, w = 480, 640 frame1_resized = cv2.resize(frame1, (w, h)) frame2_resized = cv2.resize(frame2, (w, h)) combined = np.hstack((frame1_resized, frame2_resized)) # 添加时间戳信息 ts1 = time.time() cv2.putText(combined, f"Sync: {datetime.fromtimestamp(ts1).strftime('%H:%M:%S.%f')[:-3]}", (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0, 255, 0), 2) return combined return None else: # 获取指定摄像头的单帧 if camera_id is None: camera_id = 'cam1' # 默认返回cam1 return self.get_single_frame(camera_id) def display_live_feed(self): """实时显示双摄像头画面（调试用）""" if not self.is_running: print("请先启动摄像头") return print("按 'q' 退出显示，按 's' 保存同步帧") while True: frame = self.get_notification_frame(use_sync=True) if frame is not None: cv2.imshow("Dual Camera Feed", frame) key = cv2.waitKey(1) & 0xFF if key == ord('q'): break elif key == ord('s'): # 保存同步帧 sync_frames = self.get_latest_frames() if sync_frames: frame1, frame2 = sync_frames timestamp = datetime.now().strftime("%Y%m%d_%H%M%S_%f")[:20] cv2.imwrite(f"cam1_{timestamp}.jpg", frame1) cv2.imwrite(f"cam2_{timestamp}.jpg", frame2) print(f"✅ 保存同步帧: cam1_{timestamp}.jpg & cam2_{timestamp}.jpg") cv2.destroyAllWindows() def release(self): """释放摄像头资源""" print("正在释放摄像头资源...") # 停止捕获线程 if self.is_running: self.stop_event.set() # 等待线程结束 for camera_id, thread in self.capture_threads.items(): if thread.is_alive(): thread.join(timeout=2) print(f"{camera_id} 捕获线程已停止") self.capture_threads.clear() self.is_running = False # 释放摄像头 for camera_id, cap in self.cameras.items(): if cap is not None: cap.release() print(f"摄像头 {camera_id} 已释放") self.cameras.clear() self.frame_queues.clear() print("摄像头资源释放完成") def __del__(self): """析构函数，确保资源释放""" self.release() # 类方法：快速创建和启动 @classmethod def create_and_start(cls, camera_configs: Dict[str, Dict[str, Any]]) -> Optional['DualCameraController']: """快速创建并启动双摄像头控制器""" controller = cls(camera_configs) if controller.start_cameras(): return controller else: return None # 向后兼容的单摄像头控制器 class CameraController: """单摄像头控制器 - 向后兼容""" def __init__(self): self.dual_controller = DualCameraController() self.default_camera = 'cam1' def set_config(self, camera_type="ip", ip=None, port=None, username=None, password=None, channel=1): """设置摄像头配置 - 兼容旧接口""" self.dual_controller.set_camera_config( 'cam1', ip or "192.168.1.51", username or "admin", password or "XJ123456", port or 554, channel ) def setup_capture(self, camera_index=0): """设置摄像头捕获 - 兼容旧接口""" return self.dual_controller.start_cameras() def capture_frame(self): """捕获当前帧 - 兼容旧接口""" return self.dual_controller.capture_frame(self.default_camera) def capture_frame_bak(self): """捕获当前帧（备用） - 兼容旧接口""" return self.dual_controller.capture_frame_bak(self.default_camera) def release(self): """释放摄像头资源""" self.dual_controller.release() def __del__(self): """析构函数""" self.release() # 使用示例和测试 if __name__ == "__main__": # 创建双摄像头控制器 camera_configs = { 'cam2': { 'type': 'ip', 'ip': '192.168.250.61', 'port': 554, 'username': 'admin', 'password': 'XJ123456', 'channel': 1 } } controller = DualCameraController.create_and_start(camera_configs) if controller: print("双摄像头系统启动成功！") # 示例1：获取同步帧对 print("\n=== 获取同步帧 ===") while True: single_frame = controller.get_single_latest_frame() if single_frame is not None: print(f"获取到帧形状: {single_frame.shape}") cv2.imshow("Single Camera Frame", single_frame) else: print("未获取到帧") key = cv2.waitKey(1) & 0xFF if key == ord('s') and single_frame is not None: timestamp = datetime.now().strftime("%Y%m%d_%H%M%S_%f")[:20] cv2.imwrite(f"single_frame_{timestamp}.jpg", single_frame) print(f"✅ 保存单帧: single_frame_{timestamp}.jpg") if key == ord('q'): break time.sleep(1) # controller.get_single_latest_frame('cam2') # sync_frames = controller.get_latest_frames(sync_threshold_ms=50) # if sync_frames: # frame1, frame2 = sync_frames # print(f"获取到同步帧对 - 帧1形状: {frame1.shape}, 帧2形状: {frame2.shape}") # else: # print("未获取到同步帧对") # 示例2：根据通知参数获取帧 # print("\n=== 根据通知参数获取帧 ===") # # 获取同步拼接帧（用于显示） # combined_frame = controller.get_notification_frame(use_sync=True) # if combined_frame is not None: # print(f"获取到同步拼接帧，形状: {combined_frame.shape}") # cv2.imshow("Sync Frame", combined_frame) # cv2.waitKey(1000) # 显示1秒 # cv2.destroyAllWindows() # # 获取单个摄像头帧 # single_frame = controller.get_notification_frame(camera_id='cam1', use_sync=False) # if single_frame is not None: # print(f"获取到cam1单帧，形状: {single_frame.shape}") # # 示例3：实时显示 # print("\n=== 实时显示模式 ===") # print("按 'q' 退出显示，按 's' 保存同步帧") # # controller.display_live_feed() # 取消注释以启用实时显示 # # 示例4：兼容性测试 # print("\n=== 兼容性测试 ===") # old_frame = controller.capture_frame('cam1') # if old_frame is not None: # print(f"旧接口兼容 - 帧形状: {old_frame.shape}") # 清理 controller.release() print("\n摄像头资源已释放") else: print("双摄像头系统启动失败！")