#!/usr/bin/env python3
"""
live-whisper.py — Self-contained live transcription worker using openai-whisper.

Fallback for transcribe-cli on platforms where faster-whisper / ONNX fails
(e.g. linux-arm64). Auto-creates a Python venv and installs openai-whisper.

Architecture (EGG voice/whisper lineage):
  - Utterance-chunked transcription: speech-start → sustained silence →
    transcribe the COMPLETE utterance once. No sliding-window re-transcription
    (which duplicated partials, fed whisper silence-heavy windows, and
    hallucinated on quiet audio).
  - ReSpeaker hardware VAD/AEC/DOA: when a ReSpeaker 4-Mic array is present
    (USB 2886:0018), the XMOS DSP's SPEECHDETECTED register gates utterances,
    on-chip echo cancellation is enabled, and the DOA angle is attached to
    transcripts. Falls back to adaptive energy VAD otherwise.
  - Optional dual-model consensus: a second whisper model can validate
    finalized utterances when explicitly enabled.

Protocol:
  stdin  — raw PCM16 audio (16kHz, mono, 16-bit signed LE)
  stdout — JSON lines:
    {"type":"status","message":"Installing dependencies..."}
    {"type":"status","message":"Loading model..."}
    {"type":"ready"}
    {"type":"vad","speech":true,"doa":90}
    {"type":"transcript","text":"hello world","isFinal":false}
    {"type":"transcript","text":"hello world how are you","isFinal":true,"doa":90}
    {"type":"consensus_rejected","primary":"...","secondary":"...","reason":"..."}
    {"type":"error","message":"..."}

Usage:
  arecord -f S16_LE -r 16000 -c 1 -t raw -q - | python3 live-whisper.py --model base
"""

import sys
import os
import json
import subprocess
import struct
import time
import threading
from collections import deque
from pathlib import Path

# ---------------------------------------------------------------------------
# Configuration
# ---------------------------------------------------------------------------

SCRIPT_DIR = Path(__file__).resolve().parent
VENV = SCRIPT_DIR / ".whisper-venv"
PY = VENV / "bin" / "python"
PIP = VENV / "bin" / "pip"

SAMPLE_RATE = 16000
CHANNELS = 1
SAMPLE_WIDTH = 2  # 16-bit
BLOCK_SECONDS = 0.2
BLOCK_SAMPLES = int(SAMPLE_RATE * BLOCK_SECONDS)


def whisper_model_dir() -> Path:
    """Use Omnius' durable model store instead of a package-local cache."""
    configured = os.environ.get("OMNIUS_ASR_MODEL_DIR", "").strip()
    if configured:
        return Path(configured).expanduser()
    return Path.home() / ".omnius" / "models" / "asr" / "openai-whisper"


def env_float(name: str, default: float) -> float:
    raw = os.environ.get(name, "").strip()
    if not raw:
        return default
    try:
        return float(raw)
    except ValueError:
        return default


def env_bool(name: str, default: bool = False) -> bool:
    raw = os.environ.get(name, "").strip().lower()
    if not raw:
        return default
    return raw in ("1", "true", "yes", "on", "enabled")

# ---------------------------------------------------------------------------
# Output helpers (JSON lines to stdout)
# ---------------------------------------------------------------------------

def emit(event: dict):
    """Write a JSON event to stdout and flush."""
    sys.stdout.write(json.dumps(event) + "\n")
    sys.stdout.flush()


def emit_status(msg: str):
    emit({"type": "status", "message": msg})


def emit_error(msg: str):
    emit({"type": "error", "message": msg})


def emit_transcript(text: str, is_final: bool = False, doa=None, consensus=None):
    event = {"type": "transcript", "text": text, "isFinal": is_final}
    if doa is not None:
        event["doa"] = doa
    if consensus is not None:
        event["consensus"] = bool(consensus)
    emit(event)


def emit_consensus_rejected(primary: str, secondary: str, reason: str):
    emit({
        "type": "consensus_rejected",
        "primary": primary[:200],
        "secondary": (secondary or "")[:200],
        "reason": reason,
    })


def cuda_memory_reserved_mb(device: str) -> int:
    if not str(device).startswith("cuda"):
        return 0
    try:
        import torch
        raw = str(device).split(":", 1)[1] if ":" in str(device) else ""
        index = int(raw) if raw.strip().isdigit() else torch.cuda.current_device()
        try:
            torch.cuda.synchronize(index)
        except Exception:
            pass
        reserved = int(torch.cuda.memory_reserved(index))
        allocated = int(torch.cuda.memory_allocated(index))
        return max(0, round(max(reserved, allocated) / (1024 * 1024)))
    except Exception:
        return 0


def host_cuda_hardware_present() -> bool:
    """Best-effort CUDA hardware hint, including Jetson/Orin nodes."""
    for path in (
        "/dev/nvidiactl",
        "/dev/nvidia0",
        "/dev/nvhost-gpu",
        "/proc/driver/nvidia/gpus",
        "/sys/devices/gpu.0",
    ):
        if os.path.exists(path):
            return True
    visible = os.environ.get("NVIDIA_VISIBLE_DEVICES", "").strip().lower()
    return bool(visible and visible not in ("none", "void", "no"))


def asr_cpu_allowed() -> bool:
    raw = os.environ.get("OMNIUS_ASR_ALLOW_CPU", "").strip().lower()
    return raw in ("1", "true", "yes", "on")


def jetson_host_hint() -> str:
    try:
        model = Path("/proc/device-tree/model").read_text(errors="ignore").replace("\x00", " ").strip()
    except Exception:
        model = ""
    if model and any(token in model.lower() for token in ("jetson", "orin", "tegra")):
        return (
            f" Detected {model}. Install a JetPack/L4T-compatible CUDA PyTorch wheel in the Omnius ASR venv; "
            "the generic pip torch wheel is often CPU-only or CUDA-incompatible on Jetson."
        )
    return ""


def cuda_required_error(reason: str):
    emit_error(
        f"CUDA-only ASR is enabled and Whisper cannot use CUDA: {reason}."
        f"{jetson_host_hint()} Set OMNIUS_ASR_ALLOW_CPU=1 only for explicit emergency CPU fallback."
    )
    sys.exit(1)


def select_whisper_device() -> str:
    """Force Whisper onto CUDA whenever PyTorch can see a CUDA device.

    CPU is only allowed when CUDA is genuinely unavailable to this Python
    runtime. If CUDA is visible but unusable, fail loudly instead of silently
    degrading into slow CPU transcription.
    """
    try:
        import torch
    except Exception as e:
        if host_cuda_hardware_present() or not asr_cpu_allowed():
            cuda_required_error(f"PyTorch could not be imported ({e})")
        emit_status(f"PyTorch unavailable for CUDA probe ({e}); using CPU because CUDA is not available to Whisper.")
        return "cpu"

    try:
        cuda_count = int(torch.cuda.device_count())
        cuda_available = bool(torch.cuda.is_available() and cuda_count > 0)
    except Exception as e:
        if host_cuda_hardware_present() or not asr_cpu_allowed():
            cuda_required_error(f"PyTorch CUDA probe failed ({e})")
        emit_status(f"CUDA probe failed ({e}); using CPU because CUDA is not available to Whisper.")
        return "cpu"

    if not cuda_available:
        cuda_version = getattr(getattr(torch, "version", None), "cuda", None)
        if host_cuda_hardware_present() or not asr_cpu_allowed():
            cuda_required_error(f"PyTorch reports CUDA unavailable (torch.version.cuda={cuda_version}, device_count={cuda_count})")
        emit_status(f"CUDA unavailable to PyTorch (torch.version.cuda={cuda_version}, device_count={cuda_count}); using CPU fallback.")
        return "cpu"

    raw_index = os.environ.get("OMNIUS_ASR_CUDA_DEVICE", "0").strip() or "0"
    try:
        device_index = int(raw_index)
    except ValueError:
        emit_error(f"Invalid OMNIUS_ASR_CUDA_DEVICE={raw_index!r}; refusing CPU fallback because CUDA is available.")
        sys.exit(1)
    if device_index < 0 or device_index >= cuda_count:
        emit_error(f"OMNIUS_ASR_CUDA_DEVICE={device_index} outside available CUDA device range 0..{cuda_count - 1}; refusing CPU fallback.")
        sys.exit(1)

    try:
        torch.cuda.set_device(device_index)
        name = torch.cuda.get_device_name(device_index)
    except Exception as e:
        emit_error(f"CUDA is available but cuda:{device_index} could not be selected ({e}); refusing CPU fallback.")
        sys.exit(1)

    emit_status(f"CUDA available; forcing Whisper onto cuda:{device_index} ({name}); torch={getattr(torch, '__version__', 'unknown')} cuda={getattr(getattr(torch, 'version', None), 'cuda', None)}.")
    return f"cuda:{device_index}"

# ---------------------------------------------------------------------------
# Venv bootstrap
# ---------------------------------------------------------------------------

def _in_venv() -> bool:
    return sys.prefix != sys.base_prefix


def _ensure_venv():
    if VENV.exists():
        return
    emit_status("Creating Python venv for Whisper...")
    import venv
    venv.EnvBuilder(with_pip=True).create(str(VENV))
    subprocess.check_call(
        [str(PY), "-m", "pip", "install", "--upgrade", "pip"],
        stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
    )


def _ensure_deps():
    """Check and install missing dependencies."""
    need = []
    try:
        import numpy
    except ImportError:
        need.append("numpy")
    try:
        import whisper
    except ImportError:
        need.append("openai-whisper")

    if need:
        emit_status(f"Installing: {', '.join(need)}...")
        subprocess.check_call(
            [str(PIP), "install", *need],
            stdout=subprocess.DEVNULL, stderr=subprocess.PIPE,
        )
        for mod in ["numpy", "whisper"]:
            if mod in sys.modules:
                del sys.modules[mod]

    # pyusb is optional — only needed for ReSpeaker DSP tuning (hardware
    # VAD/AEC/DOA). Failure is tolerated; we fall back to energy VAD.
    try:
        import usb.core  # noqa: F401
    except ImportError:
        try:
            subprocess.check_call(
                [str(PIP), "install", "pyusb"],
                stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
            )
        except Exception:
            pass

# ---------------------------------------------------------------------------
# Bootstrap: ensure we're in the venv with deps
# ---------------------------------------------------------------------------

if not _in_venv():
    _ensure_venv()
    os.execv(str(PY), [str(PY)] + sys.argv)

_ensure_deps()

import numpy as np

# ---------------------------------------------------------------------------
# ReSpeaker DSP tuning (EGG voice/whisper lineage)
# ---------------------------------------------------------------------------
#
# The ReSpeaker 4-Mic array (USB 2886:0018) exposes its XMOS DSP over USB
# vendor control transfers. The on-chip beamformed VAD (SPEECHDETECTED) is far
# more robust than software energy gating on quiet capture chains, the echo
# canceller stops TTS playback from leaking into ASR, and DOAANGLE localizes
# the speaker.

RESPEAKER_PARAMS = {
    "SPEECHDETECTED":  (19, 22, "int"),
    "DOAANGLE":        (21, 0,  "int"),
    "AECFREEZEONOFF":  (18, 7,  "int"),
    "AECSILENCELEVEL": (18, 30, "float"),
    "ECHOONOFF":       (19, 14, "int"),
}


class ReSpeakerTuning:
    TIMEOUT = 1000  # ms — polled every block; must never stall the FSM

    def __init__(self, dev, usb_util):
        self.dev = dev
        self.usb_util = usb_util
        self.failures = 0

    def write(self, name, value):
        param_id, offset, kind = RESPEAKER_PARAMS[name]
        if kind == "int":
            payload = struct.pack(b"iii", offset, int(value), 1)
        else:
            payload = struct.pack(b"ifi", offset, float(value), 0)
        self.dev.ctrl_transfer(0x40, 0, 0, param_id, payload, self.TIMEOUT)

    def read(self, name):
        param_id, offset, kind = RESPEAKER_PARAMS[name]
        cmd = 0x80 | offset
        if kind == "int":
            cmd |= 0x40
        response = self.dev.ctrl_transfer(0xC0, 0, cmd, param_id, 8, self.TIMEOUT)
        low, high = struct.unpack(b"ii", response.tobytes())
        return low if kind == "int" else low * (2.0 ** high)

    def speech_detected(self):
        """True/False from the DSP VAD, or None when the read fails."""
        try:
            value = self.read("SPEECHDETECTED")
            self.failures = 0
            return bool(value)
        except Exception:
            self.failures += 1
            return None

    def doa(self):
        try:
            return int(self.read("DOAANGLE"))
        except Exception:
            return None

    def close(self):
        try:
            self.usb_util.dispose_resources(self.dev)
        except Exception:
            pass


def find_respeaker():
    """Locate the ReSpeaker array and enable on-chip echo cancellation."""
    try:
        import usb.core
        import usb.util
    except ImportError:
        return None
    try:
        dev = usb.core.find(idVendor=0x2886, idProduct=0x0018)
        if dev is None:
            return None
        tuning = ReSpeakerTuning(dev, usb.util)
        # EGG-recommended AEC settings: adaptive echo canceller on, echo
        # suppression on, raised silence level to mask self-generated audio.
        try:
            tuning.write("AECFREEZEONOFF", 0)
            tuning.write("ECHOONOFF", 1)
            tuning.write("AECSILENCELEVEL", 1e-4)
        except Exception:
            pass  # tuning writes need udev permissions; VAD reads may still work
        # Probe the VAD once — if it doesn't answer, don't use the device.
        if tuning.speech_detected() is None:
            tuning.close()
            return None
        return tuning
    except Exception:
        return None

# ---------------------------------------------------------------------------
# Transcript quality helpers
# ---------------------------------------------------------------------------

def _norm_tokens(text):
    """Normalize a transcript for cross-model comparison: lowercase, strip
    punctuation, collapse whitespace, drop empty tokens."""
    import re
    return [t for t in re.sub(r"[^\w\s]", " ", text.lower()).split() if t]


def _texts_agree(a: str, b: str, threshold: float) -> bool:
    """Consensus check between two model transcripts. Real speech produces
    similar output from different whisper models; hallucinations on noise are
    high-variance and virtually never match across models."""
    ta, tb = _norm_tokens(a), _norm_tokens(b)
    if not ta or not tb:
        return False
    ja, jb = " ".join(ta), " ".join(tb)
    if ja in jb or jb in ja:
        return True
    import difflib
    return difflib.SequenceMatcher(None, ta, tb).ratio() >= threshold


def _segment_filtered_text(result) -> str:
    """Rebuild transcript from segments, dropping low-confidence and
    likely-non-speech segments (classic hallucination carriers)."""
    segments = result.get("segments") or []
    if not segments:
        return result.get("text", "").strip()
    kept = []
    for seg in segments:
        if float(seg.get("no_speech_prob", 0.0)) > 0.6:
            continue
        if float(seg.get("avg_logprob", 0.0)) < -1.2:
            continue
        kept.append(str(seg.get("text", "")))
    return " ".join(part.strip() for part in kept if part.strip()).strip()


def amplify(samples):
    """Bring quiet-but-real speech toward a healthy level for whisper.

    Gain targets the SPEECH RMS (median of the loudest half of 0.2s blocks),
    not the raw peak — a single click/pop no longer suppresses amplification
    of otherwise-quiet speech. Gain is capped at 40x (~32dB) and bounded so
    the peak never clips."""
    if len(samples) == 0:
        return samples
    peak = float(np.max(np.abs(samples)))
    if peak <= 0:
        return samples
    block = max(1, SAMPLE_RATE // 5)
    energies = sorted(
        float(np.sqrt(np.mean(samples[i:i + block] ** 2)))
        for i in range(0, max(1, len(samples) - block + 1), block)
    )
    loud_half = energies[len(energies) // 2:] or energies
    speech_rms = loud_half[len(loud_half) // 2]  # median of the loud half
    target_rms = 0.08  # ≈ -22dBFS — comfortable whisper input level
    if speech_rms >= target_rms:
        return samples
    # No-clip bound uses the 99th-percentile amplitude, not the absolute
    # peak — an isolated click/pop may clip (harmless to ASR) instead of
    # suppressing amplification of the actual speech.
    p99 = float(np.percentile(np.abs(samples), 99.0))
    gain = min(40.0, target_rms / max(speech_rms, 1e-5), 0.98 / max(p99, 1e-4))
    if gain <= 1.05:
        return samples
    return np.clip(samples * gain, -1.0, 1.0)


def pcm16_signal_metrics(samples):
    """Authoritative stats for the raw PCM16 intake before any ASR call."""
    count = int(len(samples))
    if count <= 0:
        return {
            "sampleRateHz": SAMPLE_RATE, "channels": CHANNELS, "bitsPerSample": 16,
            "sampleCount": 0, "durationMs": 0.0, "peakPcm16": 0,
            "rmsPcm16": 0.0, "activeSampleCount": 0, "activeSampleRatio": 0.0,
        }
    # Input originates in int16 stdin. Rounding only restores that exact integer
    # scale for reporting; it never modifies the samples supplied to Whisper.
    pcm = np.rint(samples * 32768.0).astype(np.int32, copy=False)
    magnitude = np.abs(pcm)
    peak = int(np.max(magnitude))
    active = int(np.count_nonzero(magnitude >= 8))
    return {
        "sampleRateHz": SAMPLE_RATE, "channels": CHANNELS, "bitsPerSample": 16,
        "sampleCount": count, "durationMs": count * 1000.0 / SAMPLE_RATE,
        "peakPcm16": peak, "rmsPcm16": float(np.sqrt(np.mean(pcm.astype(np.float64) ** 2))),
        "activeSampleCount": active, "activeSampleRatio": active / count,
    }


def digitally_silent_pcm16(samples):
    metrics = pcm16_signal_metrics(samples)
    return (
        metrics["peakPcm16"] <= 4
        and metrics["rmsPcm16"] <= 1.0
        and metrics["activeSampleCount"] == 0
    ), metrics

# ---------------------------------------------------------------------------
# Main transcription loop — utterance FSM
# ---------------------------------------------------------------------------

def main():
    import argparse
    parser = argparse.ArgumentParser(description="Live Whisper transcription worker")
    parser.add_argument("--model", default="medium", help="Whisper model size (tiny/base/small/medium/large)")
    # Accepted for backward compatibility with older launchers (sliding-window era).
    parser.add_argument("--chunk-seconds", type=float, default=3, help=argparse.SUPPRESS)
    parser.add_argument("--window-seconds", type=float, default=10, help=argparse.SUPPRESS)
    parser.add_argument("--language", default=None, help="Language code (e.g. en, es, fr). Auto-detect if omitted.")
    parser.add_argument("--consensus-model", default="off",
                        help="Second whisper model run in parallel on finalized utterances; transcripts are only emitted when both models agree. 'off' disables.")
    parser.add_argument("--consensus-threshold", type=float, default=0.55,
                        help="Token similarity (0-1) required between the two models' transcripts.")
    parser.add_argument("--silence-ms", type=float, default=3000,
                        help="Sustained silence that finalizes an utterance.")
    parser.add_argument("--preroll-ms", type=float, default=600,
                        help="Audio kept from before speech onset.")
    parser.add_argument("--max-utterance-seconds", type=float, default=22,
                        help="Force-finalize utterances longer than this.")
    parser.add_argument("--partials", action="store_true", default=env_bool("OMNIUS_ASR_PARTIALS", False),
                        help="Emit interim Whisper transcripts during long utterances. Disabled by default to avoid silence/noise hallucinations.")
    parser.add_argument("--min-speech-ms", type=float, default=env_float("OMNIUS_ASR_MIN_SPEECH_MS", 450.0),
                        help="Minimum VAD-positive speech duration required before an utterance is sent to Whisper.")
    parser.add_argument("--start-rms", type=float, default=env_float("OMNIUS_ASR_START_RMS", 0.0025),
                        help="Minimum block RMS required for software VAD speech start.")
    parser.add_argument("--energy-ratio", type=float, default=env_float("OMNIUS_ASR_ENERGY_RATIO", 3.5),
                        help="Software VAD speech gate as a multiple of the adaptive noise floor.")
    parser.add_argument("--min-utterance-rms", type=float, default=env_float("OMNIUS_ASR_MIN_UTTERANCE_RMS", 0.003),
                        help="Minimum loud-block RMS required before final transcription.")
    parser.add_argument("--min-utterance-peak", type=float, default=env_float("OMNIUS_ASR_MIN_UTTERANCE_PEAK", 0.015),
                        help="Minimum utterance peak amplitude required when RMS is low.")
    args = parser.parse_args()

    import whisper

    device = select_whisper_device()
    emit_status(f"Loading Whisper {args.model} model on {device}...")
    try:
        model_dir = whisper_model_dir()
        model_dir.mkdir(parents=True, exist_ok=True)
        model = whisper.load_model(args.model, device=device, download_root=str(model_dir))
    except Exception as e:
        if str(device).startswith("cuda"):
            emit_error(f"Failed to load model on {device}; refusing CPU fallback because CUDA is available: {e}")
        else:
            emit_error(f"Failed to load model on CPU fallback: {e}")
        sys.exit(1)

    # Consensus model: hallucinations (mixed-language garbage on noisy or
    # quiet audio) don't reproduce across models, so disagreement = reject.
    consensus_model = None
    consensus_name = (args.consensus_model or "off").strip().lower()
    if consensus_name not in ("off", "none", "", args.model):
        emit_status(f"Loading consensus Whisper {consensus_name} model on {device}...")
        try:
            consensus_model = whisper.load_model(
                consensus_name,
                device=device,
                download_root=str(whisper_model_dir()),
            )
        except Exception as e:
            emit_status(f"Consensus model unavailable ({e}); running single-model.")
            consensus_model = None

    # ReSpeaker DSP: hardware VAD + echo cancellation + DOA
    tuning = find_respeaker()
    if tuning is not None:
        emit_status("ReSpeaker DSP active: hardware VAD, echo cancellation, DOA.")

    fp16 = str(device).startswith("cuda")
    emit({
        "type": "ready",
        "model": args.model,
        "device": str(device),
        "cuda": bool(fp16),
        "consensusModel": consensus_name if consensus_model is not None else None,
        "vramUsedMB": cuda_memory_reserved_mb(str(device)),
    })

    def run_transcribe(m, samples):
        # Every live PCM ASR call goes through this exact gate. It is stricter
        # than VAD only for digital silence, so it cannot reject quiet speech.
        silent, _ = digitally_silent_pcm16(samples)
        if silent:
            return {"text": "", "segments": []}
        return m.transcribe(
            samples,
            fp16=fp16,
            language=args.language,
            no_speech_threshold=0.6,
            condition_on_previous_text=False,
        )

    # Shared PCM intake
    audio_buf = np.zeros(0, dtype=np.float32)
    buf_lock = threading.Lock()
    running = True

    def read_stdin():
        nonlocal audio_buf, running
        try:
            while running:
                data = sys.stdin.buffer.read(BLOCK_SAMPLES * SAMPLE_WIDTH)
                if not data:
                    break  # EOF
                samples = np.frombuffer(data, dtype=np.int16).astype(np.float32) / 32768.0
                with buf_lock:
                    audio_buf = np.concatenate([audio_buf, samples])
        except Exception:
            pass
        finally:
            running = False

    reader = threading.Thread(target=read_stdin, daemon=True)
    reader.start()

    # Utterance FSM state
    preroll_blocks = max(1, int(round(args.preroll_ms / 1000.0 / BLOCK_SECONDS)))
    finalize_blocks = max(2, int(round(args.silence_ms / 1000.0 / BLOCK_SECONDS)))
    max_utter_samples = int(args.max_utterance_seconds * SAMPLE_RATE)

    preroll = deque(maxlen=preroll_blocks)
    utterance = []       # list of float32 blocks while capturing
    capturing = False
    silence_run = 0
    speech_blocks = 0
    utterance_rms_values = []
    utterance_peak = 0.0
    noise_floor = None
    last_partial_at = 0.0
    last_final_text = ""
    last_vad_emit = (False, 0.0)
    pending = np.zeros(0, dtype=np.float32)
    cursor = 0

    def energy_speech(block_rms):
        nonlocal noise_floor
        if noise_floor is None:
            noise_floor = block_rms
        elif block_rms < noise_floor:
            noise_floor = 0.7 * noise_floor + 0.3 * block_rms  # drop fast
        else:
            noise_floor = 0.995 * noise_floor + 0.005 * block_rms  # creep up slowly
        gate = max(float(args.start_rms), noise_floor * float(args.energy_ratio))
        return block_rms >= gate

    def transcribe_partial(samples):
        try:
            result = run_transcribe(model, amplify(samples))
            text = _segment_filtered_text(result)
            if text:
                emit_transcript(
                    text,
                    is_final=False,
                    consensus=False if consensus_model is not None else None,
                )
        except Exception as e:
            emit_error(f"Transcription error: {e}")

    def utterance_quality_ok(samples, speech_block_count, rms_values, peak) -> bool:
        if len(samples) < int(0.4 * SAMPLE_RATE):
            return False
        speech_ms = speech_block_count * BLOCK_SECONDS * 1000.0
        if speech_ms < float(args.min_speech_ms):
            if env_bool("OMNIUS_ASR_VERBOSE_VAD", False):
                emit_status(f"Dropped short VAD blob before Whisper: speech_ms={speech_ms:.0f}")
            return False
        if rms_values:
            ordered = sorted(float(v) for v in rms_values)
            loud_half = ordered[len(ordered) // 2:] or ordered
            loud_rms = loud_half[len(loud_half) // 2]
        else:
            loud_rms = float(np.sqrt(np.mean(samples ** 2))) if len(samples) else 0.0
        if loud_rms < float(args.min_utterance_rms) and peak < float(args.min_utterance_peak):
            if env_bool("OMNIUS_ASR_VERBOSE_VAD", False):
                emit_status(
                    f"Dropped low-energy VAD blob before Whisper: loud_rms={loud_rms:.5f} peak={peak:.5f}"
                )
            return False
        return True

    def transcribe_final(samples, speech_block_count=0, rms_values=None, peak=0.0):
        nonlocal last_final_text
        silent, signal = digitally_silent_pcm16(samples)
        if silent:
            emit({"type": "no_speech", "kind": "no_speech", "reason": "digital_silence", "text": "", "signal": signal})
            return
        if not utterance_quality_ok(samples, speech_block_count, rms_values or [], peak):
            return
        samples = amplify(samples)
        try:
            result = run_transcribe(model, samples)

            text = _segment_filtered_text(result)
            if not text:
                return
            if text == last_final_text:
                return
            doa = tuning.doa() if tuning else None
            if consensus_model is None:
                last_final_text = text
                emit_transcript(text, is_final=True, doa=doa, consensus=None)
                return

            # Fast UX path: surface the primary model's finalized text
            # immediately as a non-final transcript. The secondary model then
            # validates whether it is safe to accept as a final utterance.
            emit_transcript(text, is_final=False, doa=doa, consensus=False)

            def validate_consensus(primary_text, primary_result, audio_samples, primary_doa):
                nonlocal last_final_text
                try:
                    secondary = run_transcribe(consensus_model, audio_samples)
                    secondary_text = _segment_filtered_text(secondary)
                    lang_a = str(primary_result.get("language", "")).lower()
                    lang_b = str(secondary.get("language", "")).lower()
                    lang_disagree = bool(lang_a and lang_b and lang_a != lang_b and not args.language)
                    if lang_disagree or not _texts_agree(primary_text, secondary_text, args.consensus_threshold):
                        emit_consensus_rejected(primary_text, secondary_text, "language_mismatch" if lang_disagree else "low_similarity")
                        return
                    last_final_text = primary_text
                    emit_transcript(primary_text, is_final=True, doa=primary_doa, consensus=True)
                except Exception as e:
                    emit_consensus_rejected(primary_text, "", "validator_error")
                    if env_bool("OMNIUS_ASR_VERBOSE_CONSENSUS", False):
                        emit_status(f"Consensus validation error: {e}")

            threading.Thread(
                target=validate_consensus,
                args=(text, result, samples.copy(), doa),
                daemon=True,
            ).start()
        except Exception as e:
            emit_error(f"Transcription error: {e}")

    try:
        while running:
            time.sleep(BLOCK_SECONDS / 2)

            with buf_lock:
                if len(audio_buf) > cursor:
                    pending = np.concatenate([pending, audio_buf[cursor:]])
                    cursor = len(audio_buf)
                # Bound memory: keep the buffer from growing without limit.
                if len(audio_buf) > 120 * SAMPLE_RATE:
                    audio_buf = audio_buf[-SAMPLE_RATE:]
                    cursor = len(audio_buf)

            while len(pending) >= BLOCK_SAMPLES:
                block = pending[:BLOCK_SAMPLES]
                pending = pending[BLOCK_SAMPLES:]

                block_rms = float(np.sqrt(np.mean(block ** 2)))
                energy = energy_speech(block_rms)
                hw = tuning.speech_detected() if tuning is not None else None
                if tuning is not None and tuning.failures > 5:
                    # Repeated USB read failures (unplug, permission loss) —
                    # stop polling so control-transfer timeouts can't stall
                    # the FSM; energy VAD takes over.
                    emit_status("ReSpeaker DSP unresponsive — falling back to energy VAD.")
                    tuning.close()
                    tuning = None
                    hw = None
                # Hardware VAD is authoritative when the DSP answers; energy
                # VAD covers non-ReSpeaker mics and USB read failures.
                speech = hw if hw is not None else energy

                now = time.time()
                if speech != last_vad_emit[0] and now - last_vad_emit[1] > 0.3:
                    last_vad_emit = (speech, now)
                    emit({"type": "vad", "speech": bool(speech), "doa": tuning.doa() if tuning else None})

                if not capturing:
                    preroll.append(block)
                    if speech:
                        capturing = True
                        silence_run = 0
                        speech_blocks = 1
                        utterance_rms_values = [block_rms]
                        utterance_peak = float(np.max(np.abs(block))) if len(block) else 0.0
                        utterance = list(preroll)
                        preroll.clear()
                    continue

                utterance.append(block)
                utterance_rms_values.append(block_rms)
                utterance_peak = max(utterance_peak, float(np.max(np.abs(block))) if len(block) else 0.0)
                if speech:
                    speech_blocks += 1
                    silence_run = 0
                else:
                    silence_run += 1

                utter_samples = sum(len(b) for b in utterance)
                if silence_run >= finalize_blocks or utter_samples >= max_utter_samples:
                    samples = np.concatenate(utterance)
                    final_speech_blocks = speech_blocks
                    final_rms_values = list(utterance_rms_values)
                    final_peak = utterance_peak
                    capturing = False
                    utterance = []
                    silence_run = 0
                    speech_blocks = 0
                    utterance_rms_values = []
                    utterance_peak = 0.0
                    transcribe_final(samples, final_speech_blocks, final_rms_values, final_peak)
                elif args.partials and utter_samples >= int(3.5 * SAMPLE_RATE) and now - last_partial_at >= 2.5:
                    last_partial_at = now
                    transcribe_partial(np.concatenate(utterance))
    except KeyboardInterrupt:
        pass

    # EOF: finalize any in-flight utterance.
    if utterance:
        try:
            transcribe_final(
                np.concatenate(utterance),
                speech_blocks,
                list(utterance_rms_values),
                utterance_peak,
            )
        except Exception:
            pass
    if tuning is not None:
        tuning.close()
    running = False


if __name__ == "__main__":
    main()