music-assistant-server

1"""
2AirPlay Receiver plugin for Music Assistant.
3
4This plugin allows Music Assistant to receive AirPlay audio streams
5and use them as a source for any player. It uses shairport-sync to
6receive the AirPlay streams and outputs them as PCM audio.
7
8The provider has multi-instance support, so multiple AirPlay receivers
9can be configured with different names.
10"""
11
12from __future__ import annotations
13
14import asyncio
15import hashlib
16import os
17import time
18from collections.abc import Callable
19from contextlib import suppress
20from typing import TYPE_CHECKING, Any, cast
21
22from music_assistant_models.enums import (
23    ContentType,
24    ImageType,
25    MediaType,
26    PlaybackState,
27    ProviderFeature,
28    SourceControl,
29    StreamType,
30)
31from music_assistant_models.errors import (
32    AudioError,
33    MediaNotFoundError,
34    UnsupportedFeaturedException,
35)
36from music_assistant_models.media_items import (
37    AudioFormat,
38    AudioSource,
39    MediaItemImage,
40    ProviderMapping,
41)
42from music_assistant_models.streamdetails import StreamDetails, StreamMetadata
43
44from music_assistant.constants import CONF_ENTRY_WARN_PREVIEW, VERBOSE_LOG_LEVEL
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 music_assistant_models.config_entries import ConfigEntry, ProviderConfig
54    from music_assistant_models.provider import ProviderManifest
55
56    from music_assistant.mass import MusicAssistant
57    from music_assistant.models import ProviderInstanceType
58
59CONF_MASS_PLAYER_ID = "mass_player_id"
60CONF_AIRPLAY_NAME = "airplay_name"
61DEFAULT_AIRPLAY_NAME = "Music Assistant"
62
63# Special value for auto player selection
64PLAYER_ID_AUTO = "__auto__"
65
66SUPPORTED_FEATURES = {ProviderFeature.AUDIO_SOURCE}
67
68# stable id for the single AudioSource this provider exposes;
69# combined with the provider instance_id this forms the persistent uri
70AUDIO_SOURCE_ID = "main"
71
72# seconds the silence nudge waits for the audio pipe's consumer to reattach
73AUDIO_PIPE_READER_TIMEOUT = 1.0
74
75
76def airplay_receiver_port(instance_id: str) -> int:
77    """
78    Return the AirPlay port used by a receiver instance.
79
80    Deterministically derived from the instance id, so it stays the same across
81    server restarts (Python's built-in ``hash()`` is salted per process).
82
83    :param instance_id: The provider instance id of the AirPlay receiver.
84    """
85    digest = hashlib.md5(instance_id.encode(), usedforsecurity=False).hexdigest()
86    return 7000 + int(digest, 16) % 1000
87
88
89async def setup(
90    mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
91) -> ProviderInstanceType:
92    """Initialize provider(instance) with given configuration."""
93    return AirPlayReceiverProvider(mass, manifest, config)
94
95
96class AirPlayReceiverProvider(PluginProvider):
97    """Implementation of an AirPlay Receiver Plugin."""
98
99    reload_on_streams_network_change = True
100
101    def __init__(
102        self, mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
103    ) -> None:
104        """Initialize MusicProvider."""
105        super().__init__(mass, manifest, config, SUPPORTED_FEATURES)
106        # Configured default player (PLAYER_ID_AUTO or a specific player id)
107        self._default_player_id: str = (
108            cast("str", self.get_setup_value(CONF_MASS_PLAYER_ID)) or PLAYER_ID_AUTO
109        )
110        self._airplay_name = (
111            cast("str", self.get_setup_value(CONF_AIRPLAY_NAME)) or DEFAULT_AIRPLAY_NAME
112        )
113        # Currently active player (the one currently playing or selected)
114        self._active_player_id: str | None = None
115        self._shairport_bin: str | None = None
116        self._stop_called: bool = False
117        self._runner_task: asyncio.Task[None] | None = None
118        self._shairport_proc: AsyncProcess | None = None
119        self._shairport_started = asyncio.Event()
120        # Initialize named pipe helpers
121        audio_pipe_path = f"/tmp/ma_airplay_audio_{self.instance_id}"  # noqa: S108
122        metadata_pipe_path = f"/tmp/ma_airplay_metadata_{self.instance_id}"  # noqa: S108
123        self.audio_pipe = AsyncNamedPipeWriter(audio_pipe_path)
124        self.metadata_pipe = AsyncNamedPipeWriter(metadata_pipe_path)
125        self.config_file = f"/tmp/ma_shairport_sync_{self.instance_id}.conf"  # noqa: S108
126        # Use port 7000+ for AirPlay 2 compatibility, one unique port per instance.
127        # The port must be stable across restarts: the AirPlay provider uses it to
128        # recognize (and ignore) our own shairport-sync advertisement in discovery.
129        self.airplay_port = airplay_receiver_port(self.instance_id)
130        # _audio_format describes the original AirPlay source (ALAC at 44.1/16,
131        # the protocol-native format AirPlay senders use) and is what we
132        # advertise to clients for source-format display.
133        self._audio_format = AudioFormat(
134            content_type=ContentType.ALAC,
135            codec_type=ContentType.ALAC,
136            sample_rate=44100,
137            bit_depth=16,
138            channels=2,
139        )
140        # _decoded_audio_format is what shairport-sync actually pipes into MA
141        # after decoding the ALAC stream; the streams controller hands this to
142        # ffmpeg as the input format so it can read the FIFO correctly.
143        self._decoded_audio_format = AudioFormat(
144            content_type=ContentType.PCM_S16LE,
145            codec_type=ContentType.PCM_S16LE,
146            sample_rate=44100,
147            bit_depth=16,
148            channels=2,
149        )
150        self._stream_metadata = StreamMetadata(title=f"AirPlay | {self._airplay_name}")
151        self._audio_source = AudioSource(
152            item_id=AUDIO_SOURCE_ID,
153            provider=self.instance_id,
154            name=self.name,
155            provider_mappings={
156                ProviderMapping(
157                    item_id=AUDIO_SOURCE_ID,
158                    provider_domain=self.domain,
159                    provider_instance=self.instance_id,
160                    audio_format=self._audio_format,
161                )
162            },
163            can_play_pause=False,
164            can_seek=False,
165            can_next_previous=False,
166            exclusive=True,
167            allow_external_trigger=True,
168            # passive: only flows when an external AirPlay client is connected
169            can_initiate=False,
170        )
171        # _in_use_by_queue: the queue currently streaming us. Claimed in
172        # on_source_selected (NOT in get_stream_details — that path also runs
173        # from queue preload, where claiming would block a later cross-queue
174        # handoff). Released in on_source_unselected when the session id
175        # matches, or in _clear_active_player on external session disconnect.
176        self._in_use_by_queue: str | None = None
177        # _active_session_id is the controller-provided token for the current
178        # stream request — used to reject stale on_source_unselected callbacks
179        # after a same-queue reconnect supersedes the previous request.
180        self._active_session_id: str | None = None
181        self._pending_stop_task: asyncio.Task[None] | None = None
182        self._on_unload_callbacks: list[Callable[..., None]] = []
183        self._runner_error_count = 0
184        self._metadata_reader: MetadataReader | None = None
185        self._first_volume_event_received = False  # Track if we've received the first volume event
186
187    @property
188    def instance_name_postfix(self) -> str | None:
189        """Return the advertised receiver name as the multi-instance postfix."""
190        return self._airplay_name if self._airplay_name != DEFAULT_AIRPLAY_NAME else None
191
192    async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
193        """Return runtime options for this provider."""
194        return (CONF_ENTRY_WARN_PREVIEW,)
195
196    async def handle_async_init(self) -> None:
197        """Handle async initialization of the provider."""
198        self._shairport_bin = await get_shairport_sync_binary()
199        # Always start the daemon - we always have a default player configured
200        self._setup_shairport_daemon()
201
202    async def unload(self, is_removed: bool = False) -> None:
203        """Handle close/cleanup of the provider."""
204        self._stop_called = True
205
206        # Stop shairport-sync daemon
207        await self._stop_shairport_daemon()
208
209        # Cleanup callbacks
210        for callback in self._on_unload_callbacks:
211            callback()
212
213    async def get_audio_sources(self) -> list[AudioSource]:
214        """Return the AudioSources this plugin currently exposes."""
215        return [self._audio_source]
216
217    async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
218        """
219        Return StreamDetails for streaming the AirPlay audio to a queue.
220
221        Side-effect-free: ownership is claimed in on_source_selected (which the
222        streams controller fires before this method on the actual stream
223        request). Keeping this idempotent means preload paths like
224        player_queues._load_item can fetch streamdetails without claiming the
225        source and blocking a subsequent cross-queue handoff.
226
227        Raises AudioError when no AirPlay client is currently connected.
228        """
229        if item_id != AUDIO_SOURCE_ID:
230            raise MediaNotFoundError(f"Unknown AudioSource: {item_id}")
231        if not self._active_player_id:
232            raise AudioError(
233                "AirPlay receiver has no active client — start playback from your "
234                "AirPlay-capable device first"
235            )
236        return StreamDetails(
237            provider=self.instance_id,
238            item_id=item_id,
239            audio_format=self._audio_format,
240            decoded_audio_format=self._decoded_audio_format,
241            media_type=MediaType.AUDIO_SOURCE,
242            stream_type=StreamType.NAMED_PIPE,
243            path=self.audio_pipe.path,
244            stream_metadata=self._stream_metadata,
245        )
246
247    async def on_source_control(
248        self,
249        source_id: str,
250        action: SourceControl,
251        value: SourceControlValue = None,
252    ) -> None:
253        """
254        Handle source control commands (no-op: AirPlay receiver is passive).
255
256        The AudioSource advertises no control capabilities, so MA will not invoke
257        any actions here. Override exists only to satisfy the contract.
258        """
259        del source_id, action
260
261    @property
262    def active_player_id(self) -> str | None:
263        """Return the currently active player ID for this plugin."""
264        return self._active_player_id
265
266    async def on_source_selected(
267        self,
268        source_id: str,
269        player_id: str,
270        queue_id: str,
271        stream_session_id: str,
272    ) -> None:
273        """Handle callback when this AudioSource is selected/started on a player."""
274        if source_id != AUDIO_SOURCE_ID or not player_id:
275            return
276
277        # Cache the queue_id (user-facing MA player) rather than the protocol-
278        # level player_id; protocol bridges (e.g. Sendspin's spb_…) can tear
279        # down between streams and their ID is then invalid for play_media.
280        active_player_id = queue_id
281
282        # If there's already an active player and it's different, kick it out.
283        # The lock claim a few lines below replaces the previous queue's claim;
284        # the prior stream's on_source_unselected may fire later, but its
285        # session-id guard keeps it from clobbering the new claim.
286        if self._active_player_id and self._active_player_id != active_player_id:
287            prev_player_id = self._active_player_id
288            self.logger.info(
289                "Source selected on player %s, stopping playback on %s",
290                active_player_id,
291                prev_player_id,
292            )
293            try:
294                await self.mass.players.cmd_stop(prev_player_id)
295            except Exception as err:
296                self.logger.debug("Failed to stop previous player %s: %s", prev_player_id, err)
297
298        # Claim ownership for this queue. The lock lives here (not in
299        # get_stream_details) so preload paths can fetch streamdetails without
300        # accidentally blocking a subsequent cross-queue handoff at the actual
301        # stream request.
302        self._in_use_by_queue = queue_id
303        # Record this request's session id so a later on_source_unselected can
304        # tell whether it is the live teardown or a stale callback from a
305        # superseded same-queue request.
306        self._active_session_id = stream_session_id
307
308        # Update the active player
309        self._active_player_id = active_player_id
310        self.logger.debug("Active player set to: %s", active_player_id)
311
312        # Only persist the selected player as the new default if not in auto mode
313        if self._default_player_id != PLAYER_ID_AUTO:
314            self._save_last_player_id(active_player_id)
315
316    async def on_source_unselected(
317        self, source_id: str, queue_id: str, stream_session_id: str
318    ) -> None:
319        """Release the queue-scoped exclusive claim when MA tears down the stream."""
320        if source_id != AUDIO_SOURCE_ID:
321            return
322        # Reject stale callbacks: only release if this is still the active
323        # session. A queue_id check alone is not sufficient — same-queue
324        # reconnects (player drops + reopens the same stream URL before the
325        # original request's finally fires) would otherwise let the old
326        # request's late callback clear the live claim of the new stream.
327        if self._active_session_id != stream_session_id:
328            return
329        self._active_session_id = None
330        if self._in_use_by_queue == queue_id:
331            self._in_use_by_queue = None
332
333    async def resolve_image(self, path: str) -> bytes:
334        """
335        Resolve an image from an image path.
336
337        This returns raw bytes of the cover art image received from AirPlay metadata.
338
339        :param path: The image path including the current cover art content hash suffix.
340        """
341        if not (self._metadata_reader and self._metadata_reader.cover_art_bytes):
342            return b""
343        current_hash = hashlib.md5(
344            self._metadata_reader.cover_art_bytes, usedforsecurity=False
345        ).hexdigest()[:8]
346        # Only serve when the suffix matches the current artwork's hash, so a
347        # stale request can't cache new bytes under an old hash key.
348        if path == f"cover_art_{current_hash}":
349            return self._metadata_reader.cover_art_bytes
350        return b""
351
352    async def _stop_shairport_daemon(self) -> None:
353        """
354        Stop the shairport-sync daemon without unloading the provider.
355
356        This allows the provider to restart shairport-sync later when needed.
357        """
358        # Stop metadata reader
359        if self._metadata_reader:
360            await self._metadata_reader.stop()
361            self._metadata_reader = None
362
363        # Stop shairport-sync process
364        if self._runner_task and not self._runner_task.done():
365            self._runner_task.cancel()
366            with suppress(asyncio.CancelledError):
367                await self._runner_task
368            self._runner_task = None
369
370        # Reset the shairport process reference
371        self._shairport_proc = None
372        self._shairport_started.clear()
373
374    def _get_target_player_id(self) -> str | None:
375        """
376        Determine the target player ID for playback.
377
378        Returns the player ID to use based on the following priority:
379        1. If a player was explicitly selected (source selected on a player), use that
380        2. If default is 'auto': prefer playing player, then first available
381        3. If a specific default player is configured, use that
382
383        :return: The player ID to use for playback, or None if no player available.
384        """
385        # If there's an active player (source was selected on a player), use it
386        if self._active_player_id:
387            # Validate that the active player still exists
388            if self.mass.players.get_player(self._active_player_id):
389                return self._active_player_id
390            # Active player no longer exists, clear it
391            self._active_player_id = None
392
393        # Handle auto selection
394        if self._default_player_id == PLAYER_ID_AUTO:
395            all_players = list(self.mass.players.all_players(False, False))
396            # First, try to find a playing player
397            for player in all_players:
398                if player.state.playback_state == PlaybackState.PLAYING:
399                    self.logger.debug("Auto-selecting playing player: %s", player.display_name)
400                    return player.player_id
401            # Fallback to first available player
402            if all_players:
403                first_player = all_players[0]
404                self.logger.debug(
405                    "Auto-selecting first available player: %s", first_player.display_name
406                )
407                return first_player.player_id
408            # No player available
409            return None
410
411        # Use the specific default player if configured and it still exists
412        if self.mass.players.get_player(self._default_player_id):
413            return self._default_player_id
414        self.logger.warning(
415            "Configured default player '%s' no longer exists", self._default_player_id
416        )
417        return None
418
419    def _clear_active_player(self) -> None:
420        """
421        Clear the active player and revert to default if configured.
422
423        Called when playback ends to reset the plugin state.
424        """
425        prev_player_id = self._active_player_id
426        self._active_player_id = None
427        self._in_use_by_queue = None
428        self._active_session_id = None
429
430        if prev_player_id:
431            self.logger.debug("Playback ended on player %s, clearing active player", prev_player_id)
432            # Trigger update for the player that was using this source
433            self.mass.players.trigger_player_update(prev_player_id)
434
435    def _save_last_player_id(self, player_id: str) -> None:
436        """Persist the selected player ID as the new default."""
437        if self._default_player_id == player_id:
438            return  # No change needed
439        try:
440            self._update_setup_data(CONF_MASS_PLAYER_ID, player_id)
441            self._default_player_id = player_id
442        except Exception as err:
443            self.logger.debug("Failed to persist player ID: %s", err)
444
445    async def _create_config_file(self) -> None:
446        """Create shairport-sync configuration file from template."""
447        # Read template
448        template_path = os.path.join(os.path.dirname(__file__), "bin", "shairport-sync.conf")
449
450        def _read_template() -> str:
451            with open(template_path, encoding="utf-8") as f:
452                return f.read()
453
454        template = await asyncio.to_thread(_read_template)
455
456        # Replace placeholders
457        config_content = template.replace("{AIRPLAY_NAME}", self._airplay_name)
458        config_content = config_content.replace("{METADATA_PIPE}", self.metadata_pipe.path)
459        config_content = config_content.replace("{AUDIO_PIPE}", self.audio_pipe.path)
460        config_content = config_content.replace("{PORT}", str(self.airplay_port))
461        config_content = config_content.replace(
462            "{INTERFACE_LINE}", await self._get_mdns_interface_line()
463        )
464
465        # Set default volume based on default player's current volume if available
466        # Convert player volume (0-100) to AirPlay volume (-30.0 to 0.0 dB)
467        player_volume = 100  # Default to 100%
468        if self._default_player_id and self._default_player_id != PLAYER_ID_AUTO:
469            if _player := self.mass.players.get_player(self._default_player_id):
470                if _player.volume_level is not None:
471                    player_volume = _player.volume_level
472        # Map 0-100 to -30.0...0.0
473        airplay_volume = (player_volume / 100.0) * 30.0 - 30.0
474        config_content = config_content.replace("{DEFAULT_VOLUME}", f"{airplay_volume:.1f}")
475
476        # Write config file
477        def _write_config() -> None:
478            with open(self.config_file, "w", encoding="utf-8") as f:
479                f.write(config_content)
480
481        await asyncio.to_thread(_write_config)
482
483    async def _get_mdns_interface_line(self) -> str:
484        """
485        Build the shairport-sync ``general.interface`` directive, or an empty string.
486
487        When the stream server is bound to a specific interface (not 0.0.0.0), pin
488        the AirPlay mDNS advertisement to that same interface so the receiver is
489        announced on the intended network instead of an unrelated one (e.g. a
490        Docker bridge). Returns an empty string to advertise on all interfaces.
491        """
492        bind_ip = await self.mass.streams.get_source_ip()
493        if not bind_ip:
494            return ""
495        iface_name = interface_name_for_ip(bind_ip)
496        if not iface_name:
497            self.logger.debug(
498                "No interface found for stream bind IP %s; advertising on all interfaces",
499                bind_ip,
500            )
501            return ""
502        return f'\tinterface = "{iface_name}";\n'
503
504    async def _setup_pipes_and_config(self) -> None:
505        """
506        Set up named pipes and configuration file for shairport-sync.
507
508        :raises: OSError if pipe or config file creation fails.
509        """
510        # Remove any existing pipes and config
511        await self._cleanup_pipes_and_config()
512
513        # Create named pipes for audio and metadata
514        await self.audio_pipe.create()
515        await self.metadata_pipe.create()
516
517        # Create configuration file
518        await self._create_config_file()
519
520    async def _cleanup_pipes_and_config(self) -> None:
521        """Clean up named pipes and configuration file."""
522        await self.audio_pipe.remove()
523        await self.metadata_pipe.remove()
524        await check_output("rm", "-f", self.config_file)
525
526    async def _write_silence_to_unblock_stream(self) -> None:
527        """
528        Write silence to the audio pipe to unblock ffmpeg.
529
530        When shairport-sync stops writing but ffmpeg is still reading,
531        writing silence will cause ffmpeg to output a chunk, which lets the
532        outer consumer make forward progress so the queue's cmd_stop can
533        close the stream cleanly.
534
535        We write enough silence to ensure ffmpeg outputs at least one chunk.
536        PCM_S16LE format: 2 bytes per sample, 2 channels, 44100 Hz
537        Writing 1 second of silence = 44100 * 2 * 2 = 176400 bytes
538        """
539        self.logger.debug("Writing silence to audio pipe to unblock stream")
540        silence = b"\x00" * 176400  # 1 second of silence in PCM_S16LE stereo 44.1kHz
541        # the consumer reopens the pipe shortly after shairport-sync drops it, so the
542        # nudge waits for it to come back instead of landing in that gap
543        if not await self.audio_pipe.wait_for_reader(AUDIO_PIPE_READER_TIMEOUT):
544            self.logger.debug("No reader on the audio pipe, skipping the silence write")
545            return
546        await self.audio_pipe.write(silence)
547
548    def _process_shairport_log_line(self, line: str) -> None:
549        """
550        Process a log line from shairport-sync stderr.
551
552        :param line: The log line to process.
553        """
554        # Check for fatal errors (log them, but process will exit on its own)
555        if "fatal error:" in line.lower() or "unknown option" in line.lower():
556            self.logger.error("Fatal error from shairport-sync: %s", line)
557            return
558        # Log connection messages at INFO level, everything else at DEBUG
559        if "connection from" in line:
560            self.logger.info("AirPlay client connected: %s", line)
561        else:
562            # Note: Play begin/stop events are now handled via sessioncontrol hooks
563            # through the metadata pipe, so we don't need to parse stderr logs
564            self.logger.debug(line)
565        if not self._shairport_started.is_set():
566            self._shairport_started.set()
567
568    async def _shairport_runner(self) -> None:
569        """Run the shairport-sync daemon in a background task."""
570        assert self._shairport_bin
571        self.logger.info("Starting AirPlay Receiver background daemon")
572        await self._setup_pipes_and_config()
573
574        try:
575            args: list[str] = [
576                self._shairport_bin,
577                "--configfile",
578                self.config_file,
579            ]
580            self._shairport_proc = shairport = AsyncProcess(
581                args, stderr=True, name=f"shairport-sync[{self.name}]"
582            )
583
584            # Open the FIFO before shairport-sync can invoke session-control hooks.
585            self._metadata_reader = MetadataReader(
586                self.metadata_pipe.path, self.logger, self._on_metadata_update
587            )
588            await self._metadata_reader.start()
589
590            await shairport.start()
591
592            # Check if process started successfully
593            await asyncio.sleep(0.1)
594            if shairport.returncode is not None:
595                self.logger.error(
596                    "shairport-sync exited immediately with code %s", shairport.returncode
597                )
598                return
599
600            # Keep reading logging from stderr until exit
601            self.logger.debug("Starting to read shairport-sync stderr")
602            async for stderr_line in shairport.iter_stderr():
603                line = stderr_line.strip()
604                self._process_shairport_log_line(line)
605
606        finally:
607            await shairport.close()
608            self.logger.info(
609                "AirPlay Receiver background daemon stopped for %s (exit code: %s)",
610                self.name,
611                shairport.returncode,
612            )
613
614            # Stop metadata reader
615            if self._metadata_reader:
616                await self._metadata_reader.stop()
617
618            # Clean up pipes and config
619            await self._cleanup_pipes_and_config()
620
621            if not self._shairport_started.is_set():
622                self.unload_with_error("Unable to initialize shairport-sync daemon.")
623            # Auto restart if not stopped manually
624            elif not self._stop_called and self._runner_error_count >= 5:
625                self.unload_with_error("shairport-sync daemon failed to start multiple times.")
626            elif not self._stop_called:
627                self._runner_error_count += 1
628                self.mass.call_later(2, self._setup_shairport_daemon)
629
630    def _setup_shairport_daemon(self) -> None:
631        """Handle setup of the shairport-sync daemon for a player."""
632        self._shairport_started.clear()
633        self._runner_task = self.mass.create_task(self._shairport_runner())
634
635    def _on_metadata_update(self, metadata: dict[str, Any]) -> None:
636        """
637        Handle metadata updates from shairport-sync.
638
639        :param metadata: Dictionary containing metadata updates.
640        """
641        self.logger.log(VERBOSE_LOG_LEVEL, "Received metadata update: %s", metadata)
642
643        # Handle play state changes from sessioncontrol hooks
644        if "play_state" in metadata:
645            self._handle_play_state_change(metadata["play_state"])
646            return
647
648        # Handle metadata start (new track starting)
649        if "metadata_start" in metadata:
650            return
651
652        # Handle volume changes from AirPlay client
653        if "volume" in metadata and self._in_use_by_queue:
654            self._handle_volume_change(metadata["volume"])
655
656        # Update source metadata fields
657        self._update_source_metadata(metadata)
658
659        # Handle cover art updates
660        self._update_cover_art(metadata)
661
662        # Push the metadata update through to the active queue item's streamdetails
663        if self._in_use_by_queue:
664            self.mass.streams.update_stream_metadata(
665                self._in_use_by_queue,
666                AUDIO_SOURCE_ID,
667                self.instance_id,
668                self._stream_metadata,
669            )
670
671    def _handle_play_state_change(self, play_state: str) -> None:
672        """
673        Handle play state changes from sessioncontrol hooks.
674
675        :param play_state: The new play state ("playing" or "stopped").
676        """
677        if play_state == "playing":
678            # Reset volume event flag for new playback session
679            self._first_volume_event_received = False
680            # Initiate playback via the standard play_media flow on the target player
681            if not self._in_use_by_queue:
682                target_player_id = self._get_target_player_id()
683                if target_player_id:
684                    self.logger.info("Starting AirPlay playback on player %s", target_player_id)
685                    self._active_player_id = target_player_id
686                    self.mass.create_task(self._start_playback(target_player_id))
687                else:
688                    self.logger.warning(
689                        "AirPlay playback started but no player available. "
690                        "Start it from the Live Inputs browse view to pick a player."
691                    )
692        elif play_state == "stopped":
693            self.logger.info("AirPlay playback stopped")
694            # Reset volume event flag for next session
695            self._first_volume_event_received = False
696            # Get the current player before clearing
697            current_player_id = self._in_use_by_queue
698            # Clear active player state (also clears _in_use_by_queue)
699            self._clear_active_player()
700            # Write silence to the pipe so ffmpeg can produce a chunk and notice the
701            # stream has stopped; the stop command below closes the generator path.
702            self.mass.create_task(self._write_silence_to_unblock_stream())
703            # Track the stop so a new session cannot overtake it.
704            if current_player_id:
705                self._pending_stop_task = self.mass.create_task(
706                    self.mass.players.cmd_stop(current_player_id)
707                )
708
709    async def _start_playback(self, target_player_id: str) -> None:
710        """Start playback after any pending stop completes."""
711        pending_stop_task = self._pending_stop_task
712        if pending_stop_task is not None:
713            # Await (even if already done) so a failed stop's exception is retrieved,
714            # and continue regardless of how it failed: a stop that can't complete must
715            # not keep the next session from starting. The reference is cleared only
716            # after the await so concurrent starts (rapid "playing" events before the
717            # stream is claimed) all await the same stop instead of racing past it.
718            try:
719                await pending_stop_task
720            except Exception as err:
721                self.logger.warning("Failed to stop previous AirPlay playback: %s", err)
722            # Don't clear a newer stop that replaced ours while we were awaiting.
723            if self._pending_stop_task is pending_stop_task:
724                self._pending_stop_task = None
725        await self.mass.player_queues.play_media(target_player_id, str(self._audio_source.uri))
726
727    def _handle_volume_change(self, volume: int) -> None:
728        """
729        Handle volume changes from AirPlay client (iOS/macOS device).
730
731        ignore_volume_control = "yes" means shairport-sync doesn't do software volume control,
732        but we still receive volume level changes from the client to apply to the player.
733
734        :param volume: The new volume level (0-100).
735        """
736        # Skip the first volume event as it's the initial sync from default_airplay_volume
737        # We don't want to override the player's current volume on startup
738        if not self._first_volume_event_received:
739            self._first_volume_event_received = True
740            self.logger.debug(
741                "Received initial AirPlay volume (%s%%), skipping to preserve player volume",
742                volume,
743            )
744            return
745
746        # Type check: ensure we have a valid player ID; queue_id == player_id by convention
747        player_id = self._in_use_by_queue
748        if not player_id:
749            return
750
751        self.logger.debug(
752            "AirPlay client volume changed to %s%%, applying to player %s",
753            volume,
754            player_id,
755        )
756        try:
757            self.mass.create_task(self.mass.players.cmd_volume_set(player_id, volume))
758        except UnsupportedFeaturedException:
759            self.logger.debug("Player %s does not support volume control", player_id)
760
761    def _update_source_metadata(self, metadata: dict[str, Any]) -> None:
762        """
763        Update source metadata fields from AirPlay metadata.
764
765        :param metadata: Dictionary containing metadata updates.
766        """
767        # Update individual metadata fields
768        if "title" in metadata:
769            self._stream_metadata.title = metadata["title"]
770
771        if "artist" in metadata:
772            self._stream_metadata.artist = metadata["artist"]
773
774        if "album" in metadata:
775            self._stream_metadata.album = metadata["album"]
776
777        if "duration" in metadata:
778            self._stream_metadata.duration = metadata["duration"]
779
780        if "elapsed_time" in metadata:
781            self._stream_metadata.elapsed_time = metadata["elapsed_time"]
782            # Always set elapsed_time_last_updated to current time when we receive elapsed_time
783            self._stream_metadata.elapsed_time_last_updated = time.time()
784
785    def _update_cover_art(self, metadata: dict[str, Any]) -> None:
786        """
787        Update cover art image URL from AirPlay metadata.
788
789        :param metadata: Dictionary containing metadata updates.
790        """
791        if (
792            "cover_art_timestamp" in metadata
793            and self._metadata_reader
794            and self._metadata_reader.cover_art_bytes
795        ):
796            # Use a content hash in the path so each unique image gets its own
797            # thumbnail cache entry (the thumbnail cache is keyed on provider+path).
798            img_hash = hashlib.md5(
799                self._metadata_reader.cover_art_bytes, usedforsecurity=False
800            ).hexdigest()[:8]
801            image = MediaItemImage(
802                type=ImageType.THUMB,
803                path=f"cover_art_{img_hash}",
804                provider=self.instance_id,
805                remotely_accessible=False,
806            )
807            self._stream_metadata.image_url = self.mass.metadata.get_image_url(image)
808        elif self._metadata_reader and self._metadata_reader.cover_art_bytes:
809            if not self._stream_metadata.image_url:
810                img_hash = hashlib.md5(
811                    self._metadata_reader.cover_art_bytes, usedforsecurity=False
812                ).hexdigest()[:8]
813                image = MediaItemImage(
814                    type=ImageType.THUMB,
815                    path=f"cover_art_{img_hash}",
816                    provider=self.instance_id,
817                    remotely_accessible=False,
818                )
819                self._stream_metadata.image_url = self.mass.metadata.get_image_url(image)
820