extra offline file

app.py

@@ -1,25 +1,18 @@
 import os
 import time
-from typing import Optional, Any
 
 import gradio as gr
 from loguru import logger
-from online import transcribe, clear_ui, change_stt_model
-from constants import (
-    MINUTES_KEEP,
-)
-from sdk import SDKWrapper
-from audio_tools import spec_image
-import shutil
-import tempfile
-from aic_dataset import ALL_FILES, get_local_mix_path, download_transcript
-from transcribe import transcribe_and_evaluate, transcribe_file
+
+from constants import MINUTES_KEEP
+from aic_dataset import ALL_FILES, get_local_mix_path
+
+from online import transcribe as online_transcribe, clear_ui as online_clear_ui, change_stt_model
+from offline import transcribe_with_original, transcribe_no_original, denoise_audio, cleanup, start_processing
 
 # ===============================
 # Temporary File & Cache Management
 # ===============================
-
-
 def cleanup_tmp(minutes_keep: int = MINUTES_KEEP, filter: list[str] = []):
     skipped = 0
     removed = 0
@@ -29,10 +22,7 @@ def cleanup_tmp(minutes_keep: int = MINUTES_KEEP, filter: list[str] = []):
                 f = os.path.join(root, name)
                 is_old = (time.time() - os.path.getmtime(f)) / 60 > minutes_keep
                 filtered = any(filt in f for filt in filter)
-                if filtered:
-                    skipped += 1
-                    continue
-                if not is_old:
+                if filtered or not is_old:
                     skipped += 1
                     continue
                 try:
@@ -43,321 +33,138 @@ def cleanup_tmp(minutes_keep: int = MINUTES_KEEP, filter: list[str] = []):
     logger.info(f"Cleanup tmp complete. Removed {removed} files, skipped {skipped} files.")
 
 
-# ===============================
-# Interface Logic
-# ===============================
-
-def transcribe_with_original(audio_file_path: str, file_stem: str, streamer_type: str = "deepgram") -> tuple[str, Any, Any]:
-    """
-    Transcribe an audio file and compute WER against a reference transcript.
-
-    Args:
-        audio_file_path (str): Path to WAV file
-        file_stem (str): Base filename for the audio file
-        streamer_type (str): "soniox" or "deepgram"
-    Returns:
-        tuple[str, float, Any]: Transcript text, Word Error Rate (WER), and visibility update for the original transcript
-    """
-    original_transcript = download_transcript(file_stem)
-    transcript, wer = transcribe_and_evaluate(audio_file_path, original_transcript, streamer_type)
-    wer_box = gr.update(value=wer, visible=True)
-    original_transcript_update = gr.update(value=original_transcript, visible=True)
-    return transcript, wer_box, original_transcript_update
-
-def transcribe_no_original(audio_file_path: str, stt_model: str = "deepgram") -> tuple[str, Any, Any]:
-    """
-    Transcribe an audio file without a reference transcript.
-
-    Args:
-        audio_file_path (str): Path to WAV file
-        stt_model (str): "soniox" or "deepgram"
-    Returns:
-        tuple[str, float, Any]: Transcript text, Word Error Rate (WER), and visibility update for the original transcript
-    """
-    transcript = transcribe_file(audio_file_path, stt_model)
-    visibility = gr.update(visible=False)
-    return transcript, visibility, visibility
-
-
-def create_results_title(path: str) -> str:
-    file_name = os.path.basename(path)
-    if not file_name:
-        return "## Results"
-    return f"## Results for {file_name}"
-
-
-def denoise_audio(
-    sample_path: str,
-    enhancement_level: float = 50.0,
-) -> tuple[Optional[str], Optional[str], Optional[str]]:
-    gr.Info(
-        "Processing started. This may take a moment. Please do not refresh or close the window."
-    )
-    base, ext = os.path.splitext(sample_path)
-    enhanced_path = f"{base}_enhanced{ext}"
-    noisy_path = f"{base}_noisy{ext}"
-    noisy_spec_path = f"{base}_noisy_spectrogram.png"
-    enhanced_spec_path = f"{base}_enhanced_spectrogram.png"
-    noisy_path = sample_path
-    try:
-        sdk = SDKWrapper(os.getenv("SECRET_SDK_KEY"))
-        sdk.init_processor(sample_rate=16000, enhancement_level=enhancement_level / 100)
-        sdk.process_file(noisy_path, enhanced_path)
-    except Exception as e:
-        gr.Warning(f"{e}")
-        delete_related_files(sample_path)
-        return None, None, None
-    noisy_im = spec_image(noisy_path)
-    noisy_im.save(noisy_spec_path)
-    enhanced_im = spec_image(enhanced_path)
-    enhanced_im.save(enhanced_spec_path)
-    print(f"Enhancement complete. id: {base}")
-    return enhanced_path, enhanced_spec_path, noisy_spec_path
-
-
-def toggle_SNR(choice: str):
-    if choice == "None":
-        return gr.update(visible=False, value="None")
-    else:
-        return gr.update(visible=True, value="10")
-
-
-def delete_related_files(path: str):
-    """
-    Deletes all files in /tmp containing the base filename of the given path.
-    """
-    filename_no_ext = os.path.splitext(os.path.basename(path))[0]
-    base_dir = "/tmp"
-    deleted = 0
-    for root, _, files in os.walk(base_dir):
-        for f in files:
-            if filename_no_ext in f:
-                full_path = os.path.join(root, f)
-                try:
-                    os.remove(full_path)
-                    deleted += 1
-                except Exception as e:
-                    logger.warning(f"Failed to delete file {full_path}: {e}")
-    if deleted == 0:
-        logger.info(f"No files found to delete containing '{filename_no_ext}' in {base_dir}")
-    logger.info(f"Deleted {deleted} files related to the last enhancement '{filename_no_ext}'.")
-
-
-def cleanup(last_enhancement: str, last_audio_file: str = "", new_audio_file: str = ""):
-    """
-    Deletes
-        - all enhancement files (usually .png & .wav) in the /tmp directory from the last enhancement
-        - the last uploaded audio file if a new one is uploaded
-        - other files in /tmp older than 2 hours to help manage disk space.
-
-    Note:
-        Consider adding a flag to files to indicate whether files can be deleted, if disk usage becomes an issue.
-    """
-    # Delete the last uploaded audio file if a new one is uploaded
-    if last_audio_file and last_audio_file != new_audio_file and os.path.exists(last_audio_file):
-        try:
-            os.remove(last_audio_file)
-            logger.info(f"Deleted last uploaded audio file: {last_audio_file}")
-        except Exception as e:
-            logger.warning(f"Failed to delete last uploaded audio file {last_audio_file}: {e}")
-    if last_enhancement:
-        delete_related_files(last_enhancement)
-    cleanup_tmp(minutes_keep=120)
-
-
-def start_processing(sample_path: str) -> tuple[str, str, Any, str]:
-    success = True
-    result_title = create_results_title(sample_path)
-    if not sample_path or not os.path.exists(sample_path):
-        gr.Warning("Please upload an audio sample or use the microphone input.")
-        success = False
-    if not os.getenv("SECRET_SDK_KEY"):
-        gr.Warning("No SDK key provided. Please contact us at https://ai-coustics.com/contact/.")
-        success = False
-    if not success:
-        raise ValueError("Missing audio sample or API/SDK key. Processing cannot start.")
-    # Generate a new /tmp path with a unique filename using tempfile
-    ext = os.path.splitext(sample_path)[1]
-    with tempfile.NamedTemporaryFile(delete=False, suffix=ext, dir="/tmp") as tmp_file:
-        shutil.copy(sample_path, tmp_file.name)
-        input_enhancement_path = tmp_file.name
-    return (        
-        input_enhancement_path,
-        sample_path,
-        gr.update(visible=False),
-        result_title,
-    )
-
-
-
 # ===============================
 # Gradio UI Layout
 # ===============================
 with gr.Blocks() as demo:
     input_enhancement = gr.State()
     last_audio_file = gr.State()
-    
+
     gr.HTML(
-            '<a href="https://ai-coustics.com/" target="_blank">'
-            '<img src="https://mintcdn.com/ai-coustics/Sxcrv8jVSE2qWMR1/logo/dark.svg?fit=max&auto=format&n=Sxcrv8jVSE2qWMR1&q=85&s=7f26caaf21e963912961cbd8541e6d84" alt="ai-coustics Logo" width="400" style="display: block; margin: 0 auto;">'
-            "</a>"
-        )
+        '<a href="https://ai-coustics.com/" target="_blank">'
+        '<img src="https://mintcdn.com/ai-coustics/Sxcrv8jVSE2qWMR1/logo/dark.svg?fit=max&auto=format&n=Sxcrv8jVSE2qWMR1&q=85&s=7f26caaf21e963912961cbd8541e6d84" '
+        'alt="ai-coustics Logo" width="400" style="display: block; margin: 0 auto;">'
+        "</a>"
+    )
     gr.Markdown(open("docs/intro.md", "r", encoding="utf-8").read())
+
+    # ✅ Global controls (shared by both tabs)
     stt_model = gr.Radio(label="STT Model", choices=["Deepgram", "Soniox"], value="Deepgram", interactive=True)
     enhancement_level = gr.Slider(
-                        minimum=0,
-                        maximum=100,
-                        step=1,
-                        value=100,
-                        label="Enhancement level (%)",
-                        scale=2,
+        minimum=0,
+        maximum=100,
+        step=1,
+        value=100,
+        label="Enhancement level (%)",
+        scale=2,
     )
-    # ===== Tabs =====
+
+    # Online STT streamer swap uses the same global control
+    stt_model.change(fn=change_stt_model, inputs=stt_model, outputs=[])
+
     with gr.Tabs(elem_classes="main-tabs"):
         # =========================
-        # OFFLINE TAB (FULL APP)
+        # OFFLINE TAB
         # =========================
         with gr.Tab("Offline", elem_classes="tab-offline"):
-            # ---- Input ----
             with gr.Group(elem_classes="panel"):
                 with gr.Tab("Upload", elem_classes="upload-tab"):
-                    audio_file_upload = gr.Audio(
-                        type="filepath",
-                        sources=["upload", "microphone"],
-                        
-                    )
-
+                    audio_file_upload = gr.Audio(type="filepath", sources=["upload", "microphone"])
                     enhance_btn_for_upload = gr.Button("Enhance", scale=2)
+
                 with gr.Tab("AIC Dataset", elem_classes="dataset-tab"):
-                    dataset_dropdown = gr.Dropdown(
-                        choices=ALL_FILES,
-                        label="Choose sample",
-                        value=None,
-                    )
-                    audio_file_from_dataset = gr.Audio(
-                        type="filepath",
-                        interactive=False,
-                    )
-                    
+                    dataset_dropdown = gr.Dropdown(choices=ALL_FILES, label="Choose sample", value=None)
+                    audio_file_from_dataset = gr.Audio(type="filepath", interactive=False)
                     enhance_btn_for_dataset = gr.Button("Enhance", scale=2)
 
             with gr.Group(elem_classes="panel results-card", visible=False) as results_card:
                 result_title = gr.Markdown("")
+                enhanced_audio = gr.Audio(type="filepath", interactive=False)
 
-                # Full-width audio at top of the card
-                enhanced_audio = gr.Audio(
-                    type="filepath",
-                    interactive=False
-                )
-
-                # Two columns inside one big card
                 with gr.Row(equal_height=True, elem_classes="results-row"):
-                    # LEFT: Spectrograms
                     with gr.Column(scale=5, min_width=320, elem_classes="results-left"):
-                        noisy_image = gr.Image(
-                            label="Input spectrogram",
-                            format="png",
-                            type="filepath",
-                        )
-                        enhanced_image = gr.Image(
-                            label="Enhanced spectrogram",
-                            format="png",
-                            type="filepath",
-                        )
-                        
-                    # RIGHT: Text + WER
+                        noisy_image = gr.Image(label="Input spectrogram", format="png", type="filepath")
+                        enhanced_image = gr.Image(label="Enhanced spectrogram", format="png", type="filepath")
+
                     with gr.Column(scale=5, min_width=320, elem_classes="results-right"):
-                        original_transcript = gr.Textbox(
-                            label="Original transcript",
-                            lines=3,
-                            interactive=False,
-                        )
-                        enhanced_transcript = gr.Textbox(
-                            label="Enhanced transcript",
-                            lines=3,
-                            interactive=False,
-                        )
-                        wer_box = gr.Number(
-                            label="Word Error Rate (WER)",
-                            interactive=False,
-                        )
+                        original_transcript = gr.Textbox(label="Original transcript", lines=3, interactive=False)
+                        enhanced_transcript = gr.Textbox(label="Enhanced transcript", lines=3, interactive=False)
+                        wer_box = gr.Number(label="Word Error Rate (WER)", interactive=False)
+
+            # Wiring (offline)
+            dataset_dropdown.change(get_local_mix_path, inputs=dataset_dropdown, outputs=[audio_file_from_dataset])
+
+            enhance_btn_for_dataset.click(
+                cleanup,
+                inputs=[input_enhancement, last_audio_file, audio_file_from_dataset],
+                outputs=None,
+            ).then(
+                start_processing,
+                inputs=audio_file_from_dataset,
+                outputs=[input_enhancement, last_audio_file, results_card, result_title],
+            ).success(
+                denoise_audio,
+                inputs=[input_enhancement, enhancement_level],
+                outputs=[enhanced_audio, enhanced_image, noisy_image],
+            ).then(
+                transcribe_with_original,
+                inputs=[enhanced_audio, dataset_dropdown, stt_model],
+                outputs=[enhanced_transcript, wer_box, original_transcript],
+            ).then(
+                lambda: gr.update(visible=True),
+                inputs=None,
+                outputs=results_card,
+            )
+
+            enhance_btn_for_upload.click(
+                cleanup,
+                inputs=[input_enhancement, last_audio_file, audio_file_upload],
+                outputs=None,
+            ).then(
+                start_processing,
+                inputs=audio_file_upload,
+                outputs=[input_enhancement, last_audio_file, results_card, result_title],
+            ).success(
+                denoise_audio,
+                inputs=[input_enhancement, enhancement_level],
+                outputs=[enhanced_audio, enhanced_image, noisy_image],
+            ).then(
+                transcribe_no_original,
+                inputs=[enhanced_audio, stt_model],
+                outputs=[enhanced_transcript, wer_box, original_transcript],
+            ).then(
+                lambda: gr.update(visible=True),
+                inputs=None,
+                outputs=results_card,
+            )
 
         # =========================
-        # ONLINE TAB (PLACEHOLDER)
+        # ONLINE TAB
         # =========================
         with gr.Tab("Online", elem_classes="tab-online"):
             with gr.Group(elem_classes="panel"):
                 stream_state = gr.State(None)
-                audio = gr.Audio(sources=["microphone"], streaming=True)
+                audio_stream = gr.Audio(sources=["microphone"], streaming=True)
+                with gr.Group(elem_classes="panel"):
+                    with gr.Column(scale=5, min_width=320):
+                        enhanced_text = gr.Textbox(label="Enhanced Transcribed Text", lines=6)
+                    with gr.Column(scale=5, min_width=320):
+                        raw_text = gr.Textbox(label="Raw Transcribed Text", lines=6)
                 clear_btn = gr.Button("Clear")
-                enhanced_text = gr.Textbox(label="Enhanced Transcribed Text", lines=6)
-                raw_text = gr.Textbox(label="Raw Transcribed Text", lines=6)
-                audio.stream(
-                    fn=transcribe,
-                    inputs=[stream_state, audio, enhancement_level],
+            
+
+                audio_stream.stream(
+                    fn=online_transcribe,
+                    inputs=[stream_state, audio_stream, enhancement_level],
                     outputs=[stream_state, enhanced_text, raw_text],
                     stream_every=0.05,
                 )
 
                 clear_btn.click(
-                    fn=clear_ui,
+                    fn=online_clear_ui,
                     outputs=[stream_state, enhanced_text, raw_text],
                 )
-                stt_model.change(
-                    fn=change_stt_model,
-                    inputs=stt_model,
-                )
-                # ===== Events / Wiring =====
-    dataset_dropdown.change(get_local_mix_path, inputs=dataset_dropdown, outputs=[audio_file_from_dataset])
 
-    #dataset_dropdown.change(create_results_title, inputs=dataset_dropdown, outputs=result_title)
 
-
-    enhance_btn_for_dataset.click(
-        cleanup,
-        inputs=[input_enhancement, last_audio_file, audio_file_from_dataset],
-        outputs=None,
-    ).then(
-        start_processing,
-        inputs=audio_file_from_dataset,
-        outputs=[input_enhancement, last_audio_file, results_card,result_title],
-    ).success(
-        denoise_audio,
-        inputs=[input_enhancement, enhancement_level],
-        outputs=[enhanced_audio, enhanced_image, noisy_image],
-    ).then(transcribe_with_original, 
-           inputs=[enhanced_audio, dataset_dropdown, stt_model], 
-           outputs=[enhanced_transcript, wer_box, original_transcript]
-    ).then(
-        lambda: gr.update(visible=True),
-        inputs=None,
-        outputs=results_card,
-    )
-    
-    
-    enhance_btn_for_upload.click(
-        cleanup,
-        inputs=[input_enhancement, last_audio_file, audio_file_upload],
-        outputs=None,
-    ).then(
-        start_processing,
-        inputs=audio_file_upload,
-        outputs=[input_enhancement, last_audio_file, results_card,result_title],
-    ).success(
-        denoise_audio,
-        inputs=[input_enhancement, enhancement_level],
-        outputs=[enhanced_audio, enhanced_image, noisy_image],
-    ).then(
-        transcribe_no_original, 
-        inputs=[enhanced_audio, stt_model], 
-        outputs=[enhanced_transcript, wer_box, original_transcript]
-    ).then(
-        lambda: gr.update(visible=True),
-        inputs=None,
-        outputs=results_card,
-    )       
-    
-    
-    
 cleanup_tmp(minutes_keep=0, filter=[])
 demo.launch(allowed_paths=["/tmp", "/"])

offline.py

@@ -0,0 +1,112 @@
+import os
+from typing import Optional, Any
+
+import gradio as gr
+from loguru import logger
+
+from sdk import SDKWrapper
+from audio_tools import spec_image
+import shutil
+import tempfile
+from aic_dataset import download_transcript
+from transcribe import transcribe_and_evaluate, transcribe_file
+
+
+def transcribe_with_original(
+    audio_file_path: str,
+    file_stem: str,
+    streamer_type: str = "deepgram",
+) -> tuple[str, Any, Any]:
+    original_transcript = download_transcript(file_stem)
+    transcript, wer = transcribe_and_evaluate(audio_file_path, original_transcript, streamer_type)
+    wer_box = gr.update(value=wer, visible=True)
+    original_transcript_update = gr.update(value=original_transcript, visible=True)
+    return transcript, wer_box, original_transcript_update
+
+
+def transcribe_no_original(audio_file_path: str, stt_model: str = "deepgram") -> tuple[str, Any, Any]:
+    transcript = transcribe_file(audio_file_path, stt_model)
+    hidden = gr.update(visible=False)
+    return transcript, hidden, hidden
+
+
+def create_results_title(path: str) -> str:
+    file_name = os.path.basename(path)
+    return f"## Results for {file_name}" if file_name else "## Results"
+
+
+# ===============================
+# Enhancement (offline)
+# ===============================
+def denoise_audio(
+    sample_path: str,
+    enhancement_level: float = 50.0,
+) -> tuple[Optional[str], Optional[str], Optional[str]]:
+    gr.Info("Processing started. This may take a moment. Please do not refresh or close the window.")
+
+    base, ext = os.path.splitext(sample_path)
+    enhanced_path = f"{base}_enhanced{ext}"
+    noisy_spec_path = f"{base}_noisy_spectrogram.png"
+    enhanced_spec_path = f"{base}_enhanced_spectrogram.png"
+
+    try:
+        sdk = SDKWrapper(os.getenv("SECRET_SDK_KEY"))
+        sdk.init_processor(sample_rate=16000, enhancement_level=float(enhancement_level) / 100.0)
+        sdk.process_file(sample_path, enhanced_path)
+    except Exception as e:
+        gr.Warning(f"{e}")
+        delete_related_files(sample_path)
+        return None, None, None
+
+    spec_image(sample_path).save(noisy_spec_path)
+    spec_image(enhanced_path).save(enhanced_spec_path)
+
+    return enhanced_path, enhanced_spec_path, noisy_spec_path
+
+
+def delete_related_files(path: str):
+    filename_no_ext = os.path.splitext(os.path.basename(path))[0]
+    base_dir = "/tmp"
+    deleted = 0
+    for root, _, files in os.walk(base_dir):
+        for f in files:
+            if filename_no_ext in f:
+                full_path = os.path.join(root, f)
+                try:
+                    os.remove(full_path)
+                    deleted += 1
+                except Exception as e:
+                    logger.warning(f"Failed to delete file {full_path}: {e}")
+    logger.info(f"Deleted {deleted} files related to '{filename_no_ext}'.")
+
+
+def cleanup(last_enhancement: str, last_audio_file: str = "", new_audio_file: str = ""):
+    # delete last uploaded audio if a new one is uploaded
+    if last_audio_file and last_audio_file != new_audio_file and os.path.exists(last_audio_file):
+        try:
+            os.remove(last_audio_file)
+            logger.info(f"Deleted last uploaded audio file: {last_audio_file}")
+        except Exception as e:
+            logger.warning(f"Failed to delete last uploaded audio file {last_audio_file}: {e}")
+
+    if last_enhancement:
+        delete_related_files(last_enhancement)
+
+    cleanup_tmp(minutes_keep=120)
+
+
+def start_processing(sample_path: str) -> tuple[str, str, Any, str]:
+    if not sample_path or not os.path.exists(sample_path):
+        raise ValueError("Missing audio sample. Please upload an audio sample or use the microphone input.")
+
+    if not os.getenv("SECRET_SDK_KEY"):
+        raise ValueError("No SDK key provided. Please contact us at https://ai-coustics.com/contact/.")
+
+    result_title = create_results_title(sample_path)
+
+    ext = os.path.splitext(sample_path)[1]
+    with tempfile.NamedTemporaryFile(delete=False, suffix=ext, dir="/tmp") as tmp_file:
+        shutil.copy(sample_path, tmp_file.name)
+        input_enhancement_path = tmp_file.name
+
+    return input_enhancement_path, sample_path, gr.update(visible=False), result_title