"""
Streamlit frontend for the Sunbird assessment pipeline.
Calls the FastAPI backend (see backend/main.py).
"""
from __future__ import annotations
import os
import time
from datetime import datetime, timezone
from typing import Any
from urllib.parse import urlparse
import requests
import streamlit as st
from dotenv import load_dotenv
from languages import PIPELINE_TARGET_LANGUAGES
load_dotenv()
_raw_to = (os.environ.get("PIPELINE_HTTP_TIMEOUT") or "").strip()
# Default 40 min: up to 10 min STT + summarise + translate + TTS on slow networks.
_PIPELINE_HTTP_TIMEOUT_S = int(_raw_to) if _raw_to else 2400
STEP_LABELS = ("Input", "Transcribe", "Summarise", "Translate", "Speech")
def _backend_url() -> str:
default = "http://127.0.0.1:8000"
raw = (os.environ.get("BACKEND_URL") or default).strip().rstrip("/")
if not raw:
return default
try:
host = (urlparse(raw).hostname or "").lower()
except ValueError:
return default
if host.endswith("hf.space") or "huggingface.co" in raw.lower():
return default
return raw
def _pipeline_url(backend: str) -> str:
base = (backend or "").strip().rstrip("/") or "http://127.0.0.1:8000"
return f"{base}/pipeline"
def post_pipeline(
backend: str,
*,
target_language: str,
text: str | None = None,
audio: tuple[str, bytes, str] | None = None,
) -> requests.Response:
url = _pipeline_url(backend)
if audio is not None:
name, raw, mime = audio
files = {
"target_language": (None, target_language),
"audio": (name, raw, mime or "application/octet-stream"),
}
return requests.post(url, files=files, timeout=_PIPELINE_HTTP_TIMEOUT_S)
files = {
"target_language": (None, target_language),
"text": (None, (text or "").strip()),
}
return requests.post(url, files=files, timeout=_PIPELINE_HTTP_TIMEOUT_S)
# Session snapshot so "Run audio pipeline" still works if the uploader widget returns None on the
# same rerun (common on Hugging Face). Do not put the uploader inside st.form — that often shows
# the red error state for valid OGG/WhatsApp files.
_AUDIO_WIDGET_KEY = "pipeline_audio_only"
_AUDIO_SNAP_KEY = "audio_pipeline_snap"
def _pipeline_audio_upload_changed() -> None:
f = st.session_state.get(_AUDIO_WIDGET_KEY)
if f is None:
st.session_state.pop(_AUDIO_SNAP_KEY, None)
else:
st.session_state[_AUDIO_SNAP_KEY] = {
"name": f.name,
"data": f.getvalue(),
"type": (f.type or "").strip() or "application/octet-stream",
}
def _inject_layout_css() -> None:
st.markdown(
"""
""",
unsafe_allow_html=True,
)
def _format_error(resp: requests.Response) -> str:
msg: str
try:
body: Any = resp.json()
except ValueError:
msg = resp.text[:4000] or f"HTTP {resp.status_code}"
if resp.status_code == 405:
msg += _format_error_405_hint(resp)
if getattr(resp, "url", None):
msg += f"\n\n**Request URL:**\n{resp.url}"
return msg
detail = body.get("detail")
if isinstance(detail, list):
parts = []
for err in detail:
if isinstance(err, dict):
loc = err.get("loc", ())
em = err.get("msg", "")
parts.append(f"{'/'.join(str(x) for x in loc)}: {em}")
else:
parts.append(str(err))
msg = "; ".join(parts) if parts else str(body)
elif detail is not None:
msg = str(detail)
else:
msg = str(body)
if resp.status_code == 405:
msg += _format_error_405_hint(resp)
if getattr(resp, "url", None):
msg += f"\n\n**Request URL:**\n{resp.url}"
return msg
def _format_error_405_hint(resp: requests.Response) -> str:
allow = resp.headers.get("Allow", "")
hint = (
"\n\n**HTTP 405 — Method Not Allowed.** The path exists but this HTTP method is not allowed here. "
"Typical causes: (1) **`BACKEND_URL` is your public Hugging Face Space URL** (`https://*.hf.space`) — "
"POST `/pipeline` then hits **Streamlit**, not FastAPI. Use **`http://127.0.0.1:8000`** inside the same "
"container (see `start.sh`). (2) **`BACKEND_URL` points at Streamlit** (e.g. port **8501**) instead of "
"**uvicorn** on **8000**. (3) **Stale deploy** — older builds called Sunbird legacy routes (`nllb_translate`, "
"`/tasks/tts`, RunPod-only STT); redeploy so the client uses `/tasks/translate`, `/tasks/modal/tts`, and "
"`/tasks/modal/stt`."
)
if allow:
hint += f"\n\n`Allow` response header: `{allow}`"
return hint
def _hero() -> None:
steps_html = " → ".join(
f"{s}" for s in STEP_LABELS
)
st.markdown(
f"""
Sunbird AI
Language pipeline
Summarise your content in clear English, translate it into your chosen language,
and hear it spoken — powered by the Sunbird API.
{steps_html}
""",
unsafe_allow_html=True,
)
def _card_title(label: str) -> None:
st.markdown(f'{label}
', unsafe_allow_html=True)
def _render_output_steps(
mode: str,
target_language: str,
result: dict[str, Any],
elapsed_s: float,
) -> None:
st.markdown(
f'Completed in {elapsed_s:.1f}s'
f' · target: {target_language}
',
unsafe_allow_html=True,
)
if mode == "Audio" and result.get("transcript"):
st.markdown(
f'Transcription in {target_language}
',
unsafe_allow_html=True,
)
st.markdown(
f'{result["transcript"]}
',
unsafe_allow_html=True,
)
elif mode == "Audio":
st.markdown('Audio transcription
', unsafe_allow_html=True)
st.caption("No transcript was returned for this audio.")
st.markdown('Summary
', unsafe_allow_html=True)
summary_text = result.get("summary", "")
st.markdown(
f'{summary_text}
',
unsafe_allow_html=True,
)
st.markdown('Translation
', unsafe_allow_html=True)
translated = result.get("translated_summary", "")
st.markdown(
f'{translated}
',
unsafe_allow_html=True,
)
st.markdown('Speech output
', unsafe_allow_html=True)
audio_url = result.get("audio_url")
if audio_url:
st.audio(audio_url)
st.caption("Signed audio URL from Sunbird TTS — play or download soon; it expires after a short time.")
else:
st.warning("No audio URL returned.")
report = []
if mode == "Audio" and result.get("transcript"):
report.append("=== Transcription ===\n" + str(result["transcript"]))
report.append("=== Summary ===\n" + str(result.get("summary", "")))
report.append("=== Translation ===\n" + str(result.get("translated_summary", "")))
report.append("=== Audio URL ===\n" + str(result.get("audio_url", "")))
payload = "\n\n".join(report)
st.download_button(
label="⬇ Download report (.txt)",
data=payload.encode("utf-8"),
file_name=f"sunbird_pipeline_{datetime.now(timezone.utc).strftime('%Y%m%d_%H%M%S')}.txt",
mime="text/plain",
)
with st.expander("Raw JSON response"):
st.json(result)
def main() -> None:
st.set_page_config(
page_title="Sunbird AI — Pipeline",
layout="centered",
initial_sidebar_state="collapsed",
)
_inject_layout_css()
backend = _backend_url()
_hero()
run_text = False
submitted_audio = False
text_value = ""
audio_file = None
target_lang_text = "Luganda"
target_lang_audio = "Luganda"
try:
input_panel = st.container(border=True)
except TypeError:
input_panel = st.container()
with input_panel:
_card_title("Input")
mode = st.radio(
"How would you like to provide content?",
["Text", "Audio"],
horizontal=True,
key="input_mode",
)
if mode == "Text":
st.markdown(
'Enter the text you would like to process
',
unsafe_allow_html=True,
)
text_value = st.text_area(
"Text",
height=200,
placeholder="Paste or type your text here…",
label_visibility="collapsed",
key="pipeline_text_body",
)
st.markdown(
'Target language
',
unsafe_allow_html=True,
)
target_lang_text = st.selectbox(
"Target language",
PIPELINE_TARGET_LANGUAGES,
key="target_lang_text",
)
run_text = st.button(
"Run pipeline",
type="primary",
use_container_width=True,
key="run_pipeline_text",
)
else:
st.caption(
"Supported formats: MP3, WAV, OGG, M4A, AAC · Max length: 5 minutes. "
"Pick a file, choose the language in the box below, then **Run pipeline**."
)
st.markdown(
'Upload your audio file
',
unsafe_allow_html=True,
)
audio_file = st.file_uploader(
"Audio file",
label_visibility="collapsed",
key=_AUDIO_WIDGET_KEY,
on_change=_pipeline_audio_upload_changed,
)
if audio_file is not None:
st.session_state[_AUDIO_SNAP_KEY] = {
"name": audio_file.name,
"data": audio_file.getvalue(),
"type": (audio_file.type or "").strip() or "application/octet-stream",
}
# Language + submit stay in a form so the upload survives the submit rerun (HF / Streamlit).
with st.form("audio_pipeline_form", clear_on_submit=False):
st.markdown(
'Target language
',
unsafe_allow_html=True,
)
target_lang_audio = st.selectbox(
"Target language",
PIPELINE_TARGET_LANGUAGES,
key="target_lang_audio",
)
submitted_audio = st.form_submit_button(
"Run pipeline",
type="primary",
use_container_width=True,
)
try:
results_panel = st.container(border=True)
except TypeError:
results_panel = st.container()
with results_panel:
_card_title("Results")
if not st.session_state.get("last_pipeline_result") and not st.session_state.get(
"last_pipeline_error"
):
st.info(
"Choose **Text** or **Audio**, fill in that section, then press **Run pipeline**."
)
if st.session_state.get("last_pipeline_error"):
st.error(st.session_state["last_pipeline_error"])
stored = st.session_state.get("last_pipeline_result")
if stored:
_render_output_steps(
str(stored.get("mode", "Text")),
str(stored.get("target_language", "")),
stored["result"],
float(stored.get("elapsed_s", 0)),
)
if mode == "Text" and run_text:
if not text_value.strip():
st.warning("Please enter some text before running the pipeline.")
return
st.session_state["last_pipeline_error"] = None
st.session_state["last_pipeline_result"] = None
t0 = time.perf_counter()
with st.spinner("Processing…"):
try:
resp = post_pipeline(
backend,
target_language=target_lang_text,
text=text_value.strip(),
audio=None,
)
except requests.RequestException as exc:
st.session_state["last_pipeline_result"] = None
st.session_state["last_pipeline_error"] = (
f"Could not reach the backend at `{backend}`: {exc}\n\n"
"Start the API from the project root: "
"`uvicorn backend.main:app --reload --port 8000`"
)
st.rerun()
elapsed = time.perf_counter() - t0
if not resp.ok:
st.session_state["last_pipeline_result"] = None
st.session_state["last_pipeline_error"] = _format_error(resp)
st.rerun()
result = resp.json()
st.session_state["last_pipeline_result"] = {
"result": result,
"elapsed_s": elapsed,
"mode": "Text",
"target_language": target_lang_text,
}
st.session_state["last_pipeline_error"] = None
st.rerun()
elif mode == "Audio" and submitted_audio:
snap = st.session_state.get(_AUDIO_SNAP_KEY)
if audio_file is not None:
audio_tuple = (
audio_file.name,
audio_file.getvalue(),
(audio_file.type or "").strip() or "application/octet-stream",
)
elif isinstance(snap, dict) and snap.get("data"):
audio_tuple = (str(snap["name"]), snap["data"], str(snap.get("type") or "application/octet-stream"))
else:
st.warning(
"Please upload an audio file above, then press **Run pipeline**."
)
return
st.session_state["last_pipeline_error"] = None
st.session_state["last_pipeline_result"] = None
t0 = time.perf_counter()
with st.spinner("Processing… transcription can take a moment"):
try:
resp = post_pipeline(
backend,
target_language=target_lang_audio,
text=None,
audio=audio_tuple,
)
except requests.RequestException as exc:
st.session_state["last_pipeline_result"] = None
st.session_state["last_pipeline_error"] = (
f"Could not reach the backend at `{backend}`: {exc}\n\n"
"Start the API from the project root: "
"`uvicorn backend.main:app --reload --port 8000`"
)
st.rerun()
elapsed = time.perf_counter() - t0
if not resp.ok:
st.session_state["last_pipeline_result"] = None
st.session_state["last_pipeline_error"] = _format_error(resp)
st.rerun()
result = resp.json()
st.session_state["last_pipeline_result"] = {
"result": result,
"elapsed_s": elapsed,
"mode": "Audio",
"target_language": target_lang_audio,
}
st.session_state["last_pipeline_error"] = None
st.rerun()
return
if __name__ == "__main__":
main()