/
/
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_player: 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_player: 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 owner_player_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 owner_player_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 = owner_player_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_player = owner_player_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, owner_player_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 owner_player_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_player == owner_player_id:
331 self._in_use_by_player = 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_player = 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 # the player is not playing us any more, so it should stop saying it is
433 self.mass.create_task(
434 self.mass.players.deselect_source(prev_player_id, stop_playback=False)
435 )
436
437 def _save_last_player_id(self, player_id: str) -> None:
438 """Persist the selected player ID as the new default."""
439 if self._default_player_id == player_id:
440 return # No change needed
441 try:
442 self._update_setup_data(CONF_MASS_PLAYER_ID, player_id)
443 self._default_player_id = player_id
444 except Exception as err:
445 self.logger.debug("Failed to persist player ID: %s", err)
446
447 async def _create_config_file(self) -> None:
448 """Create shairport-sync configuration file from template."""
449 # Read template
450 template_path = os.path.join(os.path.dirname(__file__), "bin", "shairport-sync.conf")
451
452 def _read_template() -> str:
453 with open(template_path, encoding="utf-8") as f:
454 return f.read()
455
456 template = await asyncio.to_thread(_read_template)
457
458 # Replace placeholders
459 config_content = template.replace("{AIRPLAY_NAME}", self._airplay_name)
460 config_content = config_content.replace("{METADATA_PIPE}", self.metadata_pipe.path)
461 config_content = config_content.replace("{AUDIO_PIPE}", self.audio_pipe.path)
462 config_content = config_content.replace("{PORT}", str(self.airplay_port))
463 config_content = config_content.replace(
464 "{INTERFACE_LINE}", await self._get_mdns_interface_line()
465 )
466
467 # Set default volume based on default player's current volume if available
468 # Convert player volume (0-100) to AirPlay volume (-30.0 to 0.0 dB)
469 player_volume = 100 # Default to 100%
470 if self._default_player_id and self._default_player_id != PLAYER_ID_AUTO:
471 if _player := self.mass.players.get_player(self._default_player_id):
472 if _player.volume_level is not None:
473 player_volume = _player.volume_level
474 # Map 0-100 to -30.0...0.0
475 airplay_volume = (player_volume / 100.0) * 30.0 - 30.0
476 config_content = config_content.replace("{DEFAULT_VOLUME}", f"{airplay_volume:.1f}")
477
478 # Write config file
479 def _write_config() -> None:
480 with open(self.config_file, "w", encoding="utf-8") as f:
481 f.write(config_content)
482
483 await asyncio.to_thread(_write_config)
484
485 async def _get_mdns_interface_line(self) -> str:
486 """
487 Build the shairport-sync ``general.interface`` directive, or an empty string.
488
489 When the stream server is bound to a specific interface (not 0.0.0.0), pin
490 the AirPlay mDNS advertisement to that same interface so the receiver is
491 announced on the intended network instead of an unrelated one (e.g. a
492 Docker bridge). Returns an empty string to advertise on all interfaces.
493 """
494 bind_ip = await self.mass.streams.get_source_ip()
495 if not bind_ip:
496 return ""
497 iface_name = interface_name_for_ip(bind_ip)
498 if not iface_name:
499 self.logger.debug(
500 "No interface found for stream bind IP %s; advertising on all interfaces",
501 bind_ip,
502 )
503 return ""
504 return f'\tinterface = "{iface_name}";\n'
505
506 async def _setup_pipes_and_config(self) -> None:
507 """
508 Set up named pipes and configuration file for shairport-sync.
509
510 :raises: OSError if pipe or config file creation fails.
511 """
512 # Remove any existing pipes and config
513 await self._cleanup_pipes_and_config()
514
515 # Create named pipes for audio and metadata
516 await self.audio_pipe.create()
517 await self.metadata_pipe.create()
518
519 # Create configuration file
520 await self._create_config_file()
521
522 async def _cleanup_pipes_and_config(self) -> None:
523 """Clean up named pipes and configuration file."""
524 await self.audio_pipe.remove()
525 await self.metadata_pipe.remove()
526 await check_output("rm", "-f", self.config_file)
527
528 async def _write_silence_to_unblock_stream(self) -> None:
529 """
530 Write silence to the audio pipe to unblock ffmpeg.
531
532 When shairport-sync stops writing but ffmpeg is still reading,
533 writing silence will cause ffmpeg to output a chunk, which lets the
534 outer consumer make forward progress so the queue's cmd_stop can
535 close the stream cleanly.
536
537 We write enough silence to ensure ffmpeg outputs at least one chunk.
538 PCM_S16LE format: 2 bytes per sample, 2 channels, 44100 Hz
539 Writing 1 second of silence = 44100 * 2 * 2 = 176400 bytes
540 """
541 self.logger.debug("Writing silence to audio pipe to unblock stream")
542 silence = b"\x00" * 176400 # 1 second of silence in PCM_S16LE stereo 44.1kHz
543 # the consumer reopens the pipe shortly after shairport-sync drops it, so the
544 # nudge waits for it to come back instead of landing in that gap
545 if not await self.audio_pipe.wait_for_reader(AUDIO_PIPE_READER_TIMEOUT):
546 self.logger.debug("No reader on the audio pipe, skipping the silence write")
547 return
548 await self.audio_pipe.write(silence)
549
550 def _process_shairport_log_line(self, line: str) -> None:
551 """
552 Process a log line from shairport-sync stderr.
553
554 :param line: The log line to process.
555 """
556 # Check for fatal errors (log them, but process will exit on its own)
557 if "fatal error:" in line.lower() or "unknown option" in line.lower():
558 self.logger.error("Fatal error from shairport-sync: %s", line)
559 return
560 # Log connection messages at INFO level, everything else at DEBUG
561 if "connection from" in line:
562 self.logger.info("AirPlay client connected: %s", line)
563 else:
564 # Note: Play begin/stop events are now handled via sessioncontrol hooks
565 # through the metadata pipe, so we don't need to parse stderr logs
566 self.logger.debug(line)
567 if not self._shairport_started.is_set():
568 self._shairport_started.set()
569
570 async def _shairport_runner(self) -> None:
571 """Run the shairport-sync daemon in a background task."""
572 assert self._shairport_bin
573 self.logger.info("Starting AirPlay Receiver background daemon")
574 await self._setup_pipes_and_config()
575
576 try:
577 args: list[str] = [
578 self._shairport_bin,
579 "--configfile",
580 self.config_file,
581 ]
582 self._shairport_proc = shairport = AsyncProcess(
583 args, stderr=True, name=f"shairport-sync[{self.name}]"
584 )
585
586 # Open the FIFO before shairport-sync can invoke session-control hooks.
587 self._metadata_reader = MetadataReader(
588 self.metadata_pipe.path, self.logger, self._on_metadata_update
589 )
590 await self._metadata_reader.start()
591
592 await shairport.start()
593
594 # Check if process started successfully
595 await asyncio.sleep(0.1)
596 if shairport.returncode is not None:
597 self.logger.error(
598 "shairport-sync exited immediately with code %s", shairport.returncode
599 )
600 return
601
602 # Keep reading logging from stderr until exit
603 self.logger.debug("Starting to read shairport-sync stderr")
604 async for stderr_line in shairport.iter_stderr():
605 line = stderr_line.strip()
606 self._process_shairport_log_line(line)
607
608 finally:
609 await shairport.close()
610 self.logger.info(
611 "AirPlay Receiver background daemon stopped for %s (exit code: %s)",
612 self.name,
613 shairport.returncode,
614 )
615
616 # Stop metadata reader
617 if self._metadata_reader:
618 await self._metadata_reader.stop()
619
620 # Clean up pipes and config
621 await self._cleanup_pipes_and_config()
622
623 if not self._shairport_started.is_set():
624 self.unload_with_error("Unable to initialize shairport-sync daemon.")
625 # Auto restart if not stopped manually
626 elif not self._stop_called and self._runner_error_count >= 5:
627 self.unload_with_error("shairport-sync daemon failed to start multiple times.")
628 elif not self._stop_called:
629 self._runner_error_count += 1
630 self.mass.call_later(2, self._setup_shairport_daemon)
631
632 def _setup_shairport_daemon(self) -> None:
633 """Handle setup of the shairport-sync daemon for a player."""
634 self._shairport_started.clear()
635 self._runner_task = self.mass.create_task(self._shairport_runner())
636
637 def _on_metadata_update(self, metadata: dict[str, Any]) -> None:
638 """
639 Handle metadata updates from shairport-sync.
640
641 :param metadata: Dictionary containing metadata updates.
642 """
643 self.logger.log(VERBOSE_LOG_LEVEL, "Received metadata update: %s", metadata)
644
645 # Handle play state changes from sessioncontrol hooks
646 if "play_state" in metadata:
647 self._handle_play_state_change(metadata["play_state"])
648 return
649
650 # Handle metadata start (new track starting)
651 if "metadata_start" in metadata:
652 return
653
654 # Handle volume changes from AirPlay client
655 if "volume" in metadata and self._in_use_by_player:
656 self._handle_volume_change(metadata["volume"])
657
658 # Update source metadata fields
659 self._update_source_metadata(metadata)
660
661 # Handle cover art updates
662 self._update_cover_art(metadata)
663
664 # Push the metadata update through to the active queue item's streamdetails
665 if self._in_use_by_player:
666 self.mass.players.update_source_metadata(
667 self._in_use_by_player,
668 AUDIO_SOURCE_ID,
669 self.instance_id,
670 self._stream_metadata,
671 )
672
673 def _handle_play_state_change(self, play_state: str) -> None:
674 """
675 Handle play state changes from sessioncontrol hooks.
676
677 :param play_state: The new play state ("playing" or "stopped").
678 """
679 if play_state == "playing":
680 # Reset volume event flag for new playback session
681 self._first_volume_event_received = False
682 # Initiate playback via the standard play_media flow on the target player
683 if not self._in_use_by_player:
684 target_player_id = self._get_target_player_id()
685 if target_player_id:
686 self.logger.info("Starting AirPlay playback on player %s", target_player_id)
687 self._active_player_id = target_player_id
688 self.mass.create_task(self._start_playback(target_player_id))
689 else:
690 self.logger.warning(
691 "AirPlay playback started but no player available. "
692 "Start it from the Live Inputs browse view to pick a player."
693 )
694 elif play_state == "stopped":
695 self.logger.info("AirPlay playback stopped")
696 # Reset volume event flag for next session
697 self._first_volume_event_received = False
698 # Get the current player before clearing
699 current_player_id = self._in_use_by_player
700 # Clear active player state (also clears _in_use_by_player)
701 self._clear_active_player()
702 # Write silence to the pipe so ffmpeg can produce a chunk and notice the
703 # stream has stopped; the stop command below closes the generator path.
704 self.mass.create_task(self._write_silence_to_unblock_stream())
705 # Track the stop so a new session cannot overtake it.
706 if current_player_id:
707 self._pending_stop_task = self.mass.create_task(
708 self.mass.players.cmd_stop(current_player_id)
709 )
710
711 async def _start_playback(self, target_player_id: str) -> None:
712 """Start playback after any pending stop completes."""
713 pending_stop_task = self._pending_stop_task
714 if pending_stop_task is not None:
715 # Await (even if already done) so a failed stop's exception is retrieved,
716 # and continue regardless of how it failed: a stop that can't complete must
717 # not keep the next session from starting. The reference is cleared only
718 # after the await so concurrent starts (rapid "playing" events before the
719 # stream is claimed) all await the same stop instead of racing past it.
720 try:
721 await pending_stop_task
722 except Exception as err:
723 self.logger.warning("Failed to stop previous AirPlay playback: %s", err)
724 # Don't clear a newer stop that replaced ours while we were awaiting.
725 if self._pending_stop_task is pending_stop_task:
726 self._pending_stop_task = None
727 await self.mass.player_queues.play_media(target_player_id, str(self._audio_source.uri))
728
729 def _handle_volume_change(self, volume: int) -> None:
730 """
731 Handle volume changes from AirPlay client (iOS/macOS device).
732
733 ignore_volume_control = "yes" means shairport-sync doesn't do software volume control,
734 but we still receive volume level changes from the client to apply to the player.
735
736 :param volume: The new volume level (0-100).
737 """
738 # Skip the first volume event as it's the initial sync from default_airplay_volume
739 # We don't want to override the player's current volume on startup
740 if not self._first_volume_event_received:
741 self._first_volume_event_received = True
742 self.logger.debug(
743 "Received initial AirPlay volume (%s%%), skipping to preserve player volume",
744 volume,
745 )
746 return
747
748 # Type check: ensure we have a valid player ID; queue_id == player_id by convention
749 player_id = self._in_use_by_player
750 if not player_id:
751 return
752
753 self.logger.debug(
754 "AirPlay client volume changed to %s%%, applying to player %s",
755 volume,
756 player_id,
757 )
758 try:
759 self.mass.create_task(self.mass.players.cmd_volume_set(player_id, volume))
760 except UnsupportedFeaturedException:
761 self.logger.debug("Player %s does not support volume control", player_id)
762
763 def _update_source_metadata(self, metadata: dict[str, Any]) -> None:
764 """
765 Update source metadata fields from AirPlay metadata.
766
767 :param metadata: Dictionary containing metadata updates.
768 """
769 # Update individual metadata fields
770 if "title" in metadata:
771 self._stream_metadata.title = metadata["title"]
772
773 if "artist" in metadata:
774 self._stream_metadata.artist = metadata["artist"]
775
776 if "album" in metadata:
777 self._stream_metadata.album = metadata["album"]
778
779 if "duration" in metadata:
780 self._stream_metadata.duration = metadata["duration"]
781
782 if "elapsed_time" in metadata:
783 self._stream_metadata.elapsed_time = metadata["elapsed_time"]
784 # Always set elapsed_time_last_updated to current time when we receive elapsed_time
785 self._stream_metadata.elapsed_time_last_updated = time.time()
786
787 def _update_cover_art(self, metadata: dict[str, Any]) -> None:
788 """
789 Update cover art image URL from AirPlay metadata.
790
791 :param metadata: Dictionary containing metadata updates.
792 """
793 if (
794 "cover_art_timestamp" in metadata
795 and self._metadata_reader
796 and self._metadata_reader.cover_art_bytes
797 ):
798 # Use a content hash in the path so each unique image gets its own
799 # thumbnail cache entry (the thumbnail cache is keyed on provider+path).
800 img_hash = hashlib.md5(
801 self._metadata_reader.cover_art_bytes, usedforsecurity=False
802 ).hexdigest()[:8]
803 image = MediaItemImage(
804 type=ImageType.THUMB,
805 path=f"cover_art_{img_hash}",
806 provider=self.instance_id,
807 remotely_accessible=False,
808 )
809 self._stream_metadata.image_url = self.mass.metadata.get_image_url(image)
810 elif self._metadata_reader and self._metadata_reader.cover_art_bytes:
811 if not self._stream_metadata.image_url:
812 img_hash = hashlib.md5(
813 self._metadata_reader.cover_art_bytes, usedforsecurity=False
814 ).hexdigest()[:8]
815 image = MediaItemImage(
816 type=ImageType.THUMB,
817 path=f"cover_art_{img_hash}",
818 provider=self.instance_id,
819 remotely_accessible=False,
820 )
821 self._stream_metadata.image_url = self.mass.metadata.get_image_url(image)
822