error : ubuntu@ubuntu-virtual-machine:~/Desktop/new$ ./main.py
/home/ubuntu/.local/lib/python3.10/site-packages/pyannote/audio/core/io.py:43: UserWarning: torchaudio._backend.set_audio_backend has been deprecated. With dispatcher enabled, this function is no-op. You can remove the function call.
  torchaudio.set_audio_backend("soundfile")
/home/ubuntu/.local/lib/python3.10/site-packages/pyannote/audio/pipelines/speaker_verification.py:43: UserWarning: torchaudio._backend.get_audio_backend has been deprecated. With dispatcher enabled, this function is no-op. You can remove the function call.
  backend = torchaudio.get_audio_backend()
INFO:speechbrain.utils.quirks:Applied quirks (see `speechbrain.utils.quirks`): [disable_jit_profiling, allow_tf32]
INFO:speechbrain.utils.quirks:Excluded quirks specified by the `SB_DISABLE_QUIRKS` environment (comma-separated list): []
/home/ubuntu/.local/lib/python3.10/site-packages/pyannote/audio/pipelines/speaker_verification.py:45: UserWarning: Module 'speechbrain.pretrained' was deprecated, redirecting to 'speechbrain.inference'. Please update your script. This is a change from SpeechBrain 1.0. See: https://github.com/speechbrain/speechbrain/releases/tag/v1.0.0
  from speechbrain.pretrained import (
/home/ubuntu/.local/lib/python3.10/site-packages/pyannote/audio/pipelines/speaker_verification.py:53: UserWarning: torchaudio._backend.set_audio_backend has been deprecated. With dispatcher enabled, this function is no-op. You can remove the function call.
  torchaudio.set_audio_backend(backend)
/home/ubuntu/.local/lib/python3.10/site-packages/pyannote/audio/tasks/segmentation/mixins.py:37: UserWarning: `torchaudio.backend.common.AudioMetaData` has been moved to `torchaudio.AudioMetaData`. Please update the import path.
  from torchaudio.backend.common import AudioMetaData
Initializing AudioProcessor with cpu
Loading Whisper v2 model: base
Lightning automatically upgraded your loaded checkpoint from v1.5.4 to v2.4.0. To apply the upgrade to your files permanently, run `python -m pytorch_lightning.utilities.upgrade_checkpoint ../../.cache/torch/whisperx-vad-segmentation.bin`
Model was trained with pyannote.audio 0.0.1, yours is 3.1.1. Bad things might happen unless you revert pyannote.audio to 0.x.
Model was trained with torch 1.10.0+cu102, yours is 2.5.1. Bad things might happen unless you revert torch to 1.x.
Loading Demucs model...
Loading VAD model...
Error initializing models: <urlopen error [Errno -3] Temporary failure in name resolution>


script :

import whisperx
import torch
import torchaudio
from demucs.pretrained import get_model
from demucs.apply import apply_model
import numpy as np
import soundfile as sf
import subprocess
from pathlib import Path
import os
from config import *
from torch.hub import load as torch_hub_load
from silero_vad import load_silero_vad, get_speech_timestamps

class AudioProcessor:
    def __init__(self, model_size=DEFAULT_MODEL_SIZE, whisper_version=DEFAULT_WHISPER_VERSION):
        torch.set_num_threads(CPU_THREADS)
        print(f"Initializing AudioProcessor with {DEVICE}")
        self.init_models(model_size, whisper_version)

    def init_models(self, model_size, whisper_version):
        try:
            print(f"Loading Whisper {whisper_version} model: {model_size}")
            self.whisper_model = whisperx.load_model(
                model_size,
                device=DEVICE,
                compute_type="float32",
                download_root=str(MODELS_DIR),
                language="en",        # Set to None for automatic language detection, or specify 'en' for English
            )

            print("Loading Demucs model...")
            self.demucs_model = get_model(name="htdemucs_ft")
            self.demucs_model.to(DEVICE)

            print("Loading VAD model...")
            # Correct way to load Silero VAD
            self.vad_model, utils = torch_hub_load(
                repo_or_dir='snakers4/silero-vad',
                model='silero_vad',
                force_reload=True
            )

            # Unpack utils properly
            (self.get_speech_timestamps,
                self.save_audio,
                self.read_audio,
                self.VADIterator,
                self.collect_chunks) = utils

            self.vad_model = self.vad_model.to(DEVICE)
            print("Models loaded successfully!")

        except Exception as e:
            print(f"Error initializing models: {e}")
            raise

    def process_file(self, input_file, output_file, beam_size=2, patience=1, temperature=0.0):
        """Main processing pipeline"""
        try:
            # 1. Extract vocals
            vocals, sr = self.extract_vocals(input_file)

            # 2. Normalize and remove silence
            normalized = self.normalize_audio(vocals, sr)

            # 3. Apply VAD
            vad_audio = self.apply_vad(normalized)

            # 4. Add markers
            marked_audio = self.add_markers(vad_audio)

            # 5. Apply compression
            compressed = self.apply_compression(marked_audio)

            # Try transcription with different approaches
            transcription = None

            # With markers
            transcription = self.transcribe(compressed, beam_size, patience, temperature)
            if self.verify_transcription(transcription):
                return self.save_transcription(transcription, output_file)

            # With inverted markers
            transcription = self.transcribe(-compressed, beam_size, patience, temperature)
            if self.verify_transcription(transcription):
                return self.save_transcription(transcription, output_file)

            # Without markers
            transcription = self.transcribe(vad_audio, beam_size, patience, temperature)
            return self.save_transcription(transcription, output_file)

        except Exception as e:
            print(f"Error processing file: {e}")
            raise
        finally:
            # Implement cleanup_temp if necessary
            pass  # Replace this with cleanup_temp() if you have that function

    def extract_vocals(self, audio_path):
        """Extract vocals using Demucs"""
        print("=== Extracting Vocals Debug Information ===")
        print(f"Input audio path: {audio_path}")

        try:
            # Load audio
            print("1. Loading audio file...")
            audio, sr = torchaudio.load(audio_path)
            print(f"- Audio loaded successfully")
            print(f"- Audio shape: {audio.shape}")
            print(f"- Sample rate: {sr}")

            # Move to device
            audio = audio.to(DEVICE)
            print(f"- Audio moved to device: {DEVICE}")

            # Chunk processing setup
            chunk_samples = int(CHUNK_DURATION * sr)
            print(f"- Chunk duration: {CHUNK_DURATION} seconds")
            print(f"- Chunk samples: {chunk_samples}")
            print(f"- Total audio samples: {audio.shape[1]}")

            # Source list in order
            sources_list = ['drums', 'bass', 'other', 'vocals']
            vocals_index = sources_list.index('vocals')  # Should be index 3

            # Process chunked or full audio
            if audio.shape[1] > chunk_samples:
                print("2. Processing audio in chunks...")
                vocals_chunks = []
                for i in range(0, audio.shape[1], chunk_samples):
                    chunk = audio[:, i:min(i+chunk_samples, audio.shape[1])]
                    print(f"  - Processing chunk {i//chunk_samples + 1}")
                    print(f"  - Chunk shape before unsqueeze: {chunk.shape}")

                    # Add batch dimension
                    chunk = chunk.unsqueeze(0)  # Shape becomes (1, channels, length)
                    print(f"  - Chunk shape after unsqueeze: {chunk.shape}")

                    try:
                        # Apply model
                        with torch.no_grad():
                            sources = apply_model(self.demucs_model, chunk, device=DEVICE)
                        print("  - apply_model result:")
                        print(f"    - Type of sources: {type(sources)}")
                        print(f"    - Sources shape: {sources.shape}")

                        # Extract vocals
                        if isinstance(sources, torch.Tensor):
                            vocals_chunk = sources[:, vocals_index, :, :]
                            vocals_chunk = vocals_chunk.squeeze(0)
                        else:
                            raise ValueError("Unexpected sources type or structure")

                        if vocals_chunk is None:
                            raise ValueError("No vocals extracted from chunk")

                        print(f"    - Vocals chunk shape: {vocals_chunk.shape}")
                        vocals_chunks.append(vocals_chunk)

                    except Exception as chunk_error:
                        print(f"  !! Error processing chunk {i//chunk_samples + 1}: {chunk_error}")
                        raise

                vocals = torch.cat(vocals_chunks, dim=1)
                print(f"- Concatenated vocals shape: {vocals.shape}")

            else:
                print("2. Processing full audio...")
                # Add batch dimension
                audio = audio.unsqueeze(0)  # Shape becomes (1, channels, length)
                print(f"  - Audio shape after unsqueeze: {audio.shape}")

                # Apply model
                with torch.no_grad():
                    sources = apply_model(self.demucs_model, audio, device=DEVICE)
                print("  - apply_model result:")
                print(f"    - Type of sources: {type(sources)}")
                print(f"    - Sources shape: {sources.shape}")

                # Extract vocals
                if isinstance(sources, torch.Tensor):
                    vocals = sources[:, vocals_index, :, :]
                    vocals = vocals.squeeze(0)
                else:
                    raise ValueError("Unexpected sources type or structure")

                if vocals is None:
                    raise ValueError("No vocals extracted from full audio")

                print(f"  - Vocals shape: {vocals.shape}")

            return vocals, sr

        except Exception as e:
            print("=== FATAL ERROR IN EXTRACT_VOCALS ===")
            print(f"Error details: {e}")
            import traceback
            traceback.print_exc()
            raise

    def normalize_audio(self, audio, sr):
        """Normalize audio and remove silence using ffmpeg"""
        import soundfile as sf
        temp_in = TEMP_DIR / "temp_in.wav"
        temp_out = TEMP_DIR / "temp_out.wav"

        try:
            # Ensure TEMP_DIR exists
            TEMP_DIR.mkdir(parents=True, exist_ok=True)
            
            # Convert audio tensor to NumPy array
            audio_np = audio.cpu().numpy()
            print(f"Audio shape before transpose: {audio_np.shape}")

            # Transpose if necessary
            if audio_np.ndim == 2:
                audio_np = audio_np.T  # Shape becomes (samples, channels)
                print(f"Transposed audio shape: {audio_np.shape}")

            # Ensure data type is float32
            if audio_np.dtype != np.float32:
                audio_np = audio_np.astype(np.float32)

            # Check for NaNs or Infs
            if np.isnan(audio_np).any() or np.isinf(audio_np).any():
                raise ValueError("Audio data contains NaNs or Infs.")

            # Print audio stats
            print(f"Audio min: {audio_np.min()}, max: {audio_np.max()}, mean: {audio_np.mean()}")

            # Write audio data to temp_in.wav
            sf.write(str(temp_in), audio_np, sr, format='WAV', subtype='PCM_16')
            print(f"Audio data written to {temp_in}")

            # Test if the file can be read
            test_audio, test_sr = sf.read(str(temp_in))
            print(f"Read test audio from file with shape {test_audio.shape} and sample rate {test_sr}")

            # Run ffmpeg to normalize and remove silence
            cmd = [
                'ffmpeg', '-y',
                '-i', str(temp_in),
                '-af', (
                    f"silenceremove=stop_periods=-1:stop_duration=1:stop_threshold={VAD_THRESHOLD}dB,"
                    "loudnorm=I=-16:TP=-1.5:LRA=11"
                ),
                '-ar', str(sr),
                str(temp_out)
            ]

            process = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
            print(f"ffmpeg stdout: {process.stdout}")
            print(f"ffmpeg stderr: {process.stderr}")

            if process.returncode != 0:
                raise Exception(f"ffmpeg error: {process.stderr}")

            normalized, sr = torchaudio.load(str(temp_out))
            normalized = normalized.squeeze(0)
            return normalized

        except Exception as e:
            print(f"Error normalizing audio: {e}")
            raise

    def apply_vad(self, audio):
        """Apply Voice Activity Detection"""
        print("Applying VAD...")
        audio_mono = audio.mean(dim=0) if audio.dim() > 1 else audio
        audio_numpy = audio_mono.cpu().numpy()

        # Process in chunks
        chunk_samples = int(CHUNK_DURATION * SAMPLE_RATE)
        speech_timestamps = []

        for i in range(0, len(audio_numpy), chunk_samples):
            chunk = audio_numpy[i:min(i + chunk_samples, len(audio_numpy))]
            chunk_tensor = torch.FloatTensor(chunk).to(DEVICE)

            # Get speech timestamps for the chunk
            chunk_timestamps = self.get_speech_timestamps(
                chunk_tensor,
                self.vad_model,
                sampling_rate=SAMPLE_RATE
            )

            # Adjust timestamps for chunk position
            for ts in chunk_timestamps:
                ts['start'] += i
                ts['end'] += i
            speech_timestamps.extend(chunk_timestamps)

        # Create masked audio
        masked_audio = torch.zeros_like(audio_mono)
        for ts in speech_timestamps:
            start_frame = ts['start']
            end_frame = ts['end']
            if start_frame < len(masked_audio) and end_frame <= len(masked_audio):
                masked_audio[start_frame:end_frame] = audio_mono[start_frame:end_frame]

        return masked_audio

    def add_markers(self, audio):
        """Add voice markers at speech boundaries"""
        print("Adding markers...")
        marked_audio = audio.clone()

        # Generate marker signal
        t = np.arange(int(MARKER_DURATION * SAMPLE_RATE))
        marker = torch.from_numpy(
            0.1 * np.sin(2 * np.pi * MARKER_FREQUENCY * t / SAMPLE_RATE)
        ).float()

        # Find speech segments
        speech_segments = self.find_speech_segments(audio)

        # Add markers at segment boundaries
        for start, end in speech_segments:
            if start + len(marker) < len(marked_audio):
                marked_audio[start:start + len(marker)] += marker
            if end - len(marker) >= 0:
                marked_audio[end - len(marker):end] += marker

        return marked_audio

    def find_speech_segments(self, audio, threshold=0.01):
        """Find speech segments in audio"""
        is_speech = torch.abs(audio) > threshold
        changes = torch.diff(is_speech.long())
        change_points = torch.nonzero(changes).flatten()

        segments = []
        for i in range(0, len(change_points)-1, 2):
            start = change_points[i].item()
            end = change_points[i+1].item() if i+1 < len(change_points) else len(audio)-1
            segments.append((start, end))

        return segments

    def apply_compression(self, audio):
        """Apply speech compression using ffmpeg"""
        print("Applying compression...")
        temp_in = TEMP_DIR / "temp_compress_in.wav"
        temp_out = TEMP_DIR / "temp_compress_out.wav"

        try:
            sf.write(str(temp_in), audio.cpu().numpy(), SAMPLE_RATE)

            cmd = [
                'ffmpeg', '-y',
                '-i', str(temp_in),
                '-af', 'acompressor=threshold=-12dB:ratio=2:attack=200:release=1000',
                '-ar', str(SAMPLE_RATE),
                str(temp_out)
            ]

            subprocess.run(cmd, check=True, capture_output=True)
            compressed, _ = torchaudio.load(str(temp_out))
            return compressed.squeeze()

        except Exception as e:
            print(f"Error applying compression: {e}")
            raise

    def transcribe(self, audio, beam_size=2, patience=1, temperature=0.0):
        """Transcribe audio using WhisperX"""
        print("Transcribing...")
        temp_file = TEMP_DIR / "temp_transcribe.wav"

        try:
            # Write audio to temporary file
            sf.write(str(temp_file), audio.cpu().numpy(), SAMPLE_RATE)

            # Load audio data
            audio_data = whisperx.load_audio(str(temp_file))

            # Transcribe audio using the underlying model
            result_segments, info = self.whisper_model.model.transcribe(
                audio_data,
                beam_size=beam_size,
                patience=patience,
                temperature=temperature,
                language="en",  # Set to None for automatic language detection
                word_timestamps=False,  # Set to True if you need word-level timestamps
            )

            # Extract the detected language
            language_code = info.language if info and hasattr(info, 'language') else None
            print(f"Detected language: {language_code}")

            # Convert segments to list of dicts expected by whisperx.align
            segments = []
            for segment in result_segments:
                segments.append({
                    'id': segment.id,
                    'seek': segment.seek,
                    'start': segment.start,
                    'end': segment.end,
                    'text': segment.text,
                    'tokens': segment.tokens,
                    'avg_logprob': segment.avg_logprob,
                    'compression_ratio': segment.compression_ratio,
                    'no_speech_prob': segment.no_speech_prob,
                })

            # Align timestamps using whisperx
            model_a, metadata = whisperx.load_align_model(
                language_code=language_code,
                device=DEVICE
            )

            aligned_result = whisperx.align(
                segments,  # Pass the list of dicts
                model_a,
                metadata,
                audio_data,
                DEVICE,
                extend_duration=0.0
            )

            return aligned_result

        except Exception as e:
            print(f"Transcription error: {e}")
            return None

    def verify_transcription(self, transcription):
        """Verify if transcription contains markers"""
        if not transcription or 'segments' not in transcription:
            return False

        marker_patterns = ["*", "beep", "tone", "[noise]", "[sound]"]
        return any(
            any(pattern in segment['text'].lower() for pattern in marker_patterns)
            for segment in transcription['segments']
        )

    def save_transcription(self, transcription, output_file):
        """Save transcription to file"""
        if not transcription:
            return False

        try:
            with open(output_file, 'w', encoding='utf-8') as f:
                for segment in transcription['segments']:
                    f.write(
                        f"[{segment['start']:.2f} - {segment['end']:.2f}] {segment['text']}\n"
                    )
            return True
        except Exception as e:
            print(f"Error saving transcription: {e}")
            return False