Fail fast when configured recording input cannot be resolved

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 "<none>"
+    items = [f"{d['index']}:{d['name']}" for d in devices[:limit]]
+    if len(devices) > limit:
+        items.append("...")
+    return ", ".join(items)

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()