diff --git a/README.md b/README.md index c6049ed..7b2626f 100644 --- a/README.md +++ b/README.md @@ -115,6 +115,7 @@ Settings are stored in `~/.config/sendspin/`: "listen_port": 8927, "use_mpris": true, "use_hardware_volume": true, + "alsa_mixer_control": "0:Digital", "hook_set_volume": "/usr/local/bin/set-avr-volume", "manufacturer": "Acme Corp", "product_name": "Living Room Speaker", @@ -149,6 +150,7 @@ Settings are stored in `~/.config/sendspin/`: | `listen_port` | integer | daemon/serve | Listen port (`--port`, default: 8927) | | `use_mpris` | boolean | TUI/daemon | Enable MPRIS integration (default: true) | | `use_hardware_volume` | boolean | TUI/daemon | Control hardware/system output volume instead of software volume (`--hardware-volume true/false`). Default: on for daemon (if available), off for TUI | +| `alsa_mixer_control` | string | TUI/daemon | ALSA mixer control as `CARD:ELEMENT` (e.g., `0:Digital`). Overrides auto-detection (`--alsa-mixer-control`) | | `hook_set_volume` | string | TUI/daemon | Script to run for external volume control (`--hook-set-volume`). Receives the effective volume 0-100 as the last argument | | `hook_start` | string | TUI/daemon | Command to run when audio stream starts | | `hook_stop` | string | TUI/daemon | Command to run when audio stream stops | diff --git a/sendspin/alsa_volume.py b/sendspin/alsa_volume.py index cf7385f..8f2ff12 100644 --- a/sendspin/alsa_volume.py +++ b/sendspin/alsa_volume.py @@ -137,8 +137,40 @@ async def find_mixer_element(card: int) -> str | None: return selected +async def _check_mixer_control_hint(mixer_control_hint: str) -> tuple[int, str] | None: + """Try to resolve an explicit mixer control hint. + + Parses ``card:element``, verifies the element has playback volume, + and returns ``(card, element)`` or None. + """ + if ":" not in mixer_control_hint: + return None + card_part, element = mixer_control_hint.rsplit(":", 1) + try: + card = int(card_part) + except ValueError: + return None + + if not await _has_playback_volume(card, element): + logger.warning( + "Mixer control hint %r (card %d, element %r) has no playback volume", + mixer_control_hint, + card, + element, + ) + return None + + logger.debug( + "Using explicit ALSA mixer control: card %d, element %r", + card, + element, + ) + return card, element + + async def async_check_alsa_available( audio_device: AudioDevice, + mixer_control_hint: str | None = None, ) -> tuple[int, str] | None: """Check if ALSA mixer volume control is available for a device. @@ -147,6 +179,11 @@ async def async_check_alsa_available( if not AVAILABLE: return None + if mixer_control_hint is not None: + resolved = await _check_mixer_control_hint(mixer_control_hint) + if resolved is not None: + return resolved + card = parse_alsa_card(audio_device.name) if card is None: return None diff --git a/sendspin/cli.py b/sendspin/cli.py index 0bead5b..6591a3b 100644 --- a/sendspin/cli.py +++ b/sendspin/cli.py @@ -204,6 +204,12 @@ def _add_player_runtime_options(target: ArgumentTarget, *, suppress_defaults: bo metavar="{true,false}", help="Enable or disable hardware/system volume control (daemon: on, TUI: off)", ) + target.add_argument( + "--alsa-mixer-control", + type=str, + default=default, + help="ALSA mixer control as CARD:ELEMENT", + ) target.add_argument( "--manufacturer", type=str, @@ -441,6 +447,12 @@ def _build_parser() -> argparse.ArgumentParser: metavar="{true,false}", help="Enable or disable hardware/system volume control (daemon: on, TUI: off)", ) + daemon_parser.add_argument( + "--alsa-mixer-control", + type=str, + default=None, + help="ALSA mixer control as CARD:ELEMENT", + ) daemon_parser.add_argument( "--hook-start", type=str, @@ -833,6 +845,8 @@ async def _run_client_mode(args: argparse.Namespace) -> int: args.hardware_volume = settings.use_hardware_volume else: args.hardware_volume = is_daemon and (HW_VOLUME_AVAILABLE or ALSA_AVAILABLE) + if args.alsa_mixer_control is None: + args.alsa_mixer_control = settings.alsa_mixer_control if args.hook_set_volume is None: args.hook_set_volume = settings.hook_set_volume if not args.hook_set_volume and args.hardware_volume and not HW_VOLUME_AVAILABLE: @@ -874,7 +888,9 @@ async def _run_client_mode(args: argparse.Namespace) -> int: volume_controller = HookVolumeController(args.hook_set_volume, settings) elif args.hardware_volume: # Try ALSA direct control first (works for hw: devices without PulseAudio). - alsa_info = await alsa_volume_check_available(audio_device) + alsa_info = await alsa_volume_check_available( + audio_device, mixer_control_hint=args.alsa_mixer_control + ) if alsa_info is not None: card, element = alsa_info LOGGER.info( diff --git a/sendspin/daemon/daemon.py b/sendspin/daemon/daemon.py index 2faa15c..8e4ffa8 100644 --- a/sendspin/daemon/daemon.py +++ b/sendspin/daemon/daemon.py @@ -49,6 +49,7 @@ class DaemonArgs: use_mpris: bool = True preferred_format: SupportedAudioFormat | None = None volume_controller: VolumeController | None = None + alsa_mixer_control: str | None = None hook_start: str | None = None hook_stop: str | None = None manufacturer: str | None = None diff --git a/sendspin/settings.py b/sendspin/settings.py index cb2f669..4381004 100644 --- a/sendspin/settings.py +++ b/sendspin/settings.py @@ -120,6 +120,7 @@ class ClientSettings(BaseSettings): use_mpris: bool = True audio_format: str | None = None use_hardware_volume: bool | None = None + alsa_mixer_control: str | None = None hook_set_volume: str | None = None hook_start: str | None = None hook_stop: str | None = None @@ -148,6 +149,7 @@ def update( use_mpris: bool | None = None, audio_format: str | None = None, use_hardware_volume: bool | None = None, + alsa_mixer_control: str | None = None, hook_set_volume: str | None = None, hook_start: str | None = None, hook_stop: str | None = None, @@ -181,6 +183,7 @@ def update( "use_mpris": use_mpris, "audio_format": audio_format, "use_hardware_volume": use_hardware_volume, + "alsa_mixer_control": alsa_mixer_control, "hook_set_volume": hook_set_volume, "hook_start": hook_start, "hook_stop": hook_stop, @@ -222,6 +225,7 @@ def _load(self) -> bool: self.use_mpris = data.get("use_mpris", True) self.audio_format = data.get("audio_format") self.use_hardware_volume = data.get("use_hardware_volume") + self.alsa_mixer_control = data.get("alsa_mixer_control") self.hook_set_volume = data.get("hook_set_volume") self.hook_start = data.get("hook_start") self.hook_stop = data.get("hook_stop") diff --git a/sendspin/tui/app.py b/sendspin/tui/app.py index ad6eb63..f6c6fa0 100644 --- a/sendspin/tui/app.py +++ b/sendspin/tui/app.py @@ -244,6 +244,7 @@ class AppArgs: use_mpris: bool = True preferred_format: SupportedAudioFormat | None = None volume_controller: VolumeController | None = None + alsa_mixer_control: str | None = None hook_start: str | None = None hook_stop: str | None = None manufacturer: str | None = None diff --git a/tests/test_alsa_volume.py b/tests/test_alsa_volume.py index 5cefd8d..505beae 100644 --- a/tests/test_alsa_volume.py +++ b/tests/test_alsa_volume.py @@ -522,3 +522,128 @@ async def test_louder_raspberry_get_volume(monkeypatch) -> None: volume, muted = await ctrl.get_state() assert volume == 57 assert muted is False + + +# -- _check_mixer_control_hint ------------------------------------------------ + + +async def test_check_mixer_control_hint_valid(monkeypatch) -> None: + """Returns (card, element) for a valid card:element hint.""" + from sendspin.alsa_volume import _check_mixer_control_hint + + sget_pvolume = " Capabilities: pvolume pswitch\n" + monkeypatch.setattr(asyncio, "create_subprocess_exec", _amixer_exec(sget_pvolume)) + + result = await _check_mixer_control_hint("0:Digital") + assert result == (0, "Digital") + + +async def test_check_mixer_control_hint_no_colon(monkeypatch) -> None: + """Returns None when the hint string has no colon.""" + from sendspin.alsa_volume import _check_mixer_control_hint + + result = await _check_mixer_control_hint("Digital") + assert result is None + + +async def test_check_mixer_control_hint_non_numeric_card(monkeypatch) -> None: + """Returns None when the card part is not numeric.""" + from sendspin.alsa_volume import _check_mixer_control_hint + + result = await _check_mixer_control_hint("sndrpijustboomd:Digital") + assert result is None + + +async def test_check_mixer_control_hint_no_playback_volume(monkeypatch) -> None: + """Returns None when the element has no playback volume capability.""" + from sendspin.alsa_volume import _check_mixer_control_hint + + # Element with no volume capability + monkeypatch.setattr( + asyncio, "create_subprocess_exec", _amixer_exec(" Capabilities: pswitch\n") + ) + + result = await _check_mixer_control_hint("0:Master") + assert result is None + + +async def test_check_mixer_control_hint_amixer_not_found(monkeypatch) -> None: + """Returns None when amixer is not found.""" + from sendspin.alsa_volume import _check_mixer_control_hint + + async def not_found(*args: object, **kwargs: object) -> NoReturn: + raise FileNotFoundError("amixer") + + monkeypatch.setattr(asyncio, "create_subprocess_exec", not_found) + result = await _check_mixer_control_hint("0:Digital") + assert result is None + + +# -- async_check_alsa_available with mixer_control_hint ----------------------- + + +async def test_alsa_available_hint_overrides_virtual_device(monkeypatch) -> None: + """A hint bypasses auto-detection for virtual devices with no hw: reference.""" + sget_pvolume = " Capabilities: pvolume pswitch\n" + monkeypatch.setattr(asyncio, "create_subprocess_exec", _amixer_exec(sget_pvolume)) + monkeypatch.setattr(_alsa_mod, "AVAILABLE", True) + + device = SimpleNamespace(name="plugdmix", is_default=False) + result = await async_check_alsa_available(device, mixer_control_hint="0:Digital") + assert result == (0, "Digital") + + +async def test_alsa_available_hint_falls_back_when_invalid(monkeypatch) -> None: + """An invalid hint falls back to auto-detection.""" + scontrols = "Simple mixer control 'Digital',0\n" + sget_pvolume = " Capabilities: pvolume pswitch\n" + + calls = [] + + async def fake_exec(*argv: object, **kwargs: object) -> _FakeProcess: + calls.append(list(argv)) + if "scontrols" in argv: + return _FakeProcess(stdout=scontrols.encode()) + return _FakeProcess(stdout=sget_pvolume.encode()) + + monkeypatch.setattr(asyncio, "create_subprocess_exec", fake_exec) + monkeypatch.setattr(_alsa_mod, "AVAILABLE", True) + device = SimpleNamespace(name="HiFiBerry DAC+: pcm512x (hw:1,0)", is_default=False) + + result = await async_check_alsa_available(device, mixer_control_hint="bad:hint") + assert result == (1, "Digital") + + +async def test_alsa_available_hint_element_no_volume_falls_back(monkeypatch) -> None: + """A hint with a non-volume element falls back to auto-detection.""" + scontrols = "Simple mixer control 'Digital',0\n" + sget_no_volume = " Capabilities: pswitch\n" # no pvolume + sget_pvolume = " Capabilities: pvolume pswitch\n" + + calls = [] + + async def fake_exec(*argv: object, **kwargs: object) -> _FakeProcess: + calls.append(list(argv)) + if "scontrols" in argv: + return _FakeProcess(stdout=scontrols.encode()) + # First sget (for hint validation) has no volume, second (for discovery) has it + if "Master" in argv: + return _FakeProcess(stdout=sget_no_volume.encode()) + return _FakeProcess(stdout=sget_pvolume.encode()) + + monkeypatch.setattr(asyncio, "create_subprocess_exec", fake_exec) + monkeypatch.setattr(_alsa_mod, "AVAILABLE", True) + device = SimpleNamespace(name="HiFiBerry DAC+: pcm512x (hw:1,0)", is_default=False) + + result = await async_check_alsa_available(device, mixer_control_hint="1:Master") + # Hint element has no volume, falls back to auto-detection + assert result == (1, "Digital") + + +async def test_alsa_available_hint_not_checked_when_unavailable(monkeypatch) -> None: + """Hint is ignored when ALSA is not available on the system.""" + monkeypatch.setattr(_alsa_mod, "AVAILABLE", False) + + device = SimpleNamespace(name="plugdmix", is_default=False) + result = await async_check_alsa_available(device, mixer_control_hint="0:Digital") + assert result is None