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

+-------------------+      +--------------------------------+      +---------------------------+
| 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.

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 .stopped:
              .cap   .cap.isOpened():
                logger.warning()
                time.sleep()
                ._connect()
                

            grabbed, frame = .cap.read()
              grabbed:
                logger.warning()
                .cap.release()
                time.sleep()
                ._connect()
                

            
             .frame_queue.full():
                :
                    .frame_queue.get_nowait()
                 queue.Empty:
                    

            .frame_queue.put(frame)

     ():
        
        :
             , .frame_queue.get(timeout=)
         queue.Empty:
             , 

     ():
        .stopped = 
         .thread.is_alive():
            .thread.join()
         .cap:
            .cap.release()
        logger.info()

---

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.

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.

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_mm = np.linalg.norm(tvec)

         {
            : (tvec[][]),
            : (tvec[][]),
            : (tvec[][]),
            : (distance_mm),
            : rmat
        }

---

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

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)
            .old_gray = frame_gray.copy()
             frame

        
        p1, st, err = cv2.calcOpticalFlowPyrLK(.old_gray, frame_gray, .p0, , **.lk_params)

        
         p1   :
            good_new = p1[st == ]
            good_old = .p0[st == ]

            
             new, old  (good_new, good_old):
                a, b = new.ravel()
                c, d = old.ravel()
                frame = cv2.line(frame, ((a), (b)), ((c), (d)), (, , ), )
                frame = cv2.circle(frame, ((a), (b)), , (, , ), -)

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

         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.