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): [allow_tf32, disable_jit_profiling]
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
No language specified, language will be first be detected for each audio file (increases inference time).
Error initializing models: TranscriptionOptions.init() missing 2 required positional arguments: 'multilingual' and 'hotwords'

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=None,        # 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:
        sf.write(str(temp_file), audio.cpu().numpy(), SAMPLE_RATE)

        # Load and transcribe
        audio_data = whisperx.load_audio(str(temp_file))
        result = self.whisper_model.transcribe(
            audio_data,
            batch_size=1,  # Optimized for CPU
            beam_size=beam_size,
            patience=patience,
            temperature=temperature,
            multilingual=True,  # Added argument
            hotwords=None       # Added argument
        )

        # Align timestamps
        model_a, metadata = whisperx.load_align_model(
            language_code=result["language"],
            device=DEVICE
        )

        result = whisperx.align(
            result["segments"],
            model_a,
            metadata,
            audio_data,
            DEVICE,
            extend_duration=0.0
        )

        return 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