music-assistant-server

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