app.py

"""
Tiny Turn Detector — Gradio demo.

SANDBOX NOTE: `gradio` and `torch`/`transformers` are not installed in the
environment this project was developed in, and cannot be installed there
(no network route to PyPI — see docs/INITIAL_ANALYSIS.md). This file is
written to run in a normal environment with those installed
(`pip install -r requirements.txt`); it has been validated here by static
syntax/import-structure checking only, not by actually starting the
Gradio server — see the final quality-check report for exactly what was
and wasn't verified.

Run with:
    python app.py
"""

from __future__ import annotations

import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent / "src"))

import numpy as np
import spaces

try:
    import gradio as gr
except ImportError as e:
    raise SystemExit(
        "gradio is not installed. Install with `pip install -r requirements.txt` "
        "in an environment with network access."
    ) from e

from turn_detector.inference import TurnDetector, TurnDetectorConfig, InferenceError, AudioValidationError

MODEL_INFO_TEXT = """
**Model:** Whisper Tiny (frozen encoder) + Logistic Regression
**Representation:** mean-pooled encoder hidden states (384-dim)
**Encoder params:** 8,208,384 (~32.8MB fp32)
**Classifier params:** 92 (~2.9KB)
**Measured F1 (75-clip real-audio validation, EXP-004):** 0.693

This is a research prototype validated on small real-audio samples from
`pipecat-ai/smart-turn-data-v3.2-train`, **not** the official held-out test
set. See `docs/RESULTS.md` and `docs/ERROR_ANALYSIS.md` for full, honestly
caveated results — including known weaknesses (filler-associated pauses,
small validation-set sizes, GPU-vs-CPU latency differences).
"""

EXPLANATION_TEXT = (
    "Predicts whether the speaker appears to have completed their turn "
    "(END) or is likely still speaking / pausing mid-thought (CONTINUE), "
    "from audio alone. Trained and validated on real clips from one "
    "public turn-detection dataset — not claimed to generalize perfectly "
    "to all accents, languages, or code-switching patterns. In "
    "particular, Hindi-English (\"Hinglish\") code-switching could not be "
    "directly verified in the source dataset (no transcripts available), "
    "so robustness to it is not a claim this demo makes."
)


_detector: TurnDetector | None = None


def get_detector() -> TurnDetector:
    global _detector
    if _detector is None:
        _detector = TurnDetector(TurnDetectorConfig())
    return _detector


@spaces.GPU(duration=30)
def predict_turn(audio) -> tuple[str, dict, str]:
    """Gradio callback. `audio` from gr.Audio(type="numpy") is either None
    or a (sample_rate, numpy_array) tuple.
    """
    if audio is None:
        return "No audio provided.", {}, ""

    sr, array = audio
    array = np.asarray(array, dtype=np.float32)
    # gr.Audio can hand back int16 PCM depending on source; normalize to
    # [-1, 1] float32 if it looks like integer-range data.
    if np.abs(array).max() > 1.5:
        array = array / 32768.0

    try:
        detector = get_detector()
    except InferenceError as e:
        return f"Model unavailable: {e}", {}, ""

    try:
        result = detector.predict(array, sr=sr)
    except AudioValidationError as e:
        return f"Invalid audio: {e}", {}, ""
    except InferenceError as e:
        return f"Inference failed: {e}", {}, ""

    decision_label = result["decision"]
    probs = {"END": result["end_probability"], "CONTINUE": result["continue_probability"]}
    detail = (
        f"**Decision:** {decision_label}\n\n"
        f"**END probability:** {result['end_probability']:.3f}\n\n"
        f"**CONTINUE probability:** {result['continue_probability']:.3f}\n\n"
        f"**Inference latency:** {result['latency_ms']:.1f} ms "
        f"(this machine — see docs/RESULTS.md for the hardware caveat)"
    )
    return decision_label, probs, detail


def build_app() -> "gr.Blocks":
    with gr.Blocks(title="Tiny Turn Detector") as demo:
        gr.Markdown("# Tiny Turn Detector")
        gr.Markdown("**Audio-based END vs CONTINUE detection for conversational voice AI**")
        gr.Markdown(EXPLANATION_TEXT)

        with gr.Row():
            with gr.Column():
                audio_input = gr.Audio(
                    sources=["upload", "microphone"],
                    type="numpy",
                    label="Audio input (WAV or microphone)",
                )
                submit_btn = gr.Button("Detect", variant="primary")
            with gr.Column():
                decision_output = gr.Label(label="Decision", elem_id="decision-output")
                probs_output = gr.Label(label="Probabilities", num_top_classes=2)
                detail_output = gr.Markdown()

        submit_btn.click(
            fn=predict_turn,
            inputs=[audio_input],
            outputs=[decision_output, probs_output, detail_output],
        )

        gr.Markdown("---")
        gr.Markdown(MODEL_INFO_TEXT)

    return demo


if __name__ == "__main__":
    app = build_app()
    app.launch()