Opencv Computer Vision

---
name: opencv-computer-vision
metadata:
  category: Computer Vision and Spatial AI
description: >-
  Implement production-grade computer vision pipelines using OpenCV (C++ and Python) with CUDA acceleration.
  Triggers when processing real-time multi-threaded RTSP camera streams, zero-copy memory transfers, image pre-processing,
  contour analysis, optical flow tracking, feature detection/matching (ORB, SIFT), camera calibration, or 3D spatial pose estimation (PnP).
compatibility: OpenCV (>= 4.8.0 with CUDA support), Python (>= 3.9) / C++17, NumPy, CUDA (>= 11.8)
---

# OpenCV Computer Vision & Spatial AI

Production implementations for GPU-accelerated image processing, multi-threaded RTSP video streaming, feature tracking, camera calibration, and 3D spatial position estimation.

---

## 1. Computer Vision Architecture

```text
+-------------------+      +--------------------------------+      +---------------------------+
| RTSP IP Camera    | ---> | Multi-Threaded VideoReader     | ---> | CUDA GpuMat Memory Transfer|
| (1080p @ 60 FPS)  |      | (Thread-safe Queue Buffer)     |      | (Zero CPU-GPU copy bottleneck)|
+-------------------+      +--------------------------------+      +---------------------------+
                                                                                 |
                                                                                 v
+-------------------+      +--------------------------------+      +---------------------------+
| Spatial 3D Pose   | <--- | Feature Detection & PnP        | <--- | GPU Pre-processing        |
| (X, Y, Z, R, P, Y)|      | (SolvePnP / ArUco Marker)      |      | (Threshold, Blur, Contours)|
+-------------------+      +--------------------------------+      +---------------------------+
```

---

## 2. Multi-Threaded RTSP Stream Reader (`rtsp_reader.py`)

A non-blocking, thread-safe RTSP video ingest pipeline with automatic reconnect logic to prevent frame dropping.

```python
import cv2
import threading
import queue
import time
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class RobustVideoSubscriber:
    def __init__(self, rtsp_url: str, max_queue_size: int = 5):
        self.rtsp_url = rtsp_url
        self.frame_queue = queue.Queue(maxsize=max_queue_size)
        self.stopped = False
        self.cap = None
        self.thread = threading.Thread(target=self._update, daemon=True)

    def start(self):
        self._connect()
        self.thread.start()
        return self

    def _connect(self):
        logger.info(f"Connecting to RTSP stream: {self.rtsp_url}")
        self.cap = cv2.VideoCapture(self.rtsp_url, cv2.CAP_FFMPEG)
        self.cap.set(cv2.CAP_PROP_BUFFERSIZE, 1) # Force minimal buffering latency

    def _update(self):
        while not self.stopped:
            if not self.cap or not self.cap.isOpened():
                logger.warning("Stream disconnected. Attempting reconnect...")
                time.sleep(2.0)
                self._connect()
                continue

            grabbed, frame = self.cap.read()
            if not grabbed:
                logger.warning("Failed to grab frame. Reconnecting...")
                self.cap.release()
                time.sleep(1.0)
                self._connect()
                continue

            # Drop oldest frame if queue is full to enforce real-time processing
            if self.frame_queue.full():
                try:
                    self.frame_queue.get_nowait()
                except queue.Empty:
                    pass

            self.frame_queue.put(frame)

    def read(self):
        """Fetch latest frame non-blocking."""
        try:
            return True, self.frame_queue.get(timeout=1.0)
        except queue.Empty:
            return False, None

    def stop(self):
        self.stopped = True
        if self.thread.is_alive():
            self.thread.join()
        if self.cap:
            self.cap.release()
        logger.info("RTSP subscriber stopped.")
```

---

## 3. CUDA GPU-Accelerated Image Pre-Processing (`gpu_pipeline.py`)

Use OpenCV's `cuda::GpuMat` interface to perform image filtering and thresholding on NVIDIA GPUs.

```python
import cv2
import numpy as np

def process_frame_cuda(cpu_frame: np.ndarray) -> np.ndarray:
    """Uploads frame to GPU, performs processing, and downloads result."""
    # 1. Upload CPU NumPy Mat to GPU GpuMat
    gpu_frame = cv2.cuda_GpuMat()
    gpu_frame.upload(cpu_frame)

    # 2. BGR to Grayscale on GPU
    gpu_gray = cv2.cuda.cvtColor(gpu_frame, cv2.COLOR_BGR2GRAY)

    # 3. Gaussian Blur on GPU
    gpu_filter = cv2.cuda.createGaussianFilter(cv2.CV_8UC1, cv2.CV_8UC1, (5, 5), 1.5)
    gpu_blurred = gpu_filter.apply(gpu_gray)

    # 4. Binary Thresholding on GPU
    _, gpu_thresh = cv2.cuda.threshold(gpu_blurred, 127, 255, cv2.THRESH_BINARY)

    # 5. Download result back to CPU memory
    output_frame = gpu_thresh.download()
    return output_frame
```

---

## 4. 3D Spatial Pose Estimation with SolvePnP (`spatial_pose.py`)

Compute the 3D position and orientation $(X, Y, Z)$ of a physical object relative to the camera lens.

```python
import cv2
import numpy as np

class SpatialPoseEstimator:
    def __init__(self, camera_matrix: np.ndarray, dist_coeffs: np.ndarray):
        self.camera_matrix = camera_matrix
        self.dist_coeffs = dist_coeffs

        # Define 3D model points of a rectangular object in physical world space (in millimeters)
        self.model_3d_points = np.array([
            [-50.0, -25.0, 0.0],  # Top-left
            [ 50.0, -25.0, 0.0],  # Top-right
            [ 50.0,  25.0, 0.0],  # Bottom-right
            [-50.0,  25.0, 0.0]   # Bottom-left
        ], dtype=np.float32)

    def estimate_pose(self, image_2d_points: np.ndarray):
        """
        Calculates Rotation vector (rvec) and Translation vector (tvec).
        image_2d_points: Array of 4 detected corner pixels [[x1, y1], [x2, y2], ...]
        """
        success, rvec, tvec = cv2.solvePnP(
            self.model_3d_points,
            image_2d_points.astype(np.float32),
            self.camera_matrix,
            self.dist_coeffs,
            flags=cv2.SOLVEPNP_ITERATIVE
        )

        if not success:
            return None

        # Convert rotation vector to 3x3 rotation matrix
        rmat, _ = cv2.Rodrigues(rvec)

        # Distance from camera lens in millimeters
        distance_mm = np.linalg.norm(tvec)

        return {
            "x_mm": float(tvec[0][0]),
            "y_mm": float(tvec[1][0]),
            "z_mm": float(tvec[2][0]),
            "distance_mm": float(distance_mm),
            "rotation_matrix": rmat
        }
```

---

## 5. Feature Detection & Optical Flow Object Tracking (`feature_tracking.py`)

```python
import cv2
import numpy as np

class LucasKanadeTracker:
    def __init__(self):
        # Parameters for ShiTomasi corner detection
        self.feature_params = dict(
            maxCorners=100,
            qualityLevel=0.3,
            minDistance=7,
            blockSize=7
        )
        # Parameters for Lucas-Kanade optical flow
        self.lk_params = dict(
            winSize=(15, 15),
            maxLevel=2,
            criteria=(cv2.TERMCRITERIA_EPS | cv2.TERMCRITERIA_COUNT, 10, 0.03)
        )
        self.old_gray = None
        self.p0 = None

    def track(self, frame: np.ndarray):
        frame_gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)

        if self.old_gray is None or self.p0 is None or len(self.p0) < 10:
            # Initialize keypoints to track
            self.p0 = cv2.goodFeaturesToTrack(frame_gray, mask=None, **self.feature_params)
            self.old_gray = frame_gray.copy()
            return frame

        # Calculate Optical Flow
        p1, st, err = cv2.calcOpticalFlowPyrLK(self.old_gray, frame_gray, self.p0, None, **self.lk_params)

        # Select good points
        if p1 is not None:
            good_new = p1[st == 1]
            good_old = self.p0[st == 1]

            # Draw vectors
            for new, old in zip(good_new, good_old):
                a, b = new.ravel()
                c, d = old.ravel()
                frame = cv2.line(frame, (int(a), int(b)), (int(c), int(d)), (0, 255, 0), 2)
                frame = cv2.circle(frame, (int(a), int(b)), 4, (0, 0, 255), -1)

            self.old_gray = frame_gray.copy()
            self.p0 = good_new.reshape(-1, 1, 2)

        return frame
```

---

## 6. Performance Benchmarks & Optimization

1. **Avoid Frequent CPU-GPU Transfers**: Maintain processing in `cuda::GpuMat` memory across sequential operations; convert back to CPU `np.ndarray` only when displaying or storing frames.
2. **Pre-allocated Output Buffers**: Pass existing arrays as output parameters (`cv2.cvtColor(src, code, dst=existing_mat)`) inside high-frequency loops to prevent memory re-allocations.
3. **Camera Matrix Calibration**: Always calibrate camera intrinsics using `cv2.calibrateCamera` with chessboard patterns prior to 3D spatial pose estimation tasks.