kyleandClaude Sonnet 5.5
ac4e09bf35
audio_state taps the microphone and the speaker monitor with parec and publishes levels, a speaking/listening/quiet activity, and trigger-locked waveform windows; audio itself stays out of ROS. The activity detector calibrates its noise floor at start-up and absorbs steady noise, so a room at -43 dBFS does not read as "listening" forever, and it ignores the mic while the robot speaks. The face arbitrates the state it shows: sleeping (agent side only, so room sound never wakes a face that means "no agent connected") > speaking > listening > thinking > idle, and draws the mouth from the real samples while fresh scope data exists, auto-scaled so a quiet voice is still visible. Checked: a tone played to the sink shows as `speaking` with the mic ignored. The microphone-to-mouth path with a live voice is not yet confirmed. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01AFN8nRRCiDiMN7ChBpHJhs
88 lines
3.4 KiB
Python
88 lines
3.4 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.2, t)
|
|
assert a.state == 'listening' # still holding
|
|
run_activity(a, -62, -100, 1.0, t + 0.2)
|
|
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, 1.0, t)
|
|
assert a.state == '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'
|