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