"""Pocket TTS (Kyutai, MIT) — the Mac voice: cloned, streaming, CPU-only. Chatterbox on Apple Silicon is structurally slow (MPS fallback ops, ~3GB, thermal throttle on the fanless Air). Pocket TTS is a 100M CALM model that runs ~7x realtime on TWO CPU cores of an M4 Air with ~40ms to first audio chunk — so the Mac gets the cloned Jarvis voice without touching the GPU or the RAM budget (~200MB). Voice: voices/jarvis_ref.wav is cloned when the gated HF weights are available (accept terms at https://huggingface.co/kyutai/pocket-tts, then `hf auth login`). Without them it falls back to a catalog voice (POCKET_VOICE, default "george") — never mute. Emotion: Pocket has no exaggeration dial, so the [emotion] tag protocol maps to *reference-clip swaps*: drop voices/jarvis_warm.wav, jarvis_excited.wav, jarvis_dry.wav, jarvis_sad.wav, jarvis_calm.wav (generate them once with Chatterbox on the RTX box, exaggeration dialed per mood) and each tag speaks through the matching clone. Missing clips just use the base voice. Emotion states preload in a background thread after startup so the first [excited] sentence doesn't stall. Env: POCKET_VOICE (catalog fallback voice), VOICE_REF (base clone wav, default voices/jarvis_ref.wav), POCKET_MAX_TOKENS (default 300 ≈ 24s cap). """ from __future__ import annotations import os import threading from pathlib import Path import numpy as np from voice.tts_chatterbox import _ANY_TAG_RE, _TAG_RE, split_emotion SAMPLE_RATE = 24_000 # pocket-tts native rate — same as Kokoro/Chatterbox _PROJECT = Path(__file__).resolve().parent.parent _DEFAULT_REF = _PROJECT / "voices" / "jarvis_ref.wav" _CATALOG_FALLBACK = os.environ.get("POCKET_VOICE", "george") _MAX_TOKENS = int(os.environ.get("POCKET_MAX_TOKENS", "300")) # emotion tag -> reference-clip stem (voices/jarvis_.wav). Tags come # from tts_chatterbox._EMOTIONS so the brain's protocol is identical on # both machines; several tags share one mood clip. _TAG_TO_STEM = { "calm": "calm", "flat": "calm", "serious": "calm", "warm": "warm", "happy": "warm", "excited": "excited", "urgent": "excited", "alarmed": "excited", "sad": "sad", "tired": "sad", "sarcastic": "dry", "dry": "dry", # richer palette (2026-07-10) — new stems need their jarvis_.wav clip "shocked": "shocked", "surprised": "shocked", "curious": "curious", "interested": "curious", "intrigued": "curious", "amused": "amused", "playful": "amused", "cheeky": "amused", "annoyed": "annoyed", "irritated": "annoyed", "impatient": "annoyed", "testy": "annoyed", "tender": "tender", "gentle": "tender", "soft": "tender", "neutral": None, # base voice } # --- tone-aware delivery (TONE_TTS, default on) ---------------------------- # The brain's [emotion] tag says how the WORDS feel; voice.prosody.tone_state # says how AHMED has been sounding. TONE_TTS fuses them (the Sesame-"voice # presence" move): a specific tag still wins, but with no usable tag his tone # picks the mood clip, and a bouncy tag softens when he's plainly flat and # tired — Jarvis shouldn't bounce at a drained man. The persona/frustration # tones (urgent/annoyed/shocked) NEVER soften: the meter's boil-over earns its # shout. TONE_TTS=0 restores today's tag-only behaviour byte-for-byte. _BOUNCE_TAGS = frozenset({"excited", "amused", "happy", "playful", "cheeky"}) _NEVER_DAMP = frozenset({"urgent", "alarmed", "annoyed", "irritated", "impatient", "testy", "shocked", "surprised"}) _DAMP_STEM = "warm" # a bounce softened for a tired Ahmed _LATE_START, _LATE_END = 22, 5 # "late night" → an extra-soft read def _late_night(hour: int | None) -> bool: return hour is not None and (hour >= _LATE_START or hour < _LATE_END) def _local_hour() -> int: from datetime import datetime return datetime.now().hour def _parse_tag(text: str) -> tuple[str | None, str | None]: """(emotion word, mood-clip stem) from a leading [emotion] tag — the exact lookup _state_for did inline, factored out so tone-biasing can reuse it. Improvised tags map via their last word; unknown ones give (word, None).""" m = _TAG_RE.match(text or "") if m: w = m.group(1).lower() return w, _TAG_TO_STEM.get(w) m = _ANY_TAG_RE.match(text or "") if m: inner = m.group(0).strip()[1:-1].strip().lower() w = inner.split()[-1] if inner else "" return (w or None), _TAG_TO_STEM.get(w) return None, None def _tone_stem(tone, hour: int | None) -> str | None: """No usable brain tag → pick a mood clip from Ahmed's vocal tone. Returns a jarvis_ key, or None for the base voice (today's untagged delivery) when his tone is unremarkable.""" if tone.laughing: return "amused" if tone.up: return "excited" if tone.high else "amused" if tone.down: if _late_night(hour) and tone.slow: return "calm" # late-night flat & slow → soft return "tender" if tone.low else "warm" return None def _biased_stem(tag_word, base_stem, tone, hour): """Fuse the brain's tag with Ahmed's live tone. `tag_word` = the brain's emotion word (or None); `base_stem` = the clip it maps to (None = neutral / no clip). A specific tag wins — except a bouncy one softens to `warm` when he's clearly down; urgent/annoyed/shocked are never softened.""" if base_stem is not None: if (tag_word in _BOUNCE_TAGS and tag_word not in _NEVER_DAMP and tone.down): return _DAMP_STEM return base_stem return _tone_stem(tone, hour) class PocketTTS: """Drop-in for KokoroTTS/ChatterboxTTS: synth(text) -> float32 @ 24kHz. Also exposes synth_stream(text) yielding ~80ms chunks (~40ms latency).""" accepts_tags = True # main.py passes the raw [emotion]-tagged sentence def __init__(self) -> None: from pocket_tts import TTSModel self._model = TTSModel.load_model() self._lock = threading.Lock() # model state isn't re-entrant self._states: dict[str, object] = {} self._cloned = False self._speed = 1.0 # kokoro-compat; pocket paces itself # Bias the mood clip by Ahmed's live vocal tone (see _biased_stem). # Pocket exposes no speed/temperature knob, so tone steers clip choice # only — no pace nudge (resampling would hurt the cloned voice). self._tone_tts = os.environ.get("TONE_TTS", "1") != "0" base_ref = os.environ.get("VOICE_REF", str(_DEFAULT_REF)) self._base_name = self._load_base(base_ref) # 'plain' voice = not the canonical Jarvis clone → skip the jarvis mood # clips (used for the local-only 'fufu' alternate voice toy). self._plain = not self._is_default(self._base_name) self.synth("Warm up.") # first call warms caches threading.Thread(target=self._preload_emotions, daemon=True).start() @staticmethod def _is_default(name: str) -> bool: """True when `name` is the canonical Jarvis clone (the only voice the mood clips belong to). Any other base voice speaks 'plain' — the jarvis mood clips would sound like a different person mid-sentence.""" try: return os.path.abspath(name) == os.path.abspath(str(_DEFAULT_REF)) except Exception: # noqa: BLE001 return False def _load_base(self, ref: str) -> str: """Clone from the ref wav if the gated weights allow it, else use a catalog voice. Returns a display name for logs/status.""" if os.path.isfile(ref): try: self._states[""] = self._model.get_state_for_audio_prompt(ref) self._cloned = True print(f" voice: cloned from {os.path.basename(ref)} (pocket)") return ref except Exception as e: # noqa: BLE001 — gated weights, bad wav… msg = str(e) if "voice cloning" in msg.lower(): print(" voice: cloning weights are GATED — accept terms " "at huggingface.co/kyutai/pocket-tts then " "`hf auth login`; using catalog voice " f"'{_CATALOG_FALLBACK}' meanwhile") else: print(f" voice: clone failed ({msg[:80]}) — catalog " f"'{_CATALOG_FALLBACK}'") self._states[""] = self._model.get_state_for_audio_prompt( _CATALOG_FALLBACK) return _CATALOG_FALLBACK def _preload_emotions(self) -> None: """Build voice states for whatever jarvis_.wav clips exist. Runs in the background after startup; each takes a few seconds.""" if not self._cloned: return # emotion clips are clones; catalog voice has no moods for stem in dict.fromkeys(s for s in _TAG_TO_STEM.values() if s): path = _PROJECT / "voices" / f"jarvis_{stem}.wav" if not path.is_file(): continue try: state = self._model.get_state_for_audio_prompt(str(path)) self._states[stem] = state except Exception as e: # noqa: BLE001 print(f" voice: mood clip {path.name} failed ({str(e)[:60]})") moods = [k for k in self._states if k] if moods: print(f" voice: mood clones ready ({', '.join(sorted(moods))})") # ---- kokoro-compat attrs so control.py hot-reload keeps working ---- @property def _voice(self): return self._base_name @_voice.setter def _voice(self, v): # config.json {"voice": "path.wav"|"catalog_name"} if not isinstance(v, str) or v == self._base_name: return try: with self._lock: self._states[""] = self._model.get_state_for_audio_prompt(v) self._base_name = v self._plain = not self._is_default(v) print(f" voice: switched to {os.path.basename(v)} " f"({'plain — moods off' if self._plain else 'jarvis + moods'})") except Exception as e: # noqa: BLE001 print(f" voice: switch to {v!r} failed ({str(e)[:60]})") def _state_for(self, text: str): """Strip the [emotion] tag and pick the matching voice state. Improvised tags ("[Mildly excited]") map via their last word. With TONE_TTS on, Ahmed's live vocal tone also steers the pick when the line is untagged and softens a bouncy tag when he's plainly flat — while urgent/annoyed/shocked always speak at full tilt (see _biased_stem).""" # A 'plain' alternate voice (e.g. fufu) has no mood clones of its own — # speak everything in its base state so it doesn't flip to jarvis's voice. if getattr(self, "_plain", False): clean, _ = split_emotion(text) return clean, self._states[""] tag_word, base_stem = _parse_tag(text or "") if getattr(self, "_tone_tts", False): from voice import prosody stem = _biased_stem(tag_word, base_stem, prosody.tone_state(), _local_hour()) else: stem = base_stem clean, _ = split_emotion(text) return clean, self._states.get(stem or "", self._states[""]) def synth(self, text: str) -> np.ndarray: """text (optionally '[emotion] ...') -> float32 mono PCM @ 24kHz.""" clean, state = self._state_for(text) if not clean.strip(): return np.zeros(0, dtype=np.float32) with self._lock: wav = self._model.generate_audio( state, clean, max_tokens=_MAX_TOKENS) return wav.detach().cpu().numpy().astype(np.float32).reshape(-1) def synth_stream(self, text: str): """Yield float32 PCM chunks (~80ms each) as they're generated — first chunk in ~40ms on an M4. Holds the model lock for the whole sentence; callers stream one sentence at a time anyway.""" clean, state = self._state_for(text) if not clean.strip(): return with self._lock: for chunk in self._model.generate_audio_stream( state, clean, max_tokens=_MAX_TOKENS): yield chunk.detach().cpu().numpy().astype( np.float32).reshape(-1)