""" Enhanced Video Accent Analyzer Supports YouTube, Loom, direct MP4 links, and local video files with improved error handling and features. """ import os import sys import tempfile import subprocess import requests import json import warnings import time from pathlib import Path from urllib.parse import urlparse from IPython.display import display, HTML, Audio import pandas as pd import matplotlib.pyplot as plt import seaborn as sns # Suppress warnings for cleaner output warnings.filterwarnings('ignore') def install_if_missing(packages): """Install packages if they're not already available in Kaggle""" for package in packages: try: package_name = package.split('==')[0].replace('-', '_') if package_name == 'yt_dlp': package_name = 'yt_dlp' __import__(package_name) except ImportError: print(f"Installing {package}...") subprocess.check_call([sys.executable, "-m", "pip", "install", package, "--quiet"]) # Required packages for Kaggle required_packages = [ "yt-dlp", "librosa", "soundfile", "transformers", "torch", "matplotlib", "seaborn" ] print("๐ง Setting up environment...") install_if_missing(required_packages) # Now import the packages import torch from transformers import Wav2Vec2FeatureExtractor, Wav2Vec2ForSequenceClassification import librosa import soundfile as sf import yt_dlp class VideoAccentAnalyzer: def __init__(self, model_name="dima806/multiple_accent_classification"): """Initialize the accent analyzer for Kaggle environment""" self.model_name = model_name # Enhanced accent labels with better mapping self.accent_labels = [ "british", "canadian", "us", "indian", "australian", "neutral" ] self.accent_display_names = { 'british': '๐ฌ๐ง British English', 'us': '๐บ๐ธ American English', 'australian': '๐ฆ๐บ Australian English', 'canadian': '๐จ๐ฆ Canadian English', 'indian': '๐ฎ๐ณ Indian English', 'neutral': '๐ Neutral English' } self.temp_dir = "/tmp/accent_analyzer" os.makedirs(self.temp_dir, exist_ok=True) self.model_loaded = False self._load_model() def _load_model(self): """Load the accent classification model with error handling""" print("๐ค Loading accent classification model...") try: self.feature_extractor = Wav2Vec2FeatureExtractor.from_pretrained(self.model_name) self.model = Wav2Vec2ForSequenceClassification.from_pretrained(self.model_name) self.device = "cuda" if torch.cuda.is_available() else "cpu" self.model.to(self.device) self.model.eval() # Set to evaluation mode self.model_loaded = True print(f"โ Model loaded successfully on {self.device}") except Exception as e: print(f"โ Error loading model: {e}") print("๐ก Tip: Check your internet connection and Kaggle environment setup") raise def _validate_url(self, url): """Validate and normalize URL""" if not url or not isinstance(url, str): return False, "Invalid URL format" url = url.strip() if not url.startswith(('http://', 'https://')): return False, "URL must start with http:// or https://" return True, url def trim_video(self, input_path, output_path, duration): try: cmd = ['ffmpeg', '-i', input_path, '-t', str(duration), '-c', 'copy', output_path, '-y'] result = subprocess.run(cmd, capture_output=True, text=True, timeout=60) if result.returncode == 0: print(f"โ๏ธ Trimmed video to {duration} seconds") return output_path else: print(f"โ Trimming failed: {result.stderr}") return input_path # fallback to original except Exception as e: print(f"โ Trimming exception: {e}") return input_path def download_video(self, url, max_duration=None): """Download video using yt-dlp with improved error handling""" is_valid, result = self._validate_url(url) if not is_valid: print(f"โ {result}") return None url = result output_path = os.path.join(self.temp_dir, "video.%(ext)s") # Enhanced yt-dlp options for better compatibility ydl_opts = { 'outtmpl': output_path, 'format': 'worst[ext=mp4]/worst', # Use worst quality for faster download and better compatibility 'quiet': False, # Show some output for debugging 'no_warnings': False, 'socket_timeout': 60, 'retries': 5, 'fragment_retries': 5, 'extract_flat': False, 'writeinfojson': False, 'writethumbnail': False, 'writesubtitles': False, } if max_duration: # More generous time limit for download ydl_opts['match_filter'] = lambda info: None if info.get('duration', 0) <= 200000 else "Video too long" try: with yt_dlp.YoutubeDL(ydl_opts) as ydl: print(f"๐ฅ Downloading video from: {url}") start_time = time.time() # Get video info first try: info = ydl.extract_info(url, download=False) print(f"๐บ Found video: {info.get('title', 'Unknown')} ({info.get('duration', 0)}s)") except Exception as e: print(f"โ ๏ธ Could not extract video info: {e}") # Download the video ydl.download([url]) download_time = time.time() - start_time # Find downloaded file (try multiple patterns) video_path = None for file in os.listdir(self.temp_dir): if file.startswith("video.") and os.path.getsize(os.path.join(self.temp_dir, file)) > 1000: # At least 1KB potential_path = os.path.join(self.temp_dir, file) print(f"๐ Found downloaded file: {file} ({os.path.getsize(potential_path)/1024:.1f}KB)") # Try basic validation - if ffprobe fails, still try to extract audio if self._is_valid_video(potential_path): print(f"โ Video validation passed: {file}") video_path = potential_path break else: print(f"โ ๏ธ Video validation failed, but continuing with: {file}") video_path = potential_path # Still try to use it break if video_path: print(f"โ Downloaded video: {os.path.basename(video_path)} ({download_time:.1f}s)") return video_path else: print("โ No video files found after download") return None except Exception as e: print(f"โ ๏ธ yt-dlp failed: {e}") return self._try_direct_download(url) def _is_valid_video(self, file_path): """Verify video file has valid structure (more lenient)""" try: # First check if file exists and has reasonable size if not os.path.exists(file_path) or os.path.getsize(file_path) < 1000: return False # Try ffprobe with more lenient settings result = subprocess.run( ['ffprobe', '-v', 'quiet', '-print_format', 'json', '-show_format', file_path], capture_output=True, text=True, timeout=15 ) if result.returncode == 0: try: # Try to parse the JSON output info = json.loads(result.stdout) # Check if we have format information if 'format' in info and 'duration' in info['format']: return True except json.JSONDecodeError: pass # If ffprobe fails, try a simpler check - just see if ffmpeg can read it result2 = subprocess.run( ['ffmpeg', '-i', file_path, '-t', '1', '-f', 'null', '-', '-v', 'quiet'], capture_output=True, text=True, timeout=15 ) return result2.returncode == 0 except subprocess.TimeoutExpired: print("โ ๏ธ Video validation timed out, assuming valid") return True # If validation times out, assume it's valid except Exception as e: print(f"โ ๏ธ Video validation error: {e}, assuming valid") return True # If validation fails, assume it's valid and let audio extraction handle it def _try_direct_download(self, url): """Enhanced fallback for direct video URLs""" try: print("๐ Trying direct download...") headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36' } response = requests.get(url, stream=True, timeout=60, headers=headers) response.raise_for_status() content_type = response.headers.get("Content-Type", "") if "text/html" in content_type: print("โ ๏ธ Received HTML instead of video - check URL access") return None video_path = os.path.join(self.temp_dir, "video.mp4") file_size = 0 with open(video_path, 'wb') as f: for chunk in response.iter_content(chunk_size=8192): if chunk: f.write(chunk) file_size += len(chunk) print(f"๐ Downloaded {file_size / (1024*1024):.1f} MB") if self._is_valid_video(video_path): print("โ Direct download successful") return video_path else: print("โ Downloaded file is not a valid video") return None except Exception as e: print(f"โ Direct download failed: {e}") return None def extract_audio(self, video_path, max_duration=None): """Extract audio with improved error handling and progress""" audio_path = os.path.join(self.temp_dir, "audio.wav") # Enhanced ffmpeg command with better error handling cmd = ['ffmpeg', '-i', video_path, '-vn', '-acodec', 'pcm_s16le', '-ar', '16000', '-ac', '1', '-y', '-v', 'warning'] if max_duration: cmd.extend(['-t', str(max_duration)]) cmd.append(audio_path) try: print(f"๐ต Extracting audio (max {max_duration}s)...") start_time = time.time() # Run ffmpeg with more detailed output for debugging result = subprocess.run(cmd, capture_output=True, text=True, timeout=180) extraction_time = time.time() - start_time if result.returncode == 0 and os.path.exists(audio_path) and os.path.getsize(audio_path) > 1000: file_size = os.path.getsize(audio_path) / (1024*1024) print(f"โ Audio extracted successfully ({extraction_time:.1f}s, {file_size:.1f}MB)") return audio_path else: print(f"โ FFmpeg stderr: {result.stderr}") print(f"โ FFmpeg stdout: {result.stdout}") # Try alternative extraction method print("๐ Trying alternative audio extraction...") cmd_alt = ['ffmpeg', '-i', video_path, '-vn', '-acodec', 'libmp3lame', '-ar', '16000', '-ac', '1', '-y', '-v', 'warning'] if max_duration: cmd_alt.extend(['-t', str(max_duration)]) audio_path_alt = os.path.join(self.temp_dir, "audio.mp3") cmd_alt.append(audio_path_alt) result_alt = subprocess.run(cmd_alt, capture_output=True, text=True, timeout=180) if result_alt.returncode == 0 and os.path.exists(audio_path_alt): # Convert mp3 to wav cmd_convert = ['ffmpeg', '-i', audio_path_alt, '-ar', '16000', '-ac', '1', audio_path, '-y', '-v', 'quiet'] result_convert = subprocess.run(cmd_convert, capture_output=True, text=True, timeout=60) if result_convert.returncode == 0 and os.path.exists(audio_path): file_size = os.path.getsize(audio_path) / (1024*1024) print(f"โ Alternative extraction successful ({file_size:.1f}MB)") return audio_path raise Exception(f"Both extraction methods failed. FFmpeg error: {result.stderr}") except subprocess.TimeoutExpired: print("โ Audio extraction timed out") return None except Exception as e: print(f"โ Audio extraction failed: {e}") return None def classify_accent(self, audio_path): """Enhanced accent classification with better preprocessing""" if not self.model_loaded: print("โ Model not loaded properly") return None try: print("๐ Loading and preprocessing audio...") audio, sr = librosa.load(audio_path, sr=16000) # Enhanced preprocessing if len(audio) == 0: print("โ Empty audio file") return None # Remove silence from beginning and end audio_trimmed, _ = librosa.effects.trim(audio, top_db=20) # Use multiple chunks for better accuracy if audio is long chunk_size = 16000 * 20 # 20 seconds chunks chunks = [] if len(audio_trimmed) > chunk_size: # Split into overlapping chunks step_size = chunk_size // 2 for i in range(0, len(audio_trimmed) - chunk_size + 1, step_size): chunks.append(audio_trimmed[i:i + chunk_size]) if len(audio_trimmed) % step_size != 0: chunks.append(audio_trimmed[-chunk_size:]) else: chunks = [audio_trimmed] print(f"๐ฏ Analyzing {len(chunks)} audio chunk(s)...") all_predictions = [] for i, chunk in enumerate(chunks[:3]): # Limit to 3 chunks for efficiency inputs = self.feature_extractor( chunk, sampling_rate=16000, return_tensors="pt", padding=True, max_length=16000 * 20, truncation=True ) inputs = {k: v.to(self.device) for k, v in inputs.items()} with torch.no_grad(): outputs = self.model(**inputs) logits = outputs.logits probabilities = torch.nn.functional.softmax(logits, dim=-1) all_predictions.append(probabilities[0].cpu().numpy()) # Average predictions across chunks avg_probabilities = sum(all_predictions) / len(all_predictions) predicted_idx = avg_probabilities.argmax() predicted_idx = min(predicted_idx, len(self.accent_labels) - 1) # Calculate English confidence (exclude 'neutral' for this calculation) english_accents = ["british", "canadian", "us", "australian", "indian"] english_confidence = sum( avg_probabilities[i] * 100 for i, label in enumerate(self.accent_labels) if label in english_accents ) results = { 'predicted_accent': self.accent_labels[predicted_idx], 'accent_confidence': avg_probabilities[predicted_idx] * 100, 'english_confidence': english_confidence, 'audio_duration': len(audio) / 16000, 'processed_duration': len(audio_trimmed) / 16000, 'chunks_analyzed': len(all_predictions), 'all_probabilities': { self.accent_labels[i]: avg_probabilities[i] * 100 for i in range(len(self.accent_labels)) }, 'is_english_likely': english_confidence > 60, 'audio_quality_score': self._assess_audio_quality(audio_trimmed) } print(f"โ Classification complete ({results['chunks_analyzed']} chunks)") return results except Exception as e: print(f"โ Classification failed: {e}") return None def _assess_audio_quality(self, audio): """Assess audio quality for better result interpretation""" try: # Simple quality metrics rms_energy = librosa.feature.rms(y=audio)[0].mean() zero_crossing_rate = librosa.feature.zero_crossing_rate(audio)[0].mean() # Normalize to 0-100 scale quality_score = min(100, (rms_energy * 1000 + (1 - zero_crossing_rate) * 50)) return max(0, quality_score) except: return 50 # Default moderate quality def analyze_video_url(self, url, max_duration=30): """Complete pipeline with enhanced error handling""" print(f"๐ฌ Starting analysis of: {url}") print(f"โฑ๏ธ Max duration: {max_duration} seconds") video_path = self.download_video(url, max_duration) if not video_path: return {"error": "Failed to download video", "url": url} audio_path = self.extract_audio(video_path, max_duration) if not audio_path: return {"error": "Failed to extract audio", "url": url} results = self.classify_accent(audio_path) if not results: return {"error": "Failed to classify accent", "url": url} results.update({ 'source_url': url, 'video_file': os.path.basename(video_path), 'audio_file': os.path.basename(audio_path), 'analysis_timestamp': time.strftime('%Y-%m-%d %H:%M:%S') }) return results def analyze_local_video(self, file_path, max_duration=30): """Enhanced local video analysis""" print(f"๐ฌ Starting analysis of local file: {file_path}") print(f"โฑ๏ธ Max duration: {max_duration} seconds") if not os.path.isfile(file_path): return {"error": f"File not found: {file_path}"} # Check file size file_size = os.path.getsize(file_path) / (1024*1024) # MB print(f"๐ File size: {file_size:.1f} MB") video_filename = os.path.basename(file_path) print(f"โ Using local video: {video_filename}") audio_path = self.extract_audio(file_path, max_duration) if not audio_path: return {"error": "Failed to extract audio"} results = self.classify_accent(audio_path) if not results: return {"error": "Failed to classify accent"} results.update({ 'source_file': file_path, 'video_file': video_filename, 'audio_file': os.path.basename(audio_path), 'file_size_mb': file_size, 'is_local': True, 'analysis_timestamp': time.strftime('%Y-%m-%d %H:%M:%S') }) return results def display_results(self, results): """Enhanced results display with visualizations""" if 'error' in results: display(HTML(f"
{accent_display}
Confidence: {confidence:.1f}%
{english_conf:.1f}%
Audio Quality: {quality_score:.0f}/100
Duration: {duration:.1f}s
Processed: {processed_duration:.1f}s
Chunks: {results.get("chunks_analyzed", 1)}