110 lines
3.7 KiB
Python
110 lines
3.7 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import struct
|
|
|
|
from app.utils import audio_backlog
|
|
|
|
FRAME_SAMPLES = 320 # 20 ms @ 16 kHz mono
|
|
BYTES_PER_FRAME = FRAME_SAMPLES * 2
|
|
|
|
|
|
def _silent() -> bytes:
|
|
return b"\x00" * BYTES_PER_FRAME
|
|
|
|
|
|
def _voiced(amp: int = 8000) -> bytes:
|
|
return struct.pack("<%dh" % FRAME_SAMPLES, *([amp] * FRAME_SAMPLES))
|
|
|
|
|
|
def _queue(frames: list[bytes]) -> asyncio.Queue:
|
|
q: asyncio.Queue = asyncio.Queue()
|
|
for frame in frames:
|
|
q.put_nowait(frame)
|
|
return q
|
|
|
|
|
|
def _drain(q: asyncio.Queue) -> list[bytes]:
|
|
out = []
|
|
while True:
|
|
try:
|
|
out.append(q.get_nowait())
|
|
except asyncio.QueueEmpty:
|
|
break
|
|
return out
|
|
|
|
|
|
def test_rms_threshold_from_dbfs() -> None:
|
|
assert audio_backlog.rms_threshold_from_dbfs(0.0) == 32768
|
|
assert audio_backlog.rms_threshold_from_dbfs(-50.0) == 103
|
|
assert audio_backlog.rms_threshold_from_dbfs(-200.0) == 0
|
|
|
|
|
|
def test_frames_from_ms_rounds_up_with_minimum() -> None:
|
|
assert audio_backlog.frames_from_ms(100, 20) == 5
|
|
assert audio_backlog.frames_from_ms(90, 20) == 5 # arredonda pra cima
|
|
assert audio_backlog.frames_from_ms(0, 20) == 1 # minimo
|
|
assert audio_backlog.frames_from_ms(40, 20) == 2
|
|
|
|
|
|
def test_is_silent_frame() -> None:
|
|
thr = audio_backlog.rms_threshold_from_dbfs(-50.0)
|
|
assert audio_backlog.is_silent_frame(_silent(), thr) is True
|
|
assert audio_backlog.is_silent_frame(_voiced(), thr) is False
|
|
|
|
|
|
def test_shed_below_keep_is_noop() -> None:
|
|
q = _queue([_voiced()] * 2)
|
|
result = audio_backlog.shed_queue_backlog(q, keep_frames=5, rms_threshold=103, blind=False)
|
|
assert result.dropped == 0
|
|
assert result.mode == "none"
|
|
assert q.qsize() == 2
|
|
|
|
|
|
def test_shed_blind_drops_oldest_and_keeps_recent_in_order() -> None:
|
|
frames = [_voiced(1000 + i) for i in range(8)]
|
|
q = _queue(frames)
|
|
result = audio_backlog.shed_queue_backlog(q, keep_frames=2, rms_threshold=103, blind=True)
|
|
assert result.dropped == 6
|
|
assert result.dropped_voiced == 6
|
|
assert result.mode == "blind"
|
|
assert _drain(q) == frames[-2:]
|
|
|
|
|
|
def test_shed_energy_drops_only_silence_preserving_speech_order() -> None:
|
|
voiced = [_voiced(2000 + i * 100) for i in range(4)]
|
|
# intercala voz e silencio: [V0, S, V1, S, V2, S, V3, S]
|
|
frames = [voiced[0], _silent(), voiced[1], _silent(), voiced[2], _silent(), voiced[3], _silent()]
|
|
q = _queue(frames)
|
|
result = audio_backlog.shed_queue_backlog(q, keep_frames=2, rms_threshold=103, blind=False)
|
|
# need = 8 - 2 = 6, mas so ha 4 frames silenciosos -> descarta os 4 silencios,
|
|
# preserva os 4 de voz na ordem (nao corta fala).
|
|
assert result.mode == "energy"
|
|
assert result.dropped == 4
|
|
assert result.dropped_silent == 4
|
|
assert result.dropped_voiced == 0
|
|
assert _drain(q) == voiced
|
|
|
|
|
|
def test_shed_energy_stops_at_need_even_with_more_silence() -> None:
|
|
voiced = [_voiced(3000 + i) for i in range(2)]
|
|
# 2 de voz + 6 de silencio, keep=4 -> need=4 -> descarta 4 silencios, sobra 4
|
|
frames = [voiced[0], _silent(), _silent(), _silent(), voiced[1], _silent(), _silent(), _silent()]
|
|
q = _queue(frames)
|
|
result = audio_backlog.shed_queue_backlog(q, keep_frames=4, rms_threshold=103, blind=False)
|
|
assert result.dropped == 4
|
|
assert result.dropped_silent == 4
|
|
remaining = _drain(q)
|
|
assert q.qsize() == 0
|
|
# sobra os 2 de voz (na ordem) + 2 silencios mais recentes
|
|
assert remaining[0] == voiced[0]
|
|
assert voiced[1] in remaining
|
|
assert len(remaining) == 4
|
|
|
|
|
|
def test_shed_disabled_via_empty_queue() -> None:
|
|
q = _queue([])
|
|
result = audio_backlog.shed_queue_backlog(q, keep_frames=2, rms_threshold=103, blind=True)
|
|
assert result.dropped == 0
|
|
assert result.mode == "none"
|