Adding CLI

This is similar to the CLI in Whisper, but it also supports
downloading URLs (also playlists), and using a VAD.

app.py

@@ -53,7 +53,7 @@ class WhisperTranscriber:
         self.inputAudioMaxDuration = inputAudioMaxDuration
         self.deleteUploadedFiles = deleteUploadedFiles
 
-    def transcribe_file(self, modelName, languageName, urlData, uploadFile, microphoneData, task, vad, vadMergeWindow, vadMaxMergeSize, vadPadding):
+    def transcribe_webui(self, modelName, languageName, urlData, uploadFile, microphoneData, task, vad, vadMergeWindow, vadMaxMergeSize, vadPadding):
         try:
             source, sourceName = self.__get_source(urlData, uploadFile, microphoneData)
             
@@ -67,54 +67,14 @@ class WhisperTranscriber:
                     model = whisper.load_model(selectedModel)
                     self.model_cache[selectedModel] = model
 
-                # Callable for processing an audio file
-                whisperCallable = lambda audio : model.transcribe(audio, language=selectedLanguage, task=task)
-
-                # The results
-                if (vad == 'silero-vad'):
-                    # Use Silero VAD and include gaps
-                    if (self.vad_model is None):
-                        self.vad_model = VadSileroTranscription()
-
-                    process_gaps = VadSileroTranscription(transcribe_non_speech = True, 
-                                    max_silent_period=vadMergeWindow, max_merge_size=vadMaxMergeSize, 
-                                    segment_padding_left=vadPadding, segment_padding_right=vadPadding, copy=self.vad_model)
-                    result = process_gaps.transcribe(source, whisperCallable)
-                elif (vad == 'silero-vad-skip-gaps'):
-                    # Use Silero VAD 
-                    if (self.vad_model is None):
-                        self.vad_model = VadSileroTranscription()
-                        
-                    skip_gaps = VadSileroTranscription(transcribe_non_speech = False, 
-                                    max_silent_period=vadMergeWindow, max_merge_size=vadMaxMergeSize, 
-                                    segment_padding_left=vadPadding, segment_padding_right=vadPadding, copy=self.vad_model)
-                    result = skip_gaps.transcribe(source, whisperCallable)
-                elif (vad == 'periodic-vad'):
-                    # Very simple VAD - mark every 5 minutes as speech. This makes it less likely that Whisper enters an infinite loop, but
-                    # it may create a break in the middle of a sentence, causing some artifacts.
-                    periodic_vad = VadPeriodicTranscription(periodic_duration=vadMaxMergeSize)
-                    result = periodic_vad.transcribe(source, whisperCallable)
-                else:
-                    # Default VAD
-                    result = whisperCallable(source)
-
-                text = result["text"]
-
-                language = result["language"]
-                languageMaxLineWidth = self.__get_max_line_width(language)
-
-                print("Max line width " + str(languageMaxLineWidth))
-                vtt = self.__get_subs(result["segments"], "vtt", languageMaxLineWidth)
-                srt = self.__get_subs(result["segments"], "srt", languageMaxLineWidth)
-
-                # Files that can be downloaded
+                # Execute whisper
+                result = self.transcribe_file(model, source, selectedLanguage, task, vad, vadMergeWindow, vadMaxMergeSize, vadPadding)
+
+                # Write result
                 downloadDirectory = tempfile.mkdtemp()
+                
                 filePrefix = slugify(sourceName, allow_unicode=True)
-
-                download = []
-                download.append(self.__create_file(srt, downloadDirectory, filePrefix + "-subs.srt"));
-                download.append(self.__create_file(vtt, downloadDirectory, filePrefix + "-subs.vtt"));
-                download.append(self.__create_file(text, downloadDirectory, filePrefix + "-transcript.txt"));
+                download, text, vtt = self.write_result(result, filePrefix, downloadDirectory)
 
                 return download, text, vtt
 
@@ -127,13 +87,68 @@ class WhisperTranscriber:
         except ExceededMaximumDuration as e:
             return [], ("[ERROR]: Maximum remote video length is " + str(e.maxDuration) + "s, file was " + str(e.videoDuration) + "s"), "[ERROR]"
 
+    def transcribe_file(self, model: whisper.Whisper, audio_path: str, language: str, task: str = None, vad: str = None, 
+                        vadMergeWindow: float = 5, vadMaxMergeSize: float = 150, vadPadding: float = 1, **decodeOptions: dict):
+        # Callable for processing an audio file
+        whisperCallable = lambda audio : model.transcribe(audio, language=language, task=task, **decodeOptions)
+
+        # The results
+        if (vad == 'silero-vad'):
+            # Use Silero VAD and include gaps
+            if (self.vad_model is None):
+                self.vad_model = VadSileroTranscription()
+
+            process_gaps = VadSileroTranscription(transcribe_non_speech = True, 
+                            max_silent_period=vadMergeWindow, max_merge_size=vadMaxMergeSize, 
+                            segment_padding_left=vadPadding, segment_padding_right=vadPadding, copy=self.vad_model)
+            result = process_gaps.transcribe(audio_path, whisperCallable)
+        elif (vad == 'silero-vad-skip-gaps'):
+            # Use Silero VAD 
+            if (self.vad_model is None):
+                self.vad_model = VadSileroTranscription()
+                
+            skip_gaps = VadSileroTranscription(transcribe_non_speech = False, 
+                            max_silent_period=vadMergeWindow, max_merge_size=vadMaxMergeSize, 
+                            segment_padding_left=vadPadding, segment_padding_right=vadPadding, copy=self.vad_model)
+            result = skip_gaps.transcribe(audio_path, whisperCallable)
+        elif (vad == 'periodic-vad'):
+            # Very simple VAD - mark every 5 minutes as speech. This makes it less likely that Whisper enters an infinite loop, but
+            # it may create a break in the middle of a sentence, causing some artifacts.
+            periodic_vad = VadPeriodicTranscription(periodic_duration=vadMaxMergeSize)
+            result = periodic_vad.transcribe(audio_path, whisperCallable)
+        else:
+            # Default VAD
+            result = whisperCallable(audio_path)
+
+        return result
+
+    def write_result(self, result: dict, source_name: str, output_dir: str):
+        if not os.path.exists(output_dir):
+            os.makedirs(output_dir)
+
+        text = result["text"]
+        language = result["language"]
+        languageMaxLineWidth = self.__get_max_line_width(language)
+
+        print("Max line width " + str(languageMaxLineWidth))
+        vtt = self.__get_subs(result["segments"], "vtt", languageMaxLineWidth)
+        srt = self.__get_subs(result["segments"], "srt", languageMaxLineWidth)
+
+        output_files = []
+        output_files.append(self.__create_file(srt, output_dir, source_name + "-subs.srt"));
+        output_files.append(self.__create_file(vtt, output_dir, source_name + "-subs.vtt"));
+        output_files.append(self.__create_file(text, output_dir, source_name + "-transcript.txt"));
+
+        return output_files, text, vtt
+
     def clear_cache(self):
         self.model_cache = dict()
+        self.vad_model = None
 
     def __get_source(self, urlData, uploadFile, microphoneData):
         if urlData:
             # Download from YouTube
-            source = download_url(urlData, self.inputAudioMaxDuration)
+            source = download_url(urlData, self.inputAudioMaxDuration)[0]
         else:
             # File input
             source = uploadFile if uploadFile is not None else microphoneData
@@ -194,7 +209,7 @@ def create_ui(inputAudioMaxDuration, share=False, server_name: str = None):
 
     ui_article = "Read the [documentation here](https://huggingface.co/spaces/aadnk/whisper-webui/blob/main/docs/options.md)"
 
-    demo = gr.Interface(fn=ui.transcribe_file, description=ui_description, article=ui_article, inputs=[
+    demo = gr.Interface(fn=ui.transcribe_webui, description=ui_description, article=ui_article, inputs=[
         gr.Dropdown(choices=["tiny", "base", "small", "medium", "large"], value="medium", label="Model"),
         gr.Dropdown(choices=sorted(LANGUAGES), label="Language"),
         gr.Text(label="URL (YouTube, etc.)"),

cli.py

@@ -0,0 +1,108 @@
+import argparse
+import os
+import pathlib
+from urllib.parse import urlparse
+import warnings
+import numpy as np
+
+import whisper
+
+import torch
+from app import LANGUAGES, WhisperTranscriber
+from src.download import download_url
+
+from src.utils import optional_float, optional_int, str2bool
+
+
+def cli():
+    parser = argparse.ArgumentParser(formatter_class=argparse.ArgumentDefaultsHelpFormatter)
+    parser.add_argument("audio", nargs="+", type=str, help="audio file(s) to transcribe")
+    parser.add_argument("--model", default="small", choices=["tiny", "base", "small", "medium", "large"], help="name of the Whisper model to use")
+    parser.add_argument("--model_dir", type=str, default=None, help="the path to save model files; uses ~/.cache/whisper by default")
+    parser.add_argument("--device", default="cuda" if torch.cuda.is_available() else "cpu", help="device to use for PyTorch inference")
+    parser.add_argument("--output_dir", "-o", type=str, default=".", help="directory to save the outputs")
+    parser.add_argument("--verbose", type=str2bool, default=True, help="whether to print out the progress and debug messages")
+
+    parser.add_argument("--task", type=str, default="transcribe", choices=["transcribe", "translate"], help="whether to perform X->X speech recognition ('transcribe') or X->English translation ('translate')")
+    parser.add_argument("--language", type=str, default=None, choices=sorted(LANGUAGES), help="language spoken in the audio, specify None to perform language detection")
+
+    parser.add_argument("--vad", type=str, default="none", choices=["none", "silero-vad", "silero-vad-skip-gaps", "periodic-vad"], help="The voice activity detection algorithm to use")
+    parser.add_argument("--vad_merge_window", type=optional_float, default=5, help="The window size (in seconds) to merge voice segments")
+    parser.add_argument("--vad_max_merge_size", type=optional_float, default=150, help="The maximum size (in seconds) of a voice segment")
+    parser.add_argument("--vad_padding", type=optional_float, default=1, help="The padding (in seconds) to add to each voice segment")
+
+    parser.add_argument("--temperature", type=float, default=0, help="temperature to use for sampling")
+    parser.add_argument("--best_of", type=optional_int, default=5, help="number of candidates when sampling with non-zero temperature")
+    parser.add_argument("--beam_size", type=optional_int, default=5, help="number of beams in beam search, only applicable when temperature is zero")
+    parser.add_argument("--patience", type=float, default=None, help="optional patience value to use in beam decoding, as in https://arxiv.org/abs/2204.05424, the default (1.0) is equivalent to conventional beam search")
+    parser.add_argument("--length_penalty", type=float, default=None, help="optional token length penalty coefficient (alpha) as in https://arxiv.org/abs/1609.08144, uses simple lengt normalization by default")
+
+    parser.add_argument("--suppress_tokens", type=str, default="-1", help="comma-separated list of token ids to suppress during sampling; '-1' will suppress most special characters except common punctuations")
+    parser.add_argument("--initial_prompt", type=str, default=None, help="optional text to provide as a prompt for the first window.")
+    parser.add_argument("--condition_on_previous_text", type=str2bool, default=True, help="if True, provide the previous output of the model as a prompt for the next window; disabling may make the text inconsistent across windows, but the model becomes less prone to getting stuck in a failure loop")
+    parser.add_argument("--fp16", type=str2bool, default=True, help="whether to perform inference in fp16; True by default")
+
+    parser.add_argument("--temperature_increment_on_fallback", type=optional_float, default=0.2, help="temperature to increase when falling back when the decoding fails to meet either of the thresholds below")
+    parser.add_argument("--compression_ratio_threshold", type=optional_float, default=2.4, help="if the gzip compression ratio is higher than this value, treat the decoding as failed")
+    parser.add_argument("--logprob_threshold", type=optional_float, default=-1.0, help="if the average log probability is lower than this value, treat the decoding as failed")
+    parser.add_argument("--no_speech_threshold", type=optional_float, default=0.6, help="if the probability of the <|nospeech|> token is higher than this value AND the decoding has failed due to `logprob_threshold`, consider the segment as silence")
+
+    args = parser.parse_args().__dict__
+    model_name: str = args.pop("model")
+    model_dir: str = args.pop("model_dir")
+    output_dir: str = args.pop("output_dir")
+    device: str = args.pop("device")
+    os.makedirs(output_dir, exist_ok=True)
+
+    if model_name.endswith(".en") and args["language"] not in {"en", "English"}:
+        warnings.warn(f"{model_name} is an English-only model but receipted '{args['language']}'; using English instead.")
+        args["language"] = "en"
+
+    temperature = args.pop("temperature")
+    temperature_increment_on_fallback = args.pop("temperature_increment_on_fallback")
+    if temperature_increment_on_fallback is not None:
+        temperature = tuple(np.arange(temperature, 1.0 + 1e-6, temperature_increment_on_fallback))
+    else:
+        temperature = [temperature]
+
+    vad = args.pop("vad")
+    vad_merge_window = args.pop("vad_merge_window")
+    vad_max_merge_size = args.pop("vad_max_merge_size")
+    vad_padding = args.pop("vad_padding")
+
+    model = whisper.load_model(model_name, device=device, download_root=model_dir)
+    transcriber = WhisperTranscriber(deleteUploadedFiles=False)
+
+    for audio_path in args.pop("audio"):
+        sources = []
+
+        # Detect URL and download the audio
+        if (uri_validator(audio_path)):
+            # Download from YouTube/URL directly
+            for source_path in  download_url(audio_path, maxDuration=-1, destinationDirectory=output_dir, playlistItems=None):
+                source_name = os.path.basename(source_path)
+                sources.append({ "path": source_path, "name": source_name })
+        else:
+            sources.append({ "path": audio_path, "name": os.path.basename(audio_path) })
+
+        for source in sources:
+            source_path = source["path"]
+            source_name = source["name"]
+
+            result = transcriber.transcribe_file(model, source_path, temperature=temperature, 
+                                                vad=vad, vadMergeWindow=vad_merge_window, vadMaxMergeSize=vad_max_merge_size, 
+                                                vadPadding=vad_padding, **args)
+            
+            transcriber.write_result(result, source_name, output_dir)
+
+    transcriber.clear_cache()
+
+def uri_validator(x):
+    try:
+        result = urlparse(x)
+        return all([result.scheme, result.netloc])
+    except:
+        return False
+
+if __name__ == '__main__':
+    cli()

src/download.py

@@ -1,4 +1,5 @@
 from tempfile import mkdtemp
+from typing import List
 from yt_dlp import YoutubeDL
 
 import yt_dlp
@@ -13,25 +14,28 @@ class FilenameCollectorPP(PostProcessor):
         self.filenames.append(information["filepath"])
         return [], information
 
-def download_url(url: str, maxDuration: int = None):
+def download_url(url: str, maxDuration: int = None, destinationDirectory: str = None, playlistItems: str = "1") -> List[str]: 
     try:
-        return _perform_download(url, maxDuration=maxDuration)
+        return _perform_download(url, maxDuration=maxDuration, outputTemplate=None, destinationDirectory=destinationDirectory, playlistItems=playlistItems)
     except yt_dlp.utils.DownloadError as e:
         # In case of an OS error, try again with a different output template
         if e.msg and e.msg.find("[Errno 36] File name too long") >= 0:
             return _perform_download(url, maxDuration=maxDuration, outputTemplate="%(title).10s %(id)s.%(ext)s")
         pass
 
-def _perform_download(url: str, maxDuration: int = None, outputTemplate: str = None):
-    destinationDirectory = mkdtemp()
+def _perform_download(url: str, maxDuration: int = None, outputTemplate: str = None, destinationDirectory: str = None, playlistItems: str = "1"):
+    # Create a temporary directory to store the downloaded files
+    if destinationDirectory is None:
+        destinationDirectory = mkdtemp()
 
     ydl_opts = {
         "format": "bestaudio/best",
-        'playlist_items': '1',
         'paths': {
             'home': destinationDirectory
         }
     }
+    if (playlistItems):
+        ydl_opts['playlist_items'] = playlistItems
 
     # Add output template if specified
     if outputTemplate:
@@ -53,8 +57,11 @@ def _perform_download(url: str, maxDuration: int = None, outputTemplate: str = N
     if len(filename_collector.filenames) <= 0:
         raise Exception("Cannot download " + url)
 
-    result = filename_collector.filenames[0]
-    print("Downloaded " + result)
+    result = []
+
+    for filename in filename_collector.filenames:
+        result.append(filename)
+        print("Downloaded " + filename)
 
     return result 
 

src/vad.py

@@ -188,7 +188,9 @@ class AbstractTranscription(ABC):
             
             result.append(current_segment)
 
-        last_segment = result[-1]
+        # Add last segment
+        last_segment = segments[-1]
+        result.append(last_segment)
 
         # Also include total duration if specified
         if (total_duration is not None):