EchoScript : app.py, Cascading language filtering, Transcript caching, Per-language translation caching, Tab visibility

app.py

@@ -9,17 +9,28 @@ Services are instantiated lazily (on first use) rather than at import time,
 so the app can start up without needing model weights on disk yet, and so
 this module stays import-safe in environments without network access.
 
-Translation languages on offer depend on whether the person supplies their
-own Anthropic API key for that session:
-
-- No key: translation runs on the local, offline MarianMT backend, so only
-  the languages it can reliably reach are offered (services.translation.
-  MARIAN_TARGET_LANGUAGES).
-- Key supplied: translation runs through Claude, which has no missing-
-  language-pair problem, so the full language list is offered (services.
-  translation.ANTHROPIC_TARGET_LANGUAGES). The key is used only for the
-  request(s) made during this session and is never written to disk,
-  logged, or cached anywhere server-side.
+Two things make repeated clicks cheap:
+
+1. Transcript caching. A `gr.State` holds the last Transcript together with
+   the exact (audio path, time window, source language) signature that
+   produced it. If "Generate Outputs" is clicked again with that signature
+   unchanged -- e.g. only the Outputs checkboxes changed -- transcription
+   is skipped entirely and the cached Transcript is reused.
+2. Per-language translation caching. A second `gr.State` dict caches each
+   Translation by language code, scoped to the current transcript
+   signature. Selecting an additional language only translates that new
+   language; languages already translated (even if briefly deselected and
+   reselected) are reused rather than recomputed.
+
+Available translation languages are cascaded from two things:
+
+- Whether an Anthropic API key is present for this session (see
+  services/translation.py for the offline-vs-Claude tradeoff). The key is
+  used only for the request(s) made during this session and is never
+  written to disk, logged, or cached anywhere server-side.
+- The source language, explicit or detected: translating a language into
+  itself isn't offered. If "Auto Detect" is used, the list is re-filtered
+  once the actual language is known, after transcription.
 """
 
 from __future__ import annotations
@@ -47,7 +58,8 @@ from services.translation import (
 # Safe to share across requests/users: TranscriptionService holds no
 # per-request state, and TranslationService resolves its backend (and
 # takes the API key, if any) fresh on every translate() call rather than
-# storing it -- see services/translation.py.
+# storing it -- see services/translation.py. Per-request caching of
+# results (not the services themselves) lives in gr.State, below.
 # ---------------------------------------------------------------------------
 
 _transcription_service: Optional[TranscriptionService] = None
@@ -81,23 +93,40 @@ def get_translation_service() -> TranslationService:
 _NAME_TO_CODE = {name: code for code, name in SUPPORTED_LANGUAGES.items()}
 SOURCE_LANGUAGE_CHOICES = ["Auto Detect"] + list(SUPPORTED_LANGUAGES.values())
 
-# "Outputs" checkbox labels, for each of the two language sets. The
-# Anthropic set is a superset of the Marian one, so switching a key in/out
-# only ever adds or removes options -- it never renames existing ones.
-_MARIAN_OUTPUT_CHOICES = ["Transcript"] + [f"{name} Translation" for name in MARIAN_TARGET_LANGUAGES.values()]
-_ANTHROPIC_OUTPUT_CHOICES = ["Transcript"] + [f"{name} Translation" for name in ANTHROPIC_TARGET_LANGUAGES.values()]
-DEFAULT_OUTPUTS = ["Transcript", "English Translation"]
-
 # Label -> ISO 639-1 code, built from the full (Anthropic) superset so it
 # resolves correctly regardless of which list is currently offered.
 _TRANSLATION_LABEL_TO_CODE = {
     f"{name} Translation": code for code, name in ANTHROPIC_TARGET_LANGUAGES.items()
 }
 
-# Tabs are pre-built for every language in the full superset (hidden by
-# default) so that toggling the API key never needs to add/remove
-# components -- only which ones are visible changes.
-_TRANSLATION_TAB_ORDER = list(_TRANSLATION_LABEL_TO_CODE.keys())
+# Sections are pre-built for every language in the full superset (hidden by
+# default) so that toggling the API key or source language never needs to
+# add/remove components -- only which ones are visible changes.
+_TRANSLATION_SECTION_ORDER = list(_TRANSLATION_LABEL_TO_CODE.keys())
+
+DEFAULT_OUTPUTS = ["Transcript", "English Translation"]
+
+
+def _compute_output_choices(has_key: bool, exclude_code: Optional[str]) -> list[str]:
+    """The Outputs checkbox list, cascaded from key presence + source language.
+
+    `exclude_code` removes "translate into the language it's already in"
+    from the list -- it's the explicit source language if one was chosen,
+    or the detected language once transcription has run.
+    """
+    pool = ANTHROPIC_TARGET_LANGUAGES if has_key else MARIAN_TARGET_LANGUAGES
+    return ["Transcript"] + [
+        f"{name} Translation" for code, name in pool.items() if code != exclude_code
+    ]
+
+
+def _on_key_or_source_change(api_key: str, source_language_label: str, current_value: list[str]):
+    """Re-cascade the Outputs choices when the API key or source language changes."""
+    has_key = bool((api_key or "").strip())
+    exclude_code = _NAME_TO_CODE.get(source_language_label)  # None when "Auto Detect"
+    choices = _compute_output_choices(has_key, exclude_code)
+    filtered_value = [v for v in (current_value or []) if v in choices] or ["Transcript"]
+    return gr.update(choices=choices, value=filtered_value)
 
 
 def _format_duration(seconds: float) -> str:
@@ -113,20 +142,6 @@ def _write_text_file(text: str, tmp_dir: Path, filename: str) -> str:
     return str(path)
 
 
-def _update_output_choices(api_key: str, current_value: list[str]):
-    """Swap the Outputs checkbox list when the API key field changes.
-
-    With a key: offer the full Anthropic language list. Without: restrict
-    to what the offline MarianMT backend can reliably translate. Any
-    currently-selected language that's no longer offered is dropped from
-    the selection (this only happens when a key is removed).
-    """
-    has_key = bool((api_key or "").strip())
-    choices = _ANTHROPIC_OUTPUT_CHOICES if has_key else _MARIAN_OUTPUT_CHOICES
-    filtered_value = [v for v in (current_value or []) if v in choices] or ["Transcript"]
-    return gr.update(choices=choices, value=filtered_value)
-
-
 # ---------------------------------------------------------------------------
 # Main processing callback
 # ---------------------------------------------------------------------------
@@ -138,6 +153,9 @@ def process_audio(
     source_language_label: str,
     selected_outputs: list[str],
     api_key: str,
+    cached_transcript: Optional[Transcript],
+    cached_signature,
+    cached_translations: dict,
 ):
     if not audio_path:
         raise gr.Error("Please upload an audio file first.")
@@ -148,24 +166,33 @@ def process_audio(
     except AudioError as exc:
         raise gr.Error(str(exc)) from exc
 
-    working_path = audio_path
-    if start is not None or end is not None:
-        try:
-            working_path = extract_window(audio_path, start, end)
-        except AudioError as exc:
-            raise gr.Error(str(exc)) from exc
-
-    language_code = _NAME_TO_CODE.get(source_language_label)  # None = auto-detect
-
-    # Step 1: Audio -> canonical Transcript. This is the only step that
-    # touches the audio; everything below works off `transcript`.
-    transcript: Transcript = get_transcription_service().transcribe(
-        working_path,
-        source_filename=Path(audio_path).name,
-        language=language_code,
-        window_start=start,
-        window_end=end,
-    )
+    source_code = _NAME_TO_CODE.get(source_language_label)  # None = auto-detect
+    signature = (audio_path, start, end, source_code)
+
+    if cached_transcript is not None and cached_signature == signature:
+        # Same file, same window, same forced source language as last time
+        # -- the canonical Transcript hasn't changed, so skip Whisper
+        # entirely and reuse it.
+        transcript = cached_transcript
+    else:
+        working_path = audio_path
+        if start is not None or end is not None:
+            try:
+                working_path = extract_window(audio_path, start, end)
+            except AudioError as exc:
+                raise gr.Error(str(exc)) from exc
+
+        # Step 1: Audio -> canonical Transcript. This is the only step
+        # that touches the audio; everything below works off `transcript`.
+        transcript = get_transcription_service().transcribe(
+            working_path,
+            source_filename=Path(audio_path).name,
+            language=source_code,
+            window_start=start,
+            window_end=end,
+        )
+        cached_signature = signature
+        cached_translations = {}  # old translations were derived from a different transcript
 
     tmp_dir = Path(tempfile.mkdtemp(prefix="echoscript_"))
 
@@ -178,55 +205,98 @@ def process_audio(
         f"**Words:** {transcript.word_count:,}"
     )
 
-    # --- Transcript tab (always computed -- it's the source of truth --
-    # but only exposed as a tab if the user kept "Transcript" checked) ---
+    # Now that the actual language is known (important when "Auto Detect"
+    # was used), re-cascade the Outputs choices to drop "translate into
+    # the language it's already in" and drop that selection if it was
+    # only chosen by default/before detection.
+    has_key = bool((api_key or "").strip())
+    updated_choices = _compute_output_choices(has_key, transcript.language)
+    effective_outputs = [v for v in selected_outputs if v in updated_choices] or ["Transcript"]
+    outputs_update = gr.update(choices=updated_choices, value=effective_outputs)
+
+    # --- Transcript section (always computed -- it's the source of truth
+    # -- but only exposed if the user kept "Transcript" checked) ---
     transcript_text = transcript.text
     transcript_file = _write_text_file(transcript_text, tmp_dir, "transcript.txt")
-    transcript_tab_visible = "Transcript" in selected_outputs
+    transcript_visible = "Transcript" in effective_outputs
 
-    # --- Translation tabs ---
+    # --- Translations: reuse anything already cached for this transcript;
+    # only compute the languages newly selected since the last click. ---
     translation_service = get_translation_service()
-    translation_updates = {}  # label -> (text, file_path)
-    for label in _TRANSLATION_TAB_ORDER:
-        if label not in selected_outputs:
-            translation_updates[label] = (None, None)
-            continue
+    section_updates = {}  # label -> (visible, text, file_path)
 
+    for label in _TRANSLATION_SECTION_ORDER:
         code = _TRANSLATION_LABEL_TO_CODE[label]
-        try:
-            translation = translation_service.translate(transcript, code, api_key=api_key)
-            text = translation.text
+        if label not in effective_outputs:
+            section_updates[label] = (False, "", None)
+            continue
+
+        if code in cached_translations:
+            text = cached_translations[code]
+        else:
+            try:
+                translation = translation_service.translate(transcript, code, api_key=api_key)
+                text = translation.text
+                cached_translations[code] = text
+            except TranslationError as exc:
+                # Surface the failure in that section rather than aborting
+                # every other output that already succeeded. Deliberately
+                # not cached, so it's retried on the next click.
+                text = f"\u26a0\ufe0f Translation failed: {exc}"
+
+        file_path = None
+        if not text.startswith("\u26a0\ufe0f"):
             file_path = _write_text_file(text, tmp_dir, f"{code}.txt")
-        except TranslationError as exc:
-            # Surface the failure in that tab rather than aborting every
-            # other output that already succeeded.
-            text = f"\u26a0\ufe0f Translation failed: {exc}"
-            file_path = None
-        translation_updates[label] = (text, file_path)
+        section_updates[label] = (True, text, file_path)
 
     # --- Subtitles (always derived from the transcript, the canonical
     # source of truth -- never regenerated from audio) ---
     srt_path = _write_text_file(generate_srt(transcript.segments), tmp_dir, "transcript.srt")
     vtt_path = _write_text_file(generate_vtt(transcript.segments), tmp_dir, "transcript.vtt")
 
-    outputs = [dashboard_md, transcript_text, transcript_file, gr.update(visible=transcript_tab_visible)]
+    outputs = [
+        dashboard_md,
+        transcript_text,
+        transcript_file,
+        gr.update(visible=transcript_visible),
+        outputs_update,
+    ]
 
-    for label in _TRANSLATION_TAB_ORDER:
-        text, file_path = translation_updates[label]
-        visible = text is not None
-        outputs += [gr.update(visible=visible), text or "", file_path]
+    for label in _TRANSLATION_SECTION_ORDER:
+        visible, text, file_path = section_updates[label]
+        outputs += [gr.update(visible=visible), text, file_path]
 
-    outputs += [srt_path, vtt_path]
+    outputs += [srt_path, vtt_path, transcript, cached_signature, cached_translations]
 
     return outputs
 
 
+def reset_session_state():
+    """Clear cached transcript/translations and the visible results."""
+    cleared = [None, None, {}]
+    ui_reset = [
+        "Upload an audio file and click **Generate Outputs** to begin.",
+        "",
+        None,
+        gr.update(visible=True),
+        gr.update(choices=_compute_output_choices(has_key=False, exclude_code=None), value=DEFAULT_OUTPUTS),
+    ]
+    for _ in _TRANSLATION_SECTION_ORDER:
+        ui_reset += [gr.update(visible=False), "", None]
+    ui_reset += [None, None]
+    return ui_reset + cleared
+
+
 # ---------------------------------------------------------------------------
 # UI layout
 # ---------------------------------------------------------------------------
 
 with gr.Blocks(title="EchoScript") as demo:
 
+    transcript_state = gr.State(value=None)
+    signature_state = gr.State(value=None)
+    translations_state = gr.State(value={})
+
     gr.Markdown(
         """
 # EchoScript
@@ -270,18 +340,24 @@ with gr.Blocks(title="EchoScript") as demo:
             )
 
             outputs_input = gr.CheckboxGroup(
-                choices=_MARIAN_OUTPUT_CHOICES,
+                choices=_compute_output_choices(has_key=False, exclude_code=None),
                 value=DEFAULT_OUTPUTS,
                 label="Outputs",
             )
 
             api_key_input.change(
-                fn=_update_output_choices,
-                inputs=[api_key_input, outputs_input],
+                fn=_on_key_or_source_change,
+                inputs=[api_key_input, language_input, outputs_input],
+                outputs=[outputs_input],
+            )
+            language_input.change(
+                fn=_on_key_or_source_change,
+                inputs=[api_key_input, language_input, outputs_input],
                 outputs=[outputs_input],
             )
 
             process_button = gr.Button("Generate Outputs", variant="primary")
+            reset_button = gr.Button("Reset (clear cache)", size="sm")
 
         with gr.Column(scale=2):
             gr.Markdown("### Results Dashboard")
@@ -297,22 +373,28 @@ with gr.Blocks(title="EchoScript") as demo:
                     )
                     transcript_download = gr.DownloadButton("Download TXT")
 
-                translation_tabs = {}
-                translation_boxes = {}
-                translation_downloads = {}
-                for label in _TRANSLATION_TAB_ORDER:
-                    short_name = label.replace(" Translation", "")
-                    with gr.Tab(short_name, visible=False) as tab:
-                        box = gr.Textbox(
-                            label=short_name,
-                            lines=16,
-                            interactive=True,
-                            buttons=["copy"],
-                        )
-                        download = gr.DownloadButton("Download TXT")
-                    translation_tabs[label] = tab
-                    translation_boxes[label] = box
-                    translation_downloads[label] = download
+                with gr.Tab("Translations"):
+                    gr.Markdown(
+                        "Every language selected in **Outputs** appears below at "
+                        "once -- nothing is hidden behind a tab you have to click "
+                        "through."
+                    )
+                    translation_groups = {}
+                    translation_boxes = {}
+                    translation_downloads = {}
+                    for label in _TRANSLATION_SECTION_ORDER:
+                        short_name = label.replace(" Translation", "")
+                        with gr.Group(visible=False) as group:
+                            box = gr.Textbox(
+                                label=short_name,
+                                lines=10,
+                                interactive=True,
+                                buttons=["copy"],
+                            )
+                            download = gr.DownloadButton("Download TXT")
+                        translation_groups[label] = group
+                        translation_boxes[label] = box
+                        translation_downloads[label] = download
 
                 with gr.Tab("Subtitles"):
                     gr.Markdown(
@@ -325,16 +407,40 @@ with gr.Blocks(title="EchoScript") as demo:
                         vtt_download = gr.DownloadButton("Download VTT")
 
     # Build the flat outputs list in the exact order process_audio() returns.
-    click_outputs = [dashboard_output, transcript_box, transcript_download, transcript_tab]
-    for label in _TRANSLATION_TAB_ORDER:
-        click_outputs += [translation_tabs[label], translation_boxes[label], translation_downloads[label]]
-    click_outputs += [srt_download, vtt_download]
+    click_outputs = [
+        dashboard_output,
+        transcript_box,
+        transcript_download,
+        transcript_tab,
+        outputs_input,
+    ]
+    for label in _TRANSLATION_SECTION_ORDER:
+        click_outputs += [translation_groups[label], translation_boxes[label], translation_downloads[label]]
+    click_outputs += [
+        srt_download,
+        vtt_download,
+        transcript_state,
+        signature_state,
+        translations_state,
+    ]
 
     process_button.click(
         fn=process_audio,
-        inputs=[audio_input, start_input, end_input, language_input, outputs_input, api_key_input],
+        inputs=[
+            audio_input,
+            start_input,
+            end_input,
+            language_input,
+            outputs_input,
+            api_key_input,
+            transcript_state,
+            signature_state,
+            translations_state,
+        ],
         outputs=click_outputs,
     )
 
+    reset_button.click(fn=reset_session_state, outputs=click_outputs)
+
 if __name__ == "__main__":
     demo.launch()