Active path for the current turn. No token or key required.
"""
def render_setup_compare() -> str:
return """
Virtual Reachy
3D Reachy on screen
Speaks the translation in your browser
Works instantly for everyone
Physical Reachy Mini
Turns its body to the speaker
Plays the voice on the robot speaker
Nods to confirm
Recommended: Virtual — instant in this browser. Physical adds motion + robot voice when you run locally with your robot.
"""
def render_local_guide() -> str:
cmd = (
'# Start the Reachy Mini daemon, then run the app pointed at your robot:\n'
'$env:ENGINE="seamless"; $env:ROBOT_CONNECTED="1"; python app.py'
)
return f"""
On the hosted Space everything stays virtual (it can't reach your robot). To unlock motion + robot voice, run locally next to your Reachy Mini:
{html.escape(cmd)}
Then use Test robot below to confirm the antennas and head move.
"""
def pick_practice_language(pair_label: str) -> str:
source_lang, target_lang = parse_pair(pair_label)
return source_lang if source_lang != "English" else target_lang
def normalize_for_match(text: str) -> str:
keep = [c for c in text.lower() if c.isalnum() or c.isspace()]
return " ".join("".join(keep).split())
def build_tutor_audio(text: str, practice_language: str) -> str | None:
if practice_language != "English":
return None
cascade_engine = getattr(engine, "cascade", None)
if cascade_engine is None:
return None
try:
return cascade_engine._synthesize_english(text)
except Exception:
return None
# ---------------------------------------------------------------------------
# Interpret turn (text or audio) — outputs:
# status, result, audio, avatar_signal, history_html, history_state
# ---------------------------------------------------------------------------
def do_turn(audio_path, text_input, active_input, pair_label, history):
source_lang, target_lang = parse_pair(pair_label)
turn_hint = "right" if source_lang == "English" else "left"
use_audio = active_input == "audio"
has_text = bool(text_input and text_input.strip())
has_audio = bool(audio_path)
history = history or []
NO = gr.update() # leave a component unchanged
if use_audio and not has_audio:
yield (
render_status("Record first", "Record your voice on the Speak tab, then press Translate."),
render_placeholder(), None, render_avatar_html(robot.idle_snapshot()),
render_history(history), history, NO, NO,
)
return
if not use_audio and not has_text:
yield (
render_status("Type first", "Type a sentence on the Type tab, then press Translate."),
render_placeholder(), None, render_avatar_html(robot.idle_snapshot()),
render_history(history), history, NO, NO,
)
return
yield (
render_status("Interpreting", "Listening and composing the translation. First run loads the model (~30s on CPU); the GPU Space is fast."),
NO, None, render_avatar_html(robot.thinking_snapshot()),
render_history(history), history, NO, NO,
)
time.sleep(0.2)
try:
if use_audio:
result = engine.process_turn(TurnRequest(audio_path=audio_path, source_lang=source_lang, target_lang=target_lang))
else:
result = engine.translate_text(source_lang, target_lang, text_input.strip())
except Exception as exc:
yield (
render_status("Engine setup needed", str(exc), "warn"),
render_placeholder(), None,
render_avatar_html(robot.encourage_snapshot("Check the logs, then try again.")),
render_history(history), history, NO, NO,
)
return
avatar = robot.perform_translation(
text=result.translated_text, audio_path=result.output_audio_path,
speech_supported=result.speech_supported, turn_hint=turn_hint,
)
new_history = history + [{
"src": source_lang, "tgt": target_lang,
"source": result.source_text, "translation": result.translated_text,
}]
if result.raw_response.get("fallback_from") == "seamless":
status_body = "Seamless was busy — used the on-device fallback."
tone = "warn"
elif result.speech_supported:
status_body = "Reachy spoke this translation aloud."
tone = "ok"
else:
status_body = "Translation ready (shown on screen)."
tone = "ok"
# clear the inputs so the next turn starts fresh (no stale re-translation)
yield (
render_status("Done", status_body, tone),
render_result(source_lang, target_lang, result.source_text, result.translated_text),
result.output_audio_path,
render_avatar_html(avatar),
render_history(new_history),
new_history,
gr.update(value=""), # clear text box
gr.update(value=None), # clear mic recording
)
def do_turn_mic(audio_path, pair_label, history):
yield from do_turn(audio_path, "", "audio", pair_label, history)
def reset_session():
return (
render_status("Reachy is ready", "Pick a route, then talk or type to start."),
render_placeholder(), None,
render_avatar_html(robot.idle_snapshot()), render_history([]), [],
gr.update(value=""), gr.update(value=None),
)
# ---------------------------------------------------------------------------
# Tutor turn
# ---------------------------------------------------------------------------
def tutor_audio_path(lang, idx):
"""Reliable pre-generated tutor audio (static file, NO GPU/engine dependency)."""
p = STATIC_DIR / "tutor" / f"{lang}_{idx}.mp3"
return str(p) if p.exists() else None
def tutor_prompt(pair_label, idx):
"""Advance to the NEXT practice phrase (rotates), and speak it if the GPU is free."""
lang = pick_practice_language(pair_label)
phrases = TUTOR_PHRASES[lang]
new_idx = ((idx if idx is not None else -1) + 1) % len(phrases)
phrase = phrases[new_idx]
audio = tutor_audio_path(lang, new_idx)
listen = (
"Listen to Reachy, then record yourself saying it."
if audio else
"Read it aloud, then record yourself."
)
return (
render_status(
f"Phrase {new_idx + 1} of {len(phrases)} · {lang}",
f"{phrase['practice']} · {phrase['guide']} — {listen}",
),
render_result(lang, "Meaning", phrase["practice"], phrase["meaning"]),
audio,
new_idx,
)
def tutor_check(audio_path, pair_label, idx):
"""Score the user's attempt against the CURRENT phrase (direct match, no extra GPU)."""
lang = pick_practice_language(pair_label)
phrases = TUTOR_PHRASES[lang]
phrase = phrases[(idx if idx is not None else 0) % len(phrases)]
if not audio_path:
return (
render_status("Record first", "Record yourself saying the phrase, then press Check."),
gr.update(), gr.update(), idx,
)
try:
heard, _ = engine.transcribe_only(audio_path, lang)
except Exception as exc:
return render_status("Tutor needs the speech model", str(exc), "warn"), gr.update(), gr.update(), idx
sim = SequenceMatcher(None, normalize_for_match(heard), normalize_for_match(phrase["practice"])).ratio()
success = sim >= 0.6
fb = "Great pronunciation!" if success else "Good try — listen again and match the sounds, then check again."
return (
render_status(
"Tutor result" + (" ✓" if success else ""),
f"{fb} · match {sim:.0%}",
"ok" if success else "warn",
),
render_result(lang, "Your attempt", phrase["practice"], heard or "(nothing heard — try again)"),
gr.update(), # keep the phrase audio; don't re-synthesize (saves GPU)
idx,
)
def run_self_test():
ok, message = robot.self_test()
return render_status("Robot test", message, "ok" if ok else "warn")
def run_system_check():
t0 = time.time()
try:
result = engine.translate_text("English", "Spanish", "Hello, this is a quick system check.", prefer_voice_output=True)
dt = time.time() - t0
voice = "voice produced" if result.output_audio_path else "text only"
body = f'{friendly_engine_label(result.engine_used)} · {voice} · {dt:.1f}s — "{result.translated_text}"'
return render_status("System check", body, "ok")
except Exception as exc:
return render_status("System check", str(exc), "warn")
def on_setup_change(setup_mode):
physical = setup_mode == SETUP_PHYSICAL
# Selecting physical attempts a real connection + self-test when armed
# (ROBOT_CONNECTED=1). self_test() short-circuits to a "robot mode off"
# message with no SDK import or motion when not armed, so this is safe on
# the hosted Space and locally without the env var.
return (
gr.update(visible=physical),
render_runtime(engine.active_label, "Idle", setup_mode),
run_self_test() if physical else "",
)
# ---------------------------------------------------------------------------
# Live Conversation (robot-only) — thread-safe buffer + worker + UI handlers.
# A background worker drives the robot mic loop and mutates a module-global,
# lock-protected buffer; a gr.Timer polls it to refresh the UI. (gr.State is
# per-session and invisible to threads, so it cannot be used here.)
# ---------------------------------------------------------------------------
_CONV_LOCK = threading.Lock()
_CONV_STATE = {
"running": False,
"avatar": "idle",
"title": "Live Conversation",
"body": "Connect your Reachy Mini to begin.",
"turns": [],
}
def _conv_update(**kw):
turns = kw.pop("turns", None)
with _CONV_LOCK:
_CONV_STATE.update(kw)
if turns is not None:
_CONV_STATE["turns"] = turns
def _conv_add_turn(source_lang, target_lang, source_text, translation):
with _CONV_LOCK:
_CONV_STATE["turns"] = (_CONV_STATE["turns"] + [{
"src": source_lang, "tgt": target_lang,
"source": source_text, "translation": translation,
}])[-8:]
def _conv_snapshot():
with _CONV_LOCK:
return dict(_CONV_STATE), list(_CONV_STATE["turns"])
def _interruptible_pause(stop_event, seconds):
end = time.time() + seconds
while time.time() < end and not stop_event.is_set():
time.sleep(0.05)
def _conversation_worker(stop_event, source_lang, target_lang):
"""Robot-mic hands-free loop: listen -> interpret (fixed pair) -> speak (female) -> repeat."""
_conv_update(running=True, avatar="speaking", title="Reachy says hi",
body="Reachy is saying hello…")
try:
engine.warm_female_voice(target_lang)
except Exception:
pass
try:
robot.speak_chatter(engine.synthesize_female(CONV_GREETING, "English"), stop_event)
except Exception:
pass
while not stop_event.is_set():
_conv_update(avatar="listening", title="Listening", body=random.choice(CONV_LISTEN))
audio_path, doa = robot.capture_utterance(stop_event)
if stop_event.is_set():
break
if not audio_path:
continue
_conv_update(avatar="thinking", title="Interpreting", body=random.choice(CONV_THINK))
try:
result = engine.process_turn(TurnRequest(
audio_path=audio_path, source_lang=source_lang,
target_lang=target_lang, prefer_voice_output=False))
robot_audio = engine.synthesize_female(result.translated_text, target_lang)
except Exception as exc:
_conv_update(avatar="encourage", title="Hiccup",
body=f"{random.choice(CONV_ERR)} ({exc})")
_interruptible_pause(stop_event, 1.4) # let the hiccup caption stay visible
continue
_conv_add_turn(source_lang, target_lang, result.source_text, result.translated_text)
_conv_update(avatar="speaking", title="Reachy speaks", body=random.choice(CONV_SPEAK))
robot.perform_conversation_turn(robot_audio, doa, stop_event)
# Terminal copy is written by conversation_stop() (single writer) after the
# worker has joined — avoids a last-writer-wins race on the final caption.
class ConversationSession:
"""Owns the worker thread for one Live Conversation run. A single module-global
instance backs the (single-user, on-the-robot) app; start/stop are serialized."""
def __init__(self):
self._thread = None
self._stop = None
self._lock = threading.Lock()
def is_running(self):
with self._lock:
return self._thread is not None and self._thread.is_alive()
def start(self, source_lang, target_lang):
with self._lock:
self._stop_locked()
if not robot.start_conversation():
return False
self._stop = threading.Event()
self._thread = threading.Thread(
target=_conversation_worker, args=(self._stop, source_lang, target_lang), daemon=True)
self._thread.start()
return True
def stop(self):
with self._lock:
self._stop_locked()
def _stop_locked(self):
if self._stop is not None:
self._stop.set()
thread = self._thread
if thread is not None and thread.is_alive():
thread.join(timeout=8.0)
try:
robot.stop_conversation()
except Exception:
pass
self._thread = None
self._stop = None
def render_conversation_avatar(state):
snap = robot.snapshot(state["avatar"], state["title"], state["body"], "Live Convo")
return render_avatar_html(snap)
def render_conversation_captions(turns):
if not turns:
return ('
Your bilingual captions appear '
'here. Press Start and talk to Reachy.
')
prev = turns[:-1][-4:]
latest = turns[-1]
small = "".join(
f'
')
runtime_html = gr.HTML(render_runtime(engine.active_label, "Idle", SETUP_VIRTUAL))
system_check_btn = gr.Button("Run system check", variant="secondary")
system_check_out = gr.HTML()
# ---------------- About ----------------
with gr.Tab("About"):
gr.Markdown(
"""
### Reachy Bridge — live voice interpreter for Reachy Mini
Speak or type in one language; Reachy **speaks the translation** in another, and a 3D Reachy Mini reacts.
**Zero setup — no token, no API key.** Every model is public and downloads automatically.
**Engine:** SeamlessM4T v2 on ZeroGPU (one model: hears speech, speaks the translation). If the GPU is busy it falls back to an on‑device, token‑free path (Whisper + NLLB + Piper). Optional Qwen phrasing only if you set `HF_TOKEN`.
**Built for families** — a grandparent speaks Hindi, Spanish, German, or Chinese, and the next listener hears it in their language.
"""
)
# ---- wiring ----
interpret_outputs = [status_html, result_html, browser_audio, avatar_html, history_html, history_state, text_input, audio_input]
translate_btn.click(do_turn, [audio_input, text_input, active_input, pair, history_state], interpret_outputs)
audio_input.stop_recording(do_turn_mic, [audio_input, pair, history_state], interpret_outputs)
clear_btn.click(reset_session, outputs=interpret_outputs)
type_tab.select(lambda: "text", None, active_input)
speak_tab.select(lambda: "audio", None, active_input)
tutor_outputs = [tutor_status, tutor_result, tutor_prompt_audio, tutor_idx]
prompt_btn.click(tutor_prompt, [tutor_pair, tutor_idx], tutor_outputs)
check_btn.click(tutor_check, [tutor_audio, tutor_pair, tutor_idx], tutor_outputs)
setup.change(on_setup_change, [setup], [local_guide_group, runtime_html, test_output])
test_robot.click(run_self_test, outputs=[test_output])
system_check_btn.click(run_system_check, outputs=[system_check_out])
convo_start_btn.click(conversation_start, [convo_pair],
[convo_card, convo_captions, convo_timer], concurrency_limit=1)
convo_stop_btn.click(conversation_stop, None,
[convo_card, convo_captions, convo_timer], concurrency_limit=1)
convo_timer.tick(conversation_tick, None, [convo_card, convo_captions], show_progress="hidden")
if __name__ == "__main__":
on_space = bool(os.getenv("SPACE_ID"))
demo.queue().launch(
server_name="0.0.0.0" if on_space else os.getenv("GRADIO_SERVER_NAME", "127.0.0.1"),
allowed_paths=[str(STATIC_DIR)],
)