/
/
/
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. HOME goes too: PA writes an
323 # auth cookie there and refuses to start when it is unwritable, as it is for
324 # a container running as a uid with no passwd entry.
325 proc = AsyncProcess(
326 [
327 "pulseaudio",
328 "--daemonize=no",
329 "-n",
330 "--exit-idle-time=-1",
331 f"--file={self._config_path}",
332 ],
333 stderr=True,
334 name="pulse-capture",
335 env={
336 "HOME": str(self._base_dir),
337 "XDG_RUNTIME_DIR": str(self._base_dir),
338 "PULSE_RUNTIME_PATH": str(self._base_dir),
339 "PULSE_STATE_PATH": str(self._base_dir),
340 },
341 )
342 self._proc = proc
343 try:
344 await proc.start()
345 await self._wait_ready(proc)
346 # verify the daemon actually accepts connections before declaring
347 # ready; the construction cannot be interrupted mid-flight, so on
348 # cancellation close whatever the worker thread still produced to
349 # avoid leaking a live threaded mainloop
350 ctor = asyncio.ensure_future(asyncio.to_thread(PAVolumeController, self.server_address))
351 try:
352 controller = await asyncio.shield(ctor)
353 except asyncio.CancelledError:
354 ctor.add_done_callback(_close_controller_result)
355 raise
356 except BaseException:
357 if self._proc is proc:
358 self._proc = None
359 with suppress(Exception):
360 await proc.close()
361 raise
362 # publish the new controller and generation before disposing of the old
363 # controller, so teardown can always reach the live one even when the
364 # disposal await is cancelled
365 old_controller = self._controller
366 self._controller = controller
367 self._generation += 1
368 if old_controller is not None:
369 with suppress(Exception):
370 await asyncio.to_thread(old_controller.close)
371 LOGGER.debug(
372 "Private PulseAudio capture daemon ready on %s (generation %d)",
373 self.server_address,
374 self._generation,
375 )
376
377 async def _wait_ready(self, proc: AsyncProcess) -> None:
378 """Wait for the daemon's native socket to appear."""
379 try:
380 async with asyncio.timeout(_READY_TIMEOUT):
381 while not await asyncio.to_thread(self._socket_path.exists):
382 if proc.returncode is not None:
383 raise RuntimeError(
384 f"pulseaudio exited during startup (code {proc.returncode})"
385 )
386 await asyncio.sleep(_READY_POLL_INTERVAL)
387 except TimeoutError:
388 raise RuntimeError("Timeout waiting for the pulseaudio daemon socket") from None
389
390 async def _supervise(self) -> None:
391 """Restart the daemon (with bounded backoff) until the supervisor is cancelled."""
392 backoff = _RESTART_BACKOFF_INITIAL
393 while True:
394 if (proc := self._proc) is not None:
395 try:
396 async for line in proc.iter_stderr():
397 LOGGER.debug("pulseaudio: %s", line)
398 except Exception as err:
399 LOGGER.debug("pulseaudio log reader stopped: %s", err)
400 await proc.close()
401 if self._proc is proc:
402 self._proc = None
403 LOGGER.warning(
404 "Private PulseAudio capture daemon exited unexpectedly, restarting in %.1fs",
405 backoff,
406 )
407 await asyncio.sleep(backoff)
408 try:
409 await self._launch_daemon()
410 except Exception as err:
411 backoff = min(backoff * 2, _RESTART_BACKOFF_MAX)
412 LOGGER.error("Failed to restart the PulseAudio capture daemon: %s", err)
413 continue
414 backoff = _RESTART_BACKOFF_INITIAL
415
416 async def _load_module(self, module_name: str, argument: str) -> int | None:
417 """Load a PA module on the private daemon (blocking libpulse call in a thread)."""
418 controller = self._require_controller()
419 return await asyncio.to_thread(controller.load_module, module_name, argument)
420
421 async def _unload_module(self, module_index: int) -> bool:
422 """Unload a PA module from the private daemon."""
423 controller = self._require_controller()
424 return await asyncio.to_thread(controller.unload_module, module_index)
425
426 async def _set_sink_volume_raw(self, sink_name: str, volume_pct: float) -> bool:
427 """Set raw (linear) volume on a sink of the private daemon."""
428 controller = self._require_controller()
429 return await asyncio.to_thread(controller.set_sink_volume_raw, sink_name, volume_pct)
430
431 def _require_controller(self) -> PAVolumeController:
432 """Return the connected controller or raise if the server is not running."""
433 if (controller := self._controller) is None:
434 raise RuntimeError("Pulse capture server is not running")
435 return controller
436
437
438class PipeSink:
439 """
440 One isolated module-pipe-sink capture sink on a :class:`PulseCaptureServer`.
441
442 The PA daemon creates a FIFO at :attr:`fifo_path` that delivers the sink's
443 audio as raw PCM in the fixed capture format (CAPTURE_SAMPLE_FORMAT /
444 CAPTURE_SAMPLE_RATE / CAPTURE_CHANNELS); read it with
445 ``helpers.named_pipe.read_named_pipe`` or by pointing ffmpeg at it.
446
447 Both consumer lifecycles are supported: a single long-lived sink per
448 provider instance, or a fresh sink per stream (create -> suspend/resume ->
449 unload). A sink does not survive a daemon restart: when the server's
450 ``generation`` no longer matches the value snapshotted at creation, drop
451 this instance and create a new one.
452 """
453
454 def __init__(
455 self,
456 server: PulseCaptureServer,
457 sink_name: str,
458 fifo_path: Path,
459 module_index: int,
460 generation: int,
461 ) -> None:
462 """Initialize the sink. Use the async :meth:`create` factory instead."""
463 self._server = server
464 self._sink_name = sink_name
465 self._fifo_path = fifo_path
466 self._module_index: int | None = module_index
467 self._generation = generation
468
469 @classmethod
470 async def create(cls, server: PulseCaptureServer, name_prefix: str) -> PipeSink:
471 """
472 Create a new uniquely-named pipe sink on the given capture server.
473
474 The pipe-sink module creates the FIFO file itself.
475
476 :param server: An acquired PulseCaptureServer.
477 :param name_prefix: Prefix for the generated sink name (e.g. a provider
478 instance id); a short unique suffix is appended.
479 """
480 sink_name = f"{name_prefix}_{uuid.uuid4().hex[:8]}"
481 fifo_path = server._base_dir / f"{sink_name}.pcm"
482 argument = (
483 f"sink_name={sink_name} file={fifo_path} "
484 f"format={CAPTURE_SAMPLE_FORMAT} rate={CAPTURE_SAMPLE_RATE} "
485 f"channels={CAPTURE_CHANNELS}"
486 )
487 # snapshot before the load: a restart during the load would otherwise
488 # pair a module index from the dead daemon with the new generation,
489 # and a later unload could hit an unrelated module on the replacement
490 generation = server.generation
491 module_index = await server._load_module("module-pipe-sink", argument)
492 if module_index is None:
493 raise RuntimeError(f"Failed to load module-pipe-sink for {sink_name}")
494 if server.generation != generation:
495 raise RuntimeError(f"capture daemon restarted while creating sink {sink_name}")
496 return cls(server, sink_name, fifo_path, module_index, generation)
497
498 @property
499 def sink_name(self) -> str:
500 """The PA sink name (pass to PulseCaptureServer.child_env for the client)."""
501 return self._sink_name
502
503 @property
504 def fifo_path(self) -> Path:
505 """Path of the FIFO delivering this sink's PCM audio."""
506 return self._fifo_path
507
508 async def set_volume(self, volume_pct: float) -> None:
509 """
510 Set the sink's raw (linear) volume.
511
512 A sink from a previous daemon generation ignores the call (recreate the
513 sink after a restart).
514
515 :param volume_pct: 100 is unity gain; values above 100 amplify (e.g. 400
516 for reciprocal cubic compensation). Clamped to MAX_RAW_VOLUME_PCT.
517 """
518 if self._generation != self._server.generation:
519 LOGGER.debug("Ignoring volume for stale sink %s", self._sink_name)
520 return
521 if not await self._server._set_sink_volume_raw(self._sink_name, volume_pct):
522 raise RuntimeError(f"Failed to set volume on capture sink {self._sink_name}")
523
524 async def suspend(self) -> None:
525 """Suspend the sink (its FIFO stops producing audio until resumed)."""
526 await self._set_suspended(True)
527
528 async def resume(self) -> None:
529 """Resume a suspended sink."""
530 await self._set_suspended(False)
531
532 async def unload(self) -> None:
533 """
534 Unload the sink's module and remove its FIFO (idempotent).
535
536 A sink whose daemon has restarted since creation is already gone; only
537 the leftover FIFO file is cleaned up in that case.
538 """
539 module_index = self._module_index
540 self._module_index = None
541 if module_index is not None and self._generation == self._server.generation:
542 # best effort: a failure usually means the daemon/module is already
543 # gone, and a restart reclaims all modules anyway
544 if not await self._server._unload_module(module_index):
545 LOGGER.warning("Failed to unload capture sink module %s", self._sink_name)
546 with suppress(OSError):
547 await asyncio.to_thread(self._fifo_path.unlink)
548
549 async def _set_suspended(self, suspended: bool) -> None:
550 """Toggle the sink's suspend state via pactl."""
551 if self._generation != self._server.generation:
552 LOGGER.debug("Ignoring suspend toggle for stale sink %s", self._sink_name)
553 return
554 returncode, output = await check_output(
555 "pactl",
556 "--server",
557 self._server.server_address,
558 "suspend-sink",
559 self._sink_name,
560 "1" if suspended else "0",
561 env={"PULSE_SERVER": self._server.server_address},
562 timeout=5,
563 )
564 if returncode != 0:
565 LOGGER.warning(
566 "pactl suspend-sink %s %d failed: %s",
567 self._sink_name,
568 int(suspended),
569 output.decode("utf-8", errors="replace").strip(),
570 )
571
572
573class PAVolumeController:
574 """
575 Shared libpulse connection for PA sink volume and module control.
576
577 One instance is shared per PA server. All calls are blocking and must be
578 invoked via run_in_executor/to_thread from async code.
579 """
580
581 def __init__(self, server: str | None = None) -> None:
582 """
583 Connect to PulseAudio and start the threaded mainloop.
584
585 :param server: PA server address to connect to (e.g. "unix:<socket>").
586 Uses env/default socket discovery when omitted.
587 """
588 self._lib = _get_full_lib()
589 self._lock = threading.Lock()
590 self._mainloop = self._lib.pa_threaded_mainloop_new()
591 if not self._mainloop:
592 raise OSError("pa_threaded_mainloop_new returned NULL")
593
594 api = self._lib.pa_threaded_mainloop_get_api(self._mainloop)
595 self._context = self._lib.pa_context_new(api, b"music-assistant-volume")
596 if not self._context:
597 self._lib.pa_threaded_mainloop_free(self._mainloop)
598 self._mainloop = None
599 raise OSError("pa_context_new returned NULL")
600
601 self._ready = threading.Event()
602 self._failed = threading.Event()
603
604 def _state_cb_impl(_ctx: int, _userdata: int) -> None:
605 state = self._lib.pa_context_get_state(self._context)
606 if state == PA_CONTEXT_READY:
607 self._ready.set()
608 elif state in (PA_CONTEXT_FAILED, PA_CONTEXT_TERMINATED):
609 self._failed.set()
610
611 self._state_cb = _CONTEXT_NOTIFY_CB(_state_cb_impl) # keep reference alive â GC
612 self._lib.pa_context_set_state_callback(self._context, self._state_cb, None)
613
614 pulse_server = server or get_default_pulse_server()
615 ret = self._lib.pa_context_connect(
616 self._context,
617 pulse_server.encode() if pulse_server else None,
618 PA_CONTEXT_NOAUTOSPAWN,
619 None,
620 )
621 if ret < 0:
622 self.close()
623 raise OSError(f"pa_context_connect failed (ret={ret})")
624
625 self._lib.pa_threaded_mainloop_start(self._mainloop)
626
627 if not self._ready.wait(timeout=5.0):
628 self.close()
629 raise OSError("Timed out connecting to PulseAudio for volume control")
630
631 def set_sink_volume(self, sink_name: str, volume_pct: int, channels: int = 2) -> bool:
632 """
633 Set hardware volume on a named PA sink.
634
635 :param sink_name: PA sink name as returned by ``enumerate_pa_sinks()``.
636 :param volume_pct: Volume level 0-100, mapped through an exponential
637 audio taper curve before being sent to PA.
638 :param channels: Channel count for the PA volume structure. Should
639 match the sink's actual channel count.
640 :returns: True if PA reported success.
641 """
642 amplitude = volume_pct_to_amplitude(volume_pct)
643 # cube root counteracts PA's own cubic volume curve
644 pa_vol = round(PA_VOLUME_NORM * amplitude ** (1.0 / 3.0))
645 return self._apply_sink_volume(sink_name, pa_vol, channels)
646
647 def set_sink_volume_raw(self, sink_name: str, volume_pct: float, channels: int = 2) -> bool:
648 """
649 Set raw (linear) volume on a named PA sink, without the audio taper.
650
651 :param sink_name: PA sink name.
652 :param volume_pct: Linear percentage where 100 maps exactly onto
653 PA_VOLUME_NORM (0 dB). Values above 100 amplify (e.g. 400 for
654 reciprocal cubic compensation); clamped to MAX_RAW_VOLUME_PCT.
655 :param channels: Channel count for the PA volume structure.
656 :returns: True if PA reported success.
657 """
658 volume_pct = min(max(volume_pct, 0.0), MAX_RAW_VOLUME_PCT)
659 pa_vol = round(PA_VOLUME_NORM * volume_pct / 100.0)
660 return self._apply_sink_volume(sink_name, pa_vol, channels)
661
662 def load_module(self, module_name: str, argument: str) -> int | None:
663 """
664 Load a PulseAudio module (e.g. module-remap-sink) via libpulse.
665
666 :param module_name: PA module name, e.g. "module-remap-sink".
667 :param argument: Module argument string, e.g.
668 "sink_name=Foo master=bar channels=2 master_channel_map=...
669 channel_map=front-left,front-right remix=no".
670
671 Blocks (up to ~2s) for PA's response.
672 :returns: The loaded module's index, or None on failure/timeout.
673 """
674 with self._lock:
675 if self._failed.is_set() or not self._mainloop or not self._context:
676 return None
677
678 done = threading.Event()
679 result: dict[str, int] = {}
680
681 def _index_cb_impl(_ctx: int, idx: int, _userdata: int) -> None:
682 result["index"] = idx
683 done.set()
684
685 index_cb = _CONTEXT_INDEX_CB(_index_cb_impl)
686
687 self._lib.pa_threaded_mainloop_lock(self._mainloop)
688 try:
689 op = self._lib.pa_context_load_module(
690 self._context,
691 module_name.encode(),
692 argument.encode(),
693 index_cb,
694 None,
695 )
696 if not op:
697 return None
698 finally:
699 self._lib.pa_threaded_mainloop_unlock(self._mainloop)
700
701 if not done.wait(timeout=2.0):
702 self._cancel_operation(op)
703 return None
704 self._lib.pa_operation_unref(op)
705 idx = result.get("index", PA_INVALID_INDEX)
706 return None if idx == PA_INVALID_INDEX else idx
707
708 def unload_module(self, module_index: int) -> bool:
709 """
710 Unload a previously-loaded PulseAudio module by index.
711
712 Blocks (up to ~2s) for PA's response.
713 :returns: True if PA reported success.
714 """
715 with self._lock:
716 if self._failed.is_set() or not self._mainloop or not self._context:
717 return False
718
719 done = threading.Event()
720 result: dict[str, int] = {}
721
722 def _success_cb_impl(_ctx: int, success: int, _userdata: int) -> None:
723 result["success"] = success
724 done.set()
725
726 success_cb = _CONTEXT_SUCCESS_CB(_success_cb_impl)
727
728 self._lib.pa_threaded_mainloop_lock(self._mainloop)
729 try:
730 op = self._lib.pa_context_unload_module(
731 self._context, module_index, success_cb, None
732 )
733 if not op:
734 return False
735 finally:
736 self._lib.pa_threaded_mainloop_unlock(self._mainloop)
737
738 if not done.wait(timeout=2.0):
739 self._cancel_operation(op)
740 return False
741 self._lib.pa_operation_unref(op)
742 return bool(result.get("success", 0))
743
744 def close(self) -> None:
745 """Disconnect and tear down the mainloop."""
746 with self._lock:
747 if self._context:
748 self._lib.pa_context_disconnect(self._context)
749 self._lib.pa_context_unref(self._context)
750 self._context = None
751 if self._mainloop:
752 self._lib.pa_threaded_mainloop_stop(self._mainloop)
753 self._lib.pa_threaded_mainloop_free(self._mainloop)
754 self._mainloop = None
755
756 def _cancel_operation(self, op: int) -> None:
757 """
758 Detach a timed-out operation's callback before dropping the reference.
759
760 PA may still deliver the response later from the mainloop thread; without
761 the cancel it would invoke the (garbage-collected) ctypes trampoline of a
762 callback that went out of scope â undefined behavior.
763 """
764 self._lib.pa_operation_cancel(op)
765 self._lib.pa_operation_unref(op)
766
767 def _apply_sink_volume(self, sink_name: str, pa_volume: int, channels: int) -> bool:
768 """
769 Send an already-mapped raw PA volume to a named sink.
770
771 :returns: True if PA reported success.
772 """
773 with self._lock:
774 if self._failed.is_set() or not self._mainloop or not self._context:
775 return False
776 cvol = _PACVolume()
777 self._lib.pa_cvolume_set(ctypes.byref(cvol), channels, pa_volume)
778
779 done = threading.Event()
780 result: dict[str, int] = {}
781
782 def _success_cb_impl(_ctx: int, success: int, _userdata: int) -> None:
783 result["success"] = success
784 done.set()
785
786 success_cb = _CONTEXT_SUCCESS_CB(_success_cb_impl)
787
788 self._lib.pa_threaded_mainloop_lock(self._mainloop)
789 try:
790 op = self._lib.pa_context_set_sink_volume_by_name(
791 self._context,
792 sink_name.encode(),
793 ctypes.byref(cvol),
794 success_cb,
795 None,
796 )
797 if not op:
798 return False
799 finally:
800 self._lib.pa_threaded_mainloop_unlock(self._mainloop)
801
802 if not done.wait(timeout=_SET_VOLUME_TIMEOUT):
803 self._cancel_operation(op)
804 return False
805 self._lib.pa_operation_unref(op)
806 return bool(result.get("success", 0))
807
808
809# One shared server per MusicAssistant instance; weak keys so a discarded mass
810# (tests) does not pin its server object forever.
811_servers: weakref.WeakKeyDictionary[MusicAssistant, PulseCaptureServer] = (
812 weakref.WeakKeyDictionary()
813)
814
815
816class _PACVolume(ctypes.Structure):
817 _fields_: ClassVar = [
818 ("channels", ctypes.c_uint8),
819 ("values", ctypes.c_uint32 * PA_CHANNELS_MAX),
820 ]
821
822
823def _log_supervisor_exit(task: asyncio.Task[None]) -> None:
824 """Surface a supervisor that died on an unexpected error (restarts stop with it)."""
825 if not task.cancelled() and task.exception() is not None:
826 LOGGER.error("PulseAudio capture supervisor died unexpectedly: %s", task.exception())
827
828
829def _close_controller_result(fut: asyncio.Future[PAVolumeController]) -> None:
830 """Close a controller whose awaiting task was cancelled mid-construction."""
831 with suppress(Exception):
832 fut.result().close()
833
834
835def _load_full_lib() -> ctypes.CDLL:
836 """
837 Load and configure libpulse for use by PAVolumeController.
838
839 Called once per process; the result is cached by _get_full_lib().
840 """
841 lib = ctypes.CDLL("libpulse.so.0")
842
843 lib.pa_threaded_mainloop_new.restype = ctypes.c_void_p
844 lib.pa_threaded_mainloop_get_api.restype = ctypes.c_void_p
845 lib.pa_threaded_mainloop_get_api.argtypes = [ctypes.c_void_p]
846 lib.pa_threaded_mainloop_start.restype = ctypes.c_int
847 lib.pa_threaded_mainloop_start.argtypes = [ctypes.c_void_p]
848 lib.pa_threaded_mainloop_stop.argtypes = [ctypes.c_void_p]
849 lib.pa_threaded_mainloop_free.argtypes = [ctypes.c_void_p]
850 lib.pa_threaded_mainloop_lock.argtypes = [ctypes.c_void_p]
851 lib.pa_threaded_mainloop_unlock.argtypes = [ctypes.c_void_p]
852
853 lib.pa_context_new.restype = ctypes.c_void_p
854 lib.pa_context_new.argtypes = [ctypes.c_void_p, ctypes.c_char_p]
855 lib.pa_context_set_state_callback.argtypes = [
856 ctypes.c_void_p,
857 _CONTEXT_NOTIFY_CB,
858 ctypes.c_void_p,
859 ]
860 lib.pa_context_connect.restype = ctypes.c_int
861 lib.pa_context_connect.argtypes = [
862 ctypes.c_void_p,
863 ctypes.c_char_p,
864 ctypes.c_int,
865 ctypes.c_void_p,
866 ]
867 lib.pa_context_get_state.restype = ctypes.c_int
868 lib.pa_context_get_state.argtypes = [ctypes.c_void_p]
869 lib.pa_context_disconnect.argtypes = [ctypes.c_void_p]
870 lib.pa_context_unref.argtypes = [ctypes.c_void_p]
871
872 lib.pa_cvolume_set.restype = ctypes.c_void_p
873 lib.pa_cvolume_set.argtypes = [ctypes.c_void_p, ctypes.c_uint, ctypes.c_uint32]
874
875 lib.pa_context_set_sink_volume_by_name.restype = ctypes.c_void_p
876 lib.pa_context_set_sink_volume_by_name.argtypes = [
877 ctypes.c_void_p,
878 ctypes.c_char_p,
879 ctypes.c_void_p,
880 _CONTEXT_SUCCESS_CB,
881 ctypes.c_void_p,
882 ]
883 lib.pa_operation_unref.argtypes = [ctypes.c_void_p]
884 lib.pa_operation_cancel.argtypes = [ctypes.c_void_p]
885
886 lib.pa_context_load_module.restype = ctypes.c_void_p
887 lib.pa_context_load_module.argtypes = [
888 ctypes.c_void_p,
889 ctypes.c_char_p,
890 ctypes.c_char_p,
891 _CONTEXT_INDEX_CB,
892 ctypes.c_void_p,
893 ]
894 lib.pa_context_unload_module.restype = ctypes.c_void_p
895 lib.pa_context_unload_module.argtypes = [
896 ctypes.c_void_p,
897 ctypes.c_uint32,
898 _CONTEXT_SUCCESS_CB,
899 ctypes.c_void_p,
900 ]
901 return lib
902
903
904_full_lib: ctypes.CDLL | None = None
905
906
907def _get_full_lib() -> ctypes.CDLL:
908 global _full_lib # noqa: PLW0603
909 if _full_lib is None:
910 _full_lib = _load_full_lib()
911 return _full_lib
912