from abc import ABC, abstractmethod from typing import Any, Dict, Optional, TextIO, List import whisper from whisper import Whisper import torch from math import floor, ceil from transformers import pipeline import numpy as np import librosa import torch from math import floor, ceil import os import argparse from pyannote.audio import Pipeline, Audio from whisper.utils import WriteSRT, WriteVTT import soundfile as sf import librosa import json from dataclasses import dataclass, asdict from jsonschema import validate, ValidationError import create_pdf import paragraphs_creator from pydub import AudioSegment import yt_dlp from slugify import slugify import uuid import yaml class ASRFacade(ABC): """Abstract base class to define an interface for transcription.""" @abstractmethod def load_audio(self, file_path: str): pass @abstractmethod def transcribe( self, start: float, end: float, options: Dict[str, Any], ) -> Dict[str, Any]: """Transcribe a portion of audio from start to end time in seconds.""" pass class WhisperFacade(ASRFacade): wmodel: Whisper audio: Any # The loaded audio array from Whisper def __init__(self, model: str, *, quantize=False) -> None: print("Initialize Whisper") whisper_model = whisper.load_model(model) if quantize == True: print("Quantize active") DTYPE = torch.qint8 qmodel: Whisper = torch.quantization.quantize_dynamic( whisper_model, {torch.nn.Linear}, dtype=DTYPE ) del whisper_model self.wmodel = qmodel else: print("Quantize non active") self.wmodel = whisper_model def load_audio(self, file_path: str): self.audio = whisper.load_audio(file_path) def _set_timing_for(self, segment: dict[str, float], offset: float) -> None: # Keep this logic the same s = segment s['start'] += offset s['end'] += offset if 'words' in s: for w in s['words']: w['start'] += offset w['end'] += offset def transcribe( self, start: float, end: float, options: dict[str, Any] ) -> dict[str, Any]: SAMPLE_RATE = 16_000 start_index = floor(start * SAMPLE_RATE) end_index = ceil(end * SAMPLE_RATE) audio_segment = self.audio[start_index:end_index] result = whisper.transcribe(self.wmodel, audio_segment, **options) for s in result['segments']: self._set_timing_for(segment=s, offset=start) return result class TransformersFacade(ASRFacade): """Use a Hugging Face ASR pipeline instead of Whisper.""" def __init__(self, model_name: str): print(f"Initialize Transformers pipeline with model {model_name}") self.asr_pipeline = pipeline("automatic-speech-recognition", model=model_name) self.sr = 16000 self.audio_data = None # We'll store the loaded waveform here def load_audio(self, file_path: str): # We'll load audio into a single waveform at 16 kHz print(f"Loading audio for Transformers from {file_path}") waveform, sr = librosa.load(file_path, sr=self.sr, mono=True) self.audio_data = waveform print(f"Audio loaded: shape={waveform.shape}, sample_rate={sr}") def transcribe( self, start: float, end: float, options: dict[str, Any] ) -> dict[str, Any]: """ We want to return a structure with 'segments' just like Whisper does. We'll treat the entire chunk as a single forward pass to the pipeline. If you prefer more advanced chunking or word-level timestamps, you can expand this. """ start_sample = floor(start * self.sr) end_sample = ceil(end * self.sr) audio_segment = self.audio_data[start_sample:end_sample] # The pipeline can handle direct numpy arrays. # Some HF pipelines let you pass 'return_timestamps=True' in generate(), but that may vary by model. # We'll do a straightforward approach here: transcription = self.asr_pipeline(audio_segment)#, sampling_rate=self.sr) # We want to unify the output structure with whisper-like dict. # Example result for consistency: result = { "language": "ca", # or fetch from pipeline if available "segments": [ { "id": 0, "start": start, "end": end, "text": transcription["text"], # If you want words/tokens, you can parse them here if your model supports it } ] } return result def create_asr_facade(model_name: str, quantize: bool = False) -> ASRFacade: """Factory function to return either a Whisper or Transformers facade.""" # Example: if user requests 'base-ca', we switch to Transformers if model_name in ['tiny.en', 'tiny', 'base.en', 'base', 'small.en', 'small', 'medium.en', 'medium', 'large-v1', 'large-v2', 'large-v3', 'large', 'large-v3-turbo', 'turbo']: return WhisperFacade(model=model_name, quantize=quantize) else: print(f"Trying to use Transformers model from {model_name}") return TransformersFacade(model_name=model_name) ############################################################################# @dataclass class Segment: start: float end: float text: str speaker: str language: str def generate_segments(transcription_data, speaker, language) -> List[Segment]: segments = [] for item in transcription_data: segment = Segment( start=item['start'], end=item['end'], text=item['text'], speaker=speaker, language=language ) segments.append(segment) return segments schema = { "type": "object", "properties": { "segments": { "type": "array", "items": { "type": "object", "properties": { "start": {"type": "number"}, "end": {"type": "number"}, "text": {"type": "string"}, "speaker": {"type": "string"}, "language": {"type": "string"} }, "required": ["start", "end", "text", "speaker", "language"] } } }, "required": ["segments"] } def validate_json(json_data: str) -> bool: try: data = json.loads(json_data) validate(instance=data, schema=schema) return True except ValidationError as e: print(f"Validation error: {e}") return False def diarize_audio(HF_AUTH_TOKEN, AUDIO_FILE): pipeline = Pipeline.from_pretrained( "pyannote/speaker-diarization-3.1", use_auth_token=HF_AUTH_TOKEN) # Send pyannote pipeline to GPU (when available) device: str = "" if torch.cuda.is_available(): device = "cuda" else: device = "cpu" pipeline.to(torch.device(device)) print(f"Diarize audio on {device}") io = Audio(mono='downmix', sample_rate=16000) waveform, sample_rate = io(AUDIO_FILE) diarization = pipeline({"waveform": waveform, "sample_rate": sample_rate}) return diarization, waveform, sample_rate class AppendResultsMixin: """Class to return srt or vtt file path and open mode of write or append to allow incremental writing. """ first_call: bool = True output_path: str = '' def get_path_and_open_mode(self, *, audio_path: str, dir: str, ext: str) -> tuple[str, str]: mode: str if self.first_call: audio_basename = os.path.basename(audio_path) audio_basename = os.path.splitext(audio_basename)[0] self.output_path: str = os.path.join(dir, audio_basename + "." + ext) self.first_call = False mode = 'w' # open for write initially else: mode = 'a' # open for append after return self.output_path, mode class WriteSRTIncremental(AppendResultsMixin, WriteSRT): """Incrementally create an SRT file with multiple calls appending new entries to the file. """ srt_index: int = 1 # Index for SRT blocks retained across multiple calls def __init__(self, output_dir: Optional[str] = None): super().__init__(output_dir=output_dir) self.extension = '.srt' self.srt_index = 1 # Move to instance variable def __call__( self, result: dict, audio_path: str, speaker: str, start_base: float, options: Optional[dict] = None, output_path: Optional[str] = None, **kwargs, ): if output_path: path = output_path mode = 'a' if os.path.exists(path) else 'w' else: audio_dir = os.path.dirname(audio_path) output_dir = self.output_dir if self.output_dir else audio_dir audio_basename = os.path.splitext(os.path.basename(audio_path))[0] path = os.path.join(output_dir, f"{audio_basename}{self.extension}") mode = 'a' if os.path.exists(path) else 'w' with open(path, mode, encoding="utf-8") as f: self.write_result(result, f, speaker, start_base, options=options, **kwargs) def write_result( self, result: dict, file: TextIO, speaker: str, start_base: float, options: Optional[dict] = None, **kwargs, ): for segment in result['segments']: start = self.format_timestamp(segment['start']) end = self.format_timestamp(segment['end']) text = f"[{speaker}]: {segment['text']}" # TODO: Add verbose option to print the SRT blocks print(f"{self.srt_index}\n{start} --> {end}\n{text}\n", file=file, flush=True) self.srt_index += 1 class WriteVTTIncremental(AppendResultsMixin, WriteVTT): """Incrementally create a VTT file with multiple calls appending new entries to the file. """ def __call__( self, result: dict, audio_path: str, speaker: str, start: float, options: Optional[dict] = None, output_path: Optional[str] = None, **kwargs, ): if output_path: path = output_path mode = 'a' if os.path.exists(path) else 'w' else: path, mode = self.get_path_and_open_mode( audio_path=audio_path, dir=".", ext="vtt" ) with open(path, mode, encoding="utf-8") as f: self.write_result(result, file=f, options=options, **kwargs) def write_result( self, result: dict, file: TextIO, options: Optional[dict] = None, **kwargs, ): if file.tell() == 0: print("WEBVTT\n", file=file) for segment in result['segments']: start = self.format_timestamp(segment['start']) end = self.format_timestamp(segment['end']) text = f"[{segment.get('speaker', 'unknown')}]: {segment['text']}" print(f"{start} --> {end}\n{text}\n", file=file, flush=True) class SegmentsJSONWriter(AppendResultsMixin): """Incrementally write segments to a JSON file.""" def __init__(self, output_dir: Optional[str] = None): self.output_dir = output_dir # Now optional self.first_call = True self.output_path = '' # Will store the output file path self.temp_output_path = '' # Will store the temporary output file path def __call__( self, segments: List[Segment], audio_path: str, output_path: Optional[str] = None, ): if output_path: # Use the provided output path directly path = output_path mode = 'a' if os.path.exists(path) else 'w' self.output_path = path self.temp_output_path = os.path.splitext(path)[0] + ".jsonl" else: if not self.output_path: audio_basename = os.path.splitext(os.path.basename(audio_path))[0] # Use output_dir if provided, else use the directory of audio_path dir = self.output_dir if self.output_dir else os.path.dirname(audio_path) self.output_path = os.path.join(dir, audio_basename + ".json") self.temp_output_path = os.path.join(dir, audio_basename + ".jsonl") path = self.output_path mode = 'a' if not self.first_call else 'w' if len(segments) > 0 and all(hasattr(segment, 'text') for segment in segments): with open(self.temp_output_path, mode, encoding='utf-8') as f: for idx, segment in enumerate(segments): segment.text = segment.text.strip() json.dump(asdict(segment), f, ensure_ascii=False) f.write('\n') self.first_call = False # with open(path, mode, encoding='utf-8') as f: # if self.first_call: # # Start the JSON structure # f.write('{"segments": [\n') # else: # f.write(',\n') # for idx, segment in enumerate(segments): # if segment: # Check if the segment is not empty # if idx > 0: # f.write(',\n') # segment.text = segment.text.strip() # json.dump(asdict(segment), f, ensure_ascii=False, indent=2) self.first_call = False def finalize(self): """Call this method after all segments have been written to close the JSON array.""" if self.temp_output_path: dict = {"segments": []} with open(self.temp_output_path, 'r', encoding='utf-8') as temp_file: for line in temp_file: segments = json.loads(line.strip()) dict["segments"].append(segments) # if self.output_path: # with open(self.output_path, 'a', encoding='utf-8') as f: # f.write('\n ]\n}\n') # This is a quick solution to the error that occurs with empty items in the list ..},,{... # However, it's not the best solution. It's better to find and fix the root cause of the issue. # One strategy would be read the json file, and append the segments to the list, then write it back. # with open(self.output_path, 'r', encoding='utf-8') as file: # content = file.read() # # Remove extra commas # fixed_content = content.replace(',\n,', ',\n') # fixed_content = content.replace(',\n,\n,', ',\n') # fixed_content = content.replace(',\n,\n,\n,', ',\n') # # Load the JSON to ensure it's valid # try: # json_data = json.loads(fixed_content) # except json.JSONDecodeError as e: # raise f"Error decoding JSON: {e}" # Write the fixed JSON back to the file with open(self.output_path, 'w', encoding='utf-8') as file: json.dump(dict, file, ensure_ascii=False, indent=2) def clip_audio(audio_file_path, sample_rate, start, end, output_path): # Ensure the output directory exists os.makedirs(os.path.dirname(output_path), exist_ok=True) # Load the audio file waveform, sr = librosa.load(audio_file_path, sr=sample_rate, mono=True) # Calculate start and end samples start_sample = int(start * sr) end_sample = int(end * sr) # Write the audio segment to the output path sf.write(output_path, waveform[start_sample:end_sample], sr, format='WAV') def convert_mpx_to_wav(file_path, output_path): if file_path.lower().endswith('.mp3'): # Load the MP3 file audio = AudioSegment.from_mp3(file_path) elif file_path.lower().endswith('.mp4'): audio = AudioSegment.from_file(file_path, format="mp4") else: raise ValueError("Input file must be an MP3 or MP4 file") # Define the output path #wav_file_path = os.path.splitext(file_path)[0] + '.wav' # Export as WAV audio.export(output_path, format='wav') return output_path def download_youtube_video(url, filename, output_path='/tmp'): ydl_opts = { 'format': 'bestaudio/best', #'outtmpl': os.path.join(output_path, '%(title)s.%(ext)s'), 'outtmpl': os.path.join(output_path, f"{filename.split('.')[0]}.%(ext)s"), 'postprocessors': [{ 'key': 'FFmpegExtractAudio', 'preferredcodec': 'wav', 'preferredquality': '192', }], } if not os.path.exists(output_path): os.makedirs(output_path) with yt_dlp.YoutubeDL(ydl_opts) as ydl: info_dict = ydl.extract_info(url, download=True) output_file = ydl.prepare_filename(info_dict) print(output_file) base, ext = os.path.splitext(output_file) new_file = base + '-original.wav' return new_file import subprocess import shutil def convert_video_to_wav(input_file, output_file): """ Convert a .rm file to .wav using FFmpeg. :param input_file: Path to the input .rm file :param output_file: Desired path to the output .wav file """ # Ensure the output directory exists (optional) os.makedirs(os.path.dirname(output_file), exist_ok=True) # Build the FFmpeg command # -i input_file : input file # -vn : ignore video (if any) # -acodec pcm_s16le: use 16-bit PCM audio codec # -ar 44100 : set sample rate to 44100 Hz # -ac 2 : set audio channels to stereo command = [ "ffmpeg", "-y", # overwrite output file if it exists (optional) "-i", input_file, "-vn", "-acodec", "pcm_s16le", "-ar", "44100", "-ac", "2", output_file ] # Run the command try: subprocess.run(command, check=True) print(f"Conversion successful! '{input_file}' has been converted to '{output_file}'.") except subprocess.CalledProcessError as e: print(f"Error during conversion: {e}") def main(args): # TODO: Take out the file_path from ODTP here if args.input_file.startswith('/odtp/odtp-input/http://') or args.input_file.startswith('/odtp/odtp-input/https://'): file_path = download_youtube_video(args.input_file.replace("/odtp/odtp-input/",""), filename=os.path.basename(args.output_file) , output_path=os.path.dirname(args.output_file)) base_slug = slugify(file_path, separator='_') elif args.input_file.lower().endswith('.mp3'): file_path = convert_mpx_to_wav(args.input_file) shutil.copy(file_path, os.path.join("/odtp/odtp-output", os.path.basename(file_path).replace('.wav', '-original.mp3'))) elif args.input_file.lower().endswith('.wav'): file_path = args.input_file shutil.copy(file_path, os.path.join("/odtp/odtp-output", os.path.basename(file_path).replace('.wav', '-original.wav'))) elif args.input_file.lower().endswith('.mp4'): file_path = "/odtp/odtp-output/" + os.path.basename(args.input_file).replace('.mp4', '-original.wav') convert_mpx_to_wav(args.input_file, file_path) elif args.input_file.lower().endswith('.rm'): file_path = "/odtp/odtp-output/" + os.path.basename(args.input_file).replace('.rm', '-original.wav') convert_video_to_wav(args.input_file, file_path) elif args.input_file.lower().endswith('.f4v'): file_path = "/odtp/odtp-output/" + os.path.basename(args.input_file).replace('.f4v', '-original.wav') convert_video_to_wav(args.input_file, file_path) elif args.input_file.lower().endswith('.mkv'): file_path = "/odtp/odtp-output/" + os.path.basename(args.input_file).replace('.mkv', '-original.wav') convert_video_to_wav(args.input_file, file_path) else: raise ValueError(f"Input file must be an MP3, WAV, RM, F4V, MKV, Youtube Link, or MP4 file. Input file: {args.input_file}") diarization, _, sample_rate = diarize_audio(args.hf_token, file_path) # Create the correct ASR facade asr_model = create_asr_facade(args.model, quantize=args.quantize) asr_model.load_audio(file_path) writer = WriteSRTIncremental() writer_json = SegmentsJSONWriter() # Whisper-like transcription options whisper_options = { "verbose": None, "word_timestamps": False, "task": args.task, "suppress_tokens": "" } if args.language: # This is only relevant for Whisper. For Transformers, # you might specify a different approach or ignore it. whisper_options["language"] = args.language writer_options = { "max_line_width": 55, "max_line_count": 2, "word_timestamps": False } if args.verbose: print("Process diarized blocks") grouped_segments = [] current_speaker = None current_start = None current_end = None for turn, _, speaker in diarization.itertracks(yield_label=True): if args.verbose: print(speaker) if turn.end - turn.start < 0.5: # ignore short utterances continue if speaker == current_speaker: current_end = turn.end else: if current_speaker is not None: grouped_segments.append((current_start, current_end, current_speaker)) current_speaker = speaker current_start = turn.start current_end = turn.end if current_speaker is not None: grouped_segments.append((current_start, current_end, current_speaker)) # Process each grouped segment for start, end, speaker in grouped_segments: clip_path = f"/tmp/speaker_{speaker}_start_{start:.1f}_end_{end:.1f}.wav" clip_audio(file_path, sample_rate, start, end, clip_path) # Important: we call asr_model instead of model result = asr_model.transcribe(start=start, end=end, options=whisper_options) language = result.get('language', args.language or 'unknown') if args.verbose: print(f"start={start:.1f}s stop={end:.1f}s lang={language} {speaker}") # Use your existing logic to write SRT, JSON, etc. writer(result, args.output_file, speaker, start, writer_options) writer_json(generate_segments(result['segments'], speaker, language), args.output_json_file) writer_json.finalize() # If you want to validate JSON, paragraphs, PDF creation, etc. paragraphs_creator.process_paragraphs( args.output_json_file, args.output_paragraphs_json_file, 3 ) create_pdf.convert_json_to_pdf( args.output_paragraphs_json_file, args.output_md_file, args.output_pdf_file ) if __name__ == '__main__': # Multiprocessing requires spawn when working with CUDA #multiprocessing.set_start_method('spawn') parser = argparse.ArgumentParser(description="Diarization and Whisper Transcription CLI") parser.add_argument('--model', type=str, required=True, help="Whisper model to use") parser.add_argument('--quantize', action='store_true', help="Whether to quantize the model") parser.add_argument('--hf-token', type=str, required=True, help="Hugging Face authentication token") parser.add_argument('--task', type=str, choices=['transcribe', 'translate'], required=True, help="Task to perform") parser.add_argument('--language', type=str, required=False, help="Language to use for transcription or translation") parser.add_argument('--input-file', type=str, required=True, help="Input audio file") parser.add_argument('--output-file', type=str, required=True, help="Output file for the results (SRT or VTT)") parser.add_argument('--output-json-file', type=str, required=True, help="Output json file.") parser.add_argument('--output-paragraphs-json-file', type=str, required=True, help="Output paragraphs file") parser.add_argument('--output-md-file', type=str, required=True, help="Output markdown file") parser.add_argument('--output-pdf-file', type=str, required=True, help="Output pdf file") parser.add_argument('--verbose', action='store_true', help="Printing status") args = parser.parse_args() main(args)