1. 多路视频流接入与统一管理

任务说明

本案例实现一个多路RTSP视频流接入管理器，支持动态添加/移除视频源，并统一跟踪各路的连接状态。核心目标是：给定若干RTSP地址，能够建立连接、读取视频帧、感知断流，并在运行中增删视频源。

环境依赖

需要 opencv-python（含cv2）与Python 3.8+。OpenCV自带RTSP支持，无需额外后端。

完整实现

import cv2
import threading
import time
from dataclasses import dataclass, field
from typing import Dict, Optional

@dataclass
class StreamSource:
    """单个视频源的状态与捕获句柄"""
    url: str
    cap: Optional[cv2.VideoCapture] = None
    connected: bool = False
    last_frame_time: float = 0.0
    error: str = ""
    lock: threading.Lock = field(default_factory=threading.Lock)

class StreamManager:
    """多路RTSP流统一管理器"""
    def __init__(self, timeout_sec: float = 5.0):
        self._streams: Dict[str, StreamSource] = {}
        self._timeout = timeout_sec
        self._monitor_thread: Optional[threading.Thread] = None
        self._running = False

    def add_stream(self, stream_id: str, url: str) -> bool:
        """动态添加视频源，返回是否成功建立连接"""
        if stream_id in self._streams:
            return False
        cap = cv2.VideoCapture(url)
        # 设置RTSP低延迟选项
        cap.set(cv2.CAP_PROP_BUFFERSIZE, 1)
        ok, _ = cap.read()
        src = StreamSource(url=url, cap=cap, connected=ok)
        if not ok:
            src.error = "Failed to open or read stream"
        self._streams[stream_id] = src
        return ok

    def remove_stream(self, stream_id: str) -> bool:
        """动态移除视频源并释放资源"""
        src = self._streams.pop(stream_id, None)
        if src is None:
            return False
        with src.lock:
            if src.cap:
                src.cap.release()
        return True

    def read_frame(self, stream_id: str):
        """读取指定流的一帧，返回(ret, frame)；断流时ret为False"""
        src = self._streams.get(stream_id)
        if src is None or not src.connected:
            return False, None
        with src.lock:
            ret, frame = src.cap.read()
            if ret:
                src.last_frame_time = time.time()
            else:
                src.connected = False
                src.error = "Stream disconnected"
            return ret, frame

    def get_status(self) -> dict:
        """返回所有流的状态快照"""
        return {
            sid: {
                "connected": s.connected,
                "last_frame_age": time.time() - s.last_frame_time if s.last_frame_time else -1,
                "error": s.error
            }
            for sid, s in self._streams.items()
        }

    def _monitor_loop(self):
        """后台监控：定期探测断流并尝试重连"""
        while self._running:
            for sid, src in list(self._streams.items()):
                if not src.connected:
                    # 尝试重连
                    with src.lock:
                        if src.cap:
                            src.cap.release()
                        new_cap = cv2.VideoCapture(src.url)
                        ok, _ = new_cap.read()
                        if ok:
                            src.cap = new_cap
                            src.connected = True
                            src.error = ""
                        else:
                            new_cap.release()
                            src.error = "Reconnect failed"
            time.sleep(self._timeout)

    def start_monitoring(self):
        """启动后台断流检测与重连线程"""
        if self._monitor_thread and self._monitor_thread.is_alive():
            return
        self._running = True
        self._monitor_thread = threading.Thread(target=self._monitor_loop, daemon=True)
        self._monitor_thread.start()

    def stop_monitoring(self):
        self._running = False
        if self._monitor_thread:
            self._monitor_thread.join(timeout=1.0)

    def shutdown(self):
        """释放所有资源"""
        self.stop_monitoring()
        for sid in list(self._streams.keys()):
            self.remove_stream(sid)

# 最小可运行示例
if __name__ == "__main__":
    # 用本地测试视频文件模拟RTSP源（实际RTSP地址替换即可）
    test_sources = {
        "cam1": "sample1.mp4",   # 替换为 rtsp://...
        "cam2": "sample2.mp4",
    }
    mgr = StreamManager(timeout_sec=2.0)
    for sid, url in test_sources.items():
        ok = mgr.add_stream(sid, url)
        print(f"Add {sid}: {'OK' if ok else 'FAIL'}")

    mgr.start_monitoring()
    # 读取几帧验证
    for _ in range(5):
        for sid in test_sources:
            ret, frame = mgr.read_frame(sid)
            if ret:
                print(f"{sid}: frame {frame.shape}")
        time.sleep(0.1)

    # 动态移除一路
    mgr.remove_stream("cam1")
    print("After remove:", mgr.get_status())

    mgr.shutdown()
    print("Shutdown complete")

运行方式

# 准备两个测试视频文件（或用真实RTSP地址替换代码中的URL）
python stream_manager.py

验证方法

• 添加成功：控制台输出 Add cam1: OK，说明连接建立并可读取首帧。
• 帧读取：循环读取时打印 frame (height, width, 3)，说明帧数据正常获取。
• 动态移除：remove_stream("cam1") 后，get_status() 中不再包含 cam1，且资源已释放。
• 断流检测：手动中断视频源（如关闭摄像头），约 timeout_sec 秒后 get_status() 中该路 connected 变为 False，并自动尝试重连。

关键说明

• 代码中 sample1.mp4 为本地文件模拟RTSP；实际部署时替换为 rtsp://user:pass@ip:port/stream 格式地址即可，无需修改其他逻辑。
• 后台监控线程每 timeout_sec 秒检查一次所有流，对断流自动重连，实现“统一管理连接状态”的目标。
• 线程安全通过每路流的 lock 保证，避免读写竞争。