engine.py

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="<i2").astype(np.float32) / 32768.0
    elif sample_width == 4:
        audio = np.frombuffer(raw, dtype="<i4").astype(np.float32) / 2147483648.0
    else:
        raise RuntimeError(f"Unsupported WAV sample width: {sample_width}")

    if channels > 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("<f4").tobytes()).decode("ascii")


def _base64_to_float32(audio_blob: str) -> np.ndarray:
    return np.frombuffer(base64.b64decode(audio_blob), dtype="<f4").astype(np.float32)


def _write_float_audio_wav(audio: np.ndarray, sample_rate: int, prefix: str) -> 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("<i2")
    with wave.open(str(output_path), "wb") as wav_file:
        wav_file.setnchannels(1)
        wav_file.setsampwidth(2)
        wav_file.setframerate(sample_rate)
        wav_file.writeframes(pcm.tobytes())
    return str(output_path)


def build_engine() -> RoutingSpeechEngine:
    return RoutingSpeechEngine()