146 lines
4.4 KiB
Python
146 lines
4.4 KiB
Python
from __future__ import annotations
|
|
|
|
import sys
|
|
import threading
|
|
import time
|
|
import types
|
|
|
|
import numpy as np
|
|
|
|
from src.main import LiveAudioSource
|
|
|
|
|
|
class FakeInputStream:
|
|
def __init__(self, **kwargs) -> None:
|
|
self.kwargs = kwargs
|
|
self.channels = kwargs["channels"]
|
|
self.samplerate = kwargs["samplerate"]
|
|
self.aborted = False
|
|
self.closed = False
|
|
self._wake = threading.Event()
|
|
|
|
def start(self) -> None:
|
|
pass
|
|
|
|
def read(self, frames: int):
|
|
self._wake.wait(0.005)
|
|
if self.aborted:
|
|
raise RuntimeError("aborted")
|
|
return np.full((frames, self.channels), 0.01, dtype=np.float32), False
|
|
|
|
def abort(self) -> None:
|
|
self.aborted = True
|
|
self._wake.set()
|
|
|
|
def close(self) -> None:
|
|
self.closed = True
|
|
|
|
|
|
def test_live_capture_uses_blocking_reads_without_callback(monkeypatch) -> None:
|
|
created: list[FakeInputStream] = []
|
|
|
|
def input_stream(**kwargs):
|
|
stream = FakeInputStream(**kwargs)
|
|
created.append(stream)
|
|
return stream
|
|
|
|
devices = [
|
|
{
|
|
"name": "Fake USB 8ch",
|
|
"max_input_channels": 8,
|
|
"hostapi": 0,
|
|
"default_samplerate": 44_100.0,
|
|
},
|
|
{
|
|
"name": "Fake USB 8ch",
|
|
"max_input_channels": 8,
|
|
"hostapi": 1,
|
|
"default_samplerate": 24_000.0,
|
|
},
|
|
]
|
|
hostapis = [{"name": "MME"}, {"name": "Windows WASAPI"}]
|
|
|
|
def query_devices(index=None):
|
|
return devices if index is None else devices[index]
|
|
|
|
fake_sounddevice = types.SimpleNamespace(
|
|
default=types.SimpleNamespace(device=(0, -1)),
|
|
query_devices=query_devices,
|
|
query_hostapis=lambda index: hostapis[index],
|
|
InputStream=input_stream,
|
|
)
|
|
monkeypatch.setitem(sys.modules, "sounddevice", fake_sounddevice)
|
|
|
|
source = LiveAudioSource(blocksize=128)
|
|
try:
|
|
deadline = time.monotonic() + 1.0
|
|
while source.frames_captured == 0 and time.monotonic() < deadline:
|
|
time.sleep(0.01)
|
|
|
|
assert source.frames_captured > 0
|
|
assert created[0].kwargs["device"] == 1
|
|
assert created[0].kwargs["samplerate"] == 24_000
|
|
assert created[0].kwargs["blocksize"] == 0
|
|
assert "callback" not in created[0].kwargs
|
|
assert source.doa is not None
|
|
assert source.sample_rate == 24_000
|
|
assert source.doa.config.sample_rate == 24_000
|
|
assert source.doa.buffered_samples > 0
|
|
assert max(source.read_levels(0.0)) > -41.0
|
|
assert source.console_status() == (
|
|
"[audio] Input stream opened\n"
|
|
" Device: #1 Fake USB 8ch\n"
|
|
" Driver interface (Host API): Windows WASAPI\n"
|
|
" Input channels: 8 used / 8 available\n"
|
|
" Sample rate: 24000 Hz (device default: 24000 Hz)"
|
|
)
|
|
finally:
|
|
source.close()
|
|
|
|
assert created[0].aborted
|
|
assert created[0].closed
|
|
|
|
|
|
def test_console_status_explains_missing_live_input(monkeypatch) -> None:
|
|
fake_sounddevice = types.SimpleNamespace(
|
|
default=types.SimpleNamespace(device=(-1, -1)),
|
|
query_devices=lambda: [],
|
|
)
|
|
monkeypatch.setitem(sys.modules, "sounddevice", fake_sounddevice)
|
|
|
|
source = LiveAudioSource()
|
|
|
|
assert not source.available
|
|
assert source.console_status() == (
|
|
"[audio] Live input unavailable; using simulated preview\n"
|
|
" Reason: No audio input device with at least 8 channels was found"
|
|
)
|
|
|
|
|
|
def test_explicit_sample_rate_overrides_detected_rate(monkeypatch) -> None:
|
|
created: list[FakeInputStream] = []
|
|
device = {
|
|
"name": "Fake USB 8ch",
|
|
"max_input_channels": 8,
|
|
"hostapi": 0,
|
|
"default_samplerate": 24_000.0,
|
|
}
|
|
|
|
fake_sounddevice = types.SimpleNamespace(
|
|
default=types.SimpleNamespace(device=(0, -1)),
|
|
query_devices=lambda index=None: [device] if index is None else device,
|
|
query_hostapis=lambda _index: {"name": "Windows WASAPI"},
|
|
InputStream=lambda **kwargs: created.append(FakeInputStream(**kwargs))
|
|
or created[-1],
|
|
)
|
|
monkeypatch.setitem(sys.modules, "sounddevice", fake_sounddevice)
|
|
|
|
source = LiveAudioSource(sample_rate=32_000, blocksize=128)
|
|
try:
|
|
assert created[0].kwargs["samplerate"] == 32_000
|
|
assert source.sample_rate == 32_000
|
|
assert source.doa is not None
|
|
assert source.doa.config.sample_rate == 32_000
|
|
finally:
|
|
source.close()
|