Detect Surface Defects in Production Video with Anomalib and Supervision

"""
Production implementation for detecting surface defects in videos using Anomalib and Supervision.
Provides secure, scalable operations with data validation, logging, and error handling.
"""
from typing import Dict, Any, List
import os
import logging
import cv2
import time
import numpy as np
from anomalib import AnomalyDetector

# Configure logging for the application
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class Config:
    """Configuration class to manage environment variables."""
    video_source: str = os.getenv('VIDEO_SOURCE', 'video.mp4')
    model_path: str = os.getenv('MODEL_PATH', 'model.pth')
    output_path: str = os.getenv('OUTPUT_PATH', 'output/')

def validate_input(data: Dict[str, Any]) -> bool:
    """Validate input data for processing.
    
    Args:
        data: Input to validate
    Returns:
        True if valid
    Raises:
        ValueError: If validation fails
    """
    if 'video_file' not in data:
        raise ValueError('Missing video_file key in input data')
    return True

def normalize_data(frame: np.ndarray) -> np.ndarray:
    """Normalize the video frame for anomaly detection.
    
    Args:
        frame: The video frame to normalize
    Returns:
        Normalized frame
    Raises:
        Exception: If normalization fails
    """
    try:
        return cv2.normalize(frame, None, 0, 255, cv2.NORM_MINMAX)
    except Exception as e:
        logger.error(f'Error in normalizing data: {e}')
        raise

def process_frame(detector: AnomalyDetector, frame: np.ndarray) -> Dict[str, Any]:
    """Process a single video frame for defects.
    
    Args:
        detector: Anomaly detector instance
        frame: The video frame to process
    Returns:
        A dictionary with detection results
    Raises:
        Exception: If processing fails
    """
    try:
        normalized_frame = normalize_data(frame)  # Normalize the frame
        result = detector.predict(normalized_frame)  # Predict anomalies
        return {'frame': frame, 'result': result}
    except Exception as e:
        logger.error(f'Error processing frame: {e}')
        raise

def save_results(results: List[Dict[str, Any]], output_path: str) -> None:
    """Save detection results to output directory.
    
    Args:
        results: List of detection results
        output_path: Path to save results
    Raises:
        Exception: If saving fails
    """
    try:
        for idx, result in enumerate(results):
            output_file = os.path.join(output_path, f'result_{idx}.png')
            cv2.imwrite(output_file, result['frame'])  # Save the frame
        logger.info('Results saved successfully.')
    except Exception as e:
        logger.error(f'Error saving results: {e}')
        raise

def fetch_video_frames(video_path: str) -> List[np.ndarray]:
    """Extract frames from the video source.
    
    Args:
        video_path: Path of the video to process
    Returns:
        List of video frames
    Raises:
        Exception: If video cannot be read
    """
    frames = []
    try:
        cap = cv2.VideoCapture(video_path)
        while cap.isOpened():
            ret, frame = cap.read()
            if not ret:
                break
            frames.append(frame)
        cap.release()
        logger.info('Frames extracted successfully.')
    except Exception as e:
        logger.error(f'Error fetching video frames: {e}')
        raise
    return frames

class SurfaceDefectDetector:
    """Main class to orchestrate defect detection workflow."""
    def __init__(self, config: Config):
        self.config = config
        self.detector = AnomalyDetector.load(self.config.model_path)

    def run(self) -> None:
        """Execute the detection workflow."""
        try:
            frames = fetch_video_frames(self.config.video_source)  # Fetch video frames
            results = []
            for frame in frames:
                result = process_frame(self.detector, frame)  # Process each frame
                results.append(result)
            save_results(results, self.config.output_path)  # Save results
        except Exception as e:
            logger.error(f'Error in the detection workflow: {e}')

if __name__ == '__main__':
    config = Config()  # Load configuration
    detector = SurfaceDefectDetector(config)  # Instantiate detector
    detector.run()  # Execute detection workflow