opt funcs

app.py

@@ -72,21 +72,19 @@ def download_youtube_audio(url, method_choice):
 
     Args:
         url (str): The YouTube URL.
-        method_choice (str): The method to use for downloading ('yt-dlp', 'pytube', 'youtube-dl').
+        method_choice (str): The method to use for downloading ('yt-dlp', 'pytube').
 
     Returns:
         str: Path to the downloaded audio file, or None if failed.
     """
     methods = {
-        'yt-dlp': youtube_dl_method,
+        'yt-dlp': yt_dlp_method,
         'pytube': pytube_method,
-        'youtube-dl': youtube_dl_classic_method,
-        'yt-dlp-alt': youtube_dl_alternative_method,
     }
     method = methods.get(method_choice)
     if method is None:
         logging.warning(f"Invalid download method for YouTube: {method_choice}. Defaulting to 'yt-dlp'.")
-        method = youtube_dl_method
+        method = yt_dlp_method
     try:
         logging.info(f"Attempting to download YouTube audio using {method_choice}")
         return method(url)
@@ -115,17 +113,64 @@ def youtube_dl_method(url):
         logging.error(f"Error in youtube_dl_method: {str(e)}")
         return None
 
+def yt_dlp_method(url):
+    """
+    Downloads audio using yt-dlp.
+
+    Args:
+        url (str): The YouTube URL.
+
+    Returns:
+        str: Path to the downloaded audio file, or None if failed.
+    """
+    logging.info("Using yt-dlp method")
+    try:
+        ydl_opts = {
+            'format': 'bestaudio/best',
+            'postprocessors': [{
+                'key': 'FFmpegExtractAudio',
+                'preferredcodec': 'mp3',
+                'preferredquality': '192',
+            }],
+            'outtmpl': '%(id)s.%(ext)s',
+            'quiet': True,
+            'no_warnings': True,
+        }
+        with yt_dlp.YoutubeDL(ydl_opts) as ydl:
+            info = ydl.extract_info(url, download=True)
+            output_file = ydl.prepare_filename(info)
+            if output_file.endswith('.webm') or output_file.endswith('.mp4'):
+                output_file = output_file.rsplit('.', 1)[0] + '.mp3'
+            logging.info(f"Downloaded YouTube audio: {output_file}")
+            return output_file
+    except Exception as e:
+        logging.error(f"Error in yt_dlp_method: {str(e)}")
+        return None
+        
 def pytube_method(url):
+    """
+    Downloads audio using pytube.
+
+    Args:
+        url (str): The YouTube URL.
+
+    Returns:
+        str: Path to the downloaded audio file, or None if failed.
+    """
     logging.info("Using pytube method")
-    from pytube import YouTube
-    yt = YouTube(url)
-    audio_stream = yt.streams.filter(only_audio=True).first()
-    out_file = audio_stream.download()
-    base, ext = os.path.splitext(out_file)
-    new_file = base + '.mp3'
-    os.rename(out_file, new_file)
-    logging.info(f"Downloaded and converted audio to: {new_file}")
-    return new_file
+    try:
+        from pytube import YouTube
+        yt = YouTube(url)
+        audio_stream = yt.streams.filter(only_audio=True).first()
+        out_file = audio_stream.download()
+        base, ext = os.path.splitext(out_file)
+        new_file = base + '.mp3'
+        os.rename(out_file, new_file)
+        logging.info(f"Downloaded and converted audio to: {new_file}")
+        return new_file
+    except Exception as e:
+        logging.error(f"Error in pytube_method: {str(e)}")
+        return None
 
 def youtube_dl_classic_method(url):
     logging.info("Using youtube-dl classic method")
@@ -179,6 +224,32 @@ def aria2_method(url):
     logging.info(f"Downloaded audio to: {output_file}")
     return output_file
 
+def requests_method(url):
+    """
+    Downloads audio using the requests library.
+
+    Args:
+        url (str): The URL of the audio file.
+
+    Returns:
+        str: Path to the downloaded audio file, or None if failed.
+    """
+    try:
+        response = requests.get(url, stream=True)
+        if response.status_code == 200:
+            with tempfile.NamedTemporaryFile(delete=False, suffix=".mp3") as temp_file:
+                for chunk in response.iter_content(chunk_size=8192):
+                    if chunk:
+                        temp_file.write(chunk)
+            logging.info(f"Downloaded direct audio to: {temp_file.name}")
+            return temp_file.name
+        else:
+            logging.error(f"Failed to download audio from {url} with status code {response.status_code}")
+            return None
+    except Exception as e:
+        logging.error(f"Error in requests_method: {str(e)}")
+        return None
+        
 def download_direct_audio(url, method_choice):
     """
     Downloads audio from a direct URL using the specified method.
@@ -191,35 +262,44 @@ def download_direct_audio(url, method_choice):
         str: Path to the downloaded audio file, or None if failed.
     """
     logging.info(f"Downloading direct audio from: {url} using method: {method_choice}")
-    if method_choice == 'wget':
-        return wget_method(url)
-    else:
-        try:
-            response = requests.get(url, stream=True)
-            if response.status_code == 200:
-                with tempfile.NamedTemporaryFile(delete=False, suffix=".mp3") as temp_file:
-                    for chunk in response.iter_content(chunk_size=8192):
-                        if chunk:
-                            temp_file.write(chunk)
-                logging.info(f"Downloaded direct audio to: {temp_file.name}")
-                return temp_file.name
-            else:
-                raise Exception(f"Failed to download audio from {url} with status code {response.status_code}")
-        except Exception as e:
-            logging.error(f"Error downloading direct audio: {str(e)}")
-            return None
+    methods = {
+        'wget': wget_method,
+        'requests': requests_method,
+    }
+    method = methods.get(method_choice)
+    if method is None:
+        logging.warning(f"Invalid download method: {method_choice}. Defaulting to 'requests'.")
+        method = requests_method
+    try:
+        return method(url)
+    except Exception as e:
+        logging.error(f"Error downloading direct audio: {str(e)}")
+        return None
 
 def wget_method(url):
+    """
+    Downloads audio using the wget command-line tool.
+
+    Args:
+        url (str): The URL of the audio file.
+
+    Returns:
+        str: Path to the downloaded audio file, or None if failed.
+    """
     logging.info("Using wget method")
-    output_file = tempfile.mktemp(suffix='.mp3')  
+    output_file = tempfile.mktemp(suffix='.mp3')
     command = ['wget', '-O', output_file, url]
-    subprocess.run(command, check=True, capture_output=True)
-    logging.info(f"Downloaded audio to: {output_file}")
-    return output_file
+    try:
+        subprocess.run(command, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+        logging.info(f"Downloaded audio to: {output_file}")
+        return output_file
+    except Exception as e:
+        logging.error(f"Error in wget_method: {str(e)}")
+        return None
 
 def trim_audio(audio_path, start_time, end_time):
     """
-    Trims an audio file to the specified start and end times.
+    Trims an audio file to the specified start and end times using pydub.
 
     Args:
         audio_path (str): Path to the audio file.
@@ -230,7 +310,7 @@ def trim_audio(audio_path, start_time, end_time):
         str: Path to the trimmed audio file.
 
     Raises:
-        gr.Error: If invalid start or end times are provided.
+        gr.Error: If invalid start or end times are provided or if FFmpeg is not found.
     """
     try:
         logging.info(f"Trimming audio from {start_time} to {end_time}")
@@ -256,6 +336,9 @@ def trim_audio(audio_path, start_time, end_time):
             trimmed_audio.export(temp_audio_file.name, format="wav")
             logging.info(f"Trimmed audio saved to: {temp_audio_file.name}")
         return temp_audio_file.name
+    except FileNotFoundError as e:
+        logging.error(f"FFmpeg not found: {str(e)}")
+        raise gr.Error("FFmpeg not found. Please ensure that FFmpeg is installed and in your system PATH.")
     except Exception as e:
         logging.error(f"Error trimming audio: {str(e)}")
         raise gr.Error(f"Error trimming audio: {str(e)}")
@@ -319,7 +402,7 @@ def transcribe_audio(input_source, pipeline_type, model_id, dtype, batch_size, d
             logging.getLogger().setLevel(logging.INFO)
         else:
             logging.getLogger().setLevel(logging.WARNING)
-                
+
         logging.info(f"Transcription parameters: pipeline_type={pipeline_type}, model_id={model_id}, dtype={dtype}, batch_size={batch_size}, download_method={download_method}")
         verbose_messages = f"Starting transcription with parameters:\nPipeline Type: {pipeline_type}\nModel ID: {model_id}\nData Type: {dtype}\nBatch Size: {batch_size}\nDownload Method: {download_method}\n"
 
@@ -371,21 +454,26 @@ def transcribe_audio(input_source, pipeline_type, model_id, dtype, batch_size, d
             elif pipeline_type == "faster-sequenced":
                 model_or_pipeline = WhisperModel(model_id, device=device, compute_type=dtype)
             elif pipeline_type == "transformers":
-                torch_dtype = torch.float16 if dtype == "float16" else torch.float32
+                # Adjust torch_dtype based on dtype and device
+                if dtype == "float16" and device == "cpu":
+                    torch_dtype = torch.float32
+                elif dtype == "float16":
+                    torch_dtype = torch.float16
+                else:
+                    torch_dtype = torch.float32
+
                 model = AutoModelForSpeechSeq2Seq.from_pretrained(
-                    model_id, torch_dtype=torch_dtype, low_cpu_mem_usage=True
+                    model_id, torch_dtype=torch_dtype
                 )
-                model.to(device)
                 processor = AutoProcessor.from_pretrained(model_id)
                 model_or_pipeline = pipeline(
-                   "automatic-speech-recognition",
+                    "automatic-speech-recognition",
                     model=model,
                     tokenizer=processor.tokenizer,
                     feature_extractor=processor.feature_extractor,
                     chunk_length_s=30,
                     batch_size=batch_size,
                     return_timestamps=True,
-                    torch_dtype=torch_dtype,
                     device=device,
                 )
             else:
@@ -432,11 +520,9 @@ def transcribe_audio(input_source, pipeline_type, model_id, dtype, batch_size, d
         yield f"An error occurred: {str(e)}", "", None
 
     finally:
-        # Clean up temporary files
+        # Clean up temporary audio files
         if audio_path and is_temp_file and os.path.exists(audio_path):
             os.remove(audio_path)
-        if 'transcription_file' in locals() and transcription_file and os.path.exists(transcription_file):
-            os.remove(transcription_file)
 
 with gr.Blocks() as iface:
     gr.Markdown("# Multi-Pipeline Transcription")