/
/
/
1"""AirPlay Receiver plugin provider implementation."""
2
3from __future__ import annotations
4
5import asyncio
6import hashlib
7import os
8import re
9import time
10from contextlib import suppress
11from dataclasses import dataclass, field
12from functools import partial
13from typing import TYPE_CHECKING, Any, cast
14
15from music_assistant_models.enums import (
16 ContentType,
17 EventType,
18 ImageType,
19 MediaType,
20 ProviderFeature,
21 SourceControl,
22 StreamType,
23)
24from music_assistant_models.errors import (
25 AudioError,
26 MediaNotFoundError,
27 UnsupportedFeaturedException,
28)
29from music_assistant_models.media_items import (
30 AudioFormat,
31 AudioSource,
32 MediaItemImage,
33 ProviderMapping,
34)
35from music_assistant_models.streamdetails import StreamDetails, StreamMetadata
36
37from music_assistant.constants import VERBOSE_LOG_LEVEL
38from music_assistant.helpers.config_entries import (
39 CONF_CONNECTED_PLAYERS,
40 CONF_PUBLISH_NAME_TEMPLATE,
41 create_connected_players_entry,
42 create_publish_name_template_entry,
43 resolve_publish_name,
44)
45from music_assistant.helpers.named_pipe import AsyncNamedPipeWriter
46from music_assistant.helpers.process import AsyncProcess, check_output
47from music_assistant.helpers.util import interface_name_for_ip
48from music_assistant.models.plugin import PluginProvider, SourceControlValue
49from music_assistant.providers.airplay_receiver.helpers import get_shairport_sync_binary
50from music_assistant.providers.airplay_receiver.metadata import MetadataReader
51
52if TYPE_CHECKING:
53 from collections.abc import Callable, Iterable
54
55 from music_assistant_models.config_entries import ConfigEntry, ProviderConfig
56 from music_assistant_models.event import MassEvent
57 from music_assistant_models.provider import ProviderManifest
58
59 from music_assistant.mass import MusicAssistant
60 from music_assistant.models.player import Player
61
62SUPPORTED_FEATURES = {ProviderFeature.AUDIO_SOURCE}
63
64# seconds the silence nudge waits for the audio pipe's consumer to reattach
65AUDIO_PIPE_READER_TIMEOUT = 1.0
66
67
68def airplay_receiver_ports(instance_id: str, player_ids: Iterable[str]) -> dict[str, int]:
69 """
70 Return the AirPlay port used for each connected player of a receiver instance.
71
72 Deterministically derived from the instance id and player id, so the ports stay
73 the same across server restarts (Python's built-in ``hash()`` is salted per
74 process). Colliding derivations probe upwards deterministically, staying within
75 the 7000-7999 AirPlay 2 range.
76
77 :param instance_id: The provider instance id of the AirPlay receiver.
78 :param player_ids: The connected player ids to derive ports for.
79 """
80 unique_player_ids = sorted(set(player_ids))
81 if len(unique_player_ids) > 1000:
82 # cannot happen through the UI; guards the probing loop against a
83 # malformed stored value hanging startup
84 raise ValueError("More connected players than available AirPlay ports")
85 ports: dict[str, int] = {}
86 claimed: set[int] = set()
87 # iterate sorted so probing resolves collisions the same way for any input order
88 for player_id in unique_player_ids:
89 digest = hashlib.md5(
90 f"{instance_id}_{player_id}".encode(), usedforsecurity=False
91 ).hexdigest()
92 port = 7000 + int(digest, 16) % 1000
93 while port in claimed:
94 port = 7000 + (port - 7000 + 1) % 1000
95 claimed.add(port)
96 ports[player_id] = port
97 return ports
98
99
100@dataclass
101class _ReceiverDaemon:
102 """State for one connected player's shairport-sync receiver."""
103
104 # the connected player this receiver plays on; doubles as the AudioSource item_id
105 player_id: str
106 # player_id sanitized for use in filesystem paths
107 safe_player_id: str
108 # the name this receiver advertises in the AirPlay device list
109 airplay_name: str
110 port: int
111 audio_pipe: AsyncNamedPipeWriter
112 metadata_pipe: AsyncNamedPipeWriter
113 config_file: str
114 audio_source: AudioSource
115 stream_metadata: StreamMetadata
116 shairport_proc: AsyncProcess | None = None
117 runner_task: asyncio.Task[None] | None = None
118 started: asyncio.Event = field(default_factory=asyncio.Event)
119 metadata_reader: MetadataReader | None = None
120 runner_error_count: int = 0
121 stop_called: bool = False
122 # Currently active player (the one currently playing or selected)
123 active_player_id: str | None = None
124 # in_use_by_player: the queue currently streaming us. Claimed in
125 # on_source_selected (NOT in get_stream_details â that path also runs
126 # from queue preload, where claiming would block a later cross-queue
127 # handoff). Released in on_source_unselected when the session id
128 # matches, or in _clear_active_player on external session disconnect.
129 in_use_by_player: str | None = None
130 # active_session_id is the controller-provided token for the current
131 # stream request â used to reject stale on_source_unselected callbacks
132 # after a same-queue reconnect supersedes the previous request.
133 active_session_id: str | None = None
134 pending_stop_task: asyncio.Task[None] | None = None
135 # the in-flight externally-triggered playback start (awaits the pending stop)
136 pending_start_task: asyncio.Task[None] | None = None
137 first_volume_event_received: bool = False # Track if we've received the first volume event
138
139 def cover_art_path(self, img_hash: str) -> str:
140 """
141 Return the provider-scoped image path for this receiver's cover art.
142
143 :param img_hash: Content hash of the current artwork bytes.
144 """
145 # the player id keeps simultaneous sessions on different receivers from
146 # serving each other's artwork through the single provider instance
147 return f"cover_art_{self.safe_player_id}_{img_hash}"
148
149
150class AirPlayReceiverProvider(PluginProvider):
151 """Implementation of an AirPlay Receiver Plugin."""
152
153 reload_on_streams_network_change = True
154
155 def __init__(
156 self, mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
157 ) -> None:
158 """Initialize MusicProvider."""
159 super().__init__(mass, manifest, config, SUPPORTED_FEATURES)
160 self._shairport_bin: str | None = None
161 self._daemons: dict[str, _ReceiverDaemon] = {}
162 # players whose daemon gave up permanently; skipped by reconcile until the
163 # player re-registers or the provider reloads
164 self._failed_player_ids: set[str] = set()
165 self._reconcile_lock = asyncio.Lock()
166 self._unload_called = False
167 self._unsubscribe: Callable[[], None] | None = None
168 # the connected players are immutable per load: config changes reload the provider
169 self._assigned_player_ids: tuple[str, ...] = tuple(
170 cast("list[str]", self.get_config_value(CONF_CONNECTED_PLAYERS) or [])
171 )
172 # One unique AirPlay 2 (7000+) port per connected player. The ports must be
173 # stable across restarts: the AirPlay provider uses them to recognize (and
174 # ignore) our own shairport-sync advertisements in discovery.
175 self._ports = airplay_receiver_ports(self.instance_id, self._assigned_player_ids)
176 # _audio_format describes the original AirPlay source (ALAC at 44.1/16,
177 # the protocol-native format AirPlay senders use) and is what we
178 # advertise to clients for source-format display.
179 self._audio_format = AudioFormat(
180 content_type=ContentType.ALAC,
181 codec_type=ContentType.ALAC,
182 sample_rate=44100,
183 bit_depth=16,
184 channels=2,
185 )
186 # _decoded_audio_format is what shairport-sync actually pipes into MA
187 # after decoding the ALAC stream; the streams controller hands this to
188 # ffmpeg as the input format so it can read the FIFO correctly.
189 self._decoded_audio_format = AudioFormat(
190 content_type=ContentType.PCM_S16LE,
191 codec_type=ContentType.PCM_S16LE,
192 sample_rate=44100,
193 bit_depth=16,
194 channels=2,
195 )
196
197 @property
198 def airplay_ports(self) -> set[int]:
199 """Return the AirPlay ports of the currently running receiver daemons."""
200 return {daemon.port for daemon in self._daemons.values()}
201
202 async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
203 """Return runtime options for this provider."""
204 return (
205 create_connected_players_entry(
206 self.mass, cast("list[str]", self.get_config_value(CONF_CONNECTED_PLAYERS) or [])
207 ),
208 create_publish_name_template_entry(self.get_config_value(CONF_PUBLISH_NAME_TEMPLATE)),
209 )
210
211 async def handle_async_init(self) -> None:
212 """Handle async initialization of the provider."""
213 self._shairport_bin = await get_shairport_sync_binary()
214
215 async def loaded_in_mass(self) -> None:
216 """Start the receiver daemons and follow the connected players' lifecycle."""
217 await super().loaded_in_mass()
218 if self._assigned_player_ids:
219 self._unsubscribe = self.mass.subscribe(
220 self._on_player_event,
221 event_filter=(
222 EventType.PLAYER_ADDED,
223 EventType.PLAYER_REMOVED,
224 EventType.PLAYER_CONFIG_UPDATED,
225 EventType.PLAYER_UPDATED,
226 ),
227 id_filter=self._assigned_player_ids,
228 )
229 # players register after plugins load, so on a cold boot this typically starts
230 # nothing yet: the PLAYER_ADDED events drive the actual daemon startups
231 await self._reconcile()
232
233 async def unload(self, is_removed: bool = False) -> None:
234 """Handle close/cleanup of the provider."""
235 self._unload_called = True
236 if self._unsubscribe is not None:
237 self._unsubscribe()
238 self._unsubscribe = None
239 async with self._reconcile_lock:
240 daemons = list(self._daemons.values())
241 self._daemons.clear()
242 if daemons:
243 await asyncio.gather(*(self._stop_receiver(daemon) for daemon in daemons))
244 # drop the standing source entries from the players' cached source lists
245 for daemon in daemons:
246 self.mass.players.trigger_player_update(daemon.player_id)
247
248 async def get_audio_sources(self) -> list[AudioSource]:
249 """Return the AudioSources this plugin currently exposes."""
250 return [daemon.audio_source for daemon in self._daemons.values()]
251
252 def get_player_audio_sources(self, player_id: str) -> list[AudioSource]:
253 """Return the AudioSource bound to the given connected player, if any."""
254 daemon = self._daemons.get(player_id)
255 return [daemon.audio_source] if daemon else []
256
257 async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
258 """
259 Return StreamDetails for streaming the AirPlay audio to a queue.
260
261 Side-effect-free: ownership is claimed in on_source_selected (which the
262 streams controller fires before this method on the actual stream
263 request). Keeping this idempotent means preload paths like
264 player_queues._load_item can fetch streamdetails without claiming the
265 source and blocking a subsequent cross-queue handoff.
266
267 Raises AudioError when no AirPlay client is currently connected.
268 """
269 daemon = self._daemons.get(item_id)
270 if daemon is None:
271 raise MediaNotFoundError(f"Unknown AudioSource: {item_id}")
272 if not daemon.active_player_id:
273 raise AudioError(
274 "AirPlay receiver has no active client â start playback from your "
275 "AirPlay-capable device first"
276 )
277 return StreamDetails(
278 provider=self.instance_id,
279 item_id=item_id,
280 audio_format=self._audio_format,
281 decoded_audio_format=self._decoded_audio_format,
282 media_type=MediaType.AUDIO_SOURCE,
283 stream_type=StreamType.NAMED_PIPE,
284 path=daemon.audio_pipe.path,
285 stream_metadata=daemon.stream_metadata,
286 )
287
288 async def on_source_control(
289 self,
290 source_id: str,
291 action: SourceControl,
292 value: SourceControlValue = None,
293 ) -> None:
294 """
295 Handle source control commands (no-op: AirPlay receiver is passive).
296
297 The AudioSource advertises no control capabilities, so MA will not invoke
298 any actions here. Override exists only to satisfy the contract.
299 """
300 del source_id, action
301
302 async def on_source_selected(
303 self,
304 source_id: str,
305 player_id: str,
306 owner_player_id: str,
307 stream_session_id: str,
308 ) -> None:
309 """Handle callback when this AudioSource is selected/started on a player."""
310 daemon = self._daemons.get(source_id)
311 if daemon is None or not player_id:
312 return
313
314 # Cache the owner_player_id (user-facing MA player) rather than the protocol-
315 # level player_id; protocol bridges (e.g. Sendspin's spb_â¦) can tear
316 # down between streams and their ID is then invalid for play_media.
317 active_player_id = owner_player_id
318
319 # If there's already an active player and it's different, kick it out.
320 # The lock claim a few lines below replaces the previous queue's claim;
321 # the prior stream's on_source_unselected may fire later, but its
322 # session-id guard keeps it from clobbering the new claim.
323 if daemon.active_player_id and daemon.active_player_id != active_player_id:
324 prev_player_id = daemon.active_player_id
325 self.logger.info(
326 "Source selected on player %s, stopping playback on %s",
327 active_player_id,
328 prev_player_id,
329 )
330 try:
331 await self.mass.players.cmd_stop(prev_player_id)
332 except Exception as err:
333 self.logger.debug("Failed to stop previous player %s: %s", prev_player_id, err)
334
335 # Claim ownership for this queue. The lock lives here (not in
336 # get_stream_details) so preload paths can fetch streamdetails without
337 # accidentally blocking a subsequent cross-queue handoff at the actual
338 # stream request.
339 daemon.in_use_by_player = owner_player_id
340 # Record this request's session id so a later on_source_unselected can
341 # tell whether it is the live teardown or a stale callback from a
342 # superseded same-queue request.
343 daemon.active_session_id = stream_session_id
344
345 # Update the active player
346 daemon.active_player_id = active_player_id
347 self.logger.debug("Active player set to: %s", active_player_id)
348
349 async def on_source_unselected(
350 self, source_id: str, owner_player_id: str, stream_session_id: str
351 ) -> None:
352 """Release the queue-scoped exclusive claim when MA tears down the stream."""
353 daemon = self._daemons.get(source_id)
354 if daemon is None:
355 return
356 # Reject stale callbacks: only release if this is still the active
357 # session. A owner_player_id check alone is not sufficient â same-queue
358 # reconnects (player drops + reopens the same stream URL before the
359 # original request's finally fires) would otherwise let the old
360 # request's late callback clear the live claim of the new stream.
361 if daemon.active_session_id != stream_session_id:
362 return
363 daemon.active_session_id = None
364 if daemon.in_use_by_player == owner_player_id:
365 daemon.in_use_by_player = None
366
367 async def resolve_image(self, path: str) -> bytes:
368 """
369 Resolve an image from an image path.
370
371 This returns raw bytes of the cover art image received from AirPlay metadata.
372
373 :param path: The image path, carrying the receiver's player id and the
374 current cover art content hash suffix.
375 """
376 for daemon in self._daemons.values():
377 if not (daemon.metadata_reader and daemon.metadata_reader.cover_art_bytes):
378 continue
379 current_hash = hashlib.md5(
380 daemon.metadata_reader.cover_art_bytes, usedforsecurity=False
381 ).hexdigest()[:8]
382 # Only serve when the suffix matches the current artwork's hash, so a
383 # stale request can't cache new bytes under an old hash key.
384 if path == daemon.cover_art_path(current_hash):
385 return daemon.metadata_reader.cover_art_bytes
386 return b""
387
388 async def _on_player_event(self, event: MassEvent) -> None:
389 """Reconcile the receiver daemons after a connected player's lifecycle event."""
390 if self._unload_called:
391 return
392 if event.event == EventType.PLAYER_REMOVED:
393 # permanent removal: stop the daemon; a temporarily unavailable player
394 # (which fires only PLAYER_UPDATED) keeps its running daemon so the
395 # advertised device identity stays stable across the outage
396 async with self._reconcile_lock:
397 if event.object_id and (daemon := self._daemons.pop(event.object_id, None)):
398 # the session may be consumed by ANOTHER player (cross-select or
399 # sync-group owner); release it so that player is not left bound
400 # to a source that can no longer stream
401 self._clear_active_player(daemon)
402 await self._stop_receiver(daemon)
403 return
404 if event.event == EventType.PLAYER_ADDED and event.object_id:
405 # a re-registered player earns a permanently failed daemon a fresh start
406 self._failed_player_ids.discard(event.object_id)
407 await self._reconcile()
408
409 async def _reconcile(self) -> None:
410 """
411 Align the running receiver daemons with the connected players.
412
413 Starts a daemon for every connected player that is registered, and restarts
414 a daemon whose advertised name drifted from the player's current name.
415 """
416 async with self._reconcile_lock:
417 if self._unload_called:
418 return
419 template = self.get_config_value(CONF_PUBLISH_NAME_TEMPLATE)
420 for player_id in self._assigned_player_ids:
421 if player_id in self._failed_player_ids:
422 # this daemon gave up permanently; blocked from restarts until the
423 # player re-registers or the provider reloads
424 continue
425 player = self.mass.players.get_player(player_id)
426 if player is None:
427 # not (yet) registered: never start a daemon for it; an already
428 # running one is deliberately kept (see _on_player_event)
429 continue
430 airplay_name = resolve_publish_name(template, player.display_name)
431 daemon = self._daemons.get(player_id)
432 if daemon is not None and daemon.airplay_name == airplay_name:
433 continue
434 if daemon is not None:
435 # the advertised name follows the player name: restart on rename.
436 # A live session is released first so the consuming player's queue
437 # is not left held by a source the replaced daemon cannot stream
438 # (unload and player removal already release via the controller).
439 self._clear_active_player(daemon)
440 del self._daemons[player_id]
441 await self._stop_receiver(daemon)
442 self._start_receiver(player, airplay_name)
443 # the standing source entry feeds the player's cached source list
444 self.mass.players.trigger_player_update(player_id)
445
446 def _start_receiver(self, player: Player, airplay_name: str) -> None:
447 """
448 Create the receiver state for a connected player and start its daemon.
449
450 :param player: The (registered) player this receiver plays on.
451 :param airplay_name: The name to advertise in the AirPlay device list.
452 """
453 player_id = player.player_id
454 safe_player_id = re.sub(r"[^A-Za-z0-9_.-]", "_", player_id)
455 receiver_key = f"{self.instance_id}_{safe_player_id}"
456 audio_source = AudioSource(
457 # the player id is stable across renames, so the source uri survives them
458 item_id=player_id,
459 provider=self.instance_id,
460 name=f"{self.name} ({player.display_name})",
461 provider_mappings={
462 ProviderMapping(
463 item_id=player_id,
464 provider_domain=self.domain,
465 provider_instance=self.instance_id,
466 audio_format=self._audio_format,
467 )
468 },
469 can_play_pause=False,
470 can_seek=False,
471 can_next_previous=False,
472 exclusive=True,
473 allow_external_trigger=True,
474 # passive: only flows when an external AirPlay client is connected
475 can_initiate=False,
476 )
477 daemon = _ReceiverDaemon(
478 player_id=player_id,
479 safe_player_id=safe_player_id,
480 airplay_name=airplay_name,
481 port=self._ports[player_id],
482 audio_pipe=AsyncNamedPipeWriter(f"/tmp/ma_airplay_audio_{receiver_key}"), # noqa: S108
483 metadata_pipe=AsyncNamedPipeWriter(
484 f"/tmp/ma_airplay_metadata_{receiver_key}" # noqa: S108
485 ),
486 config_file=f"/tmp/ma_shairport_sync_{receiver_key}.conf", # noqa: S108
487 audio_source=audio_source,
488 stream_metadata=StreamMetadata(title=f"AirPlay | {airplay_name}"),
489 )
490 self._daemons[player_id] = daemon
491 self._setup_shairport_daemon(daemon)
492
493 async def _stop_receiver(self, daemon: _ReceiverDaemon) -> None:
494 """Stop a receiver's shairport-sync daemon and release its resources."""
495 daemon.stop_called = True
496
497 # a pending stop or deferred start must not wake after the teardown and
498 # act on the replaced daemon's session; cancel BOTH before awaiting either
499 # (awaiting first could let the other progress meanwhile), and await so a
500 # cancellation-delayed player command cannot land after the daemon is gone
501 pending_tasks = [
502 task
503 for task in (daemon.pending_stop_task, daemon.pending_start_task)
504 if task and not task.done()
505 ]
506 for pending_task in pending_tasks:
507 pending_task.cancel()
508 for pending_task in pending_tasks:
509 with suppress(asyncio.CancelledError):
510 await pending_task
511 daemon.pending_stop_task = None
512 daemon.pending_start_task = None
513
514 # Stop metadata reader
515 if daemon.metadata_reader:
516 await daemon.metadata_reader.stop()
517 daemon.metadata_reader = None
518
519 # Stop shairport-sync process
520 if daemon.runner_task and not daemon.runner_task.done():
521 daemon.runner_task.cancel()
522 with suppress(asyncio.CancelledError):
523 await daemon.runner_task
524 daemon.runner_task = None
525
526 # Reset the shairport process reference
527 daemon.shairport_proc = None
528 daemon.started.clear()
529
530 async def _give_up_receiver(self, daemon: _ReceiverDaemon, error: str) -> None:
531 """
532 Permanently stop a single failed receiver, leaving the other receivers running.
533
534 :param daemon: The receiver whose daemon failed permanently.
535 :param error: The daemon's failure description.
536 """
537 async with self._reconcile_lock:
538 # a rename restart, player removal or unload may have replaced or
539 # stopped the receiver meanwhile; only the live one gives up
540 if self._daemons.get(daemon.player_id) is not daemon:
541 return
542 del self._daemons[daemon.player_id]
543 self._failed_player_ids.add(daemon.player_id)
544 self.logger.warning(
545 "Giving up on AirPlay receiver '%s' for player %s: %s "
546 "Other players are unaffected; reload the provider to retry.",
547 daemon.airplay_name,
548 daemon.player_id,
549 error,
550 )
551 # release a consuming player so it is not left bound to a dead source
552 self._clear_active_player(daemon)
553 await self._stop_receiver(daemon)
554 # drop the standing source entry from the player's cached source list
555 self.mass.players.trigger_player_update(daemon.player_id)
556
557 def _setup_shairport_daemon(self, daemon: _ReceiverDaemon) -> None:
558 """Handle setup of the shairport-sync daemon for a receiver."""
559 # a delayed restart can fire after the receiver was stopped or replaced
560 if daemon.stop_called or self._daemons.get(daemon.player_id) is not daemon:
561 return
562 daemon.started.clear()
563 daemon.runner_task = self.mass.create_task(self._shairport_runner(daemon))
564
565 async def _shairport_runner(self, daemon: _ReceiverDaemon) -> None:
566 """Run a receiver's shairport-sync daemon in a background task."""
567 assert self._shairport_bin
568 self.logger.info("Starting AirPlay Receiver background daemon for %s", daemon.airplay_name)
569 await self._setup_pipes_and_config(daemon)
570
571 try:
572 args: list[str] = [
573 self._shairport_bin,
574 "--configfile",
575 daemon.config_file,
576 ]
577 daemon.shairport_proc = shairport = AsyncProcess(
578 args, stderr=True, name=f"shairport-sync[{daemon.airplay_name}]"
579 )
580
581 # Open the FIFO before shairport-sync can invoke session-control hooks.
582 daemon.metadata_reader = MetadataReader(
583 daemon.metadata_pipe.path, self.logger, partial(self._on_metadata_update, daemon)
584 )
585 await daemon.metadata_reader.start()
586
587 await shairport.start()
588
589 # Check if process started successfully
590 await asyncio.sleep(0.1)
591 if shairport.returncode is not None:
592 self.logger.error(
593 "shairport-sync exited immediately with code %s", shairport.returncode
594 )
595 return
596
597 # Keep reading logging from stderr until exit
598 self.logger.debug("Starting to read shairport-sync stderr")
599 async for stderr_line in shairport.iter_stderr():
600 line = stderr_line.strip()
601 self._process_shairport_log_line(daemon, line)
602
603 finally:
604 await shairport.close()
605 self.logger.info(
606 "AirPlay Receiver background daemon stopped for %s (exit code: %s)",
607 daemon.airplay_name,
608 shairport.returncode,
609 )
610
611 # Stop metadata reader
612 if daemon.metadata_reader:
613 await daemon.metadata_reader.stop()
614
615 # Clean up pipes and config
616 await self._cleanup_pipes_and_config(daemon)
617
618 self._handle_runner_exit(daemon)
619
620 def _handle_runner_exit(self, daemon: _ReceiverDaemon) -> None:
621 """
622 Decide how to follow up on a receiver's daemon exit (restart, give up or unload).
623
624 :param daemon: The receiver daemon whose shairport-sync process exited.
625 """
626 if daemon.stop_called:
627 # deliberately stopped (unload, rename restart or player removal)
628 return
629 if not daemon.started.is_set():
630 # the binary produced no output at all: an environment-level problem
631 # every receiver would hit alike
632 self.unload_with_error("Unable to initialize shairport-sync daemon.")
633 # Auto restart if not stopped manually
634 elif daemon.runner_error_count >= 5:
635 # deferred task (no eager start): this runs on the daemon's own
636 # runner task, which the give-up is about to stop
637 self.mass.create_task(
638 self._give_up_receiver(
639 daemon, "shairport-sync daemon failed to start multiple times."
640 ),
641 eager_start=False,
642 )
643 else:
644 daemon.runner_error_count += 1
645 self.mass.call_later(2, self._setup_shairport_daemon, daemon)
646
647 def _process_shairport_log_line(self, daemon: _ReceiverDaemon, line: str) -> None:
648 """
649 Process a log line from shairport-sync stderr.
650
651 :param daemon: The receiver daemon the log line originates from.
652 :param line: The log line to process.
653 """
654 # any output proves the binary runs: a subsequent exit is then handled by
655 # the supervised restart path (per receiver) instead of unloading the
656 # provider as uninitializable
657 if not daemon.started.is_set():
658 daemon.started.set()
659 # Check for fatal errors (log them, but process will exit on its own)
660 if "fatal error:" in line.lower() or "unknown option" in line.lower():
661 self.logger.error("Fatal error from shairport-sync: %s", line)
662 return
663 # Log connection messages at INFO level, everything else at DEBUG
664 if "connection from" in line:
665 self.logger.info("AirPlay client connected: %s", line)
666 else:
667 # Note: Play begin/stop events are now handled via sessioncontrol hooks
668 # through the metadata pipe, so we don't need to parse stderr logs
669 self.logger.debug(line)
670
671 async def _setup_pipes_and_config(self, daemon: _ReceiverDaemon) -> None:
672 """
673 Set up named pipes and configuration file for shairport-sync.
674
675 :raises: OSError if pipe or config file creation fails.
676 """
677 # Remove any existing pipes and config
678 await self._cleanup_pipes_and_config(daemon)
679
680 # Create named pipes for audio and metadata
681 await daemon.audio_pipe.create()
682 await daemon.metadata_pipe.create()
683
684 # Create configuration file
685 await self._create_config_file(daemon)
686
687 async def _cleanup_pipes_and_config(self, daemon: _ReceiverDaemon) -> None:
688 """Clean up named pipes and configuration file."""
689 await daemon.audio_pipe.remove()
690 await daemon.metadata_pipe.remove()
691 await check_output("rm", "-f", daemon.config_file)
692
693 async def _create_config_file(self, daemon: _ReceiverDaemon) -> None:
694 """Create a receiver's shairport-sync configuration file from the template."""
695 # Read template
696 template_path = os.path.join(os.path.dirname(__file__), "bin", "shairport-sync.conf")
697
698 def _read_template() -> str:
699 with open(template_path, encoding="utf-8") as f:
700 return f.read()
701
702 template = await asyncio.to_thread(_read_template)
703
704 # Replace placeholders. The name lands inside a quoted libconfig string:
705 # escape it so a quote or backslash in a player name cannot break the config.
706 safe_name = daemon.airplay_name.replace("\\", "\\\\").replace('"', '\\"')
707 config_content = template.replace("{AIRPLAY_NAME}", safe_name)
708 config_content = config_content.replace("{METADATA_PIPE}", daemon.metadata_pipe.path)
709 config_content = config_content.replace("{AUDIO_PIPE}", daemon.audio_pipe.path)
710 config_content = config_content.replace("{PORT}", str(daemon.port))
711 config_content = config_content.replace(
712 "{INTERFACE_LINE}", await self._get_mdns_interface_line()
713 )
714
715 # Set default volume based on the connected player's current volume if available
716 # Convert player volume (0-100) to AirPlay volume (-30.0 to 0.0 dB)
717 player_volume = 100 # Default to 100%
718 if _player := self.mass.players.get_player(daemon.player_id):
719 if _player.volume_level is not None:
720 player_volume = _player.volume_level
721 # Map 0-100 to -30.0...0.0
722 airplay_volume = (player_volume / 100.0) * 30.0 - 30.0
723 config_content = config_content.replace("{DEFAULT_VOLUME}", f"{airplay_volume:.1f}")
724
725 # Write config file
726 def _write_config() -> None:
727 with open(daemon.config_file, "w", encoding="utf-8") as f:
728 f.write(config_content)
729
730 await asyncio.to_thread(_write_config)
731
732 async def _get_mdns_interface_line(self) -> str:
733 """
734 Build the shairport-sync ``general.interface`` directive, or an empty string.
735
736 When the stream server is bound to a specific interface (not 0.0.0.0), pin
737 the AirPlay mDNS advertisement to that same interface so the receiver is
738 announced on the intended network instead of an unrelated one (e.g. a
739 Docker bridge). Returns an empty string to advertise on all interfaces.
740 """
741 bind_ip = await self.mass.streams.get_source_ip()
742 if not bind_ip:
743 return ""
744 iface_name = interface_name_for_ip(bind_ip)
745 if not iface_name:
746 self.logger.debug(
747 "No interface found for stream bind IP %s; advertising on all interfaces",
748 bind_ip,
749 )
750 return ""
751 return f'\tinterface = "{iface_name}";\n'
752
753 async def _write_silence_to_unblock_stream(self, daemon: _ReceiverDaemon) -> None:
754 """
755 Write silence to a receiver's audio pipe to unblock ffmpeg.
756
757 When shairport-sync stops writing but ffmpeg is still reading,
758 writing silence will cause ffmpeg to output a chunk, which lets the
759 outer consumer make forward progress so the queue's cmd_stop can
760 close the stream cleanly.
761
762 We write enough silence to ensure ffmpeg outputs at least one chunk.
763 PCM_S16LE format: 2 bytes per sample, 2 channels, 44100 Hz
764 Writing 1 second of silence = 44100 * 2 * 2 = 176400 bytes
765 """
766 self.logger.debug("Writing silence to audio pipe to unblock stream")
767 silence = b"\x00" * 176400 # 1 second of silence in PCM_S16LE stereo 44.1kHz
768 # the consumer reopens the pipe shortly after shairport-sync drops it, so the
769 # nudge waits for it to come back instead of landing in that gap
770 if not await daemon.audio_pipe.wait_for_reader(AUDIO_PIPE_READER_TIMEOUT):
771 self.logger.debug("No reader on the audio pipe, skipping the silence write")
772 return
773 await daemon.audio_pipe.write(silence)
774
775 def _clear_active_player(self, daemon: _ReceiverDaemon) -> None:
776 """
777 Clear a receiver's active player.
778
779 Called when playback ends to reset the receiver's session state.
780 """
781 prev_player_id = daemon.active_player_id
782 source_session = (
783 self.mass.players.get_audio_source_session(prev_player_id) if prev_player_id else None
784 )
785 daemon.active_player_id = None
786 daemon.in_use_by_player = None
787 daemon.active_session_id = None
788
789 if prev_player_id:
790 self.logger.debug("Playback ended on player %s, clearing active player", prev_player_id)
791 # the player is not playing us any more, so it should stop saying it is
792 self.mass.create_task(
793 self.mass.players.deselect_source(
794 prev_player_id,
795 stop_playback=False,
796 provider_instance_id=self.instance_id,
797 source_id=daemon.player_id,
798 playback_session_id=(
799 source_session.playback_session_id if source_session else None
800 ),
801 )
802 )
803
804 def _on_metadata_update(self, daemon: _ReceiverDaemon, metadata: dict[str, Any]) -> None:
805 """
806 Handle metadata updates from a receiver's shairport-sync daemon.
807
808 :param daemon: The receiver daemon the update originates from.
809 :param metadata: Dictionary containing metadata updates.
810 """
811 self.logger.log(VERBOSE_LOG_LEVEL, "Received metadata update: %s", metadata)
812
813 # the metadata reader outlives the cancelled tasks for a moment during
814 # teardown; a stopped daemon must not schedule new work from late events
815 if daemon.stop_called:
816 return
817
818 # Handle play state changes from sessioncontrol hooks
819 if "play_state" in metadata:
820 self._handle_play_state_change(daemon, metadata["play_state"])
821 return
822
823 # Handle metadata start (new track starting)
824 if "metadata_start" in metadata:
825 return
826
827 # Handle volume changes from AirPlay client
828 if "volume" in metadata and daemon.in_use_by_player:
829 self._handle_volume_change(daemon, metadata["volume"])
830
831 # Update source metadata fields
832 self._update_source_metadata(daemon, metadata)
833
834 # Handle cover art updates
835 self._update_cover_art(daemon, metadata)
836
837 # Push the metadata update through to the active queue item's streamdetails
838 if daemon.in_use_by_player:
839 self.mass.players.update_source_metadata(
840 daemon.in_use_by_player,
841 daemon.player_id,
842 self.instance_id,
843 daemon.stream_metadata,
844 )
845
846 def _handle_play_state_change(self, daemon: _ReceiverDaemon, play_state: str) -> None:
847 """
848 Handle play state changes from sessioncontrol hooks.
849
850 :param daemon: The receiver daemon the state change originates from.
851 :param play_state: The new play state ("playing" or "stopped").
852 """
853 if play_state == "playing":
854 # Reset volume event flag for new playback session
855 daemon.first_volume_event_received = False
856 # Initiate playback via the standard play_media flow on the target player
857 if not daemon.in_use_by_player:
858 if daemon.pending_start_task and not daemon.pending_start_task.done():
859 # a start is already in flight; a second one would double play_media
860 # and leave the first task untracked for teardown cancellation
861 return
862 # an explicitly selected player wins, else the receiver's own player
863 target_player_id = daemon.active_player_id or daemon.player_id
864 self.logger.info("Starting AirPlay playback on player %s", target_player_id)
865 daemon.active_player_id = target_player_id
866 daemon.pending_start_task = self.mass.create_task(
867 self._start_playback(daemon, target_player_id)
868 )
869 elif play_state == "stopped":
870 self.logger.info("AirPlay playback stopped")
871 # Reset volume event flag for next session
872 daemon.first_volume_event_received = False
873 # Get the current player before clearing
874 current_player_id = daemon.in_use_by_player
875 # Clear active player state (also clears in_use_by_player)
876 self._clear_active_player(daemon)
877 # Write silence to the pipe so ffmpeg can produce a chunk and notice the
878 # stream has stopped; the stop command below closes the generator path.
879 self.mass.create_task(self._write_silence_to_unblock_stream(daemon))
880 # Track the stop so a new session cannot overtake it.
881 if current_player_id:
882 daemon.pending_stop_task = self.mass.create_task(
883 self.mass.players.cmd_stop(current_player_id)
884 )
885
886 async def _start_playback(self, daemon: _ReceiverDaemon, target_player_id: str) -> None:
887 """Start playback after any pending stop completes."""
888 pending_stop_task = daemon.pending_stop_task
889 if pending_stop_task is not None:
890 # Await (even if already done) so a failed stop's exception is retrieved,
891 # and continue regardless of how it failed: a stop that can't complete must
892 # not keep the next session from starting. The reference is cleared only
893 # after the await so concurrent starts (rapid "playing" events before the
894 # stream is claimed) all await the same stop instead of racing past it.
895 try:
896 await pending_stop_task
897 except Exception as err:
898 self.logger.warning("Failed to stop previous AirPlay playback: %s", err)
899 # Don't clear a newer stop that replaced ours while we were awaiting.
900 if daemon.pending_stop_task is pending_stop_task:
901 daemon.pending_stop_task = None
902 await self.mass.player_queues.play_media(target_player_id, str(daemon.audio_source.uri))
903
904 def _handle_volume_change(self, daemon: _ReceiverDaemon, volume: int) -> None:
905 """
906 Handle volume changes from AirPlay client (iOS/macOS device).
907
908 ignore_volume_control = "yes" means shairport-sync doesn't do software volume control,
909 but we still receive volume level changes from the client to apply to the player.
910
911 :param daemon: The receiver daemon the volume change originates from.
912 :param volume: The new volume level (0-100).
913 """
914 # Skip the first volume event as it's the initial sync from default_airplay_volume
915 # We don't want to override the player's current volume on startup
916 if not daemon.first_volume_event_received:
917 daemon.first_volume_event_received = True
918 self.logger.debug(
919 "Received initial AirPlay volume (%s%%), skipping to preserve player volume",
920 volume,
921 )
922 return
923
924 # Type check: ensure we have a valid player ID; queue_id == player_id by convention
925 player_id = daemon.in_use_by_player
926 if not player_id:
927 return
928
929 self.logger.debug(
930 "AirPlay client volume changed to %s%%, applying to player %s",
931 volume,
932 player_id,
933 )
934 try:
935 self.mass.create_task(self.mass.players.cmd_volume_set(player_id, volume))
936 except UnsupportedFeaturedException:
937 self.logger.debug("Player %s does not support volume control", player_id)
938
939 def _update_source_metadata(self, daemon: _ReceiverDaemon, metadata: dict[str, Any]) -> None:
940 """
941 Update a receiver's source metadata fields from AirPlay metadata.
942
943 :param daemon: The receiver daemon the update originates from.
944 :param metadata: Dictionary containing metadata updates.
945 """
946 # Update individual metadata fields
947 if "title" in metadata:
948 daemon.stream_metadata.title = metadata["title"]
949
950 if "artist" in metadata:
951 daemon.stream_metadata.artist = metadata["artist"]
952
953 if "album" in metadata:
954 daemon.stream_metadata.album = metadata["album"]
955
956 if "duration" in metadata:
957 daemon.stream_metadata.duration = metadata["duration"]
958
959 if "elapsed_time" in metadata:
960 daemon.stream_metadata.elapsed_time = metadata["elapsed_time"]
961 # Always set elapsed_time_last_updated to current time when we receive elapsed_time
962 daemon.stream_metadata.elapsed_time_last_updated = time.time()
963
964 def _update_cover_art(self, daemon: _ReceiverDaemon, metadata: dict[str, Any]) -> None:
965 """
966 Update a receiver's cover art image URL from AirPlay metadata.
967
968 :param daemon: The receiver daemon the update originates from.
969 :param metadata: Dictionary containing metadata updates.
970 """
971 if (
972 "cover_art_timestamp" in metadata
973 and daemon.metadata_reader
974 and daemon.metadata_reader.cover_art_bytes
975 ):
976 # Use a content hash in the path so each unique image gets its own
977 # thumbnail cache entry (the thumbnail cache is keyed on provider+path).
978 img_hash = hashlib.md5(
979 daemon.metadata_reader.cover_art_bytes, usedforsecurity=False
980 ).hexdigest()[:8]
981 image = MediaItemImage(
982 type=ImageType.THUMB,
983 path=daemon.cover_art_path(img_hash),
984 provider=self.instance_id,
985 remotely_accessible=False,
986 )
987 daemon.stream_metadata.image_url = self.mass.metadata.get_image_url(image)
988 elif daemon.metadata_reader and daemon.metadata_reader.cover_art_bytes:
989 if not daemon.stream_metadata.image_url:
990 img_hash = hashlib.md5(
991 daemon.metadata_reader.cover_art_bytes, usedforsecurity=False
992 ).hexdigest()[:8]
993 image = MediaItemImage(
994 type=ImageType.THUMB,
995 path=daemon.cover_art_path(img_hash),
996 provider=self.instance_id,
997 remotely_accessible=False,
998 )
999 daemon.stream_metadata.image_url = self.mass.metadata.get_image_url(image)
1000