/
/
/
1"""
2Private PulseAudio capture server for grabbing PCM audio from client applications.
3
4Runs a minimal, MA-owned classic PulseAudio daemon (no autodetected devices, no
5network) whose module-pipe-sink FIFOs let MA capture the audio output of external
6client processes that can only play via PulseAudio/PipeWire (e.g. the official
7Spotify client). Consumers read a sink's FIFO with
8``helpers.named_pipe.read_named_pipe`` or by pointing ffmpeg at it directly.
9
10Also hosts :class:`PAVolumeController`, the shared ctypes libpulse controller for
11sink volume and module control, usable against any PA server (the private capture
12daemon or the system/host one).
13"""
14
15from __future__ import annotations
16
17import asyncio
18import ctypes
19import logging
20import math
21import os
22import shutil
23import threading
24import uuid
25import weakref
26from contextlib import suppress
27from pathlib import Path
28from typing import TYPE_CHECKING, ClassVar, Final
29
30from music_assistant.constants import MASS_LOGGER_NAME
31from music_assistant.helpers.process import AsyncProcess, check_output, get_subprocess_env
32
33if TYPE_CHECKING:
34 from music_assistant.mass import MusicAssistant
35
36LOGGER = logging.getLogger(f"{MASS_LOGGER_NAME}.helpers.pulse_capture")
37
38# Fixed capture format for all pipe sinks: 32-bit little-endian PCM, CD sample
39# rate, stereo. s32le keeps full precision of any 16/24-bit source even when the
40# sink volume applies (reciprocal) gain before MA reads the samples back.
41CAPTURE_SAMPLE_FORMAT: Final[str] = "s32le"
42CAPTURE_SAMPLE_RATE: Final[int] = 44100
43CAPTURE_CHANNELS: Final[int] = 2
44
45PA_VOLUME_NORM: Final = 65536 # 100% (0 dB) on PA's raw volume scale
46PA_CHANNELS_MAX: Final = 32
47
48# Ceiling for the raw (linear) volume setter. Reciprocal cubic compensation
49# needs up to 10000% (a client stream at 1% compensates with a 100x sink gain);
50# the resulting raw value (100 * PA_VOLUME_NORM) stays far below PA_VOLUME_MAX.
51MAX_RAW_VOLUME_PCT: Final = 10000.0
52
53# Daemon startup: how long to wait for the native socket to appear, and the
54# poll interval while waiting.
55_READY_TIMEOUT: Final = 10.0
56_READY_POLL_INTERVAL: Final = 0.1
57
58# Supervised-restart backoff: starts small so a one-off crash recovers fast,
59# doubles up to the max so a permanently failing daemon doesn't spin.
60_RESTART_BACKOFF_INITIAL: Final = 1.0
61_RESTART_BACKOFF_MAX: Final = 30.0
62
63# set_sink_volume() is called frequently (on every volume/mute change, and
64# at bridge start for every player). A short timeout limits how long a
65# stuck/unresponsive PA call can occupy an executor thread â under normal
66# conditions PA responds in single-digit milliseconds, so 0.5s is generous
67# while bounding the worst case. load_module()/unload_module() (rare,
68# one-time during topology setup/teardown) keep a longer 2.0s timeout since
69# we'd rather wait than have sink creation/cleanup spuriously fail.
70_SET_VOLUME_TIMEOUT: Final = 0.5
71
72PA_CONTEXT_READY: Final = 4
73PA_CONTEXT_FAILED: Final = 5
74PA_CONTEXT_TERMINATED: Final = 6
75PA_CONTEXT_NOAUTOSPAWN: Final = 1
76
77_CONTEXT_NOTIFY_CB = ctypes.CFUNCTYPE(None, ctypes.c_void_p, ctypes.c_void_p)
78_CONTEXT_SUCCESS_CB = ctypes.CFUNCTYPE(None, ctypes.c_void_p, ctypes.c_int, ctypes.c_void_p)
79_CONTEXT_INDEX_CB = ctypes.CFUNCTYPE(None, ctypes.c_void_p, ctypes.c_uint32, ctypes.c_void_p)
80
81PA_INVALID_INDEX: Final = 0xFFFFFFFF
82
83# --- Audio taper curve (dr-lex exponential, with linear roll-off) -----------
84#
85# y = a * e^(b*x) gives constant dB-per-slider-step ("audio taper" /
86# logarithmic potentiometer behavior), unlike a plain linear-amplitude
87# mapping (y = x) where the bottom of the slider is wildly more sensitive
88# than the top. See https://www.dr-lex.be/info-stuff/volumecontrols.html
89#
90# a = 10**(-range_dB/20) sets the amplitude floor; b = ln(1/a) ensures
91# y(1.0) = 1.0 (0dB) at full volume. Below _TAPER_ROLLOFF_X, a linear ramp
92# to (0, 0) is used so volume_pct=0 is true silence rather than asymptoting
93# toward the floor.
94#
95# Reference values for common dB ranges (pick one _TAPER_A and comment out
96# the rest; _TAPER_B recalculates automatically):
97#
98# Range _TAPER_A _TAPER_B MA 70% = Notes
99# 40 dB 0.01 ~4.605 -12 dB receiver / outdoor speakers (current)
100# 50 dB 0.003162 ~5.757 -15 dB medium-range setups
101# 60 dB 0.001 ~6.908 -18 dB consumer headphones / desktop speakers
102# 70 dB 0.000316 ~8.059 -21 dB high-dynamic-range hi-fi systems
103#
104# Used for PA hardware volume (PAVolumeController.set_sink_volume), after a
105# cube-root step to counteract PA's own cubic volume curve.
106_TAPER_A: Final = 0.01 # 10**(-40/20) â 40dB range, suits receiver/outdoor setups
107# _TAPER_A: Final = 0.003162 # 10**(-50/20) â 50dB range
108# _TAPER_A: Final = 0.001 # 10**(-60/20) â 60dB range, suits headphones/desktop
109# _TAPER_A: Final = 0.000316 # 10**(-70/20) â 70dB range, hi-fi high dynamic range
110_TAPER_B: Final = math.log(1.0 / _TAPER_A) # recalculates automatically from _TAPER_A
111_TAPER_ROLLOFF_X: Final = 0.10 # below 10% slider, linear ramp to true silence
112
113
114def volume_pct_to_amplitude(volume_pct: int) -> float:
115 """
116 Map a 0-100 volume percentage to a linear amplitude scale factor.
117
118 Uses the dr-lex exponential audio taper (y = a*e^(b*x)) for
119 volume_pct >= 10, giving constant dB change per slider step. Below 10%,
120 a linear ramp to (0, 0) ensures volume_pct=0 produces true silence.
121 """
122 x = max(0, min(volume_pct, 100)) / 100.0
123 if x <= 0:
124 return 0.0
125 if x < _TAPER_ROLLOFF_X:
126 y1 = _TAPER_A * math.exp(_TAPER_B * _TAPER_ROLLOFF_X)
127 return y1 * (x / _TAPER_ROLLOFF_X)
128 return _TAPER_A * math.exp(_TAPER_B * x)
129
130
131def get_default_pulse_server() -> str:
132 """
133 Detect the system's default PulseAudio server address.
134
135 Checked fresh on each call â the socket may not exist at import time but
136 appear later once the audio host/addon has fully started.
137
138 :returns: Server address (env value or "unix:<socket path>"), or an empty
139 string when nothing was found (libpulse then uses its own defaults).
140 """
141 if server := os.environ.get("PULSE_SERVER"):
142 return server
143 for path in (
144 "/run/audio/pulse.sock",
145 "/run/pulse/native",
146 "/var/run/pulse/native",
147 ):
148 if Path(path).exists():
149 return f"unix:{path}"
150 return ""
151
152
153def get_pulse_capture_server(mass: MusicAssistant) -> PulseCaptureServer:
154 """
155 Return the shared PulseCaptureServer instance for this MusicAssistant.
156
157 All consumers must obtain the server through this function so they share a
158 single private daemon, and must pair acquire()/release() around their usage.
159 """
160 if (server := _servers.get(mass)) is None:
161 server = PulseCaptureServer(mass)
162 _servers[mass] = server
163 return server
164
165
166class PulseCaptureServer:
167 """
168 Private, MA-owned classic PulseAudio daemon for audio capture.
169
170 Lazily started on the first :meth:`acquire` and stopped again on the last
171 :meth:`release` (reference-counted) â use :func:`get_pulse_capture_server`
172 to obtain the shared per-process instance. The daemon loads only the native
173 protocol on a unix socket inside a private runtime dir under MA's cache dir;
174 audio clients are pointed at it via :meth:`child_env` and their audio is
175 captured through :class:`PipeSink` FIFOs.
176
177 If the daemon dies it is restarted automatically and :attr:`generation` is
178 bumped. All sinks created against the previous daemon are gone at that
179 point, so consumers must snapshot ``generation`` when creating a sink and
180 recreate their :class:`PipeSink` when it changes.
181 """
182
183 def __init__(self, mass: MusicAssistant) -> None:
184 """
185 Initialize the capture server (the daemon is not started yet).
186
187 :param mass: MusicAssistant instance, used for the private runtime dir
188 under its cache path.
189 """
190 # only the cache path is kept: holding mass itself would pin the
191 # WeakKeyDictionary registry entry (value -> key) forever
192 self._base_dir = Path(mass.cache_path) / "pulse_capture"
193 self._socket_path = self._base_dir / "native"
194 self._config_path = self._base_dir / "pulse_capture.pa"
195 self._lock = asyncio.Lock()
196 self._refcount = 0
197 self._generation = 0
198 self._proc: AsyncProcess | None = None
199 self._controller: PAVolumeController | None = None
200 self._supervisor_task: asyncio.Task[None] | None = None
201
202 @property
203 def generation(self) -> int:
204 """Daemon generation, bumped on every (re)start â snapshot when creating sinks."""
205 return self._generation
206
207 @property
208 def server_address(self) -> str:
209 """PA server address ("unix:<socket path>") of the private daemon."""
210 return f"unix:{self._socket_path}"
211
212 async def acquire(self) -> PulseCaptureServer:
213 """
214 Register a consumer, starting the private daemon on first use.
215
216 Every successful acquire() must be paired with exactly one release().
217
218 :returns: This server instance, for convenient chaining.
219 """
220 async with self._lock:
221 if self._refcount == 0:
222 await self._start()
223 self._refcount += 1
224 return self
225
226 async def release(self) -> None:
227 """
228 Unregister a consumer, stopping the daemon when none remain.
229
230 Safe to call more often than acquire() (extra calls are ignored).
231 """
232 async with self._lock:
233 if self._refcount == 0:
234 return
235 if self._refcount > 1:
236 self._refcount -= 1
237 return
238 # last consumer: run teardown to completion even when this call is
239 # cancelled, and only commit the zero refcount once it finished â
240 # no half-stopped daemon can be left behind or double-started
241 teardown = asyncio.ensure_future(self._stop())
242 try:
243 await asyncio.shield(teardown)
244 except asyncio.CancelledError:
245 while not teardown.done():
246 with suppress(asyncio.CancelledError):
247 await asyncio.shield(teardown)
248 self._refcount = 0
249 raise
250 self._refcount = 0
251
252 def child_env(self, sink_name: str) -> dict[str, str]:
253 """
254 Environment for an audio-client subprocess that must play into a sink.
255
256 :param sink_name: Sink the client's audio must go to (PipeSink.sink_name).
257 :returns: Full subprocess environment with PULSE_SERVER/PULSE_SINK set.
258 """
259 return get_subprocess_env(
260 {
261 "PULSE_SERVER": self.server_address,
262 "PULSE_SINK": sink_name,
263 }
264 )
265
266 async def _start(self) -> None:
267 """Start the daemon and its supervisor. Called with the lock held."""
268 await self._launch_daemon()
269 self._supervisor_task = asyncio.create_task(self._supervise())
270 self._supervisor_task.add_done_callback(_log_supervisor_exit)
271
272 async def _stop(self) -> None:
273 """Stop supervisor and daemon, clean the private dir. Called with the lock held."""
274 if (task := self._supervisor_task) is not None:
275 self._supervisor_task = None
276 task.cancel()
277 with suppress(asyncio.CancelledError):
278 await task
279 if (controller := self._controller) is not None:
280 self._controller = None
281 with suppress(Exception):
282 await asyncio.to_thread(controller.close)
283 if (proc := self._proc) is not None:
284 self._proc = None
285 await proc.close()
286
287 # the daemon's death drops all loaded modules with it, so nothing needs
288 # unloading; only the private runtime dir (socket + leftover FIFOs) remains
289 def _cleanup() -> None:
290 shutil.rmtree(self._base_dir, ignore_errors=True)
291
292 await asyncio.to_thread(_cleanup)
293
294 async def _launch_daemon(self) -> None:
295 """Start the daemon process and wait until it accepts connections."""
296 config_text = (
297 f"load-module module-native-protocol-unix socket={self._socket_path} auth-anonymous=1\n"
298 )
299
300 # PA's .pa config and module argument parsers are space-delimited with no
301 # escaping; refuse early with a clear error instead of mis-parsing later
302 if " " in str(self._base_dir):
303 raise RuntimeError(f"cache path may not contain spaces: {self._base_dir}")
304
305 def _prepare() -> None:
306 self._base_dir.mkdir(parents=True, exist_ok=True)
307 # the daemon accepts anonymous connections on its socket, so the
308 # private dir must never be accessible to other local users
309 self._base_dir.chmod(0o700)
310 self._socket_path.unlink(missing_ok=True)
311 # a stale pid file from a hard shutdown makes pulse refuse to start
312 # ("daemon already running") when that pid is reused by any process
313 (self._base_dir / "pid").unlink(missing_ok=True)
314 # a hard shutdown skips sink cleanup; sweep leftover FIFOs from
315 # previous runs (a fresh daemon has no modules yet)
316 for stale_fifo in self._base_dir.glob("*.pcm"):
317 stale_fifo.unlink(missing_ok=True)
318 self._config_path.write_text(config_text, encoding="utf-8")
319
320 await asyncio.to_thread(_prepare)
321 # runtime/state dirs are redirected into the private dir via the child
322 # environment only; os.environ is never mutated
323 proc = AsyncProcess(
324 [
325 "pulseaudio",
326 "--daemonize=no",
327 "-n",
328 "--exit-idle-time=-1",
329 f"--file={self._config_path}",
330 ],
331 stderr=True,
332 name="pulse-capture",
333 env={
334 "XDG_RUNTIME_DIR": str(self._base_dir),
335 "PULSE_RUNTIME_PATH": str(self._base_dir),
336 "PULSE_STATE_PATH": str(self._base_dir),
337 },
338 )
339 self._proc = proc
340 try:
341 await proc.start()
342 await self._wait_ready(proc)
343 # verify the daemon actually accepts connections before declaring
344 # ready; the construction cannot be interrupted mid-flight, so on
345 # cancellation close whatever the worker thread still produced to
346 # avoid leaking a live threaded mainloop
347 ctor = asyncio.ensure_future(asyncio.to_thread(PAVolumeController, self.server_address))
348 try:
349 controller = await asyncio.shield(ctor)
350 except asyncio.CancelledError:
351 ctor.add_done_callback(_close_controller_result)
352 raise
353 except BaseException:
354 if self._proc is proc:
355 self._proc = None
356 with suppress(Exception):
357 await proc.close()
358 raise
359 # publish the new controller and generation before disposing of the old
360 # controller, so teardown can always reach the live one even when the
361 # disposal await is cancelled
362 old_controller = self._controller
363 self._controller = controller
364 self._generation += 1
365 if old_controller is not None:
366 with suppress(Exception):
367 await asyncio.to_thread(old_controller.close)
368 LOGGER.debug(
369 "Private PulseAudio capture daemon ready on %s (generation %d)",
370 self.server_address,
371 self._generation,
372 )
373
374 async def _wait_ready(self, proc: AsyncProcess) -> None:
375 """Wait for the daemon's native socket to appear."""
376 try:
377 async with asyncio.timeout(_READY_TIMEOUT):
378 while not await asyncio.to_thread(self._socket_path.exists):
379 if proc.returncode is not None:
380 raise RuntimeError(
381 f"pulseaudio exited during startup (code {proc.returncode})"
382 )
383 await asyncio.sleep(_READY_POLL_INTERVAL)
384 except TimeoutError:
385 raise RuntimeError("Timeout waiting for the pulseaudio daemon socket") from None
386
387 async def _supervise(self) -> None:
388 """Restart the daemon (with bounded backoff) until the supervisor is cancelled."""
389 backoff = _RESTART_BACKOFF_INITIAL
390 while True:
391 if (proc := self._proc) is not None:
392 try:
393 async for line in proc.iter_stderr():
394 LOGGER.debug("pulseaudio: %s", line)
395 except Exception as err:
396 LOGGER.debug("pulseaudio log reader stopped: %s", err)
397 await proc.close()
398 if self._proc is proc:
399 self._proc = None
400 LOGGER.warning(
401 "Private PulseAudio capture daemon exited unexpectedly, restarting in %.1fs",
402 backoff,
403 )
404 await asyncio.sleep(backoff)
405 try:
406 await self._launch_daemon()
407 except Exception as err:
408 backoff = min(backoff * 2, _RESTART_BACKOFF_MAX)
409 LOGGER.error("Failed to restart the PulseAudio capture daemon: %s", err)
410 continue
411 backoff = _RESTART_BACKOFF_INITIAL
412
413 async def _load_module(self, module_name: str, argument: str) -> int | None:
414 """Load a PA module on the private daemon (blocking libpulse call in a thread)."""
415 controller = self._require_controller()
416 return await asyncio.to_thread(controller.load_module, module_name, argument)
417
418 async def _unload_module(self, module_index: int) -> bool:
419 """Unload a PA module from the private daemon."""
420 controller = self._require_controller()
421 return await asyncio.to_thread(controller.unload_module, module_index)
422
423 async def _set_sink_volume_raw(self, sink_name: str, volume_pct: float) -> bool:
424 """Set raw (linear) volume on a sink of the private daemon."""
425 controller = self._require_controller()
426 return await asyncio.to_thread(controller.set_sink_volume_raw, sink_name, volume_pct)
427
428 def _require_controller(self) -> PAVolumeController:
429 """Return the connected controller or raise if the server is not running."""
430 if (controller := self._controller) is None:
431 raise RuntimeError("Pulse capture server is not running")
432 return controller
433
434
435class PipeSink:
436 """
437 One isolated module-pipe-sink capture sink on a :class:`PulseCaptureServer`.
438
439 The PA daemon creates a FIFO at :attr:`fifo_path` that delivers the sink's
440 audio as raw PCM in the fixed capture format (CAPTURE_SAMPLE_FORMAT /
441 CAPTURE_SAMPLE_RATE / CAPTURE_CHANNELS); read it with
442 ``helpers.named_pipe.read_named_pipe`` or by pointing ffmpeg at it.
443
444 Both consumer lifecycles are supported: a single long-lived sink per
445 provider instance, or a fresh sink per stream (create -> suspend/resume ->
446 unload). A sink does not survive a daemon restart: when the server's
447 ``generation`` no longer matches the value snapshotted at creation, drop
448 this instance and create a new one.
449 """
450
451 def __init__(
452 self,
453 server: PulseCaptureServer,
454 sink_name: str,
455 fifo_path: Path,
456 module_index: int,
457 generation: int,
458 ) -> None:
459 """Initialize the sink. Use the async :meth:`create` factory instead."""
460 self._server = server
461 self._sink_name = sink_name
462 self._fifo_path = fifo_path
463 self._module_index: int | None = module_index
464 self._generation = generation
465
466 @classmethod
467 async def create(cls, server: PulseCaptureServer, name_prefix: str) -> PipeSink:
468 """
469 Create a new uniquely-named pipe sink on the given capture server.
470
471 The pipe-sink module creates the FIFO file itself.
472
473 :param server: An acquired PulseCaptureServer.
474 :param name_prefix: Prefix for the generated sink name (e.g. a provider
475 instance id); a short unique suffix is appended.
476 """
477 sink_name = f"{name_prefix}_{uuid.uuid4().hex[:8]}"
478 fifo_path = server._base_dir / f"{sink_name}.pcm"
479 argument = (
480 f"sink_name={sink_name} file={fifo_path} "
481 f"format={CAPTURE_SAMPLE_FORMAT} rate={CAPTURE_SAMPLE_RATE} "
482 f"channels={CAPTURE_CHANNELS}"
483 )
484 # snapshot before the load: a restart during the load would otherwise
485 # pair a module index from the dead daemon with the new generation,
486 # and a later unload could hit an unrelated module on the replacement
487 generation = server.generation
488 module_index = await server._load_module("module-pipe-sink", argument)
489 if module_index is None:
490 raise RuntimeError(f"Failed to load module-pipe-sink for {sink_name}")
491 if server.generation != generation:
492 raise RuntimeError(f"capture daemon restarted while creating sink {sink_name}")
493 return cls(server, sink_name, fifo_path, module_index, generation)
494
495 @property
496 def sink_name(self) -> str:
497 """The PA sink name (pass to PulseCaptureServer.child_env for the client)."""
498 return self._sink_name
499
500 @property
501 def fifo_path(self) -> Path:
502 """Path of the FIFO delivering this sink's PCM audio."""
503 return self._fifo_path
504
505 async def set_volume(self, volume_pct: float) -> None:
506 """
507 Set the sink's raw (linear) volume.
508
509 A sink from a previous daemon generation ignores the call (recreate the
510 sink after a restart).
511
512 :param volume_pct: 100 is unity gain; values above 100 amplify (e.g. 400
513 for reciprocal cubic compensation). Clamped to MAX_RAW_VOLUME_PCT.
514 """
515 if self._generation != self._server.generation:
516 LOGGER.debug("Ignoring volume for stale sink %s", self._sink_name)
517 return
518 if not await self._server._set_sink_volume_raw(self._sink_name, volume_pct):
519 raise RuntimeError(f"Failed to set volume on capture sink {self._sink_name}")
520
521 async def suspend(self) -> None:
522 """Suspend the sink (its FIFO stops producing audio until resumed)."""
523 await self._set_suspended(True)
524
525 async def resume(self) -> None:
526 """Resume a suspended sink."""
527 await self._set_suspended(False)
528
529 async def unload(self) -> None:
530 """
531 Unload the sink's module and remove its FIFO (idempotent).
532
533 A sink whose daemon has restarted since creation is already gone; only
534 the leftover FIFO file is cleaned up in that case.
535 """
536 module_index = self._module_index
537 self._module_index = None
538 if module_index is not None and self._generation == self._server.generation:
539 # best effort: a failure usually means the daemon/module is already
540 # gone, and a restart reclaims all modules anyway
541 if not await self._server._unload_module(module_index):
542 LOGGER.warning("Failed to unload capture sink module %s", self._sink_name)
543 with suppress(OSError):
544 await asyncio.to_thread(self._fifo_path.unlink)
545
546 async def _set_suspended(self, suspended: bool) -> None:
547 """Toggle the sink's suspend state via pactl."""
548 if self._generation != self._server.generation:
549 LOGGER.debug("Ignoring suspend toggle for stale sink %s", self._sink_name)
550 return
551 returncode, output = await check_output(
552 "pactl",
553 "--server",
554 self._server.server_address,
555 "suspend-sink",
556 self._sink_name,
557 "1" if suspended else "0",
558 env={"PULSE_SERVER": self._server.server_address},
559 timeout=5,
560 )
561 if returncode != 0:
562 LOGGER.warning(
563 "pactl suspend-sink %s %d failed: %s",
564 self._sink_name,
565 int(suspended),
566 output.decode("utf-8", errors="replace").strip(),
567 )
568
569
570class PAVolumeController:
571 """
572 Shared libpulse connection for PA sink volume and module control.
573
574 One instance is shared per PA server. All calls are blocking and must be
575 invoked via run_in_executor/to_thread from async code.
576 """
577
578 def __init__(self, server: str | None = None) -> None:
579 """
580 Connect to PulseAudio and start the threaded mainloop.
581
582 :param server: PA server address to connect to (e.g. "unix:<socket>").
583 Uses env/default socket discovery when omitted.
584 """
585 self._lib = _get_full_lib()
586 self._lock = threading.Lock()
587 self._mainloop = self._lib.pa_threaded_mainloop_new()
588 if not self._mainloop:
589 raise OSError("pa_threaded_mainloop_new returned NULL")
590
591 api = self._lib.pa_threaded_mainloop_get_api(self._mainloop)
592 self._context = self._lib.pa_context_new(api, b"music-assistant-volume")
593 if not self._context:
594 self._lib.pa_threaded_mainloop_free(self._mainloop)
595 self._mainloop = None
596 raise OSError("pa_context_new returned NULL")
597
598 self._ready = threading.Event()
599 self._failed = threading.Event()
600
601 def _state_cb_impl(_ctx: int, _userdata: int) -> None:
602 state = self._lib.pa_context_get_state(self._context)
603 if state == PA_CONTEXT_READY:
604 self._ready.set()
605 elif state in (PA_CONTEXT_FAILED, PA_CONTEXT_TERMINATED):
606 self._failed.set()
607
608 self._state_cb = _CONTEXT_NOTIFY_CB(_state_cb_impl) # keep reference alive â GC
609 self._lib.pa_context_set_state_callback(self._context, self._state_cb, None)
610
611 pulse_server = server or get_default_pulse_server()
612 ret = self._lib.pa_context_connect(
613 self._context,
614 pulse_server.encode() if pulse_server else None,
615 PA_CONTEXT_NOAUTOSPAWN,
616 None,
617 )
618 if ret < 0:
619 self.close()
620 raise OSError(f"pa_context_connect failed (ret={ret})")
621
622 self._lib.pa_threaded_mainloop_start(self._mainloop)
623
624 if not self._ready.wait(timeout=5.0):
625 self.close()
626 raise OSError("Timed out connecting to PulseAudio for volume control")
627
628 def set_sink_volume(self, sink_name: str, volume_pct: int, channels: int = 2) -> bool:
629 """
630 Set hardware volume on a named PA sink.
631
632 :param sink_name: PA sink name as returned by ``enumerate_pa_sinks()``.
633 :param volume_pct: Volume level 0-100, mapped through an exponential
634 audio taper curve before being sent to PA.
635 :param channels: Channel count for the PA volume structure. Should
636 match the sink's actual channel count.
637 :returns: True if PA reported success.
638 """
639 amplitude = volume_pct_to_amplitude(volume_pct)
640 # cube root counteracts PA's own cubic volume curve
641 pa_vol = round(PA_VOLUME_NORM * amplitude ** (1.0 / 3.0))
642 return self._apply_sink_volume(sink_name, pa_vol, channels)
643
644 def set_sink_volume_raw(self, sink_name: str, volume_pct: float, channels: int = 2) -> bool:
645 """
646 Set raw (linear) volume on a named PA sink, without the audio taper.
647
648 :param sink_name: PA sink name.
649 :param volume_pct: Linear percentage where 100 maps exactly onto
650 PA_VOLUME_NORM (0 dB). Values above 100 amplify (e.g. 400 for
651 reciprocal cubic compensation); clamped to MAX_RAW_VOLUME_PCT.
652 :param channels: Channel count for the PA volume structure.
653 :returns: True if PA reported success.
654 """
655 volume_pct = min(max(volume_pct, 0.0), MAX_RAW_VOLUME_PCT)
656 pa_vol = round(PA_VOLUME_NORM * volume_pct / 100.0)
657 return self._apply_sink_volume(sink_name, pa_vol, channels)
658
659 def load_module(self, module_name: str, argument: str) -> int | None:
660 """
661 Load a PulseAudio module (e.g. module-remap-sink) via libpulse.
662
663 :param module_name: PA module name, e.g. "module-remap-sink".
664 :param argument: Module argument string, e.g.
665 "sink_name=Foo master=bar channels=2 master_channel_map=...
666 channel_map=front-left,front-right remix=no".
667
668 Blocks (up to ~2s) for PA's response.
669 :returns: The loaded module's index, or None on failure/timeout.
670 """
671 with self._lock:
672 if self._failed.is_set() or not self._mainloop or not self._context:
673 return None
674
675 done = threading.Event()
676 result: dict[str, int] = {}
677
678 def _index_cb_impl(_ctx: int, idx: int, _userdata: int) -> None:
679 result["index"] = idx
680 done.set()
681
682 index_cb = _CONTEXT_INDEX_CB(_index_cb_impl)
683
684 self._lib.pa_threaded_mainloop_lock(self._mainloop)
685 try:
686 op = self._lib.pa_context_load_module(
687 self._context,
688 module_name.encode(),
689 argument.encode(),
690 index_cb,
691 None,
692 )
693 if not op:
694 return None
695 finally:
696 self._lib.pa_threaded_mainloop_unlock(self._mainloop)
697
698 if not done.wait(timeout=2.0):
699 self._cancel_operation(op)
700 return None
701 self._lib.pa_operation_unref(op)
702 idx = result.get("index", PA_INVALID_INDEX)
703 return None if idx == PA_INVALID_INDEX else idx
704
705 def unload_module(self, module_index: int) -> bool:
706 """
707 Unload a previously-loaded PulseAudio module by index.
708
709 Blocks (up to ~2s) for PA's response.
710 :returns: True if PA reported success.
711 """
712 with self._lock:
713 if self._failed.is_set() or not self._mainloop or not self._context:
714 return False
715
716 done = threading.Event()
717 result: dict[str, int] = {}
718
719 def _success_cb_impl(_ctx: int, success: int, _userdata: int) -> None:
720 result["success"] = success
721 done.set()
722
723 success_cb = _CONTEXT_SUCCESS_CB(_success_cb_impl)
724
725 self._lib.pa_threaded_mainloop_lock(self._mainloop)
726 try:
727 op = self._lib.pa_context_unload_module(
728 self._context, module_index, success_cb, None
729 )
730 if not op:
731 return False
732 finally:
733 self._lib.pa_threaded_mainloop_unlock(self._mainloop)
734
735 if not done.wait(timeout=2.0):
736 self._cancel_operation(op)
737 return False
738 self._lib.pa_operation_unref(op)
739 return bool(result.get("success", 0))
740
741 def close(self) -> None:
742 """Disconnect and tear down the mainloop."""
743 with self._lock:
744 if self._context:
745 self._lib.pa_context_disconnect(self._context)
746 self._lib.pa_context_unref(self._context)
747 self._context = None
748 if self._mainloop:
749 self._lib.pa_threaded_mainloop_stop(self._mainloop)
750 self._lib.pa_threaded_mainloop_free(self._mainloop)
751 self._mainloop = None
752
753 def _cancel_operation(self, op: int) -> None:
754 """
755 Detach a timed-out operation's callback before dropping the reference.
756
757 PA may still deliver the response later from the mainloop thread; without
758 the cancel it would invoke the (garbage-collected) ctypes trampoline of a
759 callback that went out of scope â undefined behavior.
760 """
761 self._lib.pa_operation_cancel(op)
762 self._lib.pa_operation_unref(op)
763
764 def _apply_sink_volume(self, sink_name: str, pa_volume: int, channels: int) -> bool:
765 """
766 Send an already-mapped raw PA volume to a named sink.
767
768 :returns: True if PA reported success.
769 """
770 with self._lock:
771 if self._failed.is_set() or not self._mainloop or not self._context:
772 return False
773 cvol = _PACVolume()
774 self._lib.pa_cvolume_set(ctypes.byref(cvol), channels, pa_volume)
775
776 done = threading.Event()
777 result: dict[str, int] = {}
778
779 def _success_cb_impl(_ctx: int, success: int, _userdata: int) -> None:
780 result["success"] = success
781 done.set()
782
783 success_cb = _CONTEXT_SUCCESS_CB(_success_cb_impl)
784
785 self._lib.pa_threaded_mainloop_lock(self._mainloop)
786 try:
787 op = self._lib.pa_context_set_sink_volume_by_name(
788 self._context,
789 sink_name.encode(),
790 ctypes.byref(cvol),
791 success_cb,
792 None,
793 )
794 if not op:
795 return False
796 finally:
797 self._lib.pa_threaded_mainloop_unlock(self._mainloop)
798
799 if not done.wait(timeout=_SET_VOLUME_TIMEOUT):
800 self._cancel_operation(op)
801 return False
802 self._lib.pa_operation_unref(op)
803 return bool(result.get("success", 0))
804
805
806# One shared server per MusicAssistant instance; weak keys so a discarded mass
807# (tests) does not pin its server object forever.
808_servers: weakref.WeakKeyDictionary[MusicAssistant, PulseCaptureServer] = (
809 weakref.WeakKeyDictionary()
810)
811
812
813class _PACVolume(ctypes.Structure):
814 _fields_: ClassVar = [
815 ("channels", ctypes.c_uint8),
816 ("values", ctypes.c_uint32 * PA_CHANNELS_MAX),
817 ]
818
819
820def _log_supervisor_exit(task: asyncio.Task[None]) -> None:
821 """Surface a supervisor that died on an unexpected error (restarts stop with it)."""
822 if not task.cancelled() and task.exception() is not None:
823 LOGGER.error("PulseAudio capture supervisor died unexpectedly: %s", task.exception())
824
825
826def _close_controller_result(fut: asyncio.Future[PAVolumeController]) -> None:
827 """Close a controller whose awaiting task was cancelled mid-construction."""
828 with suppress(Exception):
829 fut.result().close()
830
831
832def _load_full_lib() -> ctypes.CDLL:
833 """
834 Load and configure libpulse for use by PAVolumeController.
835
836 Called once per process; the result is cached by _get_full_lib().
837 """
838 lib = ctypes.CDLL("libpulse.so.0")
839
840 lib.pa_threaded_mainloop_new.restype = ctypes.c_void_p
841 lib.pa_threaded_mainloop_get_api.restype = ctypes.c_void_p
842 lib.pa_threaded_mainloop_get_api.argtypes = [ctypes.c_void_p]
843 lib.pa_threaded_mainloop_start.restype = ctypes.c_int
844 lib.pa_threaded_mainloop_start.argtypes = [ctypes.c_void_p]
845 lib.pa_threaded_mainloop_stop.argtypes = [ctypes.c_void_p]
846 lib.pa_threaded_mainloop_free.argtypes = [ctypes.c_void_p]
847 lib.pa_threaded_mainloop_lock.argtypes = [ctypes.c_void_p]
848 lib.pa_threaded_mainloop_unlock.argtypes = [ctypes.c_void_p]
849
850 lib.pa_context_new.restype = ctypes.c_void_p
851 lib.pa_context_new.argtypes = [ctypes.c_void_p, ctypes.c_char_p]
852 lib.pa_context_set_state_callback.argtypes = [
853 ctypes.c_void_p,
854 _CONTEXT_NOTIFY_CB,
855 ctypes.c_void_p,
856 ]
857 lib.pa_context_connect.restype = ctypes.c_int
858 lib.pa_context_connect.argtypes = [
859 ctypes.c_void_p,
860 ctypes.c_char_p,
861 ctypes.c_int,
862 ctypes.c_void_p,
863 ]
864 lib.pa_context_get_state.restype = ctypes.c_int
865 lib.pa_context_get_state.argtypes = [ctypes.c_void_p]
866 lib.pa_context_disconnect.argtypes = [ctypes.c_void_p]
867 lib.pa_context_unref.argtypes = [ctypes.c_void_p]
868
869 lib.pa_cvolume_set.restype = ctypes.c_void_p
870 lib.pa_cvolume_set.argtypes = [ctypes.c_void_p, ctypes.c_uint, ctypes.c_uint32]
871
872 lib.pa_context_set_sink_volume_by_name.restype = ctypes.c_void_p
873 lib.pa_context_set_sink_volume_by_name.argtypes = [
874 ctypes.c_void_p,
875 ctypes.c_char_p,
876 ctypes.c_void_p,
877 _CONTEXT_SUCCESS_CB,
878 ctypes.c_void_p,
879 ]
880 lib.pa_operation_unref.argtypes = [ctypes.c_void_p]
881 lib.pa_operation_cancel.argtypes = [ctypes.c_void_p]
882
883 lib.pa_context_load_module.restype = ctypes.c_void_p
884 lib.pa_context_load_module.argtypes = [
885 ctypes.c_void_p,
886 ctypes.c_char_p,
887 ctypes.c_char_p,
888 _CONTEXT_INDEX_CB,
889 ctypes.c_void_p,
890 ]
891 lib.pa_context_unload_module.restype = ctypes.c_void_p
892 lib.pa_context_unload_module.argtypes = [
893 ctypes.c_void_p,
894 ctypes.c_uint32,
895 _CONTEXT_SUCCESS_CB,
896 ctypes.c_void_p,
897 ]
898 return lib
899
900
901_full_lib: ctypes.CDLL | None = None
902
903
904def _get_full_lib() -> ctypes.CDLL:
905 global _full_lib # noqa: PLW0603
906 if _full_lib is None:
907 _full_lib = _load_full_lib()
908 return _full_lib
909