""" Jingle Detection Models for Stream Processor Application This module contains Django models for managing jingle detection, ad break analysis, and related functionality. These models handle the storage and management of jingle templates, detection results, and ad break timing information. """ from django.db import models from django.contrib.auth.models import User from django.utils import timezone from django.core.validators import MinValueValidator, MaxValueValidator from apps.core.models import TimestampedModel, UUIDModel, StatusModel from apps.streams.models import StreamSession, HLSSegment from apps.core.validators import ( validate_file_path, validate_slug_format, validate_name_format, validate_confidence_score, validate_template_content ) import os import json class JingleTemplate(TimestampedModel, UUIDModel): """ Model representing a jingle template for detection. Jingle templates are reference images or video frames that are used to identify specific jingles in stream content. Each template represents a unique jingle that can be detected during stream processing. Attributes: name (CharField): Human-readable name for the jingle slug (SlugField): URL-friendly identifier description (TextField): Description of the jingle content image_path (CharField): Path to the reference image file similarity_threshold (FloatField): Threshold for detection matching is_active (BooleanField): Whether this template is active category (CharField): Category classification for the jingle created_by (ForeignKey): User who uploaded this template """ # Jingle identification name = models.CharField( max_length=100, unique=True, validators=[validate_name_format], help_text="Human-readable name for the jingle" ) slug = models.SlugField( max_length=100, unique=True, validators=[validate_slug_format], help_text="URL-friendly identifier for the jingle" ) description = models.TextField( blank=True, help_text="Description of the jingle content and context" ) # File information image_path = models.CharField( max_length=500, validators=[validate_file_path], help_text="Path to the reference image file for this jingle" ) # Detection configuration similarity_threshold = models.FloatField( default=0.1, validators=[validate_confidence_score], help_text="Similarity threshold for positive detection (0.0-1.0)" ) # Status and categorization is_active = models.BooleanField( default=True, db_index=True, help_text="Whether this jingle template is active for detection" ) category = models.CharField( max_length=50, choices=[ ('ad_start', 'Advertisement Start'), ('ad_end', 'Advertisement End'), ('program_start', 'Program Start'), ('program_end', 'Program End'), ('transition', 'Transition'), ('branding', 'Channel Branding'), ('other', 'Other'), ], default='other', help_text="Category classification for this jingle" ) # Channel association channel = models.ForeignKey( 'streams.Channel', on_delete=models.CASCADE, related_name='jingle_templates', null=True, blank=True, help_text="Channel this jingle template belongs to" ) # User association created_by = models.ForeignKey( User, on_delete=models.CASCADE, related_name='created_jingle_templates', help_text="User who uploaded this jingle template" ) class Meta: verbose_name = "Jingle Template" verbose_name_plural = "Jingle Templates" ordering = ['category', 'name'] indexes = [ models.Index(fields=['slug']), models.Index(fields=['is_active']), models.Index(fields=['category']), ] def __str__(self): """String representation of the jingle template.""" return f"{self.name} ({self.category})" def image_exists(self): """Check if the reference image file exists on disk.""" return os.path.exists(self.image_path) def get_image_size(self): """Get the size of the reference image file.""" try: return os.path.getsize(self.image_path) except OSError: return None def delete_image(self): """Delete the reference image file from disk.""" try: if os.path.exists(self.image_path): os.remove(self.image_path) return True except OSError: pass return False class JingleDetection(TimestampedModel, UUIDModel): """ Model representing a detected jingle instance. This model records when a jingle has been detected in a stream, including the confidence score, frame information, and associated segment data. Attributes: session (ForeignKey): Stream session where detection occurred segment (ForeignKey): HLS segment containing the detection template (ForeignKey): Jingle template that was matched confidence_score (FloatField): Confidence level of the detection frame_path (CharField): Path to the extracted frame image detection_time (DateTimeField): When the detection occurred frame_timestamp (FloatField): Timestamp within the segment is_confirmed (BooleanField): Whether detection has been confirmed metadata (JSONField): Additional detection metadata """ # Detection relationships session = models.ForeignKey( StreamSession, on_delete=models.CASCADE, related_name='jingle_detections', help_text="Stream session where this detection occurred" ) segment = models.ForeignKey( HLSSegment, on_delete=models.CASCADE, related_name='jingle_detections', help_text="HLS segment containing this detection" ) template = models.ForeignKey( JingleTemplate, on_delete=models.CASCADE, related_name='detections', help_text="Jingle template that was matched" ) # Detection quality metrics confidence_score = models.FloatField( validators=[MinValueValidator(0.0), MaxValueValidator(1.0)], help_text="Confidence level of the detection (0.0-1.0)" ) # Frame information frame_path = models.CharField( max_length=500, help_text="Path to the extracted frame image" ) # Timing information detection_time = models.DateTimeField( auto_now_add=True, help_text="When the detection was processed" ) frame_timestamp = models.FloatField( validators=[MinValueValidator(0.0)], help_text="Timestamp within the segment where detection occurred" ) # Verification status is_confirmed = models.BooleanField( default=False, db_index=True, help_text="Whether this detection has been manually confirmed" ) # Additional data metadata = models.JSONField( default=dict, blank=True, help_text="Additional detection metadata and analysis results" ) class Meta: verbose_name = "Jingle Detection" verbose_name_plural = "Jingle Detections" ordering = ['-detection_time'] indexes = [ models.Index(fields=['session', 'detection_time']), models.Index(fields=['template', 'detection_time']), models.Index(fields=['is_confirmed']), models.Index(fields=['confidence_score']), ] def __str__(self): """String representation of the jingle detection.""" return f"{self.template.name} - {self.confidence_score:.3f} ({self.detection_time})" def frame_exists(self): """Check if the extracted frame file exists on disk.""" return os.path.exists(self.frame_path) def get_frame_url(self): """Get URL for accessing the extracted frame.""" # This would be implemented based on your media serving setup return f"/media/frames/{os.path.basename(self.frame_path)}" def delete_frame(self): """Delete the extracted frame file from disk.""" try: if os.path.exists(self.frame_path): os.remove(self.frame_path) return True except OSError: pass return False class AdBreak(TimestampedModel, UUIDModel, StatusModel): """ Model representing a detected advertisement break. Ad breaks are identified by pairs of jingle detections (start and end) and represent commercial interruptions in the stream content. Attributes: session (ForeignKey): Stream session containing the ad break channel_name (CharField): Channel identifier for external APIs region (CharField): Geographic region for the ad break start_detection (ForeignKey): Jingle detection marking the start end_detection (ForeignKey): Jingle detection marking the end start_time (DateTimeField): When the ad break started end_time (DateTimeField): When the ad break ended duration_seconds (PositiveIntegerField): Duration in seconds is_sent_to_api (BooleanField): Whether data was sent to external API api_response (JSONField): Response from external API notes (TextField): Additional notes about the ad break """ # Ad break relationships session = models.ForeignKey( StreamSession, on_delete=models.CASCADE, related_name='ad_breaks', help_text="Stream session containing this ad break" ) # External API integration channel_name = models.CharField( max_length=100, help_text="Channel identifier for external API integration" ) region = models.CharField( max_length=50, choices=[ ('Netherlands', 'Netherlands'), ('France', 'France'), ('Global', 'Global'), ], default='Global', help_text="Geographic region for this ad break" ) # Detection relationships start_detection = models.ForeignKey( JingleDetection, on_delete=models.CASCADE, related_name='ad_breaks_started', null=True, blank=True, help_text="Jingle detection that marked the start of this ad break" ) end_detection = models.ForeignKey( JingleDetection, on_delete=models.CASCADE, related_name='ad_breaks_ended', null=True, blank=True, help_text="Jingle detection that marked the end of this ad break" ) # Timing information start_time = models.DateTimeField( help_text="When the advertisement break started" ) end_time = models.DateTimeField( null=True, blank=True, help_text="When the advertisement break ended" ) duration_seconds = models.PositiveIntegerField( null=True, blank=True, validators=[MinValueValidator(1), MaxValueValidator(3600)], help_text="Duration of the ad break in seconds" ) # External API integration tracking is_sent_to_api = models.BooleanField( default=False, db_index=True, help_text="Whether this ad break data was sent to external API" ) api_response = models.JSONField( default=dict, blank=True, help_text="Response data from external API calls" ) # Additional information notes = models.TextField( blank=True, help_text="Additional notes or observations about this ad break" ) class Meta: verbose_name = "Advertisement Break" verbose_name_plural = "Advertisement Breaks" ordering = ['-start_time'] indexes = [ models.Index(fields=['session', 'start_time']), models.Index(fields=['channel_name', 'region']), models.Index(fields=['is_sent_to_api']), models.Index(fields=['status']), ] def __str__(self): """String representation of the ad break.""" duration_str = f" ({self.duration_seconds}s)" if self.duration_seconds else "" return f"{self.channel_name} - {self.start_time.strftime('%H:%M:%S')}{duration_str}" def calculate_duration(self): """Calculate the duration of the ad break.""" if self.start_time and self.end_time: delta = self.end_time - self.start_time return int(delta.total_seconds()) return None def is_complete(self): """Check if the ad break has both start and end times.""" return self.start_time is not None and self.end_time is not None def save(self, *args, **kwargs): """Override save to automatically calculate duration.""" if self.is_complete() and not self.duration_seconds: self.duration_seconds = self.calculate_duration() super().save(*args, **kwargs) class DetectionStatistics(TimestampedModel, UUIDModel): """ Model for tracking jingle detection statistics and metrics. This model aggregates detection data for reporting and analysis, providing insights into detection accuracy, frequency, and performance. Attributes: session (ForeignKey): Stream session for these statistics template (ForeignKey): Jingle template for these statistics total_detections (PositiveIntegerField): Total number of detections confirmed_detections (PositiveIntegerField): Number of confirmed detections false_positives (PositiveIntegerField): Number of false positive detections avg_confidence (FloatField): Average confidence score min_confidence (FloatField): Minimum confidence score max_confidence (FloatField): Maximum confidence score detection_rate (FloatField): Detections per hour last_detection (DateTimeField): Time of last detection """ # Statistics relationships session = models.ForeignKey( StreamSession, on_delete=models.CASCADE, related_name='detection_statistics', help_text="Stream session for these statistics" ) template = models.ForeignKey( JingleTemplate, on_delete=models.CASCADE, related_name='statistics', help_text="Jingle template for these statistics" ) # Detection counts total_detections = models.PositiveIntegerField( default=0, help_text="Total number of detections for this template" ) confirmed_detections = models.PositiveIntegerField( default=0, help_text="Number of manually confirmed detections" ) false_positives = models.PositiveIntegerField( default=0, help_text="Number of identified false positive detections" ) # Confidence metrics avg_confidence = models.FloatField( null=True, blank=True, validators=[MinValueValidator(0.0), MaxValueValidator(1.0)], help_text="Average confidence score for all detections" ) min_confidence = models.FloatField( null=True, blank=True, validators=[MinValueValidator(0.0), MaxValueValidator(1.0)], help_text="Minimum confidence score recorded" ) max_confidence = models.FloatField( null=True, blank=True, validators=[MinValueValidator(0.0), MaxValueValidator(1.0)], help_text="Maximum confidence score recorded" ) # Performance metrics detection_rate = models.FloatField( null=True, blank=True, validators=[MinValueValidator(0.0)], help_text="Number of detections per hour" ) last_detection = models.DateTimeField( null=True, blank=True, help_text="Timestamp of the most recent detection" ) class Meta: verbose_name = "Detection Statistics" verbose_name_plural = "Detection Statistics" unique_together = ['session', 'template'] ordering = ['-last_detection'] indexes = [ models.Index(fields=['session', 'template']), models.Index(fields=['last_detection']), ] def __str__(self): """String representation of the detection statistics.""" return f"{self.template.name} - {self.total_detections} detections" def accuracy_rate(self): """Calculate the accuracy rate as percentage of confirmed detections.""" if self.total_detections > 0: return (self.confirmed_detections / self.total_detections) * 100 return 0 def false_positive_rate(self): """Calculate the false positive rate as percentage.""" if self.total_detections > 0: return (self.false_positives / self.total_detections) * 100 return 0 def update_statistics(self): """Recalculate statistics based on current detection data.""" detections = JingleDetection.objects.filter( session=self.session, template=self.template ) self.total_detections = detections.count() self.confirmed_detections = detections.filter(is_confirmed=True).count() if self.total_detections > 0: confidence_scores = detections.values_list('confidence_score', flat=True) self.avg_confidence = sum(confidence_scores) / len(confidence_scores) self.min_confidence = min(confidence_scores) self.max_confidence = max(confidence_scores) latest_detection = detections.order_by('-detection_time').first() if latest_detection: self.last_detection = latest_detection.detection_time self.save()