import os import tempfile import time import gradio as gr import numpy as np import torch import yt_dlp as youtube_dl from gradio_client import Client from pyannote.audio import Pipeline from transformers.pipelines.audio_utils import ffmpeg_read YT_LENGTH_LIMIT_S = 36000 # limit to 1 hour YouTube files SAMPLING_RATE = 16000 HF_TOKEN = os.environ.get("HF_ACCESS_TOKEN") API_URL = "https://sanchit-gandhi-whisper-jax.hf.space/" # set up the Gradio client client = Client(API_URL) # set up the diarization pipeline diarization_pipeline = Pipeline.from_pretrained("pyannote/speaker-diarization", use_auth_token=HF_TOKEN) def format_string(timestamp): """ Reformat a timestamp string from (HH:)MM:SS to float seconds. Note that the hour column is optional, and is appended within the function if not input. Args: timestamp (str): Timestamp in string format, either MM:SS or HH:MM:SS. Returns: seconds (float): Total seconds corresponding to the input timestamp. """ split_time = timestamp.split(":") split_time = [float(sub_time) for sub_time in split_time] if len(split_time) == 2: split_time.insert(0, 0) seconds = split_time[0] * 3600 + split_time[1] * 60 + split_time[2] return seconds # Adapted from https://github.com/openai/whisper/blob/c09a7ae299c4c34c5839a76380ae407e7d785914/whisper/utils.py#L50 def format_timestamp(seconds: float, always_include_hours: bool = False, decimal_marker: str = "."): """ Reformat a timestamp from a float of seconds to a string in format (HH:)MM:SS. Note that the hour column is optional, and is appended in the function if the number of hours > 0. Args: seconds (float): Total seconds corresponding to the input timestamp. Returns: timestamp (str): Timestamp in string format, either MM:SS or HH:MM:SS. """ if seconds is not None: milliseconds = round(seconds * 1000.0) hours = milliseconds // 3_600_000 milliseconds -= hours * 3_600_000 minutes = milliseconds // 60_000 milliseconds -= minutes * 60_000 seconds = milliseconds // 1_000 milliseconds -= seconds * 1_000 hours_marker = f"{hours:02d}:" if always_include_hours or hours > 0 else "" return f"{hours_marker}{minutes:02d}:{seconds:02d}{decimal_marker}{milliseconds:03d}" else: # we have a malformed timestamp so just return it as is return seconds def format_as_transcription(raw_segments): return "\n\n".join( [ f"{chunk['speaker']} [{format_timestamp(chunk['timestamp'][0])} -> {format_timestamp(chunk['timestamp'][1])}] {chunk['text']}" for chunk in raw_segments ] ) def _return_yt_html_embed(yt_url): video_id = yt_url.split("?v=")[-1] HTML_str = ( f'
' "
" ) return HTML_str def download_yt_audio(yt_url, filename): info_loader = youtube_dl.YoutubeDL() try: info = info_loader.extract_info(yt_url, download=False) except youtube_dl.utils.DownloadError as err: raise gr.Error(str(err)) file_length = info["duration_string"] file_length_s = format_string(file_length) if file_length_s > YT_LENGTH_LIMIT_S: yt_length_limit_hms = time.strftime("%HH:%MM:%SS", time.gmtime(YT_LENGTH_LIMIT_S)) file_length_hms = time.strftime("%HH:%MM:%SS", time.gmtime(file_length_s)) raise gr.Error( f"To encourage fair usage of the demo, the maximum YouTube length is {yt_length_limit_hms}, " f"got {file_length_hms} YouTube video." ) ydl_opts = {"outtmpl": filename, "format": "worstvideo[ext=mp4]+bestaudio[ext=m4a]/best[ext=mp4]/best"} with youtube_dl.YoutubeDL(ydl_opts) as ydl: try: ydl.download([yt_url]) except youtube_dl.utils.ExtractorError as err: raise gr.Error(str(err)) def align(transcription, segments, group_by_speaker=True): transcription_split = transcription.split("\n") # re-format transcription from string to List[Dict] transcript = [] for chunk in transcription_split: start_end, transcription = chunk[1:].split("] ") start, end = start_end.split("->") transcript.append({"timestamp": (format_string(start), format_string(end)), "text": transcription}) # diarizer output may contain consecutive segments from the same speaker (e.g. {(0 -> 1, speaker_1), (1 -> 1.5, speaker_1), ...}) # we combine these segments to give overall timestamps for each speaker's turn (e.g. {(0 -> 1.5, speaker_1), ...}) new_segments = [] prev_segment = cur_segment = segments[0] for i in range(1, len(segments)): cur_segment = segments[i] # check if we have changed speaker ("label") if cur_segment["label"] != prev_segment["label"] and i < len(segments): # add the start/end times for the super-segment to the new list new_segments.append( { "segment": {"start": prev_segment["segment"]["start"], "end": cur_segment["segment"]["start"]}, "speaker": prev_segment["label"], } ) prev_segment = segments[i] # add the last segment(s) if there was no speaker change new_segments.append( { "segment": {"start": prev_segment["segment"]["start"], "end": cur_segment["segment"]["end"]}, "speaker": prev_segment["label"], } ) # get the end timestamps for each chunk from the ASR output end_timestamps = np.array([chunk["timestamp"][-1] for chunk in transcript]) segmented_preds = [] # align the diarizer timestamps and the ASR timestamps for segment in new_segments: # get the diarizer end timestamp end_time = segment["segment"]["end"] # find the ASR end timestamp that is closest to the diarizer's end timestamp and cut the transcript to here upto_idx = np.argmin(np.abs(end_timestamps - end_time)) if group_by_speaker: segmented_preds.append( { "speaker": segment["speaker"], "text": "".join([chunk["text"] for chunk in transcript[: upto_idx + 1]]), "timestamp": (transcript[0]["timestamp"][0], transcript[upto_idx]["timestamp"][1]), } ) else: for i in range(upto_idx + 1): segmented_preds.append({"speaker": segment["speaker"], **transcript[i]}) # crop the transcripts and timestamp lists according to the latest timestamp (for faster argmin) transcript = transcript[upto_idx + 1 :] end_timestamps = end_timestamps[upto_idx + 1 :] # final post-processing transcription = format_as_transcription(segmented_preds) return transcription def transcribe(audio_path, task="transcribe", group_by_speaker=True, progress=gr.Progress()): # run Whisper JAX asynchronously using Gradio client (endpoint) job = client.submit( audio_path, task, True, api_name="/predict_1", ) # run diarization while we wait for Whisper JAX progress(0, desc="Diarizing...") diarization = diarization_pipeline(audio_path) segments = diarization.for_json()["content"] # only fetch the transcription result after performing diarization progress(0.33, desc="Transcribing...") transcription, _ = job.result() # align the ASR transcriptions and diarization timestamps progress(0.66, desc="Aligning...") transcription = align(transcription, segments, group_by_speaker=group_by_speaker) return transcription def transcribe_yt(yt_url, task="transcribe", group_by_speaker=True, progress=gr.Progress()): # run Whisper JAX asynchronously using Gradio client (endpoint) job = client.submit( yt_url, task, True, api_name="/predict_2", ) html_embed_str = _return_yt_html_embed(yt_url) progress(0, desc="Downloading YouTube video...") with tempfile.TemporaryDirectory() as tmpdirname: filepath = os.path.join(tmpdirname, "video.mp4") download_yt_audio(yt_url, filepath) with open(filepath, "rb") as f: inputs = f.read() inputs = ffmpeg_read(inputs, SAMPLING_RATE) inputs = torch.from_numpy(inputs).float() inputs = inputs.unsqueeze(0) # run diarization while we wait for Whisper JAX progress(0.25, desc="Diarizing...") diarization = diarization_pipeline( {"waveform": inputs, "sample_rate": SAMPLING_RATE}, ) segments = diarization.for_json()["content"] # only fetch the transcription result after performing diarization progress(0.50, desc="Transcribing...") _, transcription, _ = job.result() # align the ASR transcriptions and diarization timestamps progress(0.75, desc="Aligning...") transcription = align(transcription, segments, group_by_speaker=group_by_speaker) return html_embed_str, transcription title = "Video Transcription with Speaker diarization" microphone = gr.Interface( fn=transcribe, inputs=[ gr.inputs.Audio(source="microphone", optional=True, type="filepath"), gr.inputs.Radio(["transcribe", "translate"], label="Task", default="transcribe"), gr.inputs.Checkbox(default=True, label="Group by speaker"), ], outputs=[ gr.outputs.Textbox(label="Transcription").style(show_copy_button=True), ], allow_flagging="never", title=title ) audio_file = gr.Interface( fn=transcribe, inputs=[ gr.inputs.Audio(source="upload", optional=True, label="Audio file", type="filepath"), gr.inputs.Radio(["transcribe", "translate"], label="Task", default="transcribe"), gr.inputs.Checkbox(default=True, label="Group by speaker"), ], outputs=[ gr.outputs.Textbox(label="Transcription").style(show_copy_button=True), ], allow_flagging="never", title=title, ) youtube = gr.Interface( fn=transcribe_yt, inputs=[ gr.inputs.Textbox(lines=1, placeholder="Paste the URL to a YouTube video here", label="YouTube URL"), gr.inputs.Radio(["transcribe", "translate"], label="Task", default="transcribe"), gr.inputs.Checkbox(default=True, label="Group by speaker"), ], outputs=[ gr.outputs.HTML(label="Video"), gr.outputs.Textbox(label="Transcription").style(show_copy_button=True), ], allow_flagging="never", title=title) demo = gr.Blocks() with demo: gr.TabbedInterface([microphone, audio_file, youtube], ["Microphone", "Audio File", "YouTube"]) # demo.queue(max_size=10) demo.launch()