midlajvalappil/Real-time_Object_Detection_with_YOLO

1"""2Error Handler Module3Provides comprehensive error handling and recovery mechanisms.4"""5 6import logging7import traceback8import functools9from typing import Any, Callable, Optional10import cv211import numpy as np12 13# Configure logging14logging.basicConfig(level=logging.INFO)15logger = logging.getLogger(__name__)16 17class DetectionError(Exception):18    """Custom exception for detection-related errors."""19    pass20 21class CameraError(Exception):22    """Custom exception for camera-related errors."""23    pass24 25class ModelError(Exception):26    """Custom exception for model-related errors."""27    pass28 29def handle_exceptions(default_return=None, log_error=True):30    """31    Decorator for handling exceptions in functions.32    33    Args:34        default_return: Default value to return on exception35        log_error (bool): Whether to log the error36    """37    def decorator(func: Callable) -> Callable:38        @functools.wraps(func)39        def wrapper(*args, **kwargs):40            try:41                return func(*args, **kwargs)42            except Exception as e:43                if log_error:44                    logger.error(f"Error in {func.__name__}: {str(e)}")45                    logger.debug(traceback.format_exc())46                return default_return47        return wrapper48    return decorator49 50class ErrorHandler:51    """52    Centralized error handling and recovery system.53    """54    55    def __init__(self):56        self.error_counts = {}57        self.max_retries = 358        self.recovery_strategies = {59            'camera_error': self._recover_camera,60            'model_error': self._recover_model,61            'detection_error': self._recover_detection62        }63    64    def handle_error(self, error_type: str, error: Exception, context: dict = None) -> bool:65        """66        Handle an error with appropriate recovery strategy.67        68        Args:69            error_type (str): Type of error70            error (Exception): The exception that occurred71            context (dict): Additional context information72            73        Returns:74            bool: True if recovery was successful, False otherwise75        """76        # Log the error77        logger.error(f"{error_type}: {str(error)}")78        79        # Track error count80        self.error_counts[error_type] = self.error_counts.get(error_type, 0) + 181        82        # Check if we've exceeded max retries83        if self.error_counts[error_type] > self.max_retries:84            logger.error(f"Max retries exceeded for {error_type}")85            return False86        87        # Try recovery strategy88        if error_type in self.recovery_strategies:89            try:90                return self.recovery_strategies[error_type](error, context)91            except Exception as recovery_error:92                logger.error(f"Recovery failed for {error_type}: {str(recovery_error)}")93                return False94        95        return False96    97    def _recover_camera(self, error: Exception, context: dict = None) -> bool:98        """99        Attempt to recover from camera errors.100        101        Args:102            error (Exception): The camera error103            context (dict): Context information104            105        Returns:106            bool: True if recovery successful107        """108        logger.info("Attempting camera recovery...")109        110        if context and 'webcam' in context:111            webcam = context['webcam']112            113            # Try to reinitialize camera114            try:115                webcam.stop_capture()116                return webcam.start_capture()117            except Exception as e:118                logger.error(f"Camera recovery failed: {str(e)}")119                return False120        121        return False122    123    def _recover_model(self, error: Exception, context: dict = None) -> bool:124        """125        Attempt to recover from model errors.126        127        Args:128            error (Exception): The model error129            context (dict): Context information130            131        Returns:132            bool: True if recovery successful133        """134        logger.info("Attempting model recovery...")135        136        if context and 'detector' in context:137            detector = context['detector']138            139            # Try to reload model140            try:141                return detector.load_model()142            except Exception as e:143                logger.error(f"Model recovery failed: {str(e)}")144                return False145        146        return False147    148    def _recover_detection(self, error: Exception, context: dict = None) -> bool:149        """150        Attempt to recover from detection errors.151        152        Args:153            error (Exception): The detection error154            context (dict): Context information155            156        Returns:157            bool: True if recovery successful158        """159        logger.info("Attempting detection recovery...")160        161        # For detection errors, we can try reducing confidence threshold162        if context and 'detector' in context:163            detector = context['detector']164            current_threshold = detector.confidence_threshold165            166            if current_threshold > 0.1:167                new_threshold = max(0.1, current_threshold - 0.1)168                detector.update_confidence_threshold(new_threshold)169                logger.info(f"Reduced confidence threshold to {new_threshold}")170                return True171        172        return False173    174    def reset_error_counts(self):175        """Reset all error counts."""176        self.error_counts.clear()177        logger.info("Error counts reset")178    179    def get_error_summary(self) -> dict:180        """181        Get summary of errors encountered.182        183        Returns:184            dict: Error summary185        """186        return {187            'error_counts': self.error_counts.copy(),188            'total_errors': sum(self.error_counts.values()),189            'error_types': list(self.error_counts.keys())190        }191 192class SafeDetector:193    """194    Wrapper for YOLO detector with error handling and fallbacks.195    """196    197    def __init__(self, detector, error_handler: ErrorHandler):198        self.detector = detector199        self.error_handler = error_handler200        self.fallback_frame = None201    202    @handle_exceptions(default_return=[])203    def detect_objects(self, image: np.ndarray) -> list:204        """205        Safe object detection with error handling.206        207        Args:208            image (np.ndarray): Input image209            210        Returns:211            list: List of detections or empty list on error212        """213        try:214            return self.detector.detect_objects(image)215        except Exception as e:216            # Try to recover217            context = {'detector': self.detector}218            if self.error_handler.handle_error('detection_error', e, context):219                # Retry detection after recovery220                return self.detector.detect_objects(image)221            else:222                raise DetectionError(f"Detection failed: {str(e)}")223    224    @handle_exceptions(default_return=None)225    def draw_detections(self, image: np.ndarray, detections: list) -> Optional[np.ndarray]:226        """227        Safe detection drawing with error handling.228        229        Args:230            image (np.ndarray): Input image231            detections (list): List of detections232            233        Returns:234            Optional[np.ndarray]: Annotated image or None on error235        """236        try:237            return self.detector.draw_detections(image, detections)238        except Exception as e:239            logger.error(f"Error drawing detections: {str(e)}")240            return image  # Return original image as fallback241 242class SafeWebcam:243    """244    Wrapper for webcam capture with error handling and fallbacks.245    """246    247    def __init__(self, webcam, error_handler: ErrorHandler):248        self.webcam = webcam249        self.error_handler = error_handler250        self.last_good_frame = None251    252    @handle_exceptions(default_return=None)253    def get_frame(self) -> Optional[np.ndarray]:254        """255        Safe frame capture with error handling.256        257        Returns:258            Optional[np.ndarray]: Frame or None on error259        """260        try:261            frame = self.webcam.get_frame()262            if frame is not None:263                self.last_good_frame = frame.copy()264                return frame265            else:266                # Try to recover camera267                context = {'webcam': self.webcam}268                if self.error_handler.handle_error('camera_error', CameraError("No frame received"), context):269                    return self.webcam.get_frame()270                else:271                    # Return last good frame as fallback272                    return self.last_good_frame273        except Exception as e:274            context = {'webcam': self.webcam}275            if self.error_handler.handle_error('camera_error', e, context):276                return self.webcam.get_frame()277            else:278                return self.last_good_frame279    280    def get_fps(self) -> float:281        """Get FPS with error handling."""282        try:283            return self.webcam.get_fps()284        except Exception:285            return 0.0286    287    def is_camera_available(self) -> bool:288        """Check camera availability with error handling."""289        try:290            return self.webcam.is_camera_available()291        except Exception:292            return False293 294def create_fallback_frame(width: int = 640, height: int = 480, message: str = "Camera Error") -> np.ndarray:295    """296    Create a fallback frame to display when camera fails.297    298    Args:299        width (int): Frame width300        height (int): Frame height301        message (str): Error message to display302        303    Returns:304        np.ndarray: Fallback frame305    """306    frame = np.zeros((height, width, 3), dtype=np.uint8)307    308    # Add error message309    font = cv2.FONT_HERSHEY_SIMPLEX310    font_scale = 1311    color = (0, 0, 255)  # Red312    thickness = 2313    314    # Get text size315    text_size = cv2.getTextSize(message, font, font_scale, thickness)[0]316    317    # Center the text318    text_x = (width - text_size[0]) // 2319    text_y = (height + text_size[1]) // 2320    321    cv2.putText(frame, message, (text_x, text_y), font, font_scale, color, thickness)322    323    return frame324 325# Global error handler instance326global_error_handler = ErrorHandler()327