Spaces:

abc1181
/

Jaarvis

Runtime error

Jaarvis voice Space (assembled from monorepo)

370c3e4 about 1 month ago

13.8 kB

"""
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

	"""
	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