""" Telling the user's voice apart from our own coming back through the speakers. What it is used for today ------------------------- The *echo tail*: for a moment after a reply ends, sound is still leaving the speakers, because writing to an audio device returns when the buffer accepts the audio and not when the room has finished with it. Streaming the microphone during that gap is how an assistant hears its own last sentence and answers itself. This module lets the tail drop our own voice while still passing a user who replies instantly, instead of blanket-muting. Interrupting mid-sentence by voice is the other thing it can do, and everything needed for it is here and tested — but it depends on the listener's room, so it stays switched off until it can be tried on real hardware. See the note in `_listen_audio` for how little it takes to turn back on. The problem with a level test ----------------------------- Deciding "the microphone is louder than the echo" needs to know how loud the echo *is*, and that is not a constant. It depends on the speaker volume, where the microphone sits, the room, whether headphones are plugged in, and whether the user happens to have a hand over the mic. A single tuned number is wrong for almost everyone: too eager on a loud desktop speaker, too deaf on a quiet laptop. What this does instead ---------------------- Two independent signals, neither of which needs tuning per machine: 1. **Content.** Echo is not merely loud — it is *the same sound we just played*. Both streams are reduced to a handful of band energies, and then — rather than merely compared — as much of the recent output as fits is *subtracted* from the microphone block. Pure echo cancels to almost nothing. A second voice cannot be cancelled by ours, because its formants sit in bands where ours were weak, so it survives the subtraction. That distinction holds even when the two arrive at the same loudness, which is exactly the case a level test gets wrong. 2. **A learned echo gain.** Whenever the content check says "that is definitely just echo", the observed mic-to-output ratio is folded into a running estimate. The system therefore calibrates itself to the actual room within a few seconds of the first sentence, and re-calibrates when the volume changes or a hand covers the microphone. Band energies rather than raw spectra also make the comparison sample-rate agnostic, which matters here: the microphone runs at 16 kHz and playback at 24 kHz. """ from __future__ import annotations import time import numpy as np # Log-spaced edges across the range that carries speech. Coarse on purpose: # fine bins would track pitch, and pitch is exactly what differs between two # people saying the same word — we want the *timbre* that echo preserves. _BAND_EDGES = (200, 400, 700, 1100, 1700, 2600, 3800, 5200, 7000) _HISTORY_S = 1.5 # how far back an echo could plausibly have been played _MIN_LEVEL = 0.06 # below this the mic is room noise; nothing to decide # The threshold is not a tuned constant. It is placed just above essentially ALL # the echo this particular room has been seen to produce. Residuals measured on # synthetic material, median over many blocks: # # room echo p99 a distinct voice a similar voice # headphones 0.09 0.28 0.16 # normal desk 0.13 0.28 0.16 # reverberant room 0.23 0.27 0.18 # loud, distorting, noisy 0.31 0.33 0.26 # # Read that table honestly. How good the room is decides what is possible: in a # quiet setup even a voice whose formants sit close to ours stands clear of the # echo, and in a bad one nothing reliably does. The right response to the last # row is to say so — not to interrupt at random. _MIN_USER = 0.15 # never call anything below this a voice _HEAD_Q = 97 # percentile of observed echo the threshold must clear _HEAD_MULT = 1.15 # ...with this much headroom above it # Above this echo floor, content separates the two poorly. Loudness cannot # rescue it — in exactly those rooms the echo already saturates the level meter, # so a level test can never be passed. Time can: echo wobbles around its median # while a person talking holds above it, so such rooms hold the evidence longer. _UNRELIABLE_FLOOR = 0.22 _BLOCKS_NORMAL = 5 # ~320 ms of sustained evidence _BLOCKS_NOISY = 12 # ~770 ms when the room is this hard _FLOOR_WINDOW = 60 # blocks kept to characterise the room's echo _FLOOR_Q = 35 # percentile taken as "typical echo here" _WARMUP = 16 # blocks (~1 s) of listening before judging anyone _RELEARN_RUN = 28 # a 'voice' lasting this long means the room changed def band_energies(pcm, sr: int) -> np.ndarray: """Raw energy per speech band — the fingerprint we compare. Left unnormalised on purpose: the decision below projects one of these onto another, and a projection needs real magnitudes. """ x = np.asarray(pcm, dtype=np.float32) if x.size < 64: return np.zeros(len(_BAND_EDGES) - 1, dtype=np.float32) x = x - x.mean() n = 1 << (int(x.size) - 1).bit_length() # next power of two mag = np.abs(np.fft.rfft(x * np.hanning(x.size), n=n)) freqs = np.fft.rfftfreq(n, 1.0 / sr) out = np.empty(len(_BAND_EDGES) - 1, dtype=np.float32) for i in range(len(_BAND_EDGES) - 1): m = (freqs >= _BAND_EDGES[i]) & (freqs < _BAND_EDGES[i + 1]) out[i] = float(mag[m].sum()) return out class EchoGuard: """Classifies microphone blocks while the assistant is speaking. Usage: `note_output()` from the playback path, `is_user_speech()` from the microphone callback. Both are cheap enough to sit in an audio thread — one small FFT each. """ def __init__(self) -> None: self._hist: list[tuple[float, np.ndarray, float]] = [] # (t, bands, level) self._gain = 0.6 # mic level per unit of output level; learned self._seen = 0 # how many echo blocks the estimate has seen self._last_sim = 0.0 self._last_expected = 0.0 self._residuals: list[float] = [] # recent ECHO residuals only self._floor = 0.10 # typical echo residual here; learned self._run = 0 # consecutive blocks called speech self._head = 0.13 # near-worst echo residual here; learned # ── diagnostics, for the log and for tests ────────────────────────────── @property def gain(self) -> float: return self._gain @property def calibrated(self) -> bool: """True once the estimate rests on enough real echo to be trusted.""" return self._seen >= 8 @property def floor(self) -> float: """Residual left by this room's own echo. Higher = harder to separate.""" return self._floor @property def reliable(self) -> bool: """False when the acoustics are too poor to judge on content alone. Speakers turned up in a reverberant room leave an echo that survives the subtraction almost as well as a quiet voice does. Rather than guess, the guard says so and starts demanding corroborating evidence. """ return self._floor < _UNRELIABLE_FLOOR @property def threshold(self) -> float: """The residual a block must clear right now to count as a voice.""" return max(_MIN_USER, self._head * _HEAD_MULT) @property def required_blocks(self) -> int: """Consecutive positive blocks before an interruption is believed.""" return _BLOCKS_NORMAL if self.reliable else _BLOCKS_NOISY @property def last_similarity(self) -> float: """1.0 = fully explained by our own output, 0.0 = nothing to do with it.""" return self._last_sim def reset(self) -> None: """Playback stopped — drop the history, keep what was learned.""" self._hist.clear() self._last_sim = 0.0 # ── the two entry points ──────────────────────────────────────────────── def note_output(self, pcm, sr: int, level: float, when: float | None = None) -> None: """Record a slice of what is being played, for later comparison.""" try: t = time.monotonic() if when is None else when self._hist.append((t, band_energies(pcm, sr), float(level))) cutoff = t - _HISTORY_S if len(self._hist) > 8: self._hist = [h for h in self._hist if h[0] >= cutoff] except Exception: pass # never let bookkeeping disturb playback def is_user_speech(self, pcm, sr: int, level: float, when: float | None = None) -> bool: """True if this microphone block is a different voice, not our echo.""" try: if level < _MIN_LEVEL: # We are playing and the microphone hears nothing back: # this setup returns no echo at all (headphones, or a mic # far from the speaker). Record that, or the window stays # empty and the first person to speak gets mistaken for # the calibration sample. if self._hist and max(h[2] for h in self._hist) > 0.15: self._residuals.append(0.0) if len(self._residuals) > _FLOOR_WINDOW: del self._residuals[:-_FLOOR_WINDOW] if len(self._residuals) >= _WARMUP: self._floor = float(np.percentile(self._residuals, _FLOOR_Q)) self._head = float(np.percentile(self._residuals, _HEAD_Q)) self._run = 0 return False if not self._hist: # Nothing playing that we know of — anything audible is theirs. return True t = time.monotonic() if when is None else when bands = band_energies(pcm, sr) total = float(bands.sum()) if total <= 1e-9: return False # Find the recent output slice that best explains this block, and # subtract as much of it as fits. What is left over is whatever the # microphone heard that we did not play. best_res, best_level = 1.0, 0.0 for ts, ref, ref_level in self._hist: if ts > t or t - ts > _HISTORY_S: continue denom = float(np.dot(ref, ref)) if denom < 1e-12: continue alpha = max(0.0, float(np.dot(bands, ref)) / denom) residual = np.maximum(bands - alpha * ref, 0.0) ratio = float(residual.sum()) / total if ratio < best_res: best_res, best_level = ratio, ref_level self._last_sim = 1.0 - best_res # Nothing in the window explained it at all — that is not our sound. if best_level <= 0.0: return True # Learning happens in two stages, and the order matters both times. # # While warming up, learn from EVERY block. Judging first is a trap: # in a poor room the very first echo block already sits above any # sensible starting bar, so it would be called a voice, never be # learned from, and the room would stay mis-characterised forever. # # Once warm, learn only from blocks BELOW the bar. Feeding the # user's own blocks back in is the opposite trap: they lift the # threshold over themselves and the guard goes deaf to the very # thing it is watching for. warming = len(self._residuals) < _WARMUP thr = self.threshold speech = (not warming) and best_res >= thr if speech: self._run += 1 # A run this long is not someone interrupting — nobody talks # over an assistant for seconds on end. The room changed under # us (volume, headphones unplugged), so characterise it again. if self._run > _RELEARN_RUN: self._residuals.clear() self._run = 0 return False return True self._run = 0 self._residuals.append(best_res) if len(self._residuals) > _FLOOR_WINDOW: del self._residuals[:-_FLOOR_WINDOW] if len(self._residuals) >= _WARMUP: self._floor = float(np.percentile(self._residuals, _FLOOR_Q)) self._head = float(np.percentile(self._residuals, _HEAD_Q)) if warming: return False # Comfortably just us: also a safe moment to learn how loudly this # room returns our voice. if best_res <= self._floor * 1.15 and best_level > 0.05: obs = level / max(best_level, 1e-6) self._gain += (min(obs, 3.0) - self._gain) * 0.08 self._seen = min(self._seen + 1, 999) return False except Exception: return False # any doubt: do not interrupt