kyleandClaude Sonnet 5.5
5d1ad1eff7
The face now treats /hermes/state as the agent's heartbeat: when nothing has been heard for agent_timeout_s (10 s) it logs "agent gone" and sleeps, so "sleeping" really means "no agent connected". It also logs every state change, which is how a missed wake message was found: a one-off publish before discovery matches is lost under volatile QoS. The live log showed the face flickering between listening and idle about twice a second, because the detector let go after 0.4 s. Hold is now 1.2 s so pauses between words do not drop it, and the margin 7 dB over the adaptive floor (quiet room noise peaks sit about 3 dB over it), with tests for both. Confirmed by the owner on the real display: gaze directions, and a mouth that follows their live voice proportionally. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01AFN8nRRCiDiMN7ChBpHJhs
110 lines
4.3 KiB
Python
110 lines
4.3 KiB
Python
import numpy as np
|
|
import pytest
|
|
|
|
from hermes_audio.levels import FLOOR_DB, Activity, rms_dbfs, trigger_window
|
|
|
|
|
|
def sine(f, n, rate=16000, amp=0.5, phase=0.0):
|
|
return (amp * np.sin(2 * np.pi * f * np.arange(n) / rate + phase)).astype(np.float32)
|
|
|
|
|
|
def test_rms_dbfs_known_values():
|
|
assert rms_dbfs(np.ones(100, np.float32)) == pytest.approx(0.0, abs=0.01)
|
|
assert rms_dbfs(sine(440, 16000, amp=1.0)) == pytest.approx(-3.01, abs=0.05)
|
|
assert rms_dbfs(np.zeros(100, np.float32)) == FLOOR_DB or rms_dbfs(np.zeros(100, np.float32)) < -100 + 1
|
|
assert rms_dbfs(np.array([], np.float32)) == FLOOR_DB
|
|
|
|
|
|
def test_trigger_window_starts_on_a_rising_zero_crossing_and_is_phase_locked():
|
|
a = trigger_window(sine(200, 2000, phase=0.3), 256, 512)
|
|
b = trigger_window(sine(200, 2000, phase=2.1), 256, 512) # a different arrival phase
|
|
assert len(a) == 256 and a[0] >= 0 and abs(a[0]) < 0.2 # begins near a rising zero crossing
|
|
assert np.allclose(a, b, atol=0.12) # so both draw the same picture
|
|
|
|
|
|
def test_trigger_window_handles_silence_and_short_buffers():
|
|
assert trigger_window(np.zeros(2000, np.float32)).shape == (256,)
|
|
short = trigger_window(sine(200, 100), 256, 512)
|
|
assert short.shape == (256,)
|
|
|
|
|
|
def run_activity(act, mic_db, spk_db, seconds, t0=0.0, dt=0.05):
|
|
t = t0
|
|
for _ in range(int(seconds / dt)):
|
|
t += dt
|
|
act.update(mic_db, spk_db, t, dt)
|
|
return t
|
|
|
|
|
|
def test_listens_to_a_voice_and_lets_go_after_the_hold():
|
|
a = Activity()
|
|
t = run_activity(a, -62, -100, 3) # a quiet room sets the floor
|
|
assert a.state == 'quiet'
|
|
t = run_activity(a, -35, -100, 0.5, t)
|
|
assert a.state == 'listening'
|
|
run_activity(a, -62, -100, 0.8, t)
|
|
assert a.state == 'listening' # still holding, through a pause between words
|
|
run_activity(a, -62, -100, 1.0, t + 0.8)
|
|
assert a.state == 'quiet'
|
|
|
|
|
|
def test_a_click_is_not_speech():
|
|
a = Activity()
|
|
t = run_activity(a, -62, -100, 2)
|
|
t = run_activity(a, -30, -100, 0.05, t) # one 50 ms burst, shorter than on_s
|
|
run_activity(a, -62, -100, 0.1, t)
|
|
assert a.state == 'quiet'
|
|
|
|
|
|
def test_speaking_wins_and_ignores_the_mic():
|
|
a = Activity()
|
|
t = run_activity(a, -62, -100, 2)
|
|
t = run_activity(a, -30, -40, 0.5, t) # loud mic AND loud speaker monitor
|
|
assert a.state == 'speaking'
|
|
run_activity(a, -30, -100, 0.1, t)
|
|
assert a.state == 'speaking' # held
|
|
t2 = run_activity(a, -30, -100, 1.0, t + 0.1)
|
|
assert a.state == 'listening' # speaker stopped; the mic is real speech now
|
|
assert t2 > t
|
|
|
|
|
|
def test_a_room_that_is_already_noisy_calibrates_instead_of_listening_forever():
|
|
a = Activity()
|
|
run_activity(a, -43, -100, 5) # a steady -43 dBFS noise floor from the start
|
|
assert a.state == 'quiet' and a.mic_threshold_db > -40
|
|
t = run_activity(a, -25, -100, 0.5, 5.0) # a voice 18 dB above it
|
|
assert a.state == 'listening'
|
|
run_activity(a, -43, -100, 2.0, t)
|
|
assert a.state == 'quiet'
|
|
|
|
|
|
def test_pauses_between_words_do_not_flicker():
|
|
a = Activity()
|
|
t = run_activity(a, -43, -100, 5)
|
|
states = []
|
|
for _ in range(6): # six words, 0.3 s each with 0.3 s gaps
|
|
for level, secs in ((-25, 0.3), (-43, 0.3)):
|
|
for _ in range(int(secs / 0.05)):
|
|
t += 0.05
|
|
states.append(a.update(level, -100, t, 0.05))
|
|
assert states[8:].count('quiet') == 0 # once it starts listening it stays there
|
|
|
|
|
|
def test_quiet_room_noise_peaks_stay_below_the_threshold():
|
|
a = Activity()
|
|
t = run_activity(a, -43, -100, 5)
|
|
flicker = []
|
|
for i in range(200):
|
|
t += 0.05
|
|
flicker.append(a.update(-43 + (3 if i % 7 == 0 else 0), -100, t, 0.05)) # peaks 3 dB over the floor
|
|
assert set(flicker) == {'quiet'}
|
|
|
|
|
|
def test_a_new_steady_noise_is_absorbed_after_a_while():
|
|
a = Activity()
|
|
t = run_activity(a, -60, -100, 5) # quiet room
|
|
t = run_activity(a, -35, -100, 1.0, t)
|
|
assert a.state == 'listening' # something started
|
|
run_activity(a, -35, -100, 20, t) # ...and never stops: a hum, not speech
|
|
assert a.state == 'quiet'
|