jonathan/ISR
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: collapse the single-implementation backend abstraction, fix device resolution

Browse Source 
isr.py carried a plugin architecture — AudioBackend ABC, a _backend_classes
registry, _discover_backends, available_backends/get_backend, a
preferred_backend config knob — for exactly one implementation. It was
also the source of three real bugs:

* find_device() returned None before reaching the ALSA PCM-name fallback
  whenever `arecord -l` enumerated nothing, so the documented
  `device = shared_mic` (dsnoop from asound.conf) case failed outright on
  any host with no hardware card. The fallback is the whole point: those
  PCMs never appear in `arecord -l`.

* `device = monitor` could never resolve. ALSA has no generic monitor
  source and list_devices hardcoded is_monitor=False, so the keyword —
  advertised in config.example.ini as the "EASY" way to capture system
  audio — only ever produced "device not found". Removed; it is a
  PulseAudio idea that outlived the PulseAudio backend.

* list_all_devices() deduplicated by lowercased name "in case a device
  appears in multiple backends". With one backend what it actually did
  was delete the second of two identical USB cards. Dedup dropped.

find_device() now always returns a device, so _resolve_device's
unreachable "Available devices:" error is gone; an unknown spec is
passed to ALSA verbatim and logged. AudioDevice loses backend/
is_monitor plus channels/sample_rate — the latter were hardcoded 2/44100
and only ever printed by --list-devices, i.e. numbers with no bearing on
what gets recorded.

--list-devices now says outright that it lists hardware cards only, which
is where the shared_mic confusion started.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
admin
2026-07-29 20:28:30 +02:00
co-authored by Claude Opus 5
parent 4aa7f7253e
commit 1541110ffb
4 changed files with 155 additions and 281 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+1 -1
View File
@@ -107,7 +107,7 @@ format = wav # wav | flac
| `hw:X,Y` | Exact ALSA hardware ID | | `hw:X,Y` | Exact ALSA hardware ID |
| `<pcm name>` | Any ALSA PCM defined in `asound.conf` (e.g. the `shared_mic` dsnoop device), even if it doesn't appear in `arecord -l` | | `<pcm name>` | Any ALSA PCM defined in `asound.conf` (e.g. the `shared_mic` dsnoop device), even if it doesn't appear in `arecord -l` |
Run `python isr.py --list-devices` (or `arecord -l`) to see available devices and their IDs. Run `python isr.py --list-devices` (or `arecord -l`) to see available devices and their IDs. It lists **hardware cards only** — a PCM name from `asound.conf` never appears there but can still be used directly, including on hosts where no hardware card is enumerable at all. An unrecognised value is passed to ALSA verbatim, so a typo shows up as an `arecord` error in the log rather than at startup.
FLAC output requires `pip install soundfile numpy`. FLAC output requires `pip install soundfile numpy`.
config.example.ini
+3 -13
View File
@@ -67,16 +67,14 @@ format = auto
# #
# # Device selection (use one of these methods): # # Device selection (use one of these methods):
# # device = default - Use system default input device # # device = default - Use system default input device
# # device = monitor - Auto-select first monitor/loopback source (for system audio)
# # device = <name> - Match device by partial name (e.g., "Stereo Mix") # # device = <name> - Match device by partial name (e.g., "Stereo Mix")
# # device = <id> - Use exact device ID from --list-devices # # device = <id> - Use exact device ID from --list-devices
# # device = <pcm> - Any ALSA PCM name from asound.conf (e.g. shared_mic);
# # these never show up in --list-devices but work anyway
# # # #
# # To list available devices, run: python isr.py --list-devices # # To list available devices, run: python isr.py --list-devices
# device = default # device = default
# #
# # Audio backend (only 'alsa' is supported):
# # backend = alsa
#
# # Sample rate in Hz (common: 44100, 48000, 96000) # # Sample rate in Hz (common: 44100, 48000, 96000)
# sample_rate = 44100 # sample_rate = 44100
# #
@@ -122,17 +120,10 @@ format = auto
# #
# Run: python isr.py --list-devices (or: arecord -l) to list devices. # Run: python isr.py --list-devices (or: arecord -l) to list devices.
# #
# EASY: Use the "monitor" keyword to capture system/loopback audio: # Specify a device by ALSA hardware ID from --list-devices:
# [system_audio]
# type = soundcard
# device = monitor
# format = wav
#
# MANUAL: Specify a device by ALSA hardware ID from --list-devices:
# [usb_mic] # [usb_mic]
# type = soundcard # type = soundcard
# device = hw:1,0 # device = hw:1,0
# backend = alsa
# sample_rate = 44100 # sample_rate = 44100
# channels = 2 # channels = 2
# format = flac # format = flac
@@ -148,7 +139,6 @@ format = auto
# [usb_mic_shared] # [usb_mic_shared]
# type = soundcard # type = soundcard
# device = shared_mic # device = shared_mic
# backend = alsa
# sample_rate = 48000 # sample_rate = 48000
# channels = 2 # channels = 2
# format = flac # format = flac
isr.py
+117 -221
View File
@@ -50,113 +50,24 @@ FILENAME_FORMAT = '%Y%m%d_%H%M%S'
# ============================================================================= # =============================================================================
# Audio Device & Backend System # Audio Device Discovery (ALSA)
# ============================================================================= # =============================================================================
@dataclass @dataclass
class AudioDevice: class AudioDevice:
"""Represents an audio input device.""" """An ALSA capture device.
id: str # Backend-specific identifier
name: str # Human-readable name Only the identity matters: capture format comes from config (sample_rate,
channels: int # Max input channels channels), not from the device, so this deliberately carries no
sample_rate: int # Default sample rate capability fields — reporting a card's channel count here would just be a
backend: str # Backend name ('alsa') number that has no bearing on what actually gets recorded.
is_default: bool = False # Is system default """
is_monitor: bool = False # Is a monitor/loopback source id: str # ALSA device spec: 'hw:1,0', or a PCM name
name: str # Human-readable name from `arecord -l`
is_default: bool = False
def __str__(self): def __str__(self):
flags = [] return f"{self.name} ({self.id})" + (" [DEFAULT]" if self.is_default else "")
if self.is_default:
flags.append("DEFAULT")
if self.is_monitor:
flags.append("MONITOR")
flag_str = f" [{', '.join(flags)}]" if flags else ""
return f"{self.name}{flag_str} ({self.backend})"
class AudioBackend(ABC):
"""Abstract base for audio capture backends."""
name: str = "base"
@classmethod
@abstractmethod
def is_available(cls) -> bool:
"""Check if this backend can be used on the current system."""
pass
@classmethod
@abstractmethod
def list_devices(cls) -> List[AudioDevice]:
"""List all available input devices."""
pass
@abstractmethod
def open_stream(self, device: AudioDevice, sample_rate: int, channels: int,
callback: Callable[[bytes], None]) -> Any:
"""Open an audio capture stream. Returns a context manager."""
pass
class ALSABackend(AudioBackend):
"""ALSA backend using arecord (raw PCM output, no sound server required)."""
name = "alsa"
@classmethod
def is_available(cls) -> bool:
return shutil.which('arecord') is not None
@classmethod
def list_devices(cls) -> List[AudioDevice]:
devices = []
try:
result = subprocess.run(
['arecord', '-l'],
capture_output=True,
text=True,
timeout=5
)
if result.returncode != 0:
return devices
for line in result.stdout.split('\n'):
if not line.startswith('card '):
continue
try:
card_part, rest = line.split(':', 1)
card_num = card_part.replace('card', '').strip()
if ', device ' in rest:
dev_part = rest.split(', device ')[1]
dev_num = dev_part.split(':')[0].strip()
else:
dev_num = '0'
long_name = rest.split('[')[1].split(']')[0] if '[' in rest else rest.strip()
hw_id = f"hw:{card_num},{dev_num}"
devices.append(AudioDevice(
id=hw_id,
name=long_name,
channels=2,
sample_rate=44100,
backend=cls.name,
is_default=(card_num == '0' and dev_num == '0'),
is_monitor=False,
))
except (IndexError, ValueError):
continue
except Exception:
pass
return devices
def __init__(self, logger: logging.Logger):
self.logger = logger
def open_stream(self, device: AudioDevice, sample_rate: int, channels: int,
callback: Callable[[bytes], None]):
return ALSAStream(device, sample_rate, channels, callback, self.logger)
class ALSAStream: class ALSAStream:
@@ -258,95 +169,94 @@ class ALSAStream:
class AudioSystem: class AudioSystem:
"""Manages audio backends and device discovery.""" """Discovers ALSA capture devices and opens capture streams.
# All available backend classes ALSA (via `arecord`) is the only backend. This was once a plugin system
_backend_classes: List[type] = [ALSABackend] with a registry and an abstract base class; there has only ever been one
implementation, so it is a plain class now.
"""
def __init__(self, logger: logging.Logger): def __init__(self, logger: logging.Logger):
self.logger = logger self.logger = logger
self._backends: Dict[str, AudioBackend] = {} self.available = shutil.which('arecord') is not None
self._discover_backends()
def _discover_backends(self): def list_devices(self) -> List[AudioDevice]:
"""Find and initialize available backends.""" """Hardware capture devices, parsed out of `arecord -l`.
for cls in self._backend_classes:
if cls.is_available():
self._backends[cls.name] = cls(self.logger)
self.logger.debug(f"Audio backend available: {cls.name}")
@property Virtual PCMs defined in asound.conf (dsnoop, plug, ...) are NOT listed
def available_backends(self) -> List[str]: by `arecord -l` and so never appear here — see find_device().
return list(self._backends.keys()) """
devices = []
try:
result = subprocess.run(
['arecord', '-l'],
capture_output=True,
text=True,
timeout=5
)
if result.returncode != 0:
return devices
def get_backend(self, name: str) -> Optional[AudioBackend]: for line in result.stdout.split('\n'):
return self._backends.get(name) if not line.startswith('card '):
continue
try:
card_part, rest = line.split(':', 1)
card_num = card_part.replace('card', '').strip()
def list_all_devices(self) -> List[AudioDevice]: if ', device ' in rest:
"""List devices from all available backends.""" dev_part = rest.split(', device ')[1]
all_devices = [] dev_num = dev_part.split(':')[0].strip()
seen_names = set() else:
dev_num = '0'
for cls in self._backend_classes: long_name = rest.split('[')[1].split(']')[0] if '[' in rest else rest.strip()
if cls.name in self._backends:
for dev in cls.list_devices():
# Deduplicate by name (same device may appear in multiple backends)
key = dev.name.lower()
if key not in seen_names:
all_devices.append(dev)
seen_names.add(key)
return all_devices devices.append(AudioDevice(
id=f"hw:{card_num},{dev_num}",
name=long_name,
is_default=(card_num == '0' and dev_num == '0'),
))
except (IndexError, ValueError):
continue
except Exception:
pass
return devices
def find_device(self, spec: str, preferred_backend: str = None) -> Optional[AudioDevice]: def find_device(self, spec: str) -> AudioDevice:
"""Find a device by ID, name, or pattern.""" """Resolve a config `device =` value to an ALSA device.
devices = self.list_all_devices()
if not devices:
return None
# "default" - get the default device from preferred backend
if spec.lower() == 'default':
for dev in devices:
if dev.is_default:
if preferred_backend is None or dev.backend == preferred_backend:
return dev
return devices[0] if devices else None
# "monitor" - get first monitor source
if spec.lower() == 'monitor':
for dev in devices:
if dev.is_monitor:
if preferred_backend is None or dev.backend == preferred_backend:
return dev
return None
Never returns None. An unrecognised spec is passed to ALSA verbatim,
because a valid PCM name from asound.conf (`shared_mic`, `plughw:0`)
does not appear in `arecord -l` — and that must keep working even on a
host where no hardware card is enumerable at all, which is exactly the
case an earlier "no devices, give up" guard broke.
"""
devices = self.list_devices()
spec_lower = spec.lower() spec_lower = spec.lower()
# Try exact ID match first if spec_lower == 'default':
for dev in devices: for dev in devices:
if dev.id == spec: if dev.is_default:
return dev return dev
# 'default' is itself a valid ALSA PCM when no card is enumerable.
return devices[0] if devices else AudioDevice(id='default', name='default')
# Try exact name match exact_id = [d for d in devices if d.id == spec]
for dev in devices: exact_name = [d for d in devices if d.name.lower() == spec_lower]
if dev.name.lower() == spec_lower: partial = [d for d in devices if spec_lower in d.name.lower()]
return dev for candidates in (exact_id, exact_name, partial):
if candidates:
return candidates[0]
# Try partial name match self.logger.info(
for dev in devices: f"Device '{spec}' is not in `arecord -l` — using it as an ALSA PCM name")
if spec_lower in dev.name.lower(): return AudioDevice(id=spec, name=spec)
return dev
# No hardware device matched. Treat spec as a direct ALSA PCM name so def open_stream(self, device: AudioDevice, sample_rate: int, channels: int,
# virtual devices defined in asound.conf (dsnoop, plug, etc.) work without callback: Callable[[bytes], None]) -> 'ALSAStream':
# appearing in `arecord -l`. """Open a capture stream. Returns a context manager."""
if 'alsa' in self._backends: return ALSAStream(device, sample_rate, channels, callback, self.logger)
return AudioDevice(
id=spec, name=spec, channels=2, sample_rate=44100, backend='alsa'
)
return None
class BaseRecorder(ABC): class BaseRecorder(ABC):
@@ -675,7 +585,7 @@ class _AudioFileWriter:
class SoundcardRecorder(BaseRecorder): class SoundcardRecorder(BaseRecorder):
"""Records from soundcard/audio devices using multiple backends.""" """Records from a soundcard/audio device via ALSA."""
def __init__(self, name: str, config: Dict[str, Any], logger: logging.Logger, def __init__(self, name: str, config: Dict[str, Any], logger: logging.Logger,
audio_system: 'AudioSystem' = None, audio_system: 'AudioSystem' = None,
@@ -685,9 +595,9 @@ class SoundcardRecorder(BaseRecorder):
# Initialize audio system if not provided # Initialize audio system if not provided
self.audio_system = audio_system or AudioSystem(logger) self.audio_system = audio_system or AudioSystem(logger)
if not self.audio_system.available_backends: if not self.audio_system.available:
raise RuntimeError( raise RuntimeError(
"No audio backends available.\n" "arecord not found — soundcard capture is unavailable.\n"
"Install ALSA utilities: sudo apt install alsa-utils\n" "Install ALSA utilities: sudo apt install alsa-utils\n"
"(In Docker, also ensure /dev/snd is mapped into the container)" "(In Docker, also ensure /dev/snd is mapped into the container)"
) )
@@ -696,7 +606,6 @@ class SoundcardRecorder(BaseRecorder):
self.device_spec = config.get('device', 'default') self.device_spec = config.get('device', 'default')
self.sample_rate = config.get('sample_rate', 44100) self.sample_rate = config.get('sample_rate', 44100)
self.channels = config.get('channels', 2) self.channels = config.get('channels', 2)
self.preferred_backend = config.get('backend', None)
# Validate format # Validate format
if self.file_format not in ('wav', 'flac') and self.file_format != 'auto': if self.file_format not in ('wav', 'flac') and self.file_format != 'auto':
@@ -713,23 +622,19 @@ class SoundcardRecorder(BaseRecorder):
self.buffer_lock = threading.Lock() self.buffer_lock = threading.Lock()
def _resolve_device(self) -> AudioDevice: def _resolve_device(self) -> AudioDevice:
"""Resolve device specification to AudioDevice.""" """Resolve the configured device spec to an AudioDevice.
device = self.audio_system.find_device(self.device_spec, self.preferred_backend)
if device is None:
# List available devices in error
available = self.audio_system.list_all_devices()
device_list = "\n ".join(str(d) for d in available) if available else "None found"
raise ValueError(
f"[{self.name}] Device '{self.device_spec}' not found.\n"
f"Available devices:\n {device_list}"
)
find_device() always succeeds — an unknown spec becomes a literal ALSA
PCM name — so a typo surfaces as an arecord failure at capture time
rather than here. That is the price of supporting asound.conf PCMs,
which are invisible to `arecord -l`.
"""
device = self.audio_system.find_device(self.device_spec)
self.logger.info(f"[{self.name}] Using device: {device}") self.logger.info(f"[{self.name}] Using device: {device}")
return device return device
def _audio_callback(self, data: bytes): def _audio_callback(self, data: bytes):
"""Callback for audio data from backend.""" """Callback for audio data from the capture thread."""
with self.buffer_lock: with self.buffer_lock:
self.audio_buffer.append(data) self.audio_buffer.append(data)
@@ -767,13 +672,8 @@ class SoundcardRecorder(BaseRecorder):
retry_count = 0 retry_count = 0
next_split = self.get_next_split_time() next_split = self.get_next_split_time()
backend = self.audio_system.get_backend(self.device.backend)
if not backend:
self.logger.error(f"[{self.name}] Backend '{self.device.backend}' not available")
return
self.logger.info(f"[{self.name}] Starting recorder - Split every {self.split_duration} min") self.logger.info(f"[{self.name}] Starting recorder - Split every {self.split_duration} min")
self.logger.info(f"[{self.name}] Backend: {self.device.backend}, Device: {self.device.name}") self.logger.info(f"[{self.name}] Device: {self.device}")
self.logger.info(f"[{self.name}] Sample rate: {self.sample_rate}, Channels: {self.channels}") self.logger.info(f"[{self.name}] Sample rate: {self.sample_rate}, Channels: {self.channels}")
self.logger.info(f"[{self.name}] Next split: {next_split.strftime('%H:%M:%S')}") self.logger.info(f"[{self.name}] Next split: {next_split.strftime('%H:%M:%S')}")
@@ -782,7 +682,7 @@ class SoundcardRecorder(BaseRecorder):
if self.current_file is None: if self.current_file is None:
self._open_output_file() self._open_output_file()
with backend.open_stream( with self.audio_system.open_stream(
self.device, self.sample_rate, self.channels, self._audio_callback self.device, self.sample_rate, self.channels, self._audio_callback
) as stream: ) as stream:
self.logger.info(f"[{self.name}] Recording started") self.logger.info(f"[{self.name}] Recording started")
@@ -862,8 +762,8 @@ class RecorderManager:
# Initialize shared audio system for soundcard recorders # Initialize shared audio system for soundcard recorders
self.audio_system = AudioSystem(self.logger) self.audio_system = AudioSystem(self.logger)
if self.audio_system.available_backends: if self.audio_system.available:
self.logger.info(f"Audio backends: {', '.join(self.audio_system.available_backends)}") self.logger.info("Audio capture available: ALSA (arecord)")
# Parse source sections # Parse source sections
for section in config.sections(): for section in config.sections():
@@ -1005,7 +905,7 @@ class RecorderManager:
def list_audio_devices(): def list_audio_devices():
"""List available audio input devices from all backends.""" """Print the ALSA capture devices usable as `device =` in config.ini."""
# Create a minimal logger for device listing # Create a minimal logger for device listing
logger = logging.getLogger('isr-devices') logger = logging.getLogger('isr-devices')
logger.setLevel(logging.WARNING) # Suppress debug output logger.setLevel(logging.WARNING) # Suppress debug output
@@ -1014,40 +914,36 @@ def list_audio_devices():
print(" ISR Audio Device Discovery") print(" ISR Audio Device Discovery")
print("=" * 70) print("=" * 70)
if not ALSABackend.is_available(): audio_system = AudioSystem(logger)
print("\n No audio backends available!") if not audio_system.available:
print("\n arecord not found!")
print("\n Install ALSA utilities:") print("\n Install ALSA utilities:")
print(" sudo apt install alsa-utils") print(" sudo apt install alsa-utils")
print() print()
return return
print("\n Available Backends:") devices = audio_system.list_devices()
print(" - ALSA (arecord)") if devices:
print("\n Hardware Input Devices:")
# Initialize audio system and list devices print(" " + "-" * 68)
audio_system = AudioSystem(logger) for dev in devices:
devices = audio_system.list_all_devices() flag_str = " [DEFAULT]" if dev.is_default else ""
print(f"\n {dev.name}{flag_str}")
if not devices: print(f" ID: {dev.id}")
print("\n No input devices found!") else:
print() print("\n No hardware capture devices found.")
return
print("\n Input Devices:")
print(" " + "-" * 68)
for dev in devices:
flag_str = " [DEFAULT]" if dev.is_default else ""
print(f"\n {dev.name}{flag_str}")
print(f" ID: {dev.id} | Backend: {dev.backend}")
print(f" Channels: {dev.channels} | Sample Rate: {dev.sample_rate} Hz")
# This listing only ever shows hardware cards, so say so — a working
# asound.conf PCM will never appear here and users have gone looking.
print("\n" + "=" * 70) print("\n" + "=" * 70)
print(" Configuration Examples:") print(" Configuration Examples:")
print("-" * 70) print("-" * 70)
print(" device = default # Use system default input") print(" device = default # Use system default input")
print(" device = <name> # Match by partial name") print(" device = <name> # Match by partial name")
print(" device = <id> # Use exact ALSA ID, e.g. hw:0,0") print(" device = <id> # Use exact ALSA ID, e.g. hw:0,0")
print(" device = <pcm> # Any ALSA PCM name from asound.conf, e.g. shared_mic") print(" device = <pcm> # Any ALSA PCM name from asound.conf, e.g.")
print(" # shared_mic — these are never listed above,")
print(" # but work regardless")
print("=" * 70 + "\n") print("=" * 70 + "\n")
tests/test_isr.py
+34 -46
View File
@@ -51,15 +51,12 @@ class TestAudioSystemFindDevice:
def _make_system(self, devices: List[isr.AudioDevice]) -> isr.AudioSystem: def _make_system(self, devices: List[isr.AudioDevice]) -> isr.AudioSystem:
system = isr.AudioSystem.__new__(isr.AudioSystem) system = isr.AudioSystem.__new__(isr.AudioSystem)
system.logger = make_logger() system.logger = make_logger()
system._backends = {} system.available = True
system.list_all_devices = lambda: devices system.list_devices = lambda: devices
return system return system
def _dev(self, id_, name, *, is_default=False, is_monitor=False, backend="portaudio"): def _dev(self, id_, name, *, is_default=False):
return isr.AudioDevice( return isr.AudioDevice(id=id_, name=name, is_default=is_default)
id=id_, name=name, channels=2, sample_rate=44100,
backend=backend, is_default=is_default, is_monitor=is_monitor,
)
def test_default_returns_default_device(self): def test_default_returns_default_device(self):
devices = [ devices = [
@@ -76,19 +73,9 @@ class TestAudioSystemFindDevice:
result = system.find_device("default") result = system.find_device("default")
assert result.id == "0" assert result.id == "0"
def test_monitor_returns_monitor_device(self): def test_default_with_no_cards_uses_the_alsa_default_pcm(self):
devices = [ system = self._make_system([])
self._dev("0", "Microphone"), assert system.find_device("default").id == "default"
self._dev("1", "Monitor of Built-in Audio", is_monitor=True),
]
system = self._make_system(devices)
result = system.find_device("monitor")
assert result.id == "1"
def test_monitor_returns_none_when_no_monitor(self):
devices = [self._dev("0", "Mic")]
system = self._make_system(devices)
assert system.find_device("monitor") is None
def test_exact_id_match(self): def test_exact_id_match(self):
devices = [self._dev("42", "Device 42"), self._dev("7", "Device 7")] devices = [self._dev("42", "Device 42"), self._dev("7", "Device 7")]
@@ -105,19 +92,26 @@ class TestAudioSystemFindDevice:
system = self._make_system(devices) system = self._make_system(devices)
assert system.find_device("stereo").id == "0" assert system.find_device("stereo").id == "0"
def test_returns_none_for_unknown_spec(self): def test_exact_id_wins_over_partial_name(self):
devices = [self._dev("0", "Mic")] # Two identical USB cards: both must stay reachable. An earlier
# dedup-by-name dropped the second one entirely.
devices = [self._dev("hw:1,0", "USB Audio"), self._dev("hw:2,0", "USB Audio")]
system = self._make_system(devices) system = self._make_system(devices)
assert system.find_device("nonexistent_device_xyz") is None assert system.find_device("hw:2,0").id == "hw:2,0"
def test_preferred_backend_filter_on_default(self): def test_unknown_spec_becomes_a_literal_alsa_pcm_name(self):
devices = [ # This is how `device = shared_mic` (a dsnoop PCM from asound.conf)
self._dev("0", "Mic", is_default=True, backend="pulseaudio"), # resolves — it is never listed by `arecord -l`.
self._dev("1", "Mic", is_default=True, backend="portaudio"), devices = [self._dev("hw:0,0", "Mic")]
]
system = self._make_system(devices) system = self._make_system(devices)
result = system.find_device("default", preferred_backend="portaudio") assert system.find_device("shared_mic").id == "shared_mic"
assert result.backend == "portaudio"
def test_pcm_fallback_works_when_no_cards_are_enumerable(self):
# Regression: a `if not devices: return None` guard used to fire before
# the fallback, so shared_mic failed outright on a host with no
# enumerable hardware card.
system = self._make_system([])
assert system.find_device("shared_mic").id == "shared_mic"
# =========================================================================== # ===========================================================================
@@ -433,7 +427,7 @@ class TestALSAStreamLiveness:
def _stream(self, poll_result=None) -> isr.ALSAStream: def _stream(self, poll_result=None) -> isr.ALSAStream:
device = isr.AudioDevice( device = isr.AudioDevice(
id="hw:0,0", name="Test", channels=2, sample_rate=44100, backend="alsa", id="hw:0,0", name="Test",
) )
s = isr.ALSAStream(device, 44100, 2, lambda data: None, make_logger()) s = isr.ALSAStream(device, 44100, 2, lambda data: None, make_logger())
s._process = MagicMock() s._process = MagicMock()
@@ -525,7 +519,7 @@ format = ogg
# Patch AudioSystem so it doesn't probe real hardware # Patch AudioSystem so it doesn't probe real hardware
with patch.object(isr, "AudioSystem") as mock_as: with patch.object(isr, "AudioSystem") as mock_as:
mock_as.return_value.available_backends = [] mock_as.return_value.available = False
mgr = isr.RecorderManager(str(cfg_path)) mgr = isr.RecorderManager(str(cfg_path))
assert len(mgr.recorders) == 1 assert len(mgr.recorders) == 1
@@ -539,7 +533,7 @@ format = ogg
cfg_path.write_text(self._minimal_config(tmp_path)) cfg_path.write_text(self._minimal_config(tmp_path))
with patch.object(isr, "AudioSystem") as mock_as: with patch.object(isr, "AudioSystem") as mock_as:
mock_as.return_value.available_backends = [] mock_as.return_value.available = False
mgr = isr.RecorderManager(str(cfg_path)) mgr = isr.RecorderManager(str(cfg_path))
rec = mgr.recorders[0] rec = mgr.recorders[0]
@@ -566,7 +560,7 @@ split_minutes = 15
cfg_path.write_text(config_text) cfg_path.write_text(config_text)
with patch.object(isr, "AudioSystem") as mock_as: with patch.object(isr, "AudioSystem") as mock_as:
mock_as.return_value.available_backends = [] mock_as.return_value.available = False
mgr = isr.RecorderManager(str(cfg_path)) mgr = isr.RecorderManager(str(cfg_path))
rec = mgr.recorders[0] rec = mgr.recorders[0]
@@ -591,7 +585,7 @@ url = http://x.com/s2
cfg_path.write_text(config_text) cfg_path.write_text(config_text)
with patch.object(isr, "AudioSystem") as mock_as: with patch.object(isr, "AudioSystem") as mock_as:
mock_as.return_value.available_backends = [] mock_as.return_value.available = False
mgr = isr.RecorderManager(str(cfg_path)) mgr = isr.RecorderManager(str(cfg_path))
assert len(mgr.recorders) == 1 assert len(mgr.recorders) == 1
@@ -643,22 +637,18 @@ class TestStreamRecorderRecord:
# =========================================================================== # ===========================================================================
class TestSoundcardRecorder: class TestSoundcardRecorder:
def _make_device(self, backend="portaudio") -> isr.AudioDevice: def _make_device(self) -> isr.AudioDevice:
return isr.AudioDevice( return isr.AudioDevice(id="0", name="Test Device")
id="0", name="Test Device", channels=2,
sample_rate=44100, backend=backend,
)
def _make_audio_system(self, device: isr.AudioDevice) -> isr.AudioSystem: def _make_audio_system(self, device: isr.AudioDevice) -> isr.AudioSystem:
system = MagicMock(spec=isr.AudioSystem) system = MagicMock(spec=isr.AudioSystem)
system.available_backends = [device.backend] system.available = True
system.find_device.return_value = device system.find_device.return_value = device
return system return system
def test_wav_file_written_correctly(self, tmp_path): def test_wav_file_written_correctly(self, tmp_path):
device = self._make_device() device = self._make_device()
audio_system = self._make_audio_system(device) audio_system = self._make_audio_system(device)
audio_system.get_backend.return_value = MagicMock()
cfg = { cfg = {
"device": "default", "device": "default",
@@ -710,9 +700,7 @@ class TestSoundcardRecorder:
# split_minutes and writing zero bytes into each one. # split_minutes and writing zero bytes into each one.
device = self._make_device() device = self._make_device()
audio_system = self._make_audio_system(device) audio_system = self._make_audio_system(device)
backend = MagicMock() audio_system.open_stream.return_value = _DeadStream()
backend.open_stream.return_value = _DeadStream()
audio_system.get_backend.return_value = backend
cfg = { cfg = {
"device": "default", "device": "default",
@@ -728,7 +716,7 @@ class TestSoundcardRecorder:
with patch.object(isr.time, "sleep"): # don't pay the 0.5 s poll interval with patch.object(isr.time, "sleep"): # don't pay the 0.5 s poll interval
rec.record() rec.record()
assert backend.open_stream.call_count == cfg["max_retries"] assert audio_system.open_stream.call_count == cfg["max_retries"]
@pytest.mark.skipif(not isr.SOUNDFILE_AVAILABLE, reason="soundfile not installed") @pytest.mark.skipif(not isr.SOUNDFILE_AVAILABLE, reason="soundfile not installed")
def test_flac_file_written_correctly(self, tmp_path): def test_flac_file_written_correctly(self, tmp_path):