midlajvalappil/Real-time_Object_Detection_with_YOLO

1"""2Webcam Capture Module3Handles webcam initialization, frame capture, and video processing.4"""5 6import cv27import numpy as np8import threading9import time10from typing import Optional, Callable, Tuple11import logging12 13# Configure logging14logging.basicConfig(level=logging.INFO)15logger = logging.getLogger(__name__)16 17class WebcamCapture:18    """19    Webcam capture class for real-time video processing.20    """21    22    def __init__(self, camera_index: int = 0, width: int = 640, height: int = 480):23        """24        Initialize the webcam capture.25        26        Args:27            camera_index (int): Camera index (usually 0 for default camera)28            width (int): Frame width29            height (int): Frame height30        """31        self.camera_index = camera_index32        self.width = width33        self.height = height34        self.cap = None35        self.is_running = False36        self.current_frame = None37        self.frame_lock = threading.Lock()38        self.capture_thread = None39        self.fps_counter = FPSCounter()40        41    def initialize_camera(self) -> bool:42        """43        Initialize the camera.44        45        Returns:46            bool: True if camera initialized successfully, False otherwise47        """48        try:49            logger.info(f"Initializing camera {self.camera_index}")50            self.cap = cv2.VideoCapture(self.camera_index)51            52            if not self.cap.isOpened():53                logger.error(f"Failed to open camera {self.camera_index}")54                return False55            56            # Set camera properties57            self.cap.set(cv2.CAP_PROP_FRAME_WIDTH, self.width)58            self.cap.set(cv2.CAP_PROP_FRAME_HEIGHT, self.height)59            self.cap.set(cv2.CAP_PROP_FPS, 30)60            61            # Test frame capture62            ret, frame = self.cap.read()63            if not ret:64                logger.error("Failed to capture test frame")65                self.cap.release()66                return False67            68            logger.info(f"Camera initialized successfully. Frame size: {frame.shape}")69            return True70            71        except Exception as e:72            logger.error(f"Error initializing camera: {str(e)}")73            return False74    75    def start_capture(self) -> bool:76        """77        Start the video capture in a separate thread.78        79        Returns:80            bool: True if capture started successfully, False otherwise81        """82        if self.is_running:83            logger.warning("Capture is already running")84            return True85        86        if not self.initialize_camera():87            return False88        89        self.is_running = True90        self.capture_thread = threading.Thread(target=self._capture_loop, daemon=True)91        self.capture_thread.start()92        93        logger.info("Video capture started")94        return True95    96    def stop_capture(self):97        """98        Stop the video capture.99        """100        if not self.is_running:101            return102        103        logger.info("Stopping video capture")104        self.is_running = False105        106        if self.capture_thread:107            self.capture_thread.join(timeout=2.0)108        109        if self.cap:110            self.cap.release()111            self.cap = None112        113        logger.info("Video capture stopped")114    115    def _capture_loop(self):116        """117        Main capture loop running in a separate thread.118        """119        while self.is_running and self.cap and self.cap.isOpened():120            try:121                ret, frame = self.cap.read()122                if ret:123                    with self.frame_lock:124                        self.current_frame = frame.copy()125                    self.fps_counter.update()126                else:127                    logger.warning("Failed to capture frame")128                    time.sleep(0.01)  # Small delay to prevent busy waiting129                    130            except Exception as e:131                logger.error(f"Error in capture loop: {str(e)}")132                break133    134    def get_frame(self) -> Optional[np.ndarray]:135        """136        Get the current frame.137        138        Returns:139            Optional[np.ndarray]: Current frame or None if no frame available140        """141        with self.frame_lock:142            return self.current_frame.copy() if self.current_frame is not None else None143    144    def get_fps(self) -> float:145        """146        Get the current FPS.147        148        Returns:149            float: Current FPS150        """151        return self.fps_counter.get_fps()152    153    def is_camera_available(self) -> bool:154        """155        Check if camera is available and working.156        157        Returns:158            bool: True if camera is available, False otherwise159        """160        return self.is_running and self.cap is not None and self.cap.isOpened()161    162    def get_frame_size(self) -> Tuple[int, int]:163        """164        Get the frame size.165        166        Returns:167            Tuple[int, int]: (width, height) of frames168        """169        return (self.width, self.height)170    171    def __del__(self):172        """173        Destructor to ensure proper cleanup.174        """175        self.stop_capture()176 177 178class FPSCounter:179    """180    FPS counter for measuring frame rate.181    """182    183    def __init__(self, window_size: int = 30):184        """185        Initialize FPS counter.186        187        Args:188            window_size (int): Number of frames to average over189        """190        self.window_size = window_size191        self.frame_times = []192        self.last_time = time.time()193    194    def update(self):195        """196        Update the FPS counter with a new frame.197        """198        current_time = time.time()199        self.frame_times.append(current_time - self.last_time)200        self.last_time = current_time201        202        # Keep only the last window_size frame times203        if len(self.frame_times) > self.window_size:204            self.frame_times.pop(0)205    206    def get_fps(self) -> float:207        """208        Get the current FPS.209        210        Returns:211            float: Current FPS212        """213        if len(self.frame_times) < 2:214            return 0.0215        216        avg_frame_time = sum(self.frame_times) / len(self.frame_times)217        return 1.0 / avg_frame_time if avg_frame_time > 0 else 0.0218 219 220def test_camera_availability(camera_index: int = 0) -> bool:221    """222    Test if a camera is available.223    224    Args:225        camera_index (int): Camera index to test226        227    Returns:228        bool: True if camera is available, False otherwise229    """230    try:231        cap = cv2.VideoCapture(camera_index)232        if cap.isOpened():233            ret, _ = cap.read()234            cap.release()235            return ret236        return False237    except Exception:238        return False239 240 241def get_available_cameras(max_cameras: int = 5) -> list:242    """243    Get list of available camera indices.244    245    Args:246        max_cameras (int): Maximum number of cameras to check247        248    Returns:249        list: List of available camera indices250    """251    available_cameras = []252    for i in range(max_cameras):253        if test_camera_availability(i):254            available_cameras.append(i)255    return available_cameras256