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; cancel BOTH before awaiting either
489        # (awaiting first could let the other progress meanwhile), and await so a
490        # cancellation-delayed player command cannot land after the daemon is gone
491        pending_tasks = [
492            task
493            for task in (daemon.pending_stop_task, daemon.pending_start_task)
494            if task and not task.done()
495        ]
496        for pending_task in pending_tasks:
497            pending_task.cancel()
498        for pending_task in pending_tasks:
499            with suppress(asyncio.CancelledError):
500                await pending_task
501        daemon.pending_stop_task = None
502        daemon.pending_start_task = None
503
504        # Stop metadata reader
505        if daemon.metadata_reader:
506            await daemon.metadata_reader.stop()
507            daemon.metadata_reader = None
508
509        # Stop shairport-sync process
510        if daemon.runner_task and not daemon.runner_task.done():
511            daemon.runner_task.cancel()
512            with suppress(asyncio.CancelledError):
513                await daemon.runner_task
514            daemon.runner_task = None
515
516        # Reset the shairport process reference
517        daemon.shairport_proc = None
518        daemon.started.clear()
519
520    def _setup_shairport_daemon(self, daemon: _ReceiverDaemon) -> None:
521        """Handle setup of the shairport-sync daemon for a receiver."""
522        # a delayed restart can fire after the receiver was stopped or replaced
523        if daemon.stop_called or self._daemons.get(daemon.player_id) is not daemon:
524            return
525        daemon.started.clear()
526        daemon.runner_task = self.mass.create_task(self._shairport_runner(daemon))
527
528    async def _shairport_runner(self, daemon: _ReceiverDaemon) -> None:
529        """Run a receiver's shairport-sync daemon in a background task."""
530        assert self._shairport_bin
531        self.logger.info("Starting AirPlay Receiver background daemon for %s", daemon.airplay_name)
532        await self._setup_pipes_and_config(daemon)
533
534        try:
535            args: list[str] = [
536                self._shairport_bin,
537                "--configfile",
538                daemon.config_file,
539            ]
540            daemon.shairport_proc = shairport = AsyncProcess(
541                args, stderr=True, name=f"shairport-sync[{daemon.airplay_name}]"
542            )
543
544            # Open the FIFO before shairport-sync can invoke session-control hooks.
545            daemon.metadata_reader = MetadataReader(
546                daemon.metadata_pipe.path, self.logger, partial(self._on_metadata_update, daemon)
547            )
548            await daemon.metadata_reader.start()
549
550            await shairport.start()
551
552            # Check if process started successfully
553            await asyncio.sleep(0.1)
554            if shairport.returncode is not None:
555                self.logger.error(
556                    "shairport-sync exited immediately with code %s", shairport.returncode
557                )
558                return
559
560            # Keep reading logging from stderr until exit
561            self.logger.debug("Starting to read shairport-sync stderr")
562            async for stderr_line in shairport.iter_stderr():
563                line = stderr_line.strip()
564                self._process_shairport_log_line(daemon, line)
565
566        finally:
567            await shairport.close()
568            self.logger.info(
569                "AirPlay Receiver background daemon stopped for %s (exit code: %s)",
570                daemon.airplay_name,
571                shairport.returncode,
572            )
573
574            # Stop metadata reader
575            if daemon.metadata_reader:
576                await daemon.metadata_reader.stop()
577
578            # Clean up pipes and config
579            await self._cleanup_pipes_and_config(daemon)
580
581            if daemon.stop_called:
582                # deliberately stopped (unload, rename restart or player removal)
583                pass
584            elif not daemon.started.is_set():
585                self.unload_with_error("Unable to initialize shairport-sync daemon.")
586            # Auto restart if not stopped manually
587            elif daemon.runner_error_count >= 5:
588                self.unload_with_error("shairport-sync daemon failed to start multiple times.")
589            else:
590                daemon.runner_error_count += 1
591                self.mass.call_later(2, self._setup_shairport_daemon, daemon)
592
593    def _process_shairport_log_line(self, daemon: _ReceiverDaemon, line: str) -> None:
594        """
595        Process a log line from shairport-sync stderr.
596
597        :param daemon: The receiver daemon the log line originates from.
598        :param line: The log line to process.
599        """
600        # Check for fatal errors (log them, but process will exit on its own)
601        if "fatal error:" in line.lower() or "unknown option" in line.lower():
602            self.logger.error("Fatal error from shairport-sync: %s", line)
603            return
604        # Log connection messages at INFO level, everything else at DEBUG
605        if "connection from" in line:
606            self.logger.info("AirPlay client connected: %s", line)
607        else:
608            # Note: Play begin/stop events are now handled via sessioncontrol hooks
609            # through the metadata pipe, so we don't need to parse stderr logs
610            self.logger.debug(line)
611        if not daemon.started.is_set():
612            daemon.started.set()
613
614    async def _setup_pipes_and_config(self, daemon: _ReceiverDaemon) -> None:
615        """
616        Set up named pipes and configuration file for shairport-sync.
617
618        :raises: OSError if pipe or config file creation fails.
619        """
620        # Remove any existing pipes and config
621        await self._cleanup_pipes_and_config(daemon)
622
623        # Create named pipes for audio and metadata
624        await daemon.audio_pipe.create()
625        await daemon.metadata_pipe.create()
626
627        # Create configuration file
628        await self._create_config_file(daemon)
629
630    async def _cleanup_pipes_and_config(self, daemon: _ReceiverDaemon) -> None:
631        """Clean up named pipes and configuration file."""
632        await daemon.audio_pipe.remove()
633        await daemon.metadata_pipe.remove()
634        await check_output("rm", "-f", daemon.config_file)
635
636    async def _create_config_file(self, daemon: _ReceiverDaemon) -> None:
637        """Create a receiver's shairport-sync configuration file from the template."""
638        # Read template
639        template_path = os.path.join(os.path.dirname(__file__), "bin", "shairport-sync.conf")
640
641        def _read_template() -> str:
642            with open(template_path, encoding="utf-8") as f:
643                return f.read()
644
645        template = await asyncio.to_thread(_read_template)
646
647        # Replace placeholders. The name lands inside a quoted libconfig string:
648        # escape it so a quote or backslash in a player name cannot break the config.
649        safe_name = daemon.airplay_name.replace("\\", "\\\\").replace('"', '\\"')
650        config_content = template.replace("{AIRPLAY_NAME}", safe_name)
651        config_content = config_content.replace("{METADATA_PIPE}", daemon.metadata_pipe.path)
652        config_content = config_content.replace("{AUDIO_PIPE}", daemon.audio_pipe.path)
653        config_content = config_content.replace("{PORT}", str(daemon.port))
654        config_content = config_content.replace(
655            "{INTERFACE_LINE}", await self._get_mdns_interface_line()
656        )
657
658        # Set default volume based on the connected player's current volume if available
659        # Convert player volume (0-100) to AirPlay volume (-30.0 to 0.0 dB)
660        player_volume = 100  # Default to 100%
661        if _player := self.mass.players.get_player(daemon.player_id):
662            if _player.volume_level is not None:
663                player_volume = _player.volume_level
664        # Map 0-100 to -30.0...0.0
665        airplay_volume = (player_volume / 100.0) * 30.0 - 30.0
666        config_content = config_content.replace("{DEFAULT_VOLUME}", f"{airplay_volume:.1f}")
667
668        # Write config file
669        def _write_config() -> None:
670            with open(daemon.config_file, "w", encoding="utf-8") as f:
671                f.write(config_content)
672
673        await asyncio.to_thread(_write_config)
674
675    async def _get_mdns_interface_line(self) -> str:
676        """
677        Build the shairport-sync ``general.interface`` directive, or an empty string.
678
679        When the stream server is bound to a specific interface (not 0.0.0.0), pin
680        the AirPlay mDNS advertisement to that same interface so the receiver is
681        announced on the intended network instead of an unrelated one (e.g. a
682        Docker bridge). Returns an empty string to advertise on all interfaces.
683        """
684        bind_ip = await self.mass.streams.get_source_ip()
685        if not bind_ip:
686            return ""
687        iface_name = interface_name_for_ip(bind_ip)
688        if not iface_name:
689            self.logger.debug(
690                "No interface found for stream bind IP %s; advertising on all interfaces",
691                bind_ip,
692            )
693            return ""
694        return f'\tinterface = "{iface_name}";\n'
695
696    async def _write_silence_to_unblock_stream(self, daemon: _ReceiverDaemon) -> None:
697        """
698        Write silence to a receiver's audio pipe to unblock ffmpeg.
699
700        When shairport-sync stops writing but ffmpeg is still reading,
701        writing silence will cause ffmpeg to output a chunk, which lets the
702        outer consumer make forward progress so the queue's cmd_stop can
703        close the stream cleanly.
704
705        We write enough silence to ensure ffmpeg outputs at least one chunk.
706        PCM_S16LE format: 2 bytes per sample, 2 channels, 44100 Hz
707        Writing 1 second of silence = 44100 * 2 * 2 = 176400 bytes
708        """
709        self.logger.debug("Writing silence to audio pipe to unblock stream")
710        silence = b"\x00" * 176400  # 1 second of silence in PCM_S16LE stereo 44.1kHz
711        # the consumer reopens the pipe shortly after shairport-sync drops it, so the
712        # nudge waits for it to come back instead of landing in that gap
713        if not await daemon.audio_pipe.wait_for_reader(AUDIO_PIPE_READER_TIMEOUT):
714            self.logger.debug("No reader on the audio pipe, skipping the silence write")
715            return
716        await daemon.audio_pipe.write(silence)
717
718    def _clear_active_player(self, daemon: _ReceiverDaemon) -> None:
719        """
720        Clear a receiver's active player.
721
722        Called when playback ends to reset the receiver's session state.
723        """
724        prev_player_id = daemon.active_player_id
725        source_session = (
726            self.mass.players.get_audio_source_session(prev_player_id) if prev_player_id else None
727        )
728        daemon.active_player_id = None
729        daemon.in_use_by_player = None
730        daemon.active_session_id = None
731
732        if prev_player_id:
733            self.logger.debug("Playback ended on player %s, clearing active player", prev_player_id)
734            # the player is not playing us any more, so it should stop saying it is
735            self.mass.create_task(
736                self.mass.players.deselect_source(
737                    prev_player_id,
738                    stop_playback=False,
739                    provider_instance_id=self.instance_id,
740                    source_id=daemon.player_id,
741                    playback_session_id=(
742                        source_session.playback_session_id if source_session else None
743                    ),
744                )
745            )
746
747    def _on_metadata_update(self, daemon: _ReceiverDaemon, metadata: dict[str, Any]) -> None:
748        """
749        Handle metadata updates from a receiver's shairport-sync daemon.
750
751        :param daemon: The receiver daemon the update originates from.
752        :param metadata: Dictionary containing metadata updates.
753        """
754        self.logger.log(VERBOSE_LOG_LEVEL, "Received metadata update: %s", metadata)
755
756        # the metadata reader outlives the cancelled tasks for a moment during
757        # teardown; a stopped daemon must not schedule new work from late events
758        if daemon.stop_called:
759            return
760
761        # Handle play state changes from sessioncontrol hooks
762        if "play_state" in metadata:
763            self._handle_play_state_change(daemon, metadata["play_state"])
764            return
765
766        # Handle metadata start (new track starting)
767        if "metadata_start" in metadata:
768            return
769
770        # Handle volume changes from AirPlay client
771        if "volume" in metadata and daemon.in_use_by_player:
772            self._handle_volume_change(daemon, metadata["volume"])
773
774        # Update source metadata fields
775        self._update_source_metadata(daemon, metadata)
776
777        # Handle cover art updates
778        self._update_cover_art(daemon, metadata)
779
780        # Push the metadata update through to the active queue item's streamdetails
781        if daemon.in_use_by_player:
782            self.mass.players.update_source_metadata(
783                daemon.in_use_by_player,
784                daemon.player_id,
785                self.instance_id,
786                daemon.stream_metadata,
787            )
788
789    def _handle_play_state_change(self, daemon: _ReceiverDaemon, play_state: str) -> None:
790        """
791        Handle play state changes from sessioncontrol hooks.
792
793        :param daemon: The receiver daemon the state change originates from.
794        :param play_state: The new play state ("playing" or "stopped").
795        """
796        if play_state == "playing":
797            # Reset volume event flag for new playback session
798            daemon.first_volume_event_received = False
799            # Initiate playback via the standard play_media flow on the target player
800            if not daemon.in_use_by_player:
801                if daemon.pending_start_task and not daemon.pending_start_task.done():
802                    # a start is already in flight; a second one would double play_media
803                    # and leave the first task untracked for teardown cancellation
804                    return
805                # an explicitly selected player wins, else the receiver's own player
806                target_player_id = daemon.active_player_id or daemon.player_id
807                self.logger.info("Starting AirPlay playback on player %s", target_player_id)
808                daemon.active_player_id = target_player_id
809                daemon.pending_start_task = self.mass.create_task(
810                    self._start_playback(daemon, target_player_id)
811                )
812        elif play_state == "stopped":
813            self.logger.info("AirPlay playback stopped")
814            # Reset volume event flag for next session
815            daemon.first_volume_event_received = False
816            # Get the current player before clearing
817            current_player_id = daemon.in_use_by_player
818            # Clear active player state (also clears in_use_by_player)
819            self._clear_active_player(daemon)
820            # Write silence to the pipe so ffmpeg can produce a chunk and notice the
821            # stream has stopped; the stop command below closes the generator path.
822            self.mass.create_task(self._write_silence_to_unblock_stream(daemon))
823            # Track the stop so a new session cannot overtake it.
824            if current_player_id:
825                daemon.pending_stop_task = self.mass.create_task(
826                    self.mass.players.cmd_stop(current_player_id)
827                )
828
829    async def _start_playback(self, daemon: _ReceiverDaemon, target_player_id: str) -> None:
830        """Start playback after any pending stop completes."""
831        pending_stop_task = daemon.pending_stop_task
832        if pending_stop_task is not None:
833            # Await (even if already done) so a failed stop's exception is retrieved,
834            # and continue regardless of how it failed: a stop that can't complete must
835            # not keep the next session from starting. The reference is cleared only
836            # after the await so concurrent starts (rapid "playing" events before the
837            # stream is claimed) all await the same stop instead of racing past it.
838            try:
839                await pending_stop_task
840            except Exception as err:
841                self.logger.warning("Failed to stop previous AirPlay playback: %s", err)
842            # Don't clear a newer stop that replaced ours while we were awaiting.
843            if daemon.pending_stop_task is pending_stop_task:
844                daemon.pending_stop_task = None
845        await self.mass.player_queues.play_media(target_player_id, str(daemon.audio_source.uri))
846
847    def _handle_volume_change(self, daemon: _ReceiverDaemon, volume: int) -> None:
848        """
849        Handle volume changes from AirPlay client (iOS/macOS device).
850
851        ignore_volume_control = "yes" means shairport-sync doesn't do software volume control,
852        but we still receive volume level changes from the client to apply to the player.
853
854        :param daemon: The receiver daemon the volume change originates from.
855        :param volume: The new volume level (0-100).
856        """
857        # Skip the first volume event as it's the initial sync from default_airplay_volume
858        # We don't want to override the player's current volume on startup
859        if not daemon.first_volume_event_received:
860            daemon.first_volume_event_received = True
861            self.logger.debug(
862                "Received initial AirPlay volume (%s%%), skipping to preserve player volume",
863                volume,
864            )
865            return
866
867        # Type check: ensure we have a valid player ID; queue_id == player_id by convention
868        player_id = daemon.in_use_by_player
869        if not player_id:
870            return
871
872        self.logger.debug(
873            "AirPlay client volume changed to %s%%, applying to player %s",
874            volume,
875            player_id,
876        )
877        try:
878            self.mass.create_task(self.mass.players.cmd_volume_set(player_id, volume))
879        except UnsupportedFeaturedException:
880            self.logger.debug("Player %s does not support volume control", player_id)
881
882    def _update_source_metadata(self, daemon: _ReceiverDaemon, metadata: dict[str, Any]) -> None:
883        """
884        Update a receiver's source metadata fields from AirPlay metadata.
885
886        :param daemon: The receiver daemon the update originates from.
887        :param metadata: Dictionary containing metadata updates.
888        """
889        # Update individual metadata fields
890        if "title" in metadata:
891            daemon.stream_metadata.title = metadata["title"]
892
893        if "artist" in metadata:
894            daemon.stream_metadata.artist = metadata["artist"]
895
896        if "album" in metadata:
897            daemon.stream_metadata.album = metadata["album"]
898
899        if "duration" in metadata:
900            daemon.stream_metadata.duration = metadata["duration"]
901
902        if "elapsed_time" in metadata:
903            daemon.stream_metadata.elapsed_time = metadata["elapsed_time"]
904            # Always set elapsed_time_last_updated to current time when we receive elapsed_time
905            daemon.stream_metadata.elapsed_time_last_updated = time.time()
906
907    def _update_cover_art(self, daemon: _ReceiverDaemon, metadata: dict[str, Any]) -> None:
908        """
909        Update a receiver's cover art image URL from AirPlay metadata.
910
911        :param daemon: The receiver daemon the update originates from.
912        :param metadata: Dictionary containing metadata updates.
913        """
914        if (
915            "cover_art_timestamp" in metadata
916            and daemon.metadata_reader
917            and daemon.metadata_reader.cover_art_bytes
918        ):
919            # Use a content hash in the path so each unique image gets its own
920            # thumbnail cache entry (the thumbnail cache is keyed on provider+path).
921            img_hash = hashlib.md5(
922                daemon.metadata_reader.cover_art_bytes, usedforsecurity=False
923            ).hexdigest()[:8]
924            image = MediaItemImage(
925                type=ImageType.THUMB,
926                path=daemon.cover_art_path(img_hash),
927                provider=self.instance_id,
928                remotely_accessible=False,
929            )
930            daemon.stream_metadata.image_url = self.mass.metadata.get_image_url(image)
931        elif daemon.metadata_reader and daemon.metadata_reader.cover_art_bytes:
932            if not daemon.stream_metadata.image_url:
933                img_hash = hashlib.md5(
934                    daemon.metadata_reader.cover_art_bytes, usedforsecurity=False
935                ).hexdigest()[:8]
936                image = MediaItemImage(
937                    type=ImageType.THUMB,
938                    path=daemon.cover_art_path(img_hash),
939                    provider=self.instance_id,
940                    remotely_accessible=False,
941                )
942                daemon.stream_metadata.image_url = self.mass.metadata.get_image_url(image)
943