Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand Down Expand Up @@ -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 |
Expand Down
37 changes: 37 additions & 0 deletions sendspin/alsa_volume.py
Original file line number Diff line number Diff line change
Expand Up @@ -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.

Expand All @@ -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
Expand Down
18 changes: 17 additions & 1 deletion sendspin/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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(
Expand Down
1 change: 1 addition & 0 deletions sendspin/daemon/daemon.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
4 changes: 4 additions & 0 deletions sendspin/settings.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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")
Expand Down
1 change: 1 addition & 0 deletions sendspin/tui/app.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
125 changes: 125 additions & 0 deletions tests/test_alsa_volume.py
Original file line number Diff line number Diff line change
Expand Up @@ -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