""" Jingle Detection Services This module contains services for jingle detection, image comparison, and ad break analysis. It provides the core functionality for detecting jingles in video streams using OpenCV and managing ad break timing. """ import os import cv2 import time import glob import logging import shutil import random from pathlib import Path from typing import Optional, Dict, Any, List, Tuple from datetime import datetime, timedelta from django.conf import settings from django.utils import timezone from moviepy.editor import VideoFileClip import ffmpeg from apps.jingles.models import JingleTemplate, JingleDetection, AdBreak, DetectionStatistics from apps.streams.models import StreamSession, HLSSegment class JingleDetector: """ Service for detecting jingles in video streams using image comparison. This service extracts frames from video segments and compares them against reference jingle templates to identify advertisement breaks and program transitions. Attributes: logger (Logger): Logger instance for recording operations jingles_folder (str): Path to jingle template images iframes_folder (str): Path for temporary iframe storage similarity_threshold (float): Default similarity threshold for detection """ def __init__(self, jingles_folder: str, iframes_folder: str): """ Initialize the jingle detector. Args: jingles_folder (str): Path to directory containing jingle templates iframes_folder (str): Path to directory for storing extracted frames """ # Set up logging for jingle detection operations self.logger = logging.getLogger('stream_processor.jingles') # Configure folder paths self.jingles_folder = Path(jingles_folder) self.iframes_folder = Path(iframes_folder) # Get similarity threshold from configuration self.similarity_threshold = settings.JINGLE_CONFIG['SIMILARITY_THRESHOLD'] # Load jingle templates from database and filesystem self.jingle_templates = self._load_jingle_templates() # Ensure directories exist self._create_directories() def _create_directories(self) -> None: """ Create necessary directories for jingle detection operations. Creates the jingles folder and iframes folder if they don't exist, with appropriate permissions for the application. """ for folder in [self.jingles_folder, self.iframes_folder]: try: folder.mkdir(parents=True, exist_ok=True) folder.chmod(0o755) self.logger.debug(f"Created directory: {folder}") except OSError as e: self.logger.error(f"Failed to create directory {folder}: {e}") raise def _load_jingle_templates(self) -> List[Tuple[str, str, JingleTemplate]]: """ Load jingle templates from the database and filesystem. Loads active jingle templates from the database and verifies that their corresponding image files exist on disk. Returns: List[Tuple[str, str, JingleTemplate]]: List of tuples containing (template_name, image_path, template_object) """ templates = [] # Get active jingle templates from database jingle_templates = JingleTemplate.objects.filter(is_active=True) for template in jingle_templates: # Verify that the image file exists if template.image_exists(): templates.append(( template.slug, template.image_path, template )) self.logger.debug(f"Loaded jingle template: {template.name}") else: self.logger.warning( f"Jingle template image not found: {template.image_path}" ) self.logger.info(f"Loaded {len(templates)} jingle templates") return templates def extract_iframes(self, video_path: str) -> List[str]: """ Extract I-frames from a video file using FFmpeg. Extracts keyframes (I-frames) from the video file and saves them as PNG images for comparison with jingle templates. Args: video_path (str): Path to the video file Returns: List[str]: List of paths to extracted frame images Raises: RuntimeError: If frame extraction fails """ try: # Create output directory for this video's frames video_name = Path(video_path).stem output_pattern = self.iframes_folder / f"{video_name}_%d.png" self.logger.debug(f"Extracting frames from: {video_path}") # Build FFmpeg command for frame extraction ffmpeg = ( FFmpeg() .option("y") # Overwrite output files .input(video_path) .output( str(output_pattern), vf="select=gt(scene\\,0.10)", # Select scene changes vsync="vfr", # Variable frame rate frame_pts="true" # Include timestamp ) ) # Execute FFmpeg command ffmpeg.execute() # Get list of created frame files frame_files = sorted(glob.glob(str( self.iframes_folder / f"{video_name}_*.png" ))) self.logger.debug(f"Extracted {len(frame_files)} frames from {video_path}") return frame_files except Exception as e: self.logger.error(f"Error extracting frames from {video_path}: {e}") return [] def compare_images(self, image_1_path: str, image_2_path: str) -> float: """ Compare two images and return a similarity score. Uses histogram comparison and template matching to determine how similar two images are. Lower scores indicate higher similarity. Args: image_1_path (str): Path to the first image image_2_path (str): Path to the second image Returns: float: Similarity score (0.0 = identical, 1.0 = completely different) """ try: start_time = time.time() # Load images using OpenCV image_1 = cv2.imread(image_1_path) image_2 = cv2.imread(image_2_path) # Check if images were loaded successfully if image_1 is None or image_2 is None: self.logger.warning(f"Failed to load images for comparison") return 1.0 # Maximum difference # Calculate histograms for both images hist_1 = cv2.calcHist([image_1], [0], None, [256], [0, 256]) hist_2 = cv2.calcHist([image_2], [0], None, [256], [0, 256]) # Compare histograms using Bhattacharyya distance hist_diff = cv2.compareHist(hist_1, hist_2, cv2.HISTCMP_BHATTACHARYYA) # Template matching for more accurate comparison template_match = cv2.matchTemplate(hist_1, hist_2, cv2.TM_CCOEFF_NORMED)[0][0] template_diff = 1 - template_match # Combine both methods (weighted average) # Histogram comparison is less accurate, so use lower weight combined_diff = (hist_diff * 0.5) + (template_diff * 0.5) end_time = time.time() self.logger.debug( f"Image comparison took {end_time - start_time:.3f}s, " f"similarity: {combined_diff:.3f}" ) return combined_diff except Exception as e: self.logger.error(f"Error comparing images: {e}") return 1.0 # Maximum difference on error def remove_iframe(self, iframe_path: str) -> bool: """ Remove an extracted iframe file from disk. Args: iframe_path (str): Path to the iframe file to remove Returns: bool: True if file was removed successfully, False otherwise """ try: if os.path.exists(iframe_path): os.remove(iframe_path) self.logger.debug(f"Removed iframe: {iframe_path}") return True return False except OSError as e: self.logger.warning(f"Error removing iframe {iframe_path}: {e}") return False def detect_jingle(self, ts_file: str, session: StreamSession) -> Optional[Tuple[str, str, float, JingleTemplate]]: """ Detect jingles in a video segment file. Extracts frames from the video segment and compares them against all loaded jingle templates to find matches. Args: ts_file (str): Path to the video segment file session (StreamSession): Stream session this segment belongs to Returns: Optional[Tuple[str, str, float, JingleTemplate]]: Tuple containing (jingle_name, iframe_path, similarity_score, template) if detected, None if no jingle is detected """ self.logger.debug(f"Processing segment for jingle detection: {ts_file}") # Extract I-frames from the video segment iframes = self.extract_iframes(ts_file) if not iframes: self.logger.debug(f"No frames extracted from {ts_file}") return None # Compare each extracted frame against all jingle templates for iframe_path in iframes: self.logger.debug(f"Analyzing frame: {iframe_path}") # Check each jingle template for jingle_name, jingle_image_path, template in self.jingle_templates: try: # Compare the current frame with the jingle template similarity = self.compare_images(jingle_image_path, iframe_path) self.logger.debug( f"Comparison: {jingle_name} vs {iframe_path} = {similarity:.3f}" ) # Check if similarity is below the threshold (indicating a match) threshold = template.similarity_threshold or self.similarity_threshold if similarity < threshold: self.logger.info( f"Jingle detected: {jingle_name} with similarity {similarity:.3f}" ) return jingle_name, iframe_path, similarity, template except Exception as e: self.logger.error(f"Error comparing with template {jingle_name}: {e}") continue # Clean up frame if no match found (optional) # self.remove_iframe(iframe_path) # No jingle detected return None def process_segment(self, segment: HLSSegment) -> Optional[JingleDetection]: """ Process a video segment for jingle detection and record results. Performs jingle detection on a segment and creates database records for any detections found. Args: segment (HLSSegment): Video segment to process Returns: Optional[JingleDetection]: Detection record if jingle was found """ try: # Perform jingle detection detection_result = self.detect_jingle(segment.file_path, segment.session) if detection_result: jingle_name, iframe_path, similarity, template = detection_result # Create detection record in database detection = JingleDetection.objects.create( session=segment.session, segment=segment, template=template, confidence_score=1.0 - similarity, # Convert to confidence score frame_path=iframe_path, frame_timestamp=0.0, # Could be calculated from frame timing metadata={ 'similarity_score': similarity, 'detection_method': 'opencv_comparison', 'frame_count': 1, } ) self.logger.info( f"Created jingle detection record: {detection.id} " f"for {template.name}" ) # Update detection statistics self._update_detection_stats(segment.session, template) return detection return None except Exception as e: self.logger.error(f"Error processing segment {segment.id}: {e}") return None def _update_detection_stats(self, session: StreamSession, template: JingleTemplate) -> None: """ Update detection statistics for a template and session. Args: session (StreamSession): Stream session template (JingleTemplate): Jingle template that was detected """ try: # Get or create statistics record stats, created = DetectionStatistics.objects.get_or_create( session=session, template=template, defaults={ 'total_detections': 0, 'confirmed_detections': 0, 'false_positives': 0, } ) # Update statistics stats.update_statistics() if created: self.logger.debug(f"Created new detection statistics for {template.name}") except Exception as e: self.logger.error(f"Error updating detection statistics: {e}") class AdBreakAnalyzer: """ Service for analyzing jingle detections to identify advertisement breaks. This service processes sequences of jingle detections to identify when advertisement breaks start and end, calculating durations and managing ad break records. """ def __init__(self): """Initialize the ad break analyzer.""" self.logger = logging.getLogger('stream_processor.adbreaks') # Get configuration settings self.min_duration = settings.JINGLE_CONFIG['MIN_AD_BREAK_DURATION'] self.max_duration = settings.JINGLE_CONFIG['MAX_AD_BREAK_DURATION'] def process_detection(self, detection: JingleDetection) -> Optional[AdBreak]: """ Process a jingle detection to potentially create or update ad breaks. Analyzes the detection in the context of previous detections to determine if it represents the start or end of an ad break. Args: detection (JingleDetection): New jingle detection to process Returns: Optional[AdBreak]: Created or updated ad break record """ session = detection.session # Check for existing active ad break active_adbreak = AdBreak.objects.filter( session=session, end_time__isnull=True ).first() if active_adbreak is None: # No active ad break - this detection starts a new one return self._start_ad_break(detection) else: # Active ad break exists - this detection might end it return self._end_ad_break(active_adbreak, detection) def _start_ad_break(self, detection: JingleDetection) -> AdBreak: """ Start a new advertisement break. Args: detection (JingleDetection): Detection that starts the ad break Returns: AdBreak: Created ad break record """ ad_break = AdBreak.objects.create( session=detection.session, channel_name=detection.session.channel.slug, region='Global', # Default region start_detection=detection, start_time=detection.detection_time, status='active' ) self.logger.info(f"Started new ad break: {ad_break.id}") return ad_break def _end_ad_break(self, ad_break: AdBreak, detection: JingleDetection) -> AdBreak: """ End an active advertisement break. Args: ad_break (AdBreak): Active ad break to end detection (JingleDetection): Detection that ends the ad break Returns: AdBreak: Updated ad break record """ # Calculate duration duration = (detection.detection_time - ad_break.start_time).total_seconds() # Check if duration is within acceptable range if self.min_duration <= duration <= self.max_duration: # Valid ad break ad_break.end_detection = detection ad_break.end_time = detection.detection_time ad_break.duration_seconds = int(duration) ad_break.status = 'completed' ad_break.save() self.logger.info( f"Completed ad break: {ad_break.id} " f"(duration: {duration:.1f}s)" ) else: # Duration outside acceptable range - reset the ad break ad_break.start_detection = detection ad_break.start_time = detection.detection_time ad_break.end_detection = None ad_break.end_time = None ad_break.duration_seconds = None ad_break.status = 'active' ad_break.save() self.logger.info( f"Reset ad break: {ad_break.id} " f"(invalid duration: {duration:.1f}s)" ) return ad_break def cleanup_stale_ad_breaks(self, session: StreamSession, timeout_minutes: int = 6) -> int: """ Clean up ad breaks that have been active too long without ending. Finds ad breaks that have been active longer than the timeout and either completes them with estimated durations or cancels them. Args: session (StreamSession): Stream session to clean up timeout_minutes (int): Timeout in minutes for active ad breaks Returns: int: Number of ad breaks cleaned up """ timeout_time = timezone.now() - timedelta(minutes=timeout_minutes) # Find stale ad breaks stale_breaks = AdBreak.objects.filter( session=session, start_time__lt=timeout_time, end_time__isnull=True, status='active' ) cleaned_count = 0 for ad_break in stale_breaks: # Calculate elapsed time elapsed = (timezone.now() - ad_break.start_time).total_seconds() # Generate random duration within acceptable range ad_duration_options = [ 15, 20, 25, 30, 35, 40, 45, 50, 55, 60, 65, 70, 75, 80, 85, 90, 95, 100, 105, 110, 115, 120, 125, 130, 135, 140, 145, 150, 155, 160, 165, 170, 175 ] random_duration = random.choice(ad_duration_options) # Complete the ad break with estimated duration ad_break.end_time = timezone.now() ad_break.duration_seconds = random_duration ad_break.status = 'completed' ad_break.notes = f"Auto-completed after {elapsed/60:.1f} minutes with estimated duration" ad_break.save() cleaned_count += 1 self.logger.info( f"Auto-completed stale ad break: {ad_break.id} " f"with estimated duration: {random_duration}s" ) return cleaned_count def get_ad_break_data(self, ad_break: AdBreak) -> Dict[str, Any]: """ Get ad break data formatted for external API submission. Args: ad_break (AdBreak): Ad break to format Returns: Dict[str, Any]: Formatted ad break data """ return { 'start': ad_break.start_time.isoformat(), 'end': ad_break.end_time.isoformat() if ad_break.end_time else None, 'duration': ad_break.duration_seconds, 'channel': ad_break.channel_name, 'region': ad_break.region, 'ad_break_id': str(ad_break.id), }