"""Vocal-tone reader: HOW Ahmed said it, not WHAT he said. The STT pipeline reduces every utterance to bare text and throws the voice away — pace, pitch, loudness, laughter, all gone before any brain sees it. This module recovers a few cheap acoustic facts from the SAME utterance buffer the speaker-lock already runs on (main.py) and turns them into a one-line descriptor like "fast, urgent" or "slow, flat" that rides along with the transcript into both brains. Design notes: - **Pure numpy, ~2 ms per utterance.** No model, no torch, no GPU, no numba JIT (librosa.yin would add a 6.5 s first-call compile — not worth it). About 7x cheaper than the TitaNet speaker embedding that already runs every turn. - **We do NOT classify emotion.** Sarcasm/joking aren't decidable from sound alone — the descriptor reports the delivery (e.g. "slow, flat") and Claude infers intent from delivery + words + context, the way a person does. - **Relative to the speaker.** Everyone sits at a different pitch/pace/volume, so we track a rolling EMA baseline of Ahmed's own NEUTRAL delivery and describe deviations from it. The baseline only learns from neutral, clean, loud-enough turns so one shouty or one whisper-quiet utterance can't drag it. - **Pitch uses a periodicity gate.** A frame only contributes an F0 if its autocorrelation peak is a strong fraction of zero-lag (real voicing); noise — including a quiet clip that main.py amplified ~10x — has a weak peak and is rejected, so amplified hiss can't fake a huge pitch range. Pitch is only trusted when enough voiced frames survive; otherwise pitch tags are skipped. - **Rate & pitch dynamics are the reliable dimensions** (the AEC mic's AGC flattens absolute volume, so loudness is a soft, baseline-relative hint). - **Never throws into the run loop.** Any failure returns a neutral read. """ from __future__ import annotations import os import re import threading from dataclasses import dataclass, field import numpy as np SAMPLE_RATE = 16_000 FRAME = 512 # 32 ms @ 16 kHz (same frame the VAD uses) _VOICED_GATE = 0.02 # per-frame RMS floor before a frame can be voiced _PERIODICITY = 0.35 # autocorr peak / zero-lag needed to count as voiced _F0_LO, _F0_HI = 75, 330 # human speech F0 search band (Hz) _MIN_VOICED = 8 # need this many voiced frames to trust a pitch read # Warm-up: don't emit deviation tags until the baseline is anchored on a few of # Ahmed's clean utterances. Laughter (transcript-driven) fires from turn one. _SEED_N = 4 _EMA = 0.15 # baseline adaptation rate once seeded _MIN_SPEECH_S = 0.8 # shorter clips are too brief to read prosody _SEED_MIN_LOUD = 0.08 # don't anchor the baseline on near-silent clips _LAUGH_RE = re.compile(r"(?:ha){2,}|(?:he){2,}|\bl(?:ol|mao|mfao)\b|\bhaha\b") @dataclass class Prosody: """One utterance's read. `note` is None when delivery is unremarkable — the caller injects nothing on neutral turns so the brain isn't spammed.""" note: str | None = None # e.g. "fast, urgent" | "slow, flat" | None tags: list[str] = field(default_factory=list) feats: dict = field(default_factory=dict) # --- rolling tone state (published for the TTS to read) -------------------- # read() reports ONE utterance; the TTS wants Ahmed's *sustained* tone so the # voice doesn't lurch on a single loud word. We keep a smoothed, immutable # snapshot here — the run loop folds each read into it (single writer, the # event-loop thread) and the TTS worker thread reads the latest snapshot # lock-free at synth time. This is the sensing half of "voice presence": HOW # he's been sounding, made available to shape HOW Jarvis sounds back. # Thresholds that turn the continuous EMAs into the booleans the TTS reads. # Tuned so only a clearly-sustained tone (not a one-off) trips them. _TONE_UP, _TONE_HIGH = 0.45, 0.90 # animated / very animated _TONE_DOWN, _TONE_LOW = -0.40, -0.90 # flat & low / really flat & tired _TONE_FAST, _TONE_SLOW = 0.40, -0.40 _LAUGH_HOT = 0.5 # laughed within ~1 turn _TONE_ALPHA = 0.35 # EMA rate ≈ last ~3 utterances (one word can't flip it) _LAUGH_DECAY = 0.55 # laughter recency fades per turn # per-tag contribution to the instantaneous energy / pace signal the EMA tracks _TONE_ENERGY_W = {"animated": 1.0, "loud": 0.7, "urgent": 1.0, "fast": 0.5, "laughing": 0.7, "flat": -1.0, "quiet": -0.6, "slow": -0.5} _TONE_PACE_W = {"fast": 1.0, "urgent": 1.0, "slow": -1.0} @dataclass(frozen=True) class ToneState: """A smoothed read of Ahmed's recent vocal energy — the snapshot the TTS colours delivery with. `energy` < 0 = flat/tired, > 0 = animated; `pace` < 0 = slow; `laugh` decays from 1.0 the turn he laughs. Frozen so a worker thread can read it while the run loop swaps in the next one.""" energy: float = 0.0 pace: float = 0.0 laugh: float = 0.0 n: int = 0 @property def up(self) -> bool: return self.energy >= _TONE_UP @property def high(self) -> bool: return self.energy >= _TONE_HIGH @property def down(self) -> bool: return self.energy <= _TONE_DOWN @property def low(self) -> bool: return self.energy <= _TONE_LOW @property def fast(self) -> bool: return self.pace >= _TONE_FAST @property def slow(self) -> bool: return self.pace <= _TONE_SLOW @property def laughing(self) -> bool: return self.laugh >= _LAUGH_HOT _TONE_LOCK = threading.Lock() _TONE_STATE = ToneState() def tone_state() -> ToneState: """Latest smoothed vocal-tone snapshot (never None; neutral until warmed). O(1), lock-free read — safe from the TTS worker thread at synth time.""" return _TONE_STATE def _publish_tone(pr: "Prosody") -> None: """Fold one read into the rolling tone EMA and publish the new snapshot. Side-effect only; reuses the tags read() already produced, so there's no extra DSP and no new call site anywhere.""" global _TONE_STATE tags = pr.tags or [] laughing = ("laughing" in tags) or bool(pr.feats.get("laugh")) e_inst = max(-1.2, min(1.2, sum(_TONE_ENERGY_W.get(t, 0.0) for t in tags))) p_inst = max(-1.0, min(1.0, sum(_TONE_PACE_W.get(t, 0.0) for t in tags))) with _TONE_LOCK: cur = _TONE_STATE _TONE_STATE = ToneState( energy=cur.energy + _TONE_ALPHA * (e_inst - cur.energy), pace=cur.pace + _TONE_ALPHA * (p_inst - cur.pace), laugh=max(1.0 if laughing else 0.0, cur.laugh * _LAUGH_DECAY), n=cur.n + 1) def _laughy(text: str) -> bool: t = re.sub(r"\s+", "", (text or "").lower()) return bool(_LAUGH_RE.search(t)) def _frame_rms(audio: np.ndarray) -> np.ndarray: n = len(audio) // FRAME * FRAME if n == 0: return np.zeros(0, dtype=np.float32) fr = audio[:n].reshape(-1, FRAME) return np.sqrt((fr ** 2).mean(axis=1) + 1e-9) def _pitch_track(audio: np.ndarray) -> np.ndarray: """Per-voiced-frame F0 via autocorrelation with a PERIODICITY gate: a frame only counts if its best in-band autocorr peak is >= _PERIODICITY of zero-lag. That rejects noise/unvoiced frames (weak peak) — including amplified silence — so hiss can't fake pitch. Analyses up to the first 5 s for a stable read.""" audio = audio[:5 * SAMPLE_RATE] n = len(audio) // FRAME * FRAME if n == 0: return np.zeros(0, dtype=np.float32) frames = audio[:n].reshape(-1, FRAME) lo, hi = SAMPLE_RATE // _F0_HI, SAMPLE_RATE // _F0_LO out: list[float] = [] for f in frames: if np.sqrt((f ** 2).mean()) < _VOICED_GATE: continue f = f - f.mean() ac = np.correlate(f, f, "full")[FRAME - 1:] z = ac[0] if z <= 1e-6: continue seg = ac[lo:hi] if seg.size == 0: continue k = int(seg.argmax()) if seg[k] / z < _PERIODICITY: # unvoiced/noise — no real period continue out.append(SAMPLE_RATE / (lo + k)) return np.asarray(out, dtype=np.float32) class ProsodyReader: """Reads vocal tone off raw utterance audio, relative to a rolling baseline of the owner's own neutral delivery. One instance per session; call read() per turn (mirrors SpeakerGate's construct-once/check-per-turn shape).""" def __init__(self) -> None: self._n = 0 self._base_wpm: float | None = None self._base_prange: float | None = None # baseline pitch IQR (Hz) self._base_loud: float | None = None # baseline loudness (pre-norm) self._base_evar: float | None = None # baseline energy variability self.enabled = os.environ.get("PROSODY", "1") != "0" def _update_baseline(self, wpm, prange, loud, evar) -> None: first = self._base_wpm is None def step(cur, new): if new is None: return cur return new if cur is None else cur + _EMA * (new - cur) self._base_wpm = wpm if first else step(self._base_wpm, wpm) self._base_prange = step(self._base_prange, prange) self._base_loud = loud if first else step(self._base_loud, loud) self._base_evar = evar if first else step(self._base_evar, evar) self._n += 1 def read(self, audio: np.ndarray, speech_s: float, text: str, loud: float | None = None) -> Prosody: """audio: float32 mono @ 16 kHz utterance. speech_s: net voiced seconds. text: transcript. loud: PRE-normalization peak/level (main.py has it); falls back to this clip's RMS. Never raises.""" if not self.enabled: return Prosody() try: pr = self._read(audio, speech_s, text, loud) except Exception: # never break the conversation loop over tone return Prosody() try: _publish_tone(pr) # feed the rolling tone state the TTS reads except Exception: # tone is a nicety — never let it break the turn pass return pr def _read(self, audio, speech_s, text, loud) -> Prosody: laughing = _laughy(text) n_words = len((text or "").split()) if loud is None: loud = float(np.sqrt(np.mean(audio.astype(np.float32) ** 2) + 1e-9)) if speech_s < _MIN_SPEECH_S or n_words == 0: if laughing: return Prosody("laughing", ["laughing"], {"laughter": True}) return Prosody(feats={"short": True, "words": n_words}) wpm = 60.0 * n_words / max(speech_s, 0.3) frms = _frame_rms(audio) evar = float(frms.std() / (frms.mean() + 1e-9)) if frms.size else 0.0 pitch = _pitch_track(audio) if pitch.size >= _MIN_VOICED: p25, p75 = np.percentile(pitch, [25, 75]) prange = float(p75 - p25) # robust IQR spread pitch_med = float(np.median(pitch)) else: prange = pitch_med = None # not enough voicing to trust feats = { "wpm": round(wpm), "loud": round(loud, 3), "pitch": round(pitch_med) if pitch_med else 0, "prange": round(prange) if prange is not None else None, "evar": round(evar, 2), "voiced": int(pitch.size), "laugh": laughing, "n": self._n, } # A clip is trustworthy enough to teach the baseline only if it's loud # enough and actually voiced (amplified silence must never anchor it). seedable = loud >= _SEED_MIN_LOUD and pitch.size >= _MIN_VOICED if self._n < _SEED_N or self._base_wpm is None: if seedable: self._update_baseline(wpm, prange, loud, evar) note = "laughing" if laughing else None return Prosody(note, ["laughing"] if laughing else [], feats) tags: list[str] = [] if laughing: tags.append("laughing") # --- speaking rate (AGC-proof; strongest dim). Needs >=4 words for a # trustworthy words-per-minute. --- loud_hit = bool(self._base_loud and loud > self._base_loud * 1.5) if n_words >= 4: if wpm > self._base_wpm * 1.30: tags.append("urgent" if loud_hit else "fast") elif wpm < self._base_wpm * 0.72: tags.append("slow") # --- pitch dynamics — only when this turn's pitch is trustworthy AND the # baseline has one. "flat" is CORROBORATED (monotone pitch AND even # energy) so a single noisy dim can't cry deadpan every turn. --- if prange is not None and self._base_prange and self._base_evar is not None: animated = (prange > self._base_prange * 1.5 or evar > self._base_evar * 1.6) flat = (prange < self._base_prange * 0.6 and evar < self._base_evar * 1.0) if animated: tags.append("animated") elif flat: tags.append("flat") # --- loudness (soft: AGC partly flattens it) --- if loud_hit and not any(t in tags for t in ("urgent", "fast")): tags.append("loud") elif self._base_loud and loud < self._base_loud * 0.55: tags.append("quiet") # Only fold NEUTRAL, trustworthy turns into the baseline — an urgent or # animated stretch must not drift it and desensitize the tags. Laughter # isn't a "deviation" for this (you can laugh at a normal pace/pitch). deviated = any(t in tags for t in ("fast", "slow", "urgent", "animated", "flat", "loud", "quiet")) if not deviated and seedable: self._update_baseline(wpm, prange, loud, evar) seen: list[str] = [] for t in tags: if t not in seen: seen.append(t) seen = seen[:3] note = ", ".join(seen) if seen else None return Prosody(note=note, tags=seen, feats=feats)