from __future__ import annotations import asyncio import base64 import json import os import uuid import wave from dataclasses import dataclass, field from pathlib import Path from typing import Any from urllib.parse import urlparse import numpy as np # `spaces` is a true no-op off ZeroGPU and is pre-installed on every HF Space # hardware tier, so it is imported unconditionally (see the huggingface-zerogpu # guidance). For local dev: `pip install spaces`. import spaces # --------------------------------------------------------------------------- # Language code maps # --------------------------------------------------------------------------- # Whisper / generic ISO-639-1 codes (used by faster-whisper). LANGUAGE_CODES = { "Chinese": "zh", "English": "en", "German": "de", "Hindi": "hi", "Spanish": "es", } # SeamlessM4T v2 uses 3-letter codes. SEAMLESS_LANG_CODES = { "Chinese": "cmn", "English": "eng", "German": "deu", "Hindi": "hin", "Spanish": "spa", } # NLLB-200 uses FLORES-200 codes (token-free local fallback translator). NLLB_LANG_CODES = { "Chinese": "zho_Hans", "English": "eng_Latn", "German": "deu_Latn", "Hindi": "hin_Deva", "Spanish": "spa_Latn", } OMNI_VOICE_TARGETS = {"Chinese", "English"} # Verified female Piper voices per language (research 2026-06-15). Used ONLY by the robot-only # Live Conversation feature to guarantee a warm female voice in any target language. Seamless's # speaker_id has no official gender mapping, so we synthesize the robot's speech with Piper. PIPER_FEMALE_VOICES = { "English": os.getenv("CONV_VOICE_EN", "en_US-hfc_female-medium"), "Hindi": os.getenv("CONV_VOICE_HI", "hi_IN-priyamvada-medium"), "Spanish": os.getenv("CONV_VOICE_ES", "es_MX-claude-high"), "German": os.getenv("CONV_VOICE_DE", "de_DE-kerstin-low"), "Chinese": os.getenv("CONV_VOICE_ZH", "zh_CN-huayan-medium"), } SEAMLESS_MODEL_ID = os.getenv("SEAMLESS_MODEL", "facebook/seamless-m4t-v2-large") NLLB_MODEL_ID = os.getenv("NLLB_MODEL", "facebook/nllb-200-distilled-600M") DEFAULT_RUNTIME_DIR = Path(os.getenv("REACHY_BRIDGE_CACHE", "runtime_cache")) DEFAULT_AUDIO_DIR = DEFAULT_RUNTIME_DIR / "audio" DEFAULT_VOICE_DIR = DEFAULT_RUNTIME_DIR / "voices" # --------------------------------------------------------------------------- # Hardware detection: is the SeamlessM4T (GPU) interpreter usable here? # --------------------------------------------------------------------------- def _seamless_hardware_available() -> bool: if os.environ.get("SPACES_ZERO_GPU"): return True if os.environ.get("SEAMLESS_FORCE_CPU") == "1": return True try: import torch return bool(torch.cuda.is_available()) except Exception: return False _SEAMLESS_ENABLED = _seamless_hardware_available() _seamless_model = None _seamless_processor = None def _load_seamless() -> None: """Load SeamlessM4T v2 once, at module scope, onto the resolved device. On ZeroGPU `torch.cuda.is_available()` is monkey-patched to True so the `.to("cuda")` call registers the weights for transparent device migration. """ global _seamless_model, _seamless_processor if _seamless_model is not None: return import torch from transformers import AutoProcessor, SeamlessM4Tv2Model device = "cuda" if torch.cuda.is_available() else "cpu" _seamless_processor = AutoProcessor.from_pretrained(SEAMLESS_MODEL_ID) _seamless_model = SeamlessM4Tv2Model.from_pretrained(SEAMLESS_MODEL_ID).to(device) def _decode_seamless_text(out: Any) -> str: """Decode SeamlessM4T text-generation output across transformers versions.""" try: return _seamless_processor.decode(out[0].tolist()[0], skip_special_tokens=True).strip() except Exception: sequences = getattr(out, "sequences", out) return _seamless_processor.batch_decode(sequences, skip_special_tokens=True)[0].strip() @spaces.GPU(duration=60) def _seamless_infer( audio_path: str | None, text_input: str | None, src_code: str, tgt_code: str, want_speech: bool, ) -> tuple[str, str, str | None]: """Run a full SeamlessM4T interpreter turn on the GPU. Returns (source_text, translated_text, output_audio_path). CUDA tensors are moved to CPU before crossing the ZeroGPU process boundary. """ import torch _load_seamless() device = "cuda" if torch.cuda.is_available() else "cpu" if audio_path: audio_array = _read_wave_as_float32(Path(audio_path), target_rate=16000) # transformers >=4.4x renamed `audios` -> `audio` on the SeamlessM4T processor try: inputs = _seamless_processor( audio=audio_array, sampling_rate=16000, return_tensors="pt" ).to(device) except (TypeError, ValueError): inputs = _seamless_processor( audios=audio_array, sampling_rate=16000, return_tensors="pt" ).to(device) # Source transcript via ASR (translate into the source language itself). asr_tokens = _seamless_model.generate( **inputs, tgt_lang=src_code, generate_speech=False ) source_text = _decode_seamless_text(asr_tokens) else: inputs = _seamless_processor( text=text_input, src_lang=src_code, return_tensors="pt" ).to(device) source_text = (text_input or "").strip() text_tokens = _seamless_model.generate(**inputs, tgt_lang=tgt_code, generate_speech=False) translated_text = _decode_seamless_text(text_tokens) output_audio_path: str | None = None if want_speech: speech = _seamless_model.generate(**inputs, tgt_lang=tgt_code) waveform = speech[0].cpu().numpy().squeeze() sample_rate = int(getattr(_seamless_model.config, "sampling_rate", 16000)) output_audio_path = _write_float_audio_wav(waveform, sample_rate, prefix="seamless") return source_text, translated_text, output_audio_path # Eager module-scope load on ZeroGPU / GPU hosts (cold-start paid once). if _SEAMLESS_ENABLED: try: _load_seamless() except Exception as exc: # pragma: no cover - depends on hardware/runtime _SEAMLESS_ENABLED = False print(f"[engine] SeamlessM4T unavailable, using local fallback: {exc}") @dataclass(slots=True) class TurnRequest: audio_path: str | None source_lang: str target_lang: str mode: str = "translator" prefer_voice_output: bool = True prompt_override: str | None = None text_input: str | None = None @dataclass(slots=True) class TurnResult: source_text: str translated_text: str detected_language: str output_audio_path: str | None speech_supported: bool engine_used: str error_message: str | None = None raw_response: dict[str, Any] = field(default_factory=dict) @dataclass(slots=True) class OmniRealtimeResult: translated_text: str output_audio_path: str | None session_id: str | None chunks_sent: int class SpeechEngine: def process_turn(self, request: TurnRequest) -> TurnResult: raise NotImplementedError def translate_text( self, source_lang: str, target_lang: str, text: str, prefer_voice_output: bool = True ) -> TurnResult: raise NotImplementedError def transcribe_only(self, audio_path: str | Path, source_lang: str) -> tuple[str, str]: raise NotImplementedError class SeamlessSpeechEngine(SpeechEngine): """Primary engine: one model that hears speech and speaks the translation.""" def process_turn(self, request: TurnRequest) -> TurnResult: audio_path = _require_audio_path(request.audio_path) src = SEAMLESS_LANG_CODES[request.source_lang] tgt = SEAMLESS_LANG_CODES[request.target_lang] source_text, translated_text, audio_out = _seamless_infer( str(audio_path), None, src, tgt, request.prefer_voice_output ) return TurnResult( source_text=source_text, translated_text=translated_text, detected_language=request.source_lang, output_audio_path=audio_out, speech_supported=audio_out is not None, engine_used="seamless", raw_response={"engine": "seamless", "model": SEAMLESS_MODEL_ID}, ) def translate_text( self, source_lang: str, target_lang: str, text: str, prefer_voice_output: bool = True ) -> TurnResult: src = SEAMLESS_LANG_CODES[source_lang] tgt = SEAMLESS_LANG_CODES[target_lang] source_text, translated_text, audio_out = _seamless_infer( None, text, src, tgt, prefer_voice_output ) return TurnResult( source_text=source_text or text, translated_text=translated_text, detected_language=source_lang, output_audio_path=audio_out, speech_supported=audio_out is not None, engine_used="seamless", raw_response={"engine": "seamless", "mode": "text", "model": SEAMLESS_MODEL_ID}, ) def transcribe_only(self, audio_path: str | Path, source_lang: str) -> tuple[str, str]: src = SEAMLESS_LANG_CODES[source_lang] source_text, _, _ = _seamless_infer(str(audio_path), None, src, src, False) return source_text, source_lang class CascadeSpeechEngine(SpeechEngine): """Token-free fallback: Whisper (ASR) -> NLLB/Qwen (translate) -> Piper (TTS).""" def __init__(self) -> None: self.whisper_model_size = os.getenv("WHISPER_MODEL_SIZE", "small") self.whisper_device = os.getenv("WHISPER_DEVICE", "cpu") self.whisper_compute_type = os.getenv("WHISPER_COMPUTE_TYPE", "int8") self.translation_model = os.getenv("CASCADE_TRANSLATION_MODEL", "Qwen/Qwen2.5-7B-Instruct") self.hf_token = os.getenv("HF_TOKEN") or os.getenv("HUGGINGFACEHUB_API_TOKEN") self.qwen_gguf_path = os.getenv("QWEN_GGUF_PATH") self.piper_voice = os.getenv("PIPER_VOICE", "en_US-hfc_female-medium") self.piper_model_path = os.getenv("PIPER_MODEL_PATH") self.piper_config_path = os.getenv("PIPER_CONFIG_PATH") self._whisper_model = None self._llm = None self._voice = None self._nllb_model = None self._nllb_tokenizer = None self._voices: dict[str, Any] = {} # per-language female Piper voices (conversation only) @property def translation_path(self) -> str: if self.qwen_gguf_path: return "qwen-local" if self.hf_token: return "qwen-api" return "local-nmt" def process_turn(self, request: TurnRequest) -> TurnResult: audio_path = _require_audio_path(request.audio_path) source_text, detected_language = self._transcribe(audio_path, request.source_lang) return self._finish( source_text, detected_language, request.source_lang, request.target_lang, request.prompt_override, request.prefer_voice_output, ) def translate_text( self, source_lang: str, target_lang: str, text: str, prefer_voice_output: bool = True ) -> TurnResult: clean = (text or "").strip() if not clean: raise ValueError("Type a sentence before starting translation.") return self._finish( clean, LANGUAGE_CODES.get(source_lang, ""), source_lang, target_lang, None, prefer_voice_output, ) def _finish( self, source_text: str, detected_language: str, source_lang: str, target_lang: str, prompt_override: str | None, prefer_voice_output: bool, ) -> TurnResult: translated_text = self._translate(source_text, source_lang, target_lang, prompt_override) output_audio_path: str | None = None warning_message: str | None = None speech_supported = target_lang == "English" and prefer_voice_output if speech_supported: try: output_audio_path = self._synthesize_english(translated_text) except Exception as exc: speech_supported = False warning_message = f"Translation succeeded, but English voice output is unavailable: {exc}" engine_used = "qwen-cascade" if self.translation_path.startswith("qwen") else "local-nmt" return TurnResult( source_text=source_text, translated_text=translated_text, detected_language=detected_language, output_audio_path=output_audio_path, speech_supported=speech_supported and output_audio_path is not None, engine_used=engine_used, error_message=warning_message, raw_response={ "translation_path": self.translation_path, "voice": self.piper_voice if output_audio_path else None, }, ) def transcribe_only(self, audio_path: str | Path, source_lang: str) -> tuple[str, str]: return self._transcribe(Path(audio_path), source_lang) def warm(self) -> None: """Pre-load the fallback models so the first user turn isn't a cold load.""" try: self._translate_nllb("Hello", "English", "Spanish") except Exception: pass try: from faster_whisper import WhisperModel if self._whisper_model is None: self._whisper_model = WhisperModel( self.whisper_model_size, device=self.whisper_device, compute_type=self.whisper_compute_type, ) except Exception: pass try: self._synthesize_english("Hello.") except Exception: pass def _transcribe(self, audio_path: Path, source_lang: str) -> tuple[str, str]: requested_language = LANGUAGE_CODES.get(source_lang) try: from faster_whisper import WhisperModel except ImportError as exc: raise RuntimeError( "faster-whisper is required for the local fallback. Install the project requirements first." ) from exc if self._whisper_model is None: self._whisper_model = WhisperModel( self.whisper_model_size, device=self.whisper_device, compute_type=self.whisper_compute_type, ) segments, info = self._whisper_model.transcribe( str(audio_path), beam_size=5, task="transcribe", language=requested_language, vad_filter=True, ) transcript = " ".join(segment.text.strip() for segment in segments).strip() detected = getattr(info, "language", None) or requested_language or "unknown" if not transcript: raise RuntimeError("Could not detect any speech in the recorded clip. Try again, closer to the mic.") return transcript, detected def _translate( self, source_text: str, source_lang: str, target_lang: str, prompt_override: str | None ) -> str: if source_lang == target_lang: return source_text system_prompt = prompt_override or ( "You are Reachy Bridge, a live family interpreter. " f"Translate the user's sentence from {source_lang} to {target_lang}. " "Return only the translated sentence, with no extra commentary." ) # Ladder: local Qwen GGUF -> Qwen API (only if token) -> token-free NLLB. if self.qwen_gguf_path: return self._translate_local(source_text, system_prompt) if self.hf_token: try: return self._translate_remote(source_text, system_prompt) except Exception: pass # fall through to the always-available local NMT return self._translate_nllb(source_text, source_lang, target_lang) def _translate_nllb(self, source_text: str, source_lang: str, target_lang: str) -> str: try: from transformers import AutoModelForSeq2SeqLM, AutoTokenizer except ImportError as exc: raise RuntimeError( "transformers is required for the local NLLB translator. Install the project requirements first." ) from exc if self._nllb_model is None: self._nllb_tokenizer = AutoTokenizer.from_pretrained(NLLB_MODEL_ID) self._nllb_model = AutoModelForSeq2SeqLM.from_pretrained(NLLB_MODEL_ID) src = NLLB_LANG_CODES[source_lang] tgt = NLLB_LANG_CODES[target_lang] self._nllb_tokenizer.src_lang = src encoded = self._nllb_tokenizer(source_text, return_tensors="pt", truncation=True, max_length=512) forced_bos = self._nllb_tokenizer.convert_tokens_to_ids(tgt) generated = self._nllb_model.generate( **encoded, forced_bos_token_id=forced_bos, max_new_tokens=256 ) return self._nllb_tokenizer.batch_decode(generated, skip_special_tokens=True)[0].strip() def _translate_remote(self, source_text: str, system_prompt: str) -> str: from huggingface_hub import InferenceClient client = InferenceClient(api_key=self.hf_token or None, timeout=60) response = client.chat_completion( model=self.translation_model, messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": source_text}, ], max_tokens=256, temperature=0.1, ) message = response.choices[0].message content = getattr(message, "content", None) if isinstance(content, list): return "".join( part.get("text", "") if isinstance(part, dict) else str(part) for part in content ).strip() return str(content).strip() def _translate_local(self, source_text: str, system_prompt: str) -> str: from llama_cpp import Llama if self._llm is None: self._llm = Llama( model_path=self.qwen_gguf_path, n_ctx=int(os.getenv("LLAMA_CONTEXT", "4096")), n_threads=max(1, int(os.getenv("LLAMA_THREADS", "4"))), verbose=False, ) completion = self._llm.create_chat_completion( messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": source_text}, ], temperature=0.1, max_tokens=256, ) return completion["choices"][0]["message"]["content"].strip() def _synthesize_english(self, text: str) -> str: try: from huggingface_hub import hf_hub_download from piper.voice import PiperVoice except ImportError as exc: raise RuntimeError( "Piper dependencies are required for English speech output. Install the project requirements first." ) from exc DEFAULT_AUDIO_DIR.mkdir(parents=True, exist_ok=True) DEFAULT_VOICE_DIR.mkdir(parents=True, exist_ok=True) model_path, config_path = self._resolve_piper_paths(hf_hub_download) if self._voice is None: self._voice = PiperVoice.load(str(model_path), config_path=str(config_path)) output_path = DEFAULT_AUDIO_DIR / f"cascade_{uuid.uuid4().hex}.wav" with wave.open(str(output_path), "wb") as wav_file: # piper-tts >=1.2 renamed this to synthesize_wav (which sets the # wave format itself); older builds used synthesize(text, wav_file). if hasattr(self._voice, "synthesize_wav"): self._voice.synthesize_wav(text, wav_file) else: self._voice.synthesize(text, wav_file) return str(output_path) def _resolve_piper_paths(self, hf_hub_download: Any) -> tuple[Path, Path]: if self.piper_model_path: model_path = Path(self.piper_model_path) config_path = ( Path(self.piper_config_path) if self.piper_config_path else Path(f"{self.piper_model_path}.json") ) if not model_path.exists(): raise FileNotFoundError(f"Piper model file not found: {model_path}") if not config_path.exists(): raise FileNotFoundError(f"Piper config file not found: {config_path}") return model_path, config_path locale, voice_name, quality = self._split_voice_name(self.piper_voice) language_root = locale.split("_", maxsplit=1)[0] repo_base = f"{language_root}/{locale}/{voice_name}/{quality}/{self.piper_voice}" # Use the standard HF cache (no local_dir): avoids the Windows # `.cache/huggingface/download/*.incomplete` staging bug. revision = os.getenv("PIPER_VOICE_REVISION", "v1.0.0") model_path = Path( hf_hub_download( repo_id="rhasspy/piper-voices", repo_type="model", filename=f"{repo_base}.onnx", revision=revision, ) ) config_path = Path( hf_hub_download( repo_id="rhasspy/piper-voices", repo_type="model", filename=f"{repo_base}.onnx.json", revision=revision, ) ) return model_path, config_path @staticmethod def _split_voice_name(voice_name: str) -> tuple[str, str, str]: parts = voice_name.split("-") if len(parts) < 3: raise ValueError( "PIPER_VOICE must follow the Piper naming format, for example en_US-hfc_female-medium." ) locale = parts[0] quality = parts[-1] speaker = "-".join(parts[1:-1]) return locale, speaker, quality # ------------------------------------------------------------------ # Female multi-language voice (robot-only Live Conversation). Additive: # does NOT touch the English-only _synthesize_english path used by the # Interpret/Tutor tabs. # ------------------------------------------------------------------ def synthesize_female(self, text: str, language: str) -> str: voice_name = PIPER_FEMALE_VOICES.get(language) if not voice_name: raise RuntimeError(f"No female voice is configured for {language}.") return self._synthesize_with_voice(text, voice_name) def warm_female(self, language: str) -> None: try: self.synthesize_female("Hello.", language) except Exception: pass def _synthesize_with_voice(self, text: str, voice_name: str) -> str: from huggingface_hub import hf_hub_download from piper.voice import PiperVoice DEFAULT_AUDIO_DIR.mkdir(parents=True, exist_ok=True) DEFAULT_VOICE_DIR.mkdir(parents=True, exist_ok=True) voice = self._voices.get(voice_name) if voice is None: model_path, config_path = self._resolve_voice_files(voice_name, hf_hub_download) voice = PiperVoice.load(str(model_path), config_path=str(config_path)) self._voices[voice_name] = voice output_path = DEFAULT_AUDIO_DIR / f"conv_{uuid.uuid4().hex}.wav" with wave.open(str(output_path), "wb") as wav_file: if hasattr(voice, "synthesize_wav"): voice.synthesize_wav(text, wav_file) else: voice.synthesize(text, wav_file) return str(output_path) def _resolve_voice_files(self, voice_name: str, hf_hub_download: Any) -> tuple[Path, Path]: locale, speaker, quality = self._split_voice_name(voice_name) language_root = locale.split("_", maxsplit=1)[0] repo_base = f"{language_root}/{locale}/{speaker}/{quality}/{voice_name}" revision = os.getenv("PIPER_VOICE_REVISION", "v1.0.0") model_path = Path( hf_hub_download( repo_id="rhasspy/piper-voices", repo_type="model", filename=f"{repo_base}.onnx", revision=revision, ) ) config_path = Path( hf_hub_download( repo_id="rhasspy/piper-voices", repo_type="model", filename=f"{repo_base}.onnx.json", revision=revision, ) ) return model_path, config_path class OmniSpeechEngine(SpeechEngine): def __init__(self, cascade_engine: CascadeSpeechEngine) -> None: self.cascade = cascade_engine self.provider = os.getenv("OMNI_PROVIDER", "api").lower() self.base_url = os.getenv("OPENBMB_BASE_URL", "https://minicpmo45.modelbest.cn") self.api_key = os.getenv("OPENBMB_API_KEY") self.ollama_host = os.getenv("OLLAMA_HOST", "http://127.0.0.1:11434") self.ollama_model = os.getenv("OLLAMA_MODEL", "openbmb/minicpm-o4.5:8b") self.chunk_ms = max(250, int(os.getenv("OMNI_CHUNK_MS", "250"))) def process_turn(self, request: TurnRequest) -> TurnResult: audio_path = _require_audio_path(request.audio_path) source_text, detected_language = self.cascade._transcribe(audio_path, request.source_lang) translated = self._translate_with_omni(audio_path, request, source_text) speech_supported = request.target_lang in OMNI_VOICE_TARGETS and request.prefer_voice_output if speech_supported and not translated.output_audio_path: raise RuntimeError("MiniCPM-o did not return audio for a voice-supported target language.") return TurnResult( source_text=source_text, translated_text=translated.translated_text, detected_language=detected_language, output_audio_path=translated.output_audio_path if speech_supported else None, speech_supported=speech_supported and translated.output_audio_path is not None, engine_used="omni", raw_response={ "provider": self.provider, "session_id": translated.session_id, "chunks_sent": translated.chunks_sent, }, ) def translate_text( self, source_lang: str, target_lang: str, text: str, prefer_voice_output: bool = True ) -> TurnResult: return self.cascade.translate_text(source_lang, target_lang, text, prefer_voice_output) def transcribe_only(self, audio_path: str | Path, source_lang: str) -> tuple[str, str]: return self.cascade.transcribe_only(audio_path, source_lang) def _translate_with_omni(self, audio_path: Path, request: TurnRequest, source_text: str) -> OmniRealtimeResult: instructions = request.prompt_override or ( "You are Reachy Bridge, a live family interpreter. " f"Listen to the speaker and translate from {request.source_lang} to {request.target_lang}. " "Return only the translated content, keep it natural, and do not add commentary." ) if self.provider == "ollama": raise RuntimeError( "TODO: verify MiniCPM-o 4.5 audio I/O over Ollama. Falling back until the Ollama speech contract is confirmed." ) if not self.api_key: raise RuntimeError("OPENBMB_API_KEY is required for ENGINE=omni.") return asyncio.run(self._run_realtime_audio_turn(audio_path, instructions, request.target_lang)) async def _run_realtime_audio_turn( self, audio_path: Path, instructions: str, target_lang: str ) -> OmniRealtimeResult: try: import websockets except ImportError as exc: raise RuntimeError("websockets is required for the MiniCPM-o realtime API.") from exc ws_url = self._build_ws_url() headers = {"Authorization": f"Bearer {self.api_key}"} input_audio = _read_wave_as_float32(audio_path, target_rate=16000) chunk_size = int(16000 * (self.chunk_ms / 1000)) chunks_sent = 0 text_parts: list[str] = [] audio_parts: list[np.ndarray] = [] session_id: str | None = None async with websockets.connect( ws_url, additional_headers=headers, max_size=16 * 1024 * 1024, ping_interval=20 ) as websocket: await self._await_queue_done(websocket) await websocket.send( json.dumps({"type": "session.update", "session": {"instructions": instructions}}) ) session_id = await self._await_session_created(websocket) async def send_audio() -> None: nonlocal chunks_sent for chunk in _chunk_audio(input_audio, chunk_size): await websocket.send( json.dumps( {"type": "input_audio_buffer.append", "audio": _float32_to_base64(chunk)} ) ) chunks_sent += 1 await asyncio.sleep(len(chunk) / 16000) silence = np.zeros(chunk_size, dtype=np.float32) await websocket.send( json.dumps( {"type": "input_audio_buffer.append", "audio": _float32_to_base64(silence)} ) ) chunks_sent += 1 send_task = asyncio.create_task(send_audio()) loop = asyncio.get_running_loop() turn_finished = False turn_deadline: float | None = None while True: timeout = 0.7 if not turn_finished else 0.4 try: raw_message = await asyncio.wait_for(websocket.recv(), timeout=timeout) except asyncio.TimeoutError: if send_task.done() and turn_finished: break continue message = json.loads(raw_message) message_type = message.get("type") if message_type == "response.output_audio.delta": if message.get("text"): text_parts.append(message["text"]) if target_lang in OMNI_VOICE_TARGETS and message.get("audio"): audio_parts.append(_base64_to_float32(message["audio"])) if message.get("end_of_turn"): turn_finished = True turn_deadline = loop.time() + 0.7 elif message_type == "response.listen": if send_task.done() and turn_finished: break elif message_type == "session.closed": break elif message_type == "error": error = message.get("error", {}) raise RuntimeError(f"{error.get('code', 'omni_error')}: {error.get('message', 'unknown error')}") if turn_deadline is not None and loop.time() >= turn_deadline and send_task.done(): break await send_task await websocket.send(json.dumps({"type": "session.close", "reason": "user_stop"})) translated_text = "".join(text_parts).strip() if not translated_text: raise RuntimeError("MiniCPM-o returned no translated text.") output_audio_path = None if audio_parts and target_lang in OMNI_VOICE_TARGETS: output_audio_path = _write_float_audio_wav(np.concatenate(audio_parts), 24000, prefix="omni") return OmniRealtimeResult( translated_text=translated_text, output_audio_path=output_audio_path, session_id=session_id, chunks_sent=chunks_sent, ) async def _await_queue_done(self, websocket: Any) -> None: while True: message = json.loads(await websocket.recv()) message_type = message.get("type") if message_type in {"queue_done", "session.queue_done"}: return if message_type == "error": error = message.get("error", {}) raise RuntimeError(f"{error.get('code', 'queue_error')}: {error.get('message', 'unknown error')}") async def _await_session_created(self, websocket: Any) -> str | None: while True: message = json.loads(await websocket.recv()) message_type = message.get("type") if message_type == "session.created": return message.get("session_id") if message_type == "error": error = message.get("error", {}) raise RuntimeError(f"{error.get('code', 'session_error')}: {error.get('message', 'unknown error')}") def _build_ws_url(self) -> str: parsed = urlparse(self.base_url) if parsed.scheme not in {"http", "https"}: raise ValueError("OPENBMB_BASE_URL must start with http:// or https://") scheme = "wss" if parsed.scheme == "https" else "ws" host = parsed.netloc or parsed.path return f"{scheme}://{host}/v1/realtime?mode=audio" class RoutingSpeechEngine(SpeechEngine): def __init__(self) -> None: self.cascade = CascadeSpeechEngine() self.omni = OmniSpeechEngine(self.cascade) self.seamless = SeamlessSpeechEngine() if _SEAMLESS_ENABLED else None self.requested_engine = _normalize_engine_name(os.getenv("ENGINE", "seamless")) @property def active_label(self) -> str: if self.requested_engine == "seamless" and self.seamless is not None: return "seamless" if self.requested_engine == "omni": return "omni" return self.cascade.translation_path # qwen-api / qwen-local / local-nmt def process_turn(self, request: TurnRequest) -> TurnResult: if self.requested_engine == "seamless": if self.seamless is not None: try: return self.seamless.process_turn(request) except Exception as seamless_error: return self._cascade_with_note(self.cascade.process_turn(request), seamless_error) return self.cascade.process_turn(request) if self.requested_engine in {"cascade", "qwen"}: return self.cascade.process_turn(request) try: return self.omni.process_turn(request) except Exception as omni_error: try: result = self.cascade.process_turn(request) except Exception as cascade_error: raise RuntimeError( f"Omni failed: {omni_error}. Cascade fallback also failed: {cascade_error}" ) from cascade_error fallback_note = f"MiniCPM-o failed, so Reachy switched to the local fallback: {omni_error}" result.error_message = ( f"{fallback_note}. {result.error_message}" if result.error_message else fallback_note ) result.raw_response["fallback_from"] = "omni" result.raw_response["omni_error"] = str(omni_error) return result def translate_text( self, source_lang: str, target_lang: str, text: str, prefer_voice_output: bool = True ) -> TurnResult: if self.requested_engine == "seamless" and self.seamless is not None: try: return self.seamless.translate_text(source_lang, target_lang, text, prefer_voice_output) except Exception as seamless_error: return self._cascade_with_note( self.cascade.translate_text(source_lang, target_lang, text, prefer_voice_output), seamless_error, ) return self.cascade.translate_text(source_lang, target_lang, text, prefer_voice_output) def transcribe_only(self, audio_path: str | Path, source_lang: str) -> tuple[str, str]: return self.cascade.transcribe_only(audio_path, source_lang) @staticmethod def _cascade_with_note(result: TurnResult, seamless_error: Exception) -> TurnResult: note = "The Seamless interpreter was busy, so Reachy used the on-device fallback." result.error_message = f"{note} {result.error_message}" if result.error_message else note result.raw_response["fallback_from"] = "seamless" result.raw_response["seamless_error"] = str(seamless_error) return result def synthesize_female(self, text: str, language: str) -> str: """Robot voice for Live Conversation — guaranteed female Piper voice per language.""" return self.cascade.synthesize_female(text, language) def warm_female_voice(self, language: str) -> None: self.cascade.warm_female(language) def _require_audio_path(audio_path: str | None) -> Path: if not audio_path: raise ValueError("Record a short audio message before starting translation.") resolved = Path(audio_path) if not resolved.exists(): raise FileNotFoundError(f"Audio input was not found: {resolved}") return resolved def _normalize_engine_name(engine_name: str) -> str: normalized = (engine_name or "").strip().lower() if normalized in {"seamless", "qwen", "cascade", "omni"}: return normalized if normalized == "auto": return "seamless" return normalized or "seamless" def _read_wave_as_float32(audio_path: Path, target_rate: int) -> np.ndarray: with wave.open(str(audio_path), "rb") as wav_file: channels = wav_file.getnchannels() sample_width = wav_file.getsampwidth() frame_rate = wav_file.getframerate() frame_count = wav_file.getnframes() raw = wav_file.readframes(frame_count) if sample_width == 2: audio = np.frombuffer(raw, dtype=" 1: audio = audio.reshape(-1, channels).mean(axis=1) if frame_rate != target_rate: audio = _resample_audio(audio, frame_rate, target_rate) return np.clip(audio, -1.0, 1.0).astype(np.float32) def _resample_audio(audio: np.ndarray, source_rate: int, target_rate: int) -> np.ndarray: if source_rate == target_rate: return audio.astype(np.float32) if audio.size == 0: return audio.astype(np.float32) duration = audio.shape[0] / source_rate target_size = max(1, int(round(duration * target_rate))) source_positions = np.linspace(0.0, duration, num=audio.shape[0], endpoint=False) target_positions = np.linspace(0.0, duration, num=target_size, endpoint=False) return np.interp(target_positions, source_positions, audio).astype(np.float32) def _chunk_audio(audio: np.ndarray, chunk_size: int) -> list[np.ndarray]: chunks: list[np.ndarray] = [] cursor = 0 while cursor < audio.shape[0]: chunk = audio[cursor : cursor + chunk_size] if chunk.shape[0] < chunk_size: chunk = np.pad(chunk, (0, chunk_size - chunk.shape[0])) chunks.append(chunk.astype(np.float32)) cursor += chunk_size return chunks or [np.zeros(chunk_size, dtype=np.float32)] def _float32_to_base64(audio: np.ndarray) -> str: return base64.b64encode(audio.astype(" np.ndarray: return np.frombuffer(base64.b64decode(audio_blob), dtype=" str: DEFAULT_AUDIO_DIR.mkdir(parents=True, exist_ok=True) output_path = DEFAULT_AUDIO_DIR / f"{prefix}_{uuid.uuid4().hex}.wav" clipped = np.clip(audio, -1.0, 1.0) pcm = (clipped * 32767.0).astype(" RoutingSpeechEngine: return RoutingSpeechEngine()