diff --git a/src/recorder.py b/src/recorder.py index fc4501a..e8c547c 100644 --- a/src/recorder.py +++ b/src/recorder.py @@ -36,12 +36,17 @@ def resolve_input_device(spec: str | int | None) -> int | None: if spec is None: return None if isinstance(spec, int): - return spec + if _input_device_exists(spec): + return spec + return None text = str(spec).strip() if not text: return None if text.isdigit(): - return int(text) + device_idx = int(text) + if _input_device_exists(device_idx): + return device_idx + return None lowered = text.lower() for device in list_input_devices(): name = (device.get("name") or "").lower() @@ -54,6 +59,11 @@ def start_recording(input_spec: str | int | None) -> tuple[Any, RecordResult]: sd = _sounddevice() record = RecordResult() device = resolve_input_device(input_spec) + if _is_explicit_input_spec(input_spec) and device is None: + raise ValueError( + f"recording.input '{input_spec}' did not match any input device; " + f"available inputs: {_available_input_devices_summary()}" + ) def callback(indata, _frames, _time, _status): record.frames.append(indata.copy()) @@ -105,3 +115,28 @@ def _flatten_frames(frames: Iterable[np.ndarray]) -> np.ndarray: if data.ndim > 1: data = np.squeeze(data, axis=-1) return np.asarray(data, dtype=np.float32).reshape(-1) + + +def _input_device_exists(device_idx: int) -> bool: + for device in list_input_devices(): + if int(device.get("index", -1)) == int(device_idx): + return True + return False + + +def _is_explicit_input_spec(input_spec: str | int | None) -> bool: + if input_spec is None: + return False + if isinstance(input_spec, int): + return True + return bool(str(input_spec).strip()) + + +def _available_input_devices_summary(limit: int = 8) -> str: + devices = list_input_devices() + if not devices: + return "" + items = [f"{d['index']}:{d['name']}" for d in devices[:limit]] + if len(devices) > limit: + items.append("...") + return ", ".join(items) diff --git a/tests/test_recorder.py b/tests/test_recorder.py index e6d98fb..1016a00 100644 --- a/tests/test_recorder.py +++ b/tests/test_recorder.py @@ -3,12 +3,14 @@ import unittest from pathlib import Path import numpy as np +from unittest.mock import patch ROOT = Path(__file__).resolve().parents[1] SRC = ROOT / "src" if str(SRC) not in sys.path: sys.path.insert(0, str(SRC)) +import recorder from recorder import RecordResult, stop_recording @@ -30,6 +32,40 @@ class _Stream: raise self.close_exc +class _InputStream: + def __init__(self, **kwargs): + self.kwargs = kwargs + self.started = False + + def start(self) -> None: + self.started = True + + def stop(self) -> None: + return + + def close(self) -> None: + return + + +class _FakeSoundDevice: + def __init__(self): + self.default = type("Default", (), {"device": (0, None)})() + self.created_streams = [] + self._devices = [ + {"name": "Built-in Output", "max_input_channels": 0}, + {"name": "Laptop Mic", "max_input_channels": 1}, + {"name": "USB Interface", "max_input_channels": 2}, + ] + + def query_devices(self): + return self._devices + + def InputStream(self, **kwargs): + stream = _InputStream(**kwargs) + self.created_streams.append(stream) + return stream + + class RecorderTests(unittest.TestCase): def test_stop_recording_closes_stream_when_stop_raises(self): stream = _Stream(stop_exc=RuntimeError("stop boom")) @@ -75,6 +111,40 @@ class RecorderTests(unittest.TestCase): np.testing.assert_allclose(audio, np.array([0.2, 0.4], dtype=np.float32)) + def test_start_recording_accepts_explicit_device_name(self): + sd = _FakeSoundDevice() + with patch("recorder._sounddevice", return_value=sd): + stream, record = recorder.start_recording("Laptop") + + self.assertTrue(stream.started) + self.assertEqual(stream.kwargs["device"], 1) + self.assertEqual(record.channels, 1) + + def test_start_recording_accepts_explicit_device_index(self): + sd = _FakeSoundDevice() + with patch("recorder._sounddevice", return_value=sd): + stream, _record = recorder.start_recording(2) + + self.assertTrue(stream.started) + self.assertEqual(stream.kwargs["device"], 2) + + def test_start_recording_invalid_explicit_name_raises(self): + sd = _FakeSoundDevice() + with patch("recorder._sounddevice", return_value=sd): + with self.assertRaisesRegex( + ValueError, + "did not match any input device; available inputs: 1:Laptop Mic, 2:USB Interface", + ): + recorder.start_recording("NotADevice") + + def test_start_recording_empty_input_uses_default(self): + sd = _FakeSoundDevice() + with patch("recorder._sounddevice", return_value=sd): + stream, _record = recorder.start_recording("") + + self.assertTrue(stream.started) + self.assertIsNone(stream.kwargs["device"]) + if __name__ == "__main__": unittest.main()