EchoScript : New Architecture

app.py

@@ -1,87 +1,31 @@
+"""EchoScript app entrypoint.
+
+UI is intentionally left as-is for this step -- only the internals changed.
+Previously this file talked to faster-whisper directly. It now goes through
+the service layer (services/transcription.py, services/translation.py,
+services/subtitles.py) operating on the models/transcript.py data model, so
+the Audio -> Transcript -> Outputs flow is real code, not just a diagram.
+The Gradio UI itself (upload dropzone, time window, language dropdown,
+output checkboxes, results dashboard, tabs) is the next step.
+"""
+
 import tempfile
 from pathlib import Path
 from zipfile import ZipFile
 
 import gradio as gr
-from faster_whisper import WhisperModel
 
-MODEL_SIZE = "base"
+from services.subtitles import generate_srt, generate_vtt
+from services.transcription import TranscriptionService
+from services.translation import TranslationService
 
-model = WhisperModel(
-    MODEL_SIZE,
+transcription_service = TranscriptionService(
+    model_size="base",
     device="cpu",
     compute_type="int8",
-    download_root="/tmp/whisper_models"
+    download_root="/tmp/whisper_models",
 )
-
-LANGUAGE_NAMES = {
-    "fr": "French",
-    "en": "English",
-    "de": "German",
-    "fa": "Persian",
-    "es": "Spanish",
-    "it": "Italian",
-    "pt": "Portuguese",
-    "nl": "Dutch",
-    "ar": "Arabic",
-    "ru": "Russian",
-    "tr": "Turkish"
-}
-
-
-def format_srt_timestamp(seconds):
-    hours = int(seconds // 3600)
-    minutes = int((seconds % 3600) // 60)
-    secs = int(seconds % 60)
-    millis = int((seconds - int(seconds)) * 1000)
-
-    return (
-        f"{hours:02}:{minutes:02}:{secs:02},{millis:03}"
-    )
-
-
-def format_vtt_timestamp(seconds):
-    hours = int(seconds // 3600)
-    minutes = int((seconds % 3600) // 60)
-    secs = int(seconds % 60)
-    millis = int((seconds - int(seconds)) * 1000)
-
-    return (
-        f"{hours:02}:{minutes:02}:{secs:02}.{millis:03}"
-    )
-
-
-def generate_srt(segments):
-    lines = []
-
-    for idx, segment in enumerate(segments, start=1):
-
-        start = format_srt_timestamp(segment.start)
-        end = format_srt_timestamp(segment.end)
-
-        lines.append(
-            f"{idx}\n"
-            f"{start} --> {end}\n"
-            f"{segment.text.strip()}\n"
-        )
-
-    return "\n".join(lines)
-
-
-def generate_vtt(segments):
-    lines = ["WEBVTT\n"]
-
-    for segment in segments:
-
-        start = format_vtt_timestamp(segment.start)
-        end = format_vtt_timestamp(segment.end)
-
-        lines.append(
-            f"{start} --> {end}\n"
-            f"{segment.text.strip()}\n"
-        )
-
-    return "\n".join(lines)
+translation_service = TranslationService()
 
 
 def process_files(files, mode):
@@ -90,9 +34,7 @@ def process_files(files, mode):
         return "", None
 
     tmp_dir = Path(tempfile.mkdtemp())
-
     summary_lines = []
-
     zip_path = tmp_dir / "echoscript_results.zip"
 
     with ZipFile(zip_path, "w") as zipf:
@@ -100,85 +42,45 @@ def process_files(files, mode):
         for uploaded_file in files:
 
             audio_path = uploaded_file
-
             stem = Path(audio_path).stem
 
-            task = (
-                "transcribe"
-                if mode == "Transcribe"
-                else "translate"
-            )
-
-            segments_generator, info = model.transcribe(
+            # Step 1: Audio -> canonical Transcript. Always "transcribe",
+            # never "translate" -- see services/translation.py for why.
+            transcript = transcription_service.transcribe(
                 audio_path,
-                task=task,
-                beam_size=5
+                source_filename=Path(audio_path).name,
             )
 
-            segments = list(segments_generator)
-
-            transcript = "\n".join(
-                segment.text.strip()
-                for segment in segments
-            )
-
-            srt_content = generate_srt(segments)
-            vtt_content = generate_vtt(segments)
-
-            language = LANGUAGE_NAMES.get(
-                info.language,
-                info.language
-            )
-
-            summary_lines.append(
-                f"{stem}\n"
-                f"Language: {language}\n"
-                f"Confidence: {info.language_probability:.2%}\n"
-            )
+            # Step 2: Outputs are derived from the Transcript, never from
+            # the audio again.
+            if mode == "Translate to English" and transcript.language != "en":
+                output = translation_service.translate(transcript, "en")
+            else:
+                output = transcript
 
             txt_file = tmp_dir / f"{stem}.txt"
             srt_file = tmp_dir / f"{stem}.srt"
             vtt_file = tmp_dir / f"{stem}.vtt"
 
-            txt_file.write_text(
-                transcript,
-                encoding="utf-8"
-            )
-
-            srt_file.write_text(
-                srt_content,
-                encoding="utf-8"
-            )
+            txt_file.write_text(output.text, encoding="utf-8")
+            srt_file.write_text(generate_srt(output.segments), encoding="utf-8")
+            vtt_file.write_text(generate_vtt(output.segments), encoding="utf-8")
 
-            vtt_file.write_text(
-                vtt_content,
-                encoding="utf-8"
-            )
+            for f in (txt_file, srt_file, vtt_file):
+                zipf.write(f, arcname=f.name)
 
-            zipf.write(
-                txt_file,
-                arcname=txt_file.name
-            )
-
-            zipf.write(
-                srt_file,
-                arcname=srt_file.name
-            )
-
-            zipf.write(
-                vtt_file,
-                arcname=vtt_file.name
+            language_name = transcript.language
+            summary_lines.append(
+                f"{stem}\n"
+                f"Language: {language_name}\n"
+                f"Confidence: {transcript.language_probability:.2%}\n"
+                f"Words: {transcript.word_count}\n"
             )
 
-    return (
-        "\n\n".join(summary_lines),
-        str(zip_path)
-    )
+    return "\n\n".join(summary_lines), str(zip_path)
 
 
-with gr.Blocks(
-    title="EchoScript"
-) as demo:
+with gr.Blocks(title="EchoScript") as demo:
 
     gr.Markdown(
         """
@@ -198,42 +100,23 @@ Transcribe audio files using Faster-Whisper.
         """
     )
 
-    files_input = gr.Files(
-        label="Upload Audio Files"
-    )
+    files_input = gr.Files(label="Upload Audio Files")
 
     mode_input = gr.Dropdown(
-        choices=[
-            "Transcribe",
-            "Translate to English"
-        ],
+        choices=["Transcribe", "Translate to English"],
         value="Transcribe",
-        label="Mode"
-    )
-
-    run_button = gr.Button(
-        "Start Processing"
+        label="Mode",
     )
 
-    summary_output = gr.Textbox(
-        label="Results",
-        lines=12
-    )
+    run_button = gr.Button("Start Processing")
 
-    download_output = gr.File(
-        label="Download ZIP"
-    )
+    summary_output = gr.Textbox(label="Results", lines=12)
+    download_output = gr.File(label="Download ZIP")
 
     run_button.click(
         fn=process_files,
-        inputs=[
-            files_input,
-            mode_input
-        ],
-        outputs=[
-            summary_output,
-            download_output
-        ]
+        inputs=[files_input, mode_input],
+        outputs=[summary_output, download_output],
     )
 
 if __name__ == "__main__":

models/__init__.py

@@ -0,0 +1,3 @@
+from models.transcript import Segment, Transcript, Translation
+
+__all__ = ["Segment", "Transcript", "Translation"]

models/transcript.py

@@ -0,0 +1,139 @@
+"""Core data model for EchoScript.
+
+A `Transcript` is the canonical representation of "what was said" in an
+audio file. It is produced exactly once per upload (optionally restricted
+to a start/end time window). Every other artifact -- translations,
+subtitle files, future summaries/keywords -- is derived from a Transcript
+and must never reach back into the original audio.
+
+    Audio -> Transcript -> Outputs   (allowed)
+    Audio -> Translation             (never)
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Optional
+
+
+@dataclass(frozen=True)
+class Segment:
+    """A single timed chunk of text (transcribed or translated)."""
+
+    index: int
+    start: float  # seconds, relative to the processed audio window
+    end: float    # seconds
+    text: str
+
+    @property
+    def duration(self) -> float:
+        return max(0.0, self.end - self.start)
+
+
+@dataclass
+class Transcript:
+    """The canonical transcript of an audio file.
+
+    This is the single source of truth for everything downstream. If a user
+    edits text in the UI (see v1.1: Transcript Editing), that edit happens
+    on this object, and every translation/subtitle export regenerated after
+    the edit will reflect it automatically.
+    """
+
+    source_filename: str
+    language: str                  # ISO 639-1 code detected/forced, e.g. "fr"
+    language_probability: float    # 0..1, Whisper's detection confidence
+    duration: float                # seconds, of the processed window
+    segments: list[Segment] = field(default_factory=list)
+
+    # Optional processing window applied to the source audio, in seconds,
+    # relative to the original file. None means "from the very start" /
+    # "to the very end" for that side of the window.
+    window_start: Optional[float] = None
+    window_end: Optional[float] = None
+
+    @property
+    def text(self) -> str:
+        """Full plain-text transcript, segments joined with newlines."""
+        return "\n".join(s.text for s in self.segments)
+
+    @property
+    def word_count(self) -> int:
+        return len(self.text.split())
+
+    def replace_text(self, new_text: str) -> None:
+        """Used by the (future) transcript-editing feature.
+
+        Re-flows freeform edited text back across the existing segment
+        timings as evenly as possible, so timing-dependent outputs (SRT/VTT)
+        keep working after a manual correction. Intentionally simple for
+        v1.0; a smarter alignment can replace this later without touching
+        any other service.
+        """
+        lines = new_text.split("\n")
+        if len(lines) != len(self.segments):
+            # Fallback: dump everything into the first segment rather than
+            # silently losing edited text.
+            if self.segments:
+                self.segments = [
+                    Segment(
+                        index=1,
+                        start=self.segments[0].start,
+                        end=self.segments[-1].end,
+                        text=new_text.strip(),
+                    )
+                ]
+            return
+
+        self.segments = [
+            Segment(index=seg.index, start=seg.start, end=seg.end, text=line.strip())
+            for seg, line in zip(self.segments, lines)
+        ]
+
+    def to_dict(self) -> dict:
+        return {
+            "source_filename": self.source_filename,
+            "language": self.language,
+            "language_probability": self.language_probability,
+            "duration": self.duration,
+            "window_start": self.window_start,
+            "window_end": self.window_end,
+            "word_count": self.word_count,
+            "segments": [
+                {"index": s.index, "start": s.start, "end": s.end, "text": s.text}
+                for s in self.segments
+            ],
+        }
+
+
+@dataclass
+class Translation:
+    """A translation of a Transcript into a target language.
+
+    Always derived from `Transcript.text` / per-segment text, never from
+    the original audio. Segment timings are copied 1:1 from the source
+    transcript so subtitle generation keeps working on translated output.
+    """
+
+    source_language: str
+    target_language: str
+    segments: list[Segment] = field(default_factory=list)
+
+    @property
+    def text(self) -> str:
+        return "\n".join(s.text for s in self.segments)
+
+    @property
+    def word_count(self) -> int:
+        return len(self.text.split())
+
+    def to_dict(self) -> dict:
+        return {
+            "source_language": self.source_language,
+            "target_language": self.target_language,
+            "word_count": self.word_count,
+            "segments": [
+                {"index": s.index, "start": s.start, "end": s.end, "text": s.text}
+                for s in self.segments
+            ],
+        }

requirements.txt

@@ -1,2 +1,6 @@
-gradio>=5.0
-faster-whisper>=1.1
+gradio>=4.0
+faster-whisper>=1.0
+transformers>=4.40
+sentencepiece>=0.2
+sacremoses>=0.1
+torch>=2.0

services/__init__.py

@@ -0,0 +1,22 @@
+from services.audio import AudioError, extract_window, resolve_window, validate_extension
+from services.subtitles import generate_srt, generate_vtt
+from services.transcription import SUPPORTED_LANGUAGES, TranscriptionService
+from services.translation import (
+    SUPPORTED_TARGET_LANGUAGES,
+    TranslationError,
+    TranslationService,
+)
+
+__all__ = [
+    "AudioError",
+    "extract_window",
+    "resolve_window",
+    "validate_extension",
+    "generate_srt",
+    "generate_vtt",
+    "SUPPORTED_LANGUAGES",
+    "TranscriptionService",
+    "SUPPORTED_TARGET_LANGUAGES",
+    "TranslationError",
+    "TranslationService",
+]

services/audio.py

@@ -0,0 +1,108 @@
+"""Audio input handling: extension validation and optional time-window trim.
+
+This is the only service allowed to touch the raw audio file. Once a
+Transcript exists, nothing else needs the audio again.
+"""
+
+from __future__ import annotations
+
+import re
+import subprocess
+import tempfile
+from pathlib import Path
+from typing import Optional
+
+SUPPORTED_EXTENSIONS = {".mp3", ".wav", ".m4a", ".flac"}
+
+# Accepts "HH:MM:SS", "MM:SS", with optional fractional seconds.
+_TIME_RE = re.compile(
+    r"^(?:(?P<hours>\d{1,2}):)?(?P<minutes>\d{1,2}):(?P<seconds>\d{1,2}(?:\.\d+)?)$"
+)
+
+
+class AudioError(ValueError):
+    """Raised for invalid audio files or invalid/illogical time windows."""
+
+
+def validate_extension(filename: str) -> None:
+    suffix = Path(filename).suffix.lower()
+    if suffix not in SUPPORTED_EXTENSIONS:
+        raise AudioError(
+            f"Unsupported file type '{suffix}'. "
+            f"Supported: {', '.join(sorted(SUPPORTED_EXTENSIONS))}"
+        )
+
+
+def parse_timestamp(value: Optional[str]) -> Optional[float]:
+    """Parse 'HH:MM:SS' or 'MM:SS' into seconds. Blank/None -> None."""
+    if value is None:
+        return None
+    value = value.strip()
+    if not value:
+        return None
+
+    match = _TIME_RE.match(value)
+    if not match:
+        raise AudioError(f"Invalid timestamp '{value}'. Expected HH:MM:SS.")
+
+    hours = int(match.group("hours") or 0)
+    minutes = int(match.group("minutes"))
+    seconds = float(match.group("seconds"))
+    return hours * 3600 + minutes * 60 + seconds
+
+
+def resolve_window(
+    start_value: Optional[str],
+    end_value: Optional[str],
+) -> tuple[Optional[float], Optional[float]]:
+    """Parse and sanity-check a start/end window.
+
+    Returns (start_seconds, end_seconds); either or both may be None,
+    meaning "entire file" on that side.
+    """
+    start = parse_timestamp(start_value)
+    end = parse_timestamp(end_value)
+
+    if start is not None and end is not None and end <= start:
+        raise AudioError("End time must be after start time.")
+
+    return start, end
+
+
+def extract_window(
+    audio_path: str,
+    start: Optional[float],
+    end: Optional[float],
+    workdir: Optional[Path] = None,
+) -> str:
+    """Trim `audio_path` to [start, end] seconds using ffmpeg.
+
+    Returns a path to the trimmed file. If both start and end are None,
+    returns the original path unchanged ("entire file").
+    """
+    if start is None and end is None:
+        return audio_path
+
+    workdir = workdir or Path(tempfile.mkdtemp(prefix="echoscript_audio_"))
+    workdir.mkdir(parents=True, exist_ok=True)
+
+    suffix = Path(audio_path).suffix
+    out_path = workdir / f"window{suffix}"
+
+    cmd = ["ffmpeg", "-y", "-i", str(audio_path)]
+    if start is not None:
+        cmd += ["-ss", str(start)]
+    if end is not None:
+        cmd += ["-t", str(end - (start or 0.0))]
+    cmd += ["-c", "copy", str(out_path)]
+
+    result = subprocess.run(cmd, capture_output=True, text=True)
+    if result.returncode != 0:
+        # Stream-copy can fail if the cut point isn't on a keyframe;
+        # fall back to re-encoding.
+        cmd[-2:] = ["-c:a", "pcm_s16le", str(out_path)]
+        result = subprocess.run(cmd, capture_output=True, text=True)
+        if result.returncode != 0:
+            raise AudioError(f"ffmpeg failed to trim audio: {result.stderr.strip()}")
+
+    return str(out_path)

services/subtitles.py

@@ -0,0 +1,52 @@
+"""Subtitle generation: timed segments -> SRT/VTT text.
+
+Works against anything exposing `.segments` of objects with start/end/text
+(a Transcript or a Translation), keeping with the rule that subtitles are
+always derived data and never re-recorded from audio.
+"""
+
+from __future__ import annotations
+
+from typing import Iterable, Protocol
+
+
+class HasTimedText(Protocol):
+    start: float
+    end: float
+    text: str
+
+
+def _format_srt_timestamp(seconds: float) -> str:
+    seconds = max(0.0, seconds)
+    hours = int(seconds // 3600)
+    minutes = int((seconds % 3600) // 60)
+    secs = int(seconds % 60)
+    millis = int(round((seconds - int(seconds)) * 1000))
+    return f"{hours:02}:{minutes:02}:{secs:02},{millis:03}"
+
+
+def _format_vtt_timestamp(seconds: float) -> str:
+    seconds = max(0.0, seconds)
+    hours = int(seconds // 3600)
+    minutes = int((seconds % 3600) // 60)
+    secs = int(seconds % 60)
+    millis = int(round((seconds - int(seconds)) * 1000))
+    return f"{hours:02}:{minutes:02}:{secs:02}.{millis:03}"
+
+
+def generate_srt(segments: Iterable[HasTimedText]) -> str:
+    lines = []
+    for idx, segment in enumerate(segments, start=1):
+        start = _format_srt_timestamp(segment.start)
+        end = _format_srt_timestamp(segment.end)
+        lines.append(f"{idx}\n{start} --> {end}\n{segment.text.strip()}\n")
+    return "\n".join(lines)
+
+
+def generate_vtt(segments: Iterable[HasTimedText]) -> str:
+    lines = ["WEBVTT\n"]
+    for segment in segments:
+        start = _format_vtt_timestamp(segment.start)
+        end = _format_vtt_timestamp(segment.end)
+        lines.append(f"{start} --> {end}\n{segment.text.strip()}\n")
+    return "\n".join(lines)

services/transcription.py

@@ -0,0 +1,87 @@
+"""Transcription service: Audio -> canonical Transcript.
+
+This is the only place that runs Whisper. It always uses Whisper's
+"transcribe" task, never "translate" -- translation is handled as a
+separate, text-based step (see services/translation.py) operating on the
+Transcript this produces. That split is the core v1.0 architecture
+decision: there is exactly one path from audio to text, and everything
+else branches off the resulting Transcript.
+"""
+
+from __future__ import annotations
+
+from typing import Optional
+
+from faster_whisper import WhisperModel
+
+from models.transcript import Segment, Transcript
+
+# ISO 639-1 code -> display name. Used both for the "Source Language"
+# dropdown and for labelling detected languages in the results dashboard.
+SUPPORTED_LANGUAGES: dict[str, str] = {
+    "fr": "French",
+    "en": "English",
+    "de": "German",
+    "fa": "Persian",
+    "es": "Spanish",
+}
+
+
+class TranscriptionService:
+    """Thin wrapper around a faster-whisper model.
+
+    Deliberately has no knowledge of translation, subtitles, or the UI --
+    it only knows how to turn audio into a Transcript.
+    """
+
+    def __init__(
+        self,
+        model_size: str = "base",
+        device: str = "cpu",
+        compute_type: str = "int8",
+        download_root: str = "/tmp/whisper_models",
+    ) -> None:
+        self._model = WhisperModel(
+            model_size,
+            device=device,
+            compute_type=compute_type,
+            download_root=download_root,
+        )
+
+    def transcribe(
+        self,
+        audio_path: str,
+        source_filename: str,
+        language: Optional[str] = None,
+        window_start: Optional[float] = None,
+        window_end: Optional[float] = None,
+        beam_size: int = 5,
+    ) -> Transcript:
+        """Run speech-to-text and return the canonical Transcript.
+
+        `language` is an ISO 639-1 code, or None for auto-detect (the
+        "Auto Detect" dropdown option).
+        """
+        segments_iter, info = self._model.transcribe(
+            audio_path,
+            task="transcribe",
+            language=language,
+            beam_size=beam_size,
+        )
+
+        segments = [
+            Segment(index=i, start=seg.start, end=seg.end, text=seg.text.strip())
+            for i, seg in enumerate(segments_iter, start=1)
+        ]
+
+        duration = segments[-1].end if segments else 0.0
+
+        return Transcript(
+            source_filename=source_filename,
+            language=info.language,
+            language_probability=info.language_probability,
+            duration=duration,
+            segments=segments,
+            window_start=window_start,
+            window_end=window_end,
+        )

services/translation.py

@@ -0,0 +1,98 @@
+"""Translation service: Transcript -> Translation (text only).
+
+Per the EchoScript v1.0 architecture decision, translation is ALWAYS
+derived from the canonical Transcript's text, never from the original
+audio:
+
+    Audio -> Transcript -> Translation   (allowed)
+    Audio -> Translation                 (never)
+
+This keeps a single source of truth: if a name or term is fixed once in
+the transcript (v1.1: Transcript Editing), every translation regenerated
+afterwards picks up the fix automatically, and every translation stays in
+sync with the same segment timings as the transcript (so subtitles still
+work for translated output).
+"""
+
+from __future__ import annotations
+
+from functools import lru_cache
+
+from models.transcript import Segment, Transcript, Translation
+
+# Target languages exposed as the "Outputs" checkboxes in the UI.
+SUPPORTED_TARGET_LANGUAGES: dict[str, str] = {
+    "en": "English",
+    "de": "German",
+    "fa": "Persian",
+    "es": "Spanish",
+}
+
+
+class TranslationError(RuntimeError):
+    """Raised when no translation model/engine is available for a pair."""
+
+
+@lru_cache(maxsize=None)
+def _load_pipeline(source_language: str, target_language: str):
+    """Lazily load and cache a MarianMT pipeline for one language pair.
+
+    Cached so repeated translations within a session don't reload a model
+    from disk every time. Swapping the translation backend later (a
+    different model, a hosted API, an offline engine like Argos) only
+    requires changing this one function.
+    """
+    from transformers import pipeline  # heavy import, deferred until needed
+
+    model_name = f"Helsinki-NLP/opus-mt-{source_language}-{target_language}"
+    try:
+        return pipeline("translation", model=model_name)
+    except Exception as exc:  # pragma: no cover - depends on model availability
+        raise TranslationError(
+            f"No translation model available for "
+            f"'{source_language}' -> '{target_language}': {exc}"
+        ) from exc
+
+
+class TranslationService:
+    """Translates a Transcript into one or more target languages."""
+
+    def translate(self, transcript: Transcript, target_language: str) -> Translation:
+        """Translate every segment of `transcript`, preserving timing."""
+        if target_language == transcript.language:
+            # Already in the target language -- relabel, don't re-translate.
+            return Translation(
+                source_language=transcript.language,
+                target_language=target_language,
+                segments=list(transcript.segments),
+            )
+
+        translator = _load_pipeline(transcript.language, target_language)
+
+        translated_segments = []
+        for seg in transcript.segments:
+            if not seg.text:
+                translated_segments.append(seg)
+                continue
+            result_text = translator(seg.text)[0]["translation_text"]
+            translated_segments.append(
+                Segment(index=seg.index, start=seg.start, end=seg.end, text=result_text.strip())
+            )
+
+        return Translation(
+            source_language=transcript.language,
+            target_language=target_language,
+            segments=translated_segments,
+        )
+
+    def translate_many(
+        self,
+        transcript: Transcript,
+        target_languages: list[str],
+    ) -> dict[str, Translation]:
+        """Translate into several target languages at once.
+
+        Returns a dict keyed by target-language code, in line with how the
+        UI's multi-select "Outputs" checkboxes will want to fan out.
+        """
+        return {lang: self.translate(transcript, lang) for lang in target_languages}