/
/
1"""AirPlay Player implementations."""
2
3from __future__ import annotations
4
5import asyncio
6import contextlib
7import ipaddress
8import time
9from typing import TYPE_CHECKING, cast
10
11from music_assistant_models.config_entries import ConfigEntry, ConfigValueOption, ConfigValueType
12from music_assistant_models.enums import (
13 ConfigEntryType,
14 ContentType,
15 CrossfadeMode,
16 IdentifierType,
17 MediaType,
18 PlaybackState,
19 PlayerFeature,
20 PlayerType,
21)
22from music_assistant_models.errors import PlayerCommandFailed
23from music_assistant_models.media_items import AudioFormat
24
25from music_assistant.controllers.streams.audio import overlay_active
26from music_assistant.helpers.util import get_primary_ip_address_from_zeroconf, is_valid_mac_address
27from music_assistant.models.player import DeviceInfo, Player, PlayerMedia
28from music_assistant.models.setup_flow import AbortFlow
29
30from . import announce
31from .constants import (
32 AIRPLAY_DISCOVERY_TYPE,
33 AIRPLAY_HIRES_AUDIO_FORMATS,
34 AIRPLAY_HIRES_SAMPLE_RATES,
35 AIRPLAY_PCM_FORMAT,
36 AIRPLAY_REJOIN_ATTEMPT_DELAYS,
37 AIRPLAY_VOLUME_ECHO_GRACE_S,
38 BASE_PLAYER_FEATURES,
39 CONF_AIRPLAY_CREDENTIALS,
40 CONF_BUFFER_DEPTH,
41 CONF_ENCRYPTION,
42 CONF_ENTRY_SYNC_ADJUST_AIRPLAY,
43 CONF_IGNORE_VOLUME,
44 CONF_PAIR_NOW,
45 CONF_PAIRING_PASSWORD,
46 CONF_PAIRING_PIN,
47 CONF_PASSWORD,
48 CONF_PASSWORD_INVALID,
49 CONF_RAOP_CREDENTIALS,
50 CONF_STORED_VOLUME,
51 CONF_STREAMING_MODE,
52 FALLBACK_VOLUME,
53 LEGACY_PAIRING_BIT,
54 PAIRING_PIN_FORMAT,
55 PASSWORD_BIT,
56 PIN_REQUIRED,
57 RAOP_DISCOVERY_TYPE,
58 STREAMING_MODE_AP2_COMPAT,
59 STREAMING_MODE_AP2_NTP,
60 STREAMING_MODE_AP2_PTP,
61 STREAMING_MODE_AUTO,
62 STREAMING_MODE_RAOP,
63 StreamingProtocol,
64)
65from .helpers import (
66 default_buffer_depth,
67 get_decoded_property,
68 is_apple_device,
69 is_macos_device,
70 parse_airplay_features,
71 player_id_to_mac_address,
72 supports_airplay2,
73)
74from .stream_session import AirPlayStreamSession
75
76if TYPE_CHECKING:
77 from zeroconf.asyncio import AsyncServiceInfo
78
79 from music_assistant.models.setup_flow import SetupSession
80
81 from .pairing import AirPlayPairing
82 from .provider import AirPlayProvider
83 from .stream import AirPlayStream
84
85# Docker bridge subnet, sometimes wrongly advertised via mDNS by containerized devices.
86_DOCKER_SUBNET = ipaddress.ip_network("172.16.0.0/12")
87
88
89class AirPlayPlayer(Player):
90 """Base implementation shared by all AirPlay players."""
91
92 def __init__(
93 self,
94 provider: AirPlayProvider,
95 player_id: str,
96 raop_discovery_info: AsyncServiceInfo | None,
97 airplay_discovery_info: AsyncServiceInfo | None,
98 address: str,
99 display_name: str,
100 manufacturer: str,
101 model: str,
102 initial_volume: int = FALLBACK_VOLUME,
103 ) -> None:
104 """Initialize AirPlayPlayer."""
105 self.raop_discovery_info = raop_discovery_info
106 self.airplay_discovery_info = airplay_discovery_info
107 # Audio formats the receiver advertises, learned from its /info response;
108 # zero until that lands (or when the device publishes no format tables).
109 self.advertised_audio_formats = 0
110 self._attr_enabled_by_default = not is_macos_device(manufacturer, model)
111 super().__init__(provider, player_id)
112 self.address = address
113 self.stream: AirPlayStream | None = None
114 self.last_command_sent = 0.0
115 self._volume_reports_ignored_until = 0.0
116 self._lock = asyncio.Lock()
117 self._transitioning = False # Set during stream replacement to ignore stale DACP messages
118 self._rejoin_task: asyncio.Task[None] | None = None
119 # Set (static) player attributes
120 self._attr_name = display_name
121 self._attr_available = True
122 mac_address = player_id_to_mac_address(player_id)
123 self._attr_device_info = DeviceInfo(
124 model=model,
125 manufacturer=manufacturer,
126 )
127 # Only add MAC address if it's valid (not 00:00:00:00:00:00)
128 if is_valid_mac_address(mac_address):
129 self._attr_device_info.add_identifier(IdentifierType.MAC_ADDRESS, mac_address)
130 self._attr_device_info.add_identifier(IdentifierType.IP_ADDRESS, address)
131 self._attr_device_info.add_identifier(IdentifierType.AIRPLAY_ID, player_id)
132 self._attr_volume_level = initial_volume
133 self._attr_can_group_with = {provider.instance_id}
134
135 @property
136 def protocol(self) -> StreamingProtocol:
137 """Get the streaming protocol to use/prefer for this player."""
138 # AirPlay 2 whenever the device can speak it and RAOP is not being forced;
139 # RAOP for legacy receivers (or when the RAOP streaming mode is set).
140 if self._is_airplay2_capable and self.streaming_mode != STREAMING_MODE_RAOP:
141 return StreamingProtocol.AIRPLAY2
142 return StreamingProtocol.RAOP
143
144 @property
145 def streaming_mode(self) -> str:
146 """
147 Return the effective per-player streaming mode.
148
149 Automatic unless the (advanced) streaming-mode setting pins a lane the
150 device actually offers; a stored value the device no longer advertises
151 falls back to Automatic rather than forcing an impossible route.
152 """
153 value = str(self.config.get_value(CONF_STREAMING_MODE, STREAMING_MODE_AUTO))
154 offered = {option.value for option in self.streaming_mode_options}
155 return value if value in offered else STREAMING_MODE_AUTO
156
157 @property
158 def streaming_mode_options(self) -> list[ConfigValueOption]:
159 """
160 Return the streaming-mode options this device can actually offer.
161
162 Every option is an escape from the automatic AirPlay 2 route, gated on
163 the device's own advertisements: the AirPlay 2 lanes need AirPlay 2
164 capability (PTP timing additionally needs the SupportsPTP bit), and
165 legacy RAOP needs an advertised _raop service to fall back to. A
166 RAOP-only device has no alternative lane and keeps Automatic only,
167 which hides the entry entirely. Apple receivers get every lane except
168 NTP timing — they render silence on an NTP-timed realtime stream
169 (hardware-measured). Of their lanes, the compatibility flow and
170 legacy RAOP are the escapes for networks where the PTP ports are
171 blocked; pinning PTP is an explicit choice of the normal lane.
172 """
173 options = [ConfigValueOption(STREAMING_MODE_AUTO, "Automatic (recommended)")]
174 if not self._is_airplay2_capable:
175 return options
176 apple = is_apple_device(self.device_info.manufacturer, self.device_info.model)
177 features = parse_airplay_features(self._advertised_features)
178 if (features >> 41) & 1:
179 options.append(ConfigValueOption(STREAMING_MODE_AP2_PTP, "AirPlay 2 - PTP timing"))
180 if not apple:
181 options.append(ConfigValueOption(STREAMING_MODE_AP2_NTP, "AirPlay 2 - NTP timing"))
182 options.append(
183 ConfigValueOption(STREAMING_MODE_AP2_COMPAT, "AirPlay 2 - compatibility mode")
184 )
185 if self.raop_discovery_info is not None:
186 options.append(ConfigValueOption(STREAMING_MODE_RAOP, "AirPlay 1 (RAOP)"))
187 return options
188
189 @property
190 def protocol_override(self) -> StreamingProtocol | None:
191 """
192 Return the user-forced streaming protocol, or None for automatic selection.
193
194 Only the RAOP streaming mode forces the protocol outright; the AirPlay 2
195 modes stay on the AirPlay 2 protocol and pin the flow/timing through the
196 binary's --protocol/--timing arguments instead. Otherwise the cliairplay
197 binary resolves the route itself from the mDNS TXT records (--protocol
198 auto) and the ``protocol`` property above only reflects MA's own planning
199 heuristic (timing, ports).
200 """
201 if self.streaming_mode == STREAMING_MODE_RAOP:
202 return StreamingProtocol.RAOP
203 return None
204
205 @property
206 def hires_playback_enabled(self) -> bool:
207 """Return if 24-bit hi-res playback is possible for this player."""
208 # 24-bit only works over the AirPlay 2 flow, so a device that streams RAOP
209 # (a legacy receiver, or the force-RAOP escape hatch) stays on the 16-bit
210 # base whatever it advertises.
211 return (
212 bool(self.advertised_audio_formats & AIRPLAY_HIRES_AUDIO_FORMATS)
213 and self.protocol == StreamingProtocol.AIRPLAY2
214 # the compat lane is 16-bit only, so hi-res stands down while the pin is active
215 and self.streaming_mode != STREAMING_MODE_AP2_COMPAT
216 )
217
218 @property
219 def supported_sample_rates(self) -> list[tuple[int, int]]:
220 """Return the (sample_rate, bit_depth) pairs this player natively supports."""
221 if self.hires_playback_enabled:
222 return AIRPLAY_HIRES_SAMPLE_RATES
223 return [(AIRPLAY_PCM_FORMAT.sample_rate, AIRPLAY_PCM_FORMAT.bit_depth)]
224
225 @property
226 def needs_setup(self) -> bool:
227 """Return if the player needs setup."""
228 # A stored password satisfies password protection on its own (the binary
229 # authenticates with it directly; stored credentials are only its
230 # fallback), so the password side is fully covered by the check above.
231 if self.needs_password_setup:
232 return True
233 if self._requires_pin_pairing():
234 # Credentials for either protocol keep the player usable: the binary
235 # picks the best route for the credentials it has. Re-running the setup
236 # flow from the player settings offers replacing a stored pairing.
237 if not (
238 self.get_setup_value(CONF_AIRPLAY_CREDENTIALS)
239 or self.get_setup_value(CONF_RAOP_CREDENTIALS)
240 ):
241 return True
242 return False
243
244 @property
245 def setup_reason(self) -> str | None:
246 """Return why the player needs setup, or None when it is ready to use."""
247 if not self.needs_setup:
248 return None
249 return "password_required" if self.needs_password_setup else "pairing_required"
250
251 @property
252 def password_required(self) -> bool:
253 """Return if the device announces that it is password protected."""
254 # Two announcement forms, verified against live devices (including Apple
255 # TVs, which raise the password bit only while a password is actually
256 # set): receivers publish the password bit in sf/flags and/or the classic
257 # pw boolean. Enforcement can also exist WITHOUT any announcement (stale
258 # TXT after the password was enabled); that case is caught at connect
259 # time via password_invalid.
260 if self._get_flags() & PASSWORD_BIT:
261 return True
262 if raop_info := self.raop_discovery_info:
263 return (raop_info.decoded_properties.get("pw") or "").lower() == "true"
264 return False
265
266 @property
267 def password_invalid(self) -> bool:
268 """Return if the device rejected the stored password on its last connect."""
269 return bool(
270 self.mass.config.get_raw_player_config_value(
271 self.player_id, CONF_PASSWORD_INVALID, False
272 )
273 )
274
275 @property
276 def needs_password_setup(self) -> bool:
277 """Return if the device password still has to be entered through the setup flow."""
278 # The password is only ever entered through the setup flow, so both a
279 # device that announces password protection without one stored and a
280 # password the device rejected must send the user back into that flow.
281 if self.password_invalid:
282 return True
283 return self.password_required and not self.config.get_value(CONF_PASSWORD)
284
285 def set_password_invalid(self, invalid: bool) -> None:
286 """
287 Persist (or clear) the marker that the device rejected the stored password.
288
289 :param invalid: True when the device rejected the password, False once a
290 connect succeeded or a new password was stored.
291 """
292 if self.password_invalid == invalid:
293 # keeps a successful connect from writing the config on every stream
294 return
295 self.mass.config.set_raw_player_config_value(self.player_id, CONF_PASSWORD_INVALID, invalid)
296 # needs_setup/setup_reason are part of the player's own state inputs, so a
297 # plain update publishes the (dis)appeared setup action to the clients.
298 self.update_state()
299
300 @property
301 def requires_flow_mode(self) -> bool:
302 """Return if the player requires flow mode."""
303 return True
304
305 @property
306 def supported_features(self) -> set[PlayerFeature]:
307 """Return the supported features of this player."""
308 # PAUSE is always advertised, including while synced. This keeps the AirPlay
309 # player itself as the pause control target so pause() can decide what to do:
310 # a true pause for a single player, or a full session stop for a sync group
311 # (see pause()). If PAUSE were dropped while grouped, the players controller
312 # could fall through to a linked native player's pause (e.g. a Sonos acting as
313 # an AirPlay receiver), which only pauses the sync leader while the other
314 # members keep playing.
315 features = {*BASE_PLAYER_FEATURES, PlayerFeature.PAUSE}
316 # An announcement is mixed into the audio the player is already rendering, so
317 # the feature is only offered while there is live playback to mix into. Without
318 # it the players controller plays the announcement its own way, which leaves
319 # the device to whatever else may be streaming to it.
320 if not self.has_live_audio:
321 features.discard(PlayerFeature.PLAY_ANNOUNCEMENT)
322 return features
323
324 @property
325 def has_live_audio(self) -> bool:
326 """Return True if the player is rendering audio an announcement can mix into."""
327 if self.playback_state != PlaybackState.PLAYING:
328 return False
329 return self.stream is not None and self.stream.running and self.stream.connected
330
331 @property
332 def applies_announcement_volume(self) -> bool:
333 """Return True: the announcement volume is applied around the mixed clip."""
334 return True
335
336 @property
337 def can_group_with(self) -> set[str]:
338 """
339 Return player IDs this player can group with.
340
341 RAOP and AP2 players can group with other RAOP and/or AP2 players.
342 """
343 prov = cast("AirPlayProvider", self.provider)
344 return {
345 p.player_id for p in prov.get_players() if p.available and p.player_id != self.player_id
346 }
347
348 @property
349 def native_grouping_requires_own_stream(self) -> bool:
350 """Return True: members are attached to this player's own stream session."""
351 return True
352
353 @property
354 def live_session_members(self) -> list[str]:
355 """Return the id's of the players the running stream session feeds."""
356 # group membership is bookkeeping that outlives the session: a member can be
357 # dropped from the session (write failures) or never make it in (a refused
358 # late join) while still being listed as part of the group, and without a
359 # session there is nobody to render with at all
360 if self.stream and self.stream.running and self.stream.session:
361 return [x.player_id for x in self.stream.session.sync_clients]
362 return []
363
364 async def get_config_entries(self) -> list[ConfigEntry]:
365 """Return all (provider/player specific) Config Entries for the given player (if any)."""
366 # Pairing/credentials are no longer config entries: they are collected by the
367 # interactive setup flow (run_setup_flow) and stored in the player's setup_data.
368 base_entries: list[ConfigEntry] = []
369
370 # Effective RAOP state from the current (stored) streaming mode, so the
371 # RAOP-only entries show/hide consistently with it.
372 is_raop = self.protocol == StreamingProtocol.RAOP
373
374 # Streaming-mode escape hatch: a per-device pin of the protocol/timing
375 # lane for receivers whose automatic route misbehaves. Only offered
376 # when the device actually has a lane to choose (Apple receivers are
377 # always native AirPlay 2 with PTP and get no entry).
378 mode_options = self.streaming_mode_options
379 if len(mode_options) > 1:
380 base_entries.append(
381 ConfigEntry(
382 key=CONF_STREAMING_MODE,
383 type=ConfigEntryType.STRING,
384 options=mode_options,
385 default_value=STREAMING_MODE_AUTO,
386 category="protocol_generic",
387 advanced=True,
388 )
389 )
390
391 # Regular AirPlay config entries
392 base_entries += [
393 CONF_ENTRY_SYNC_ADJUST_AIRPLAY,
394 ConfigEntry(
395 key=CONF_ENCRYPTION,
396 type=ConfigEntryType.BOOLEAN,
397 default_value=True,
398 hidden=not is_raop,
399 category="protocol_generic",
400 advanced=True,
401 ),
402 ConfigEntry(
403 key=CONF_PASSWORD,
404 type=ConfigEntryType.SECURE_STRING,
405 default_value=None,
406 required=False,
407 # Storage (and encryption) vehicle only: the device password is
408 # entered through the setup flow, which is also what a wrong
409 # password sends the user back to. A hidden entry keeps its stored
410 # value across config saves (the frontend never submits it).
411 hidden=True,
412 category="protocol_generic",
413 advanced=True,
414 ),
415 ConfigEntry(
416 key=CONF_IGNORE_VOLUME,
417 type=ConfigEntryType.BOOLEAN,
418 default_value=False,
419 category="protocol_generic",
420 advanced=True,
421 ),
422 # Receiver-queue depth presets. The range reaches past the standard
423 # 2 s receiver buffer because that figure is only what the binary
424 # assumes for a device that reports no window of its own, and the
425 # deepest starving devices ask for more than the assumption. The
426 # default comes from the device-family table, and Automatic resolves
427 # through that same table at stream time, so selecting it never
428 # downgrades an affected device.
429 ConfigEntry(
430 key=CONF_BUFFER_DEPTH,
431 type=ConfigEntryType.INTEGER,
432 options=[
433 ConfigValueOption(0),
434 ConfigValueOption(500),
435 ConfigValueOption(750),
436 ConfigValueOption(1000),
437 ConfigValueOption(1500),
438 ConfigValueOption(1750),
439 ConfigValueOption(2000),
440 ConfigValueOption(2500),
441 ConfigValueOption(3000),
442 ],
443 default_value=default_buffer_depth(
444 self.device_info.manufacturer or "",
445 self.device_info.model or "",
446 get_decoded_property(self.airplay_discovery_info, "fv")
447 if self.airplay_discovery_info
448 else None,
449 ),
450 category="protocol_generic",
451 advanced=True,
452 requires_reload=True,
453 ),
454 ]
455
456 return base_entries
457
458 async def run_setup_flow(self, session: SetupSession) -> None:
459 """
460 Run the interactive setup flow for this AirPlay player (streaming pairing).
461
462 :param session: The setup flow session used to interact with the user.
463 """
464 collected: dict[str, ConfigValueType] = {}
465 await self._run_streaming_pairing(session, collected)
466 await session.finish(collected)
467
468 async def stop(self) -> None:
469 """Send STOP command to player."""
470 # an explicit stop (including power-off routed as stop) is user intent:
471 # drop any pending automatic re-join
472 self.cancel_group_rejoin()
473 async with self._lock:
474 if self.stream and self.stream.session:
475 # forward stop to the entire stream session
476 await self.stream.session.stop()
477 elif cast("AirPlayProvider", self.provider).bridge_manager.stop_streaming(
478 self.player_id
479 ):
480 # Sendspin bridge active: it tears the transport down straight
481 # away and takes the player out of the Sendspin session
482 pass
483 elif self.stream and self.stream.running:
484 # Fallback: stop protocol directly
485 await self.stream.stop(force=True)
486 self.stream = None
487 self._attr_current_media = None
488 self.update_state()
489
490 async def play(self) -> None:
491 """Handle PLAY (unpause) command on the player."""
492 session = self.stream.session if self.stream and self.stream.running else None
493 if self.group_members or self.synced_to or (session and session.parked):
494 # Grouped pause parks the whole session (standby); unpausing one
495 # member cannot restart the group in sync, and a parked member is
496 # held with nothing being fed until a re-anchor - which ACTION=PLAY
497 # does not carry, so it would report playback over silence. The park
498 # outlives the group, so a player left alone by an ungroup is keyed
499 # on the park itself, not on its membership. Resume via the queue
500 # instead: play_media flushes and re-anchors every parked member at
501 # one shared instant. The queue can belong to a linked native parent
502 # (for example Sonos), so resolve it instead of using the AirPlay ID.
503 active_queue = self.mass.players.get_active_queue(self)
504 if active_queue is None:
505 raise PlayerCommandFailed(
506 f"Cannot resume AirPlay player {self.display_name} without an active queue"
507 )
508 await self.mass.player_queues.resume(active_queue.queue_id, fade_in=False)
509 return
510 async with self._lock:
511 if self.stream and self.stream.running:
512 if await self.stream.send_cli_command("ACTION=PLAY"):
513 # Resuming re-anchors playout; the binary zeroes its own
514 # re-anchor total on resume, so drop the tracked shift to
515 # keep the server and binary baselines aligned.
516 self.stream.reset_reanchor_shift()
517
518 async def pause(self) -> None:
519 """Send PAUSE command to player."""
520 if self.group_members or self.synced_to:
521 # A broadcast pause cannot keep independent member processes
522 # sample-aligned on resume. Instead the session is parked: every
523 # member stalls but keeps its connection (and remote control), and
524 # the queue's resume flushes and re-anchors over the live
525 # connections — the same coordinated warm restart as seek/next.
526 if (
527 self.stream
528 and self.stream.running
529 and self.stream.session
530 and await self.stream.session.standby()
531 ):
532 return
533 # Some member no longer has a live connection: full stop and let
534 # the queue controller resume from the saved position.
535 self.logger.debug("Sync group cannot be parked, using STOP instead of PAUSE")
536 await self.stop()
537 return
538
539 async with self._lock:
540 if not self.stream or not self.stream.running:
541 return
542 await self.stream.send_cli_command("ACTION=PAUSE")
543
544 async def play_media(self, media: PlayerMedia) -> None:
545 """Handle PLAY MEDIA on given player."""
546 # the player is being (re)purposed on purpose: drop any pending
547 # automatic re-join left over from an unexpected stream loss
548 self.cancel_group_rejoin()
549 async with self._lock:
550 if self.synced_to:
551 # this should not happen, but guard anyways
552 raise RuntimeError("Player is synced")
553 self._attr_current_media = media
554
555 sync_clients = self._get_sync_clients()
556 session_pcm_format = await self._get_session_pcm_format(sync_clients, media)
557
558 # Warm path: a live, compatible session absorbs the new media via a
559 # flush-refill in place (seek/next never pays the reconnect cost).
560 if (
561 self.stream
562 and self.stream.running
563 and self.stream.session
564 and self.stream.session.can_replace(sync_clients, session_pcm_format)
565 ):
566 self._transitioning = True
567 audio_source = self.mass.streams.get_stream(
568 media, session_pcm_format, self.player_id
569 )
570 if await self.stream.session.replace(audio_source, media):
571 self._transitioning = False
572 # A seek changes no media identity, so the identity-driven
573 # metadata callback stays silent and receivers would show
574 # a stale Now Playing position; nudge every member once
575 # the queue position has settled.
576 for member in self.stream.session.sync_clients:
577 self.mass.call_later(
578 1,
579 member.on_player_media_updated,
580 task_id=f"player_media_updated_{member.player_id}",
581 )
582 return
583 # warm replacement failed; fall through to a cold restart
584
585 # Cold path: stop any existing stream and set up from scratch
586 if self.stream and self.stream.running and self.stream.session:
587 # Set transitioning flag to ignore stale DACP messages (like prevent-playback)
588 self._transitioning = True
589 await self.stream.session.stop()
590 self.stream = None
591
592 # select audio source
593 audio_source = self.mass.streams.get_stream(media, session_pcm_format, self.player_id)
594
595 # setup StreamSession for player (and its sync childs if any)
596 provider = cast("AirPlayProvider", self.provider)
597 stream_session = AirPlayStreamSession(
598 provider,
599 sync_clients,
600 session_pcm_format,
601 media,
602 )
603 await stream_session.start(audio_source)
604 self._transitioning = False
605
606 async def play_announcement(
607 self, announcement: PlayerMedia, volume_level: int | None = None
608 ) -> None:
609 """
610 Play an announcement natively, mixed over the audio the player is rendering.
611
612 :param announcement: Details of the announcement that needs to be played.
613 :param volume_level: Optional volume level for the announcement.
614 """
615 # The lock windows live inside the orchestration: the dispatch decision and
616 # the arming hold self._lock like play_media does, while the multi-second
617 # clip waits run outside it (see announce.py).
618 await announce.play_announcement(self, announcement, volume_level)
619
620 async def volume_set(self, volume_level: int) -> None:
621 """Send VOLUME_SET command to given player."""
622 # Record before sending: the connect-time volume push reads this attribute,
623 # so a send that suspends first would let that push send the stale level.
624 self._attr_volume_level = volume_level
625 if self.stream and self.stream.running and self.volume_muted is not True:
626 await self.stream.send_cli_command(f"VOLUME={volume_level}")
627 self.update_state()
628 # store last state in playerconfig
629 self.mass.config.set_raw_player_config_value(
630 self.player_id, CONF_STORED_VOLUME, volume_level
631 )
632
633 async def volume_mute(self, muted: bool) -> None:
634 """Handle VOLUME_MUTE command on the player."""
635 self._attr_volume_muted = muted
636 if self.stream and self.stream.running:
637 volume = 0 if muted else (self.volume_level or 0)
638 await self.stream.send_cli_command(f"VOLUME={volume}")
639 self.update_state()
640
641 async def set_members(
642 self,
643 player_ids_to_add: list[str] | None = None,
644 player_ids_to_remove: list[str] | None = None,
645 ) -> None:
646 """Handle SET_MEMBERS command on the player."""
647 async with self._lock:
648 if self.synced_to:
649 # this should not happen, but guard anyways
650 raise RuntimeError("Player is synced, cannot set members")
651 if not player_ids_to_add and not player_ids_to_remove:
652 # nothing to do
653 return
654
655 stream_session = (
656 self.stream.session
657 if self.stream and self.stream.running and self.stream.session
658 else None
659 )
660 # handle removals first
661 if player_ids_to_remove:
662 if self.player_id in player_ids_to_remove:
663 # Callers only ask for this leader alone or for the whole group at once.
664 # A partial self+subset removal would need the other requested members
665 # released here as well, instead of returning right after the leader.
666 remaining_members = [
667 member_id
668 for member_id in self._attr_group_members
669 if member_id != self.player_id and member_id not in player_ids_to_remove
670 ]
671 if stream_session and remaining_members:
672 # Members stay behind: remove only this leader client,
673 # the session continues for the remaining players
674 await stream_session.remove_client(self, reason="leader removed from group")
675 elif stream_session:
676 # The whole group is being removed, tear the session down
677 await stream_session.stop()
678 self._attr_group_members = []
679 self.update_state()
680 return
681
682 for child_player in self._get_sync_clients():
683 if child_player.player_id in player_ids_to_remove:
684 # update group_members first to prevent race conditions
685 # where a concurrent play_media could re-include this player
686 if child_player.player_id in self._attr_group_members:
687 self._attr_group_members.remove(child_player.player_id)
688 if stream_session:
689 await stream_session.remove_client(
690 child_player, reason="child removed from group"
691 )
692 elif child_player.stream and child_player.stream.running:
693 # leader's stream is no longer running but child still has
694 # an active stream - stop it directly
695 await child_player.stream.stop(force=True)
696
697 # If group leader is left alone after removals, clear the group_members list
698 if (
699 self._attr_group_members
700 and len(self._attr_group_members) == 1
701 and self.player_id in self._attr_group_members
702 ):
703 self._attr_group_members = []
704
705 # handle additions
706 for player_id in player_ids_to_add or []:
707 if player_id == self.player_id or player_id in self.group_members:
708 # nothing to do: player is already part of the group
709 continue
710 child_player_to_add: AirPlayPlayer | None = cast(
711 "AirPlayPlayer | None", self.mass.players.get_player(player_id)
712 )
713 if not child_player_to_add:
714 # should not happen, but guard against it
715 continue
716
717 # ensure the child does not have an existing stream session active
718 if child_player_to_add := cast(
719 "AirPlayPlayer | None", self.mass.players.get_player(player_id)
720 ):
721 if (
722 child_player_to_add.playback_state == PlaybackState.PAUSED
723 and child_player_to_add.stream
724 ):
725 # Stop the paused stream to avoid a deadlock situation
726 await child_player_to_add.stream.stop()
727 if (
728 child_player_to_add.stream
729 and child_player_to_add.stream.running
730 and child_player_to_add.stream.session
731 and child_player_to_add.stream.session != stream_session
732 ):
733 await child_player_to_add.stream.session.remove_client(
734 child_player_to_add, reason="moving to different session"
735 )
736
737 # add new child to the existing stream (RAOP or AirPlay2) session (if any)
738 self._attr_group_members.append(player_id)
739 if stream_session and child_player_to_add is not None:
740 # Skip add_client if the player is already streaming in this session
741 # (e.g. after a dynamic leader switch where the stream continues)
742 if child_player_to_add not in stream_session.sync_clients:
743 await stream_session.add_client(child_player_to_add)
744 elif self.active_output_protocol not in (None, "native"):
745 # Members can only be attached to this player's own stream session, which
746 # does not exist while it renders through one of its output protocols.
747 self.logger.warning(
748 "%s joined the group of %s while that player renders through another "
749 "output protocol: there is no stream session to join, so it stays silent",
750 child_player_to_add.display_name if child_player_to_add else player_id,
751 self.display_name,
752 )
753
754 # Ensure group leader includes itself in group_members when it has members
755 # This is required for the synced_to property to work correctly
756 if self._attr_group_members and self.player_id not in self._attr_group_members:
757 self._attr_group_members.insert(0, self.player_id)
758
759 # always update the state after modifying group members
760 self.update_state()
761
762 @property
763 def ignore_volume_reports(self) -> bool:
764 """Return True if the device's own volume reports must not be acted on."""
765 if self._volume_reports_ignored_until > time.time():
766 # a level we sent ourselves is still echoing back
767 return True
768 return bool(
769 self.config.get_value(CONF_IGNORE_VOLUME)
770 or self.device_info.manufacturer.lower() == "apple"
771 )
772
773 def suppress_volume_reports(self, seconds: float = AIRPLAY_VOLUME_ECHO_GRACE_S) -> None:
774 """
775 Ignore the device's own volume reports for the given time.
776
777 :param seconds: How long from now the reports are ignored; a window that is
778 already open is only ever extended.
779 """
780 self._volume_reports_ignored_until = max(
781 self._volume_reports_ignored_until, time.time() + seconds
782 )
783
784 def update_volume_from_device(self, volume: int) -> None:
785 """Update volume from device feedback."""
786 if self.ignore_volume_reports:
787 return
788
789 cur_volume = self.volume_level or 0
790 if abs(cur_volume - volume) > 1 or (time.time() - self.last_command_sent) > 3:
791 self.mass.create_task(self._adopt_device_volume(volume))
792 else:
793 self._attr_volume_level = volume
794 self.mass.config.set_raw_player_config_value(self.player_id, CONF_STORED_VOLUME, volume)
795 self.update_state()
796
797 def set_discovery_info(self, discovery_info: AsyncServiceInfo, display_name: str) -> None:
798 """Set/update the discovery info for the player."""
799 self._attr_name = display_name
800 if discovery_info.type == AIRPLAY_DISCOVERY_TYPE:
801 self.airplay_discovery_info = discovery_info
802 elif discovery_info.type == RAOP_DISCOVERY_TYPE:
803 self.raop_discovery_info = discovery_info
804 else: # guard
805 return
806 cur_address = self.address
807 prefer_ipv6 = ":" in str(self.mass.streams.publish_ip)
808 new_address = get_primary_ip_address_from_zeroconf(discovery_info, prefer_ipv6=prefer_ipv6)
809 if new_address is None:
810 # should always be set, but guard against None
811 return
812 if cur_address != new_address:
813 # Ignore mDNS updates that replace a routable address with a Docker bridge one.
814 try:
815 if (
816 cur_address
817 and ipaddress.ip_address(new_address) in _DOCKER_SUBNET
818 and ipaddress.ip_address(cur_address) not in _DOCKER_SUBNET
819 ):
820 self.logger.warning(
821 "Ignoring mDNS update from %s to Docker address %s",
822 cur_address,
823 new_address,
824 )
825 self.update_state()
826 return
827 except ValueError:
828 pass
829 self.logger.debug("Address updated from %s to %s", cur_address, new_address)
830 self._attr_device_info.add_identifier(IdentifierType.IP_ADDRESS, new_address)
831 self.address = new_address
832 self.update_state()
833
834 def set_state_from_stream(
835 self,
836 state: PlaybackState | None = None,
837 elapsed_time: float | None = None,
838 stream: AirPlayStream | None = None,
839 ) -> None:
840 """
841 Set the playback state from stream (RAOP or AirPlay2).
842
843 :param state: New playback state (or None to keep current).
844 :param elapsed_time: New elapsed time (or None to keep current).
845 :param stream: The stream instance sending this update (for validation).
846 """
847 # Ignore state updates from old/stale streams
848 if stream is not None and stream != self.stream:
849 return
850 # The stream reclaims the device: an external (Companion-observed)
851 # source snapshot can leak in during a brief stream-restart window and
852 # would otherwise stick, freezing the UI on a stale "external source"
853 # view while we stream. While MA streams, the stream is the sole
854 # authority on this player's state.
855 active_source = getattr(self, "_attr_active_source", None)
856 if active_source is not None and active_source in getattr(self, "_external_source_ids", ()):
857 media = getattr(self, "_attr_current_media", None)
858 if media is not None and media.source_id == active_source:
859 self._attr_current_media = None
860 self._attr_active_source = None
861 if state is not None:
862 self._attr_playback_state = state
863 if elapsed_time is not None:
864 self._attr_elapsed_time = elapsed_time
865 self._attr_elapsed_time_last_updated = time.time()
866 self.update_state()
867
868 def get_stream_pcm_format(self, session_pcm_format: AudioFormat) -> AudioFormat:
869 """
870 Return the PCM format to feed this player's cliairplay process.
871
872 :param session_pcm_format: The PCM format of the (shared) stream session.
873 """
874 if not self.hires_playback_enabled:
875 return AIRPLAY_PCM_FORMAT
876 # 24-bit: the binary expects raw s32le input on stdin (--bitdepth 24)
877 # and truncates to 24-bit ALAC internally.
878 supported_rates = {sample_rate for sample_rate, _ in self.supported_sample_rates}
879 sample_rate = (
880 session_pcm_format.sample_rate
881 if session_pcm_format.sample_rate in supported_rates
882 else AIRPLAY_PCM_FORMAT.sample_rate
883 )
884 return AudioFormat(
885 content_type=ContentType.PCM_S32LE,
886 sample_rate=sample_rate,
887 bit_depth=24,
888 )
889
890 @property
891 def owns_volume(self) -> bool:
892 """
893 Return True if this output is the resolved owner of its own volume.
894
895 AirPlay volume is the receiver's own volume: setting it writes through to the
896 device and persists there after the session ends. It may therefore only be set
897 when no other control owns the volume of this output.
898 """
899 if not (parent_id := self.protocol_parent_id):
900 # a standalone AirPlay player has no other interface to defer to
901 return True
902 if not (parent_player := self.mass.players.get_player(parent_id)):
903 return True
904 return self._control_routes_to_self(parent_player.volume_control_for_output(self.player_id))
905
906 def release_foreign_mute_latch(self) -> None:
907 """Clear our mute latch when another control owns the mute of this output."""
908 if not self._attr_volume_muted:
909 # nothing latched, so nothing that could silence this stream
910 return
911 if not (parent_id := self.protocol_parent_id):
912 return
913 if not (parent_player := self.mass.players.get_player(parent_id)):
914 return
915 if self._control_routes_to_self(parent_player.mute_control_for_output(self.player_id)):
916 # our own mute, applied through the parent
917 return
918 # The mute belongs to a control that does not own this output (a sibling interface,
919 # the receiver itself, or nothing at all). Our mute is a latch that only an explicit
920 # unmute clears, so leaving it set would report a mute we do not own and turn the
921 # next volume command into a silent one.
922 self._attr_volume_muted = False
923 self.update_state()
924
925 async def on_config_updated(self) -> None:
926 """Handle logic when the player config is updated."""
927 await super().on_config_updated()
928 prov = cast("AirPlayProvider", self.provider)
929 await prov.bridge_manager.evaluate_bridge(self)
930
931 async def on_unload(self) -> None:
932 """Handle logic when the player is unloaded from the Player controller."""
933 await super().on_unload()
934 self.cancel_group_rejoin()
935 if self.stream:
936 # remove this player from the stream session if it is running
937 if self.stream.running and self.stream.session:
938 await self.stream.session.remove_client(self, reason="player unloaded")
939 self.stream = None
940
941 def schedule_group_rejoin(self, candidate_ids: list[str]) -> None:
942 """
943 Schedule a bounded automatic re-join of this player to its still-active group.
944
945 Used when this player's stream process died unexpectedly while it was part
946 of a playing sync group (e.g. the device rode out a network blackout): the
947 player is re-added to the group's live session through the regular
948 late-join path after a short backoff. Any user action on the player (or it
949 joining a session by other means) cancels the re-join; when the group is
950 no longer playing, its membership was changed meanwhile or the device is
951 offline, the re-join is abandoned and the player simply stays idle.
952
953 :param candidate_ids: Player ids that led or shared the group at the
954 moment the stream was lost, used to resolve the re-join target (the
955 leadership may transfer while the backoff runs).
956 """
957 self.cancel_group_rejoin()
958 self.logger.info(
959 "Scheduling automatic re-join of %s to its group after unexpected stream loss",
960 self.display_name,
961 )
962 self._rejoin_task = self.mass.create_task(self._group_rejoin_attempts(candidate_ids))
963
964 def cancel_group_rejoin(self) -> None:
965 """Cancel any pending automatic group re-join attempts for this player."""
966 rejoin_task = self._rejoin_task
967 self._rejoin_task = None
968 # never self-cancel: the re-join attempt itself flows through the same
969 # session (re)start paths that call this to clear stale schedules
970 if rejoin_task and not rejoin_task.done() and rejoin_task is not asyncio.current_task():
971 rejoin_task.cancel()
972
973 def on_player_media_updated(self) -> None:
974 """Handle callback when the current media of the player is updated."""
975 if not self.stream or not self.stream.running:
976 return
977 metadata = self.state.current_media
978 if not metadata:
979 return
980 progress = int(metadata.corrected_elapsed_time or 0)
981 self.mass.create_task(self.stream.send_metadata(progress, metadata))
982
983 async def _adopt_device_volume(self, volume: int) -> None:
984 """
985 Take over a level the device set itself.
986
987 :param volume: The level the device reported.
988 """
989 ignored_until = self._volume_reports_ignored_until
990 await self.volume_set(volume)
991 # Writing the level back is a volume command like any other and opens the echo
992 # window, but this one only hands the device its own level: leaving the window
993 # open would swallow the rest of a volume the user is still turning up. A longer
994 # window opened while this was in flight (an announcement) still stands.
995 if self._volume_reports_ignored_until <= time.time() + AIRPLAY_VOLUME_ECHO_GRACE_S:
996 self._volume_reports_ignored_until = ignored_until
997
998 def _control_routes_to_self(self, control: str) -> bool:
999 """Return True if the given (resolved) control routes to this player."""
1000 if control == self.player_id:
1001 return True
1002 # bridge players riding on this player (e.g. Sendspin-over-AirPlay) forward to us
1003 if control_player := self.mass.players.get_player(control):
1004 return control_player.underlying_player_id == self.player_id
1005 return False
1006
1007 def _get_flags(self) -> int:
1008 # Flags are either present via "sf" or "flags". Taken from pyatv.protocols.airplay.utils.
1009 # We combine flags from both RAOP and AirPlay discovery services because
1010 # LEGACY_PAIRING_BIT (0x200) is typically only in the RAOP service sf field
1011 # (e.g. Apple TV HD), while PIN_REQUIRED (0x8) may only appear in the AirPlay
1012 # service sf/flags field. Using only one source misses the pairing requirement.
1013 flags = 0
1014 for discovery_info in filter(None, [self.raop_discovery_info, self.airplay_discovery_info]):
1015 raw = (
1016 discovery_info.properties.get(b"sf")
1017 or discovery_info.properties.get(b"flags")
1018 or b"0x0"
1019 )
1020 with contextlib.suppress(ValueError, TypeError):
1021 flags |= int(raw, 16)
1022 return flags
1023
1024 def _requires_pin_pairing(self) -> bool:
1025 """
1026 Check if this device requires pairing.
1027
1028 Adapted from pyatv.protocols.airplay.utils.get_pairing_requirement.
1029 """
1030 return bool(self._get_flags() & (LEGACY_PAIRING_BIT | PIN_REQUIRED))
1031
1032 def _get_credentials_key(self, protocol: StreamingProtocol) -> str:
1033 """Get the config key for credentials for given protocol."""
1034 if protocol == StreamingProtocol.RAOP:
1035 return CONF_RAOP_CREDENTIALS
1036 return CONF_AIRPLAY_CREDENTIALS
1037
1038 @property
1039 def _advertised_features(self) -> str | None:
1040 """Return the AirPlay features bitmask the device advertises via mDNS."""
1041 # Prefer the _airplay service's ``features``, falling back to the _raop
1042 # service's ``ft`` when the former is absent (some devices only populate one).
1043 features: str | None = None
1044 if self.airplay_discovery_info:
1045 features = self.airplay_discovery_info.decoded_properties.get(
1046 "features"
1047 ) or self.airplay_discovery_info.decoded_properties.get("ft")
1048 if not features and self.raop_discovery_info:
1049 features = self.raop_discovery_info.decoded_properties.get("ft")
1050 return features
1051
1052 @property
1053 def _is_airplay2_capable(self) -> bool:
1054 """
1055 Return whether this device can stream over AirPlay 2.
1056
1057 Mirrors the feature-bit test the cliairplay binary uses for its own route
1058 selection: a device is AirPlay 2 capable when it exposes the _airplay
1059 service and either advertises the AirPlay 2 feature bits or offers no RAOP
1060 fallback at all (i.e. it is a pure AirPlay 2 receiver).
1061 """
1062 if not self.airplay_discovery_info:
1063 return False
1064 return supports_airplay2(self._advertised_features) or not self.raop_discovery_info
1065
1066 async def _run_streaming_pairing(
1067 self, session: SetupSession, collected: dict[str, ConfigValueType]
1068 ) -> None:
1069 """
1070 Pair the streaming protocol (RAOP or AirPlay 2) and collect the device password.
1071
1072 The two are evaluated independently: a device that is already paired can
1073 still be missing its password (or have had it rejected), which is exactly
1074 the state a receiver ends up in when it gains password protection after
1075 it was set up.
1076
1077 :param session: The setup flow session used to interact with the user.
1078 :param collected: The values collected so far; updated in place.
1079 """
1080 password_collected = await self._run_protocol_pairing(session, collected)
1081 if not password_collected and self.needs_password_setup:
1082 await self._ask_device_password(session)
1083
1084 async def _run_protocol_pairing(
1085 self, session: SetupSession, collected: dict[str, ConfigValueType]
1086 ) -> bool:
1087 """
1088 Pair the streaming protocol (RAOP or AirPlay 2), unless already paired.
1089
1090 When the device requires pairing this runs it, re-offering it as a skippable
1091 step when credentials are already stored (so a re-launched flow can replace a
1092 stale pairing). When the device requires no pairing, any leftover credentials
1093 are cleared: they would keep forcing the pair-verify route, which some
1094 receivers (e.g. HomePods after their password was removed) accept while
1095 refusing to actually output audio. The obtained credentials are added to
1096 ``collected`` under the protocol-specific key.
1097
1098 :param session: The setup flow session used to interact with the user.
1099 :param collected: The values collected so far; updated in place.
1100 :return: Whether the device password was collected as part of the pairing.
1101 """
1102 pin_pairing = self._requires_pin_pairing()
1103 # a password only replaces PIN pairing on the native AirPlay 2 flow
1104 password_pairing = self.password_required and self.protocol == StreamingProtocol.AIRPLAY2
1105 if not (pin_pairing or password_pairing):
1106 for cred_key in (CONF_AIRPLAY_CREDENTIALS, CONF_RAOP_CREDENTIALS):
1107 if self.get_setup_value(cred_key) is not None:
1108 collected[cred_key] = None
1109 return False
1110 already_paired = bool(
1111 self.get_setup_value(CONF_AIRPLAY_CREDENTIALS)
1112 or self.get_setup_value(CONF_RAOP_CREDENTIALS)
1113 )
1114 if already_paired and not await self._offer_optional_pairing(
1115 session, "streaming_repair_offer"
1116 ):
1117 return False
1118
1119 protocol = self.protocol
1120 cred_key = self._get_credentials_key(protocol)
1121 if pin_pairing:
1122 step_id, field_key, field_type, field_format = (
1123 "pair_pin",
1124 CONF_PAIRING_PIN,
1125 ConfigEntryType.PAIRING_CODE,
1126 PAIRING_PIN_FORMAT,
1127 )
1128 else:
1129 step_id, field_key, field_type, field_format = (
1130 "pair_password",
1131 CONF_PAIRING_PASSWORD,
1132 ConfigEntryType.SECURE_STRING,
1133 None,
1134 )
1135
1136 errors: dict[str, str] | None = None
1137 while True:
1138 # Each attempt uses a fresh session: finish_pairing() closes the live
1139 # subprocess/session on completion, so a rejected PIN needs a new one
1140 # (and the device re-shows its PIN).
1141 pairing = await self._prepare_streaming_pairing(protocol, pin_pairing=pin_pairing)
1142 try:
1143 values = await session.form(
1144 [
1145 ConfigEntry(
1146 key=field_key,
1147 type=field_type,
1148 required=True,
1149 category="protocol_generic",
1150 format=field_format,
1151 )
1152 ],
1153 step_id=step_id,
1154 errors=errors,
1155 )
1156 entered_value = str(values[field_key])
1157 credentials = await pairing.finish_pairing(pin=entered_value)
1158 except PlayerCommandFailed as err:
1159 # leave a default-level trace: the flow swallows the error into
1160 # the re-served form, which support logs otherwise never show
1161 self.logger.warning("Pairing with %s failed: %s", self.display_name, err)
1162 errors = {"base": err.translation_key or str(err)}
1163 continue
1164 finally:
1165 # tears down the subprocess on retry, success and abort (cancellation)
1166 await pairing.close()
1167 collected[cred_key] = credentials
1168 if password_pairing:
1169 # The device password authenticates every later stream too (the
1170 # binary's transient leg), so keep it next to the credentials
1171 # instead of discarding it with the setup form.
1172 self._store_device_password(entered_value)
1173 return password_pairing
1174
1175 async def _ask_device_password(self, session: SetupSession) -> None:
1176 """
1177 Ask for the device password and store it, without attempting any pairing.
1178
1179 Covers the devices that have no pairing to do: a legacy RAOP receiver, and
1180 an already paired device whose password is missing or was rejected. There
1181 is no live session to validate the entry against, so a wrong password only
1182 surfaces on the next connect - which marks the player as needing setup again.
1183
1184 :param session: The setup flow session used to interact with the user.
1185 """
1186 values = await session.form(
1187 [
1188 ConfigEntry(
1189 key=CONF_PAIRING_PASSWORD,
1190 type=ConfigEntryType.SECURE_STRING,
1191 required=True,
1192 category="protocol_generic",
1193 )
1194 ],
1195 step_id="pair_password",
1196 )
1197 self._store_device_password(str(values[CONF_PAIRING_PASSWORD]))
1198
1199 async def _offer_optional_pairing(self, session: SetupSession, step_id: str) -> bool:
1200 """
1201 Ask whether to run the offered (optional) pairing now.
1202
1203 :param session: The setup flow session used to interact with the user.
1204 :param step_id: The (i18n) step id describing the offered pairing.
1205 """
1206 values = await session.form(
1207 [
1208 ConfigEntry(
1209 key=CONF_PAIR_NOW,
1210 type=ConfigEntryType.BOOLEAN,
1211 default_value=False,
1212 category="protocol_generic",
1213 )
1214 ],
1215 step_id=step_id,
1216 )
1217 return bool(values[CONF_PAIR_NOW])
1218
1219 async def _prepare_streaming_pairing(
1220 self, protocol: StreamingProtocol, *, pin_pairing: bool
1221 ) -> AirPlayPairing:
1222 """
1223 Build and start a streaming pairing session (the device shows its PIN).
1224
1225 A failure here cannot be recovered by re-prompting the user, so it aborts the
1226 flow; a partially started session is torn down first.
1227
1228 :param protocol: The streaming protocol to pair (RAOP or AirPlay 2).
1229 :param pin_pairing: Whether the device shows a PIN the user must enter.
1230 """
1231 pairing: AirPlayPairing | None = None
1232 started = False
1233 try:
1234 pairing = self._build_streaming_pairing(protocol)
1235 await pairing.start_pairing_session()
1236 if pin_pairing:
1237 await pairing.start_pin_pairing()
1238 started = True
1239 except Exception as err:
1240 # a failure starting the session (device unreachable, binary/system
1241 # issue, ...) cannot be fixed by re-prompting, so abort with a clear
1242 # reason instead of letting it surface as a generic internal error
1243 self.logger.warning("Could not start AirPlay pairing session: %s", err)
1244 raise AbortFlow("pairing_failed") from err
1245 finally:
1246 if not started and pairing is not None:
1247 await pairing.close()
1248 assert pairing is not None # reached only when started, i.e. a live session
1249 return pairing
1250
1251 def _build_streaming_pairing(self, protocol: StreamingProtocol) -> AirPlayPairing:
1252 """
1253 Build an AirPlayPairing for the given streaming protocol.
1254
1255 :param protocol: The streaming protocol to pair (RAOP or AirPlay 2).
1256 """
1257 from .pairing import AirPlayPairing # noqa: PLC0415
1258
1259 # For Apple devices pairing always happens on the AirPlay port (7000) even
1260 # when streaming will use RAOP; the RAOP port (5000) is only for streaming.
1261 port: int | None = None
1262 if self.airplay_discovery_info:
1263 port = self.airplay_discovery_info.port or 7000
1264 elif self.raop_discovery_info:
1265 port = self.raop_discovery_info.port or 5000
1266 provider = cast("AirPlayProvider", self.provider)
1267 device_id = provider.dacp_id
1268 pairing_address = self.address
1269 if protocol == StreamingProtocol.AIRPLAY2 and not isinstance(
1270 ipaddress.ip_address(pairing_address), ipaddress.IPv4Address
1271 ):
1272 if self.airplay_discovery_info:
1273 discovered_address = get_primary_ip_address_from_zeroconf(
1274 self.airplay_discovery_info
1275 )
1276 if discovered_address and isinstance(
1277 ipaddress.ip_address(discovered_address), ipaddress.IPv4Address
1278 ):
1279 pairing_address = discovered_address
1280 if not isinstance(ipaddress.ip_address(pairing_address), ipaddress.IPv4Address):
1281 raise PlayerCommandFailed("AirPlay pairing requires an IPv4 device address")
1282 return AirPlayPairing(
1283 address=pairing_address,
1284 name=self.display_name,
1285 protocol=protocol,
1286 logger=self.logger,
1287 port=port,
1288 device_id=device_id,
1289 )
1290
1291 async def _get_session_pcm_format(
1292 self, sync_clients: list[AirPlayPlayer], media: PlayerMedia
1293 ) -> AudioFormat:
1294 """
1295 Select the shared PCM format for a new stream session.
1296
1297 :param sync_clients: All players that will take part in the session.
1298 :param media: The media that is about to be played.
1299 """
1300 queue = self.mass.player_queues.get(media.source_id) if media.source_id else None
1301 queue_item = (
1302 self.mass.player_queues.get_item(media.source_id, media.queue_item_id)
1303 if media.source_id and media.queue_item_id
1304 else None
1305 )
1306 streamdetails = queue_item.streamdetails if queue_item else None
1307 crossfade_enabled = bool(
1308 queue
1309 and media.media_type == MediaType.TRACK
1310 and self.mass.streams.get_crossfade_mode(queue) != CrossfadeMode.DISABLED
1311 )
1312 return await self.mass.streams.audio.select_flow_pcm_format(
1313 self,
1314 start_streamdetails=streamdetails,
1315 crossfade_enabled=crossfade_enabled,
1316 overlay_active=bool(queue and overlay_active(queue)),
1317 fallback_sample_rate=AIRPLAY_PCM_FORMAT.sample_rate,
1318 output_players=sync_clients,
1319 )
1320
1321 def _get_sync_clients(self) -> list[AirPlayPlayer]:
1322 """Get all sync clients for a player."""
1323 sync_clients: list[AirPlayPlayer] = []
1324 # we need to return the player itself too
1325 group_child_ids = {self.player_id}
1326 group_child_ids.update(self.group_members)
1327 for child_id in group_child_ids:
1328 if client := cast("AirPlayPlayer | None", self.mass.players.get_player(child_id)):
1329 sync_clients.append(client)
1330 return sync_clients
1331
1332 async def _group_rejoin_attempts(self, candidate_ids: list[str]) -> None:
1333 """Re-join this player to its group's live session after a bounded backoff."""
1334 max_attempts = len(AIRPLAY_REJOIN_ATTEMPT_DELAYS)
1335 for attempt, delay in enumerate(AIRPLAY_REJOIN_ATTEMPT_DELAYS, start=1):
1336 await asyncio.sleep(delay)
1337 if (
1338 self.group_members
1339 or (self.stream and self.stream.running)
1340 or self.playback_state != PlaybackState.IDLE
1341 # synced into a group outside the original one = deliberate regroup.
1342 # Still pointing at an original candidate is fine: a static group
1343 # keeps the sync membership while only the session lost this player.
1344 or (self.synced_to and self.synced_to not in candidate_ids)
1345 ):
1346 # the player was grouped or repurposed by other means meanwhile
1347 self.logger.debug(
1348 "Automatic group re-join for %s cancelled: player is active again",
1349 self.display_name,
1350 )
1351 return
1352 if not self.available:
1353 # the device is offline: an attempt cannot succeed and the user
1354 # may well have switched it off on purpose
1355 self.logger.debug(
1356 "Automatic group re-join for %s cancelled: player is unavailable",
1357 self.display_name,
1358 )
1359 return
1360 target = self._resolve_rejoin_target(candidate_ids)
1361 if target is None:
1362 # the group may be between sessions (e.g. a track change); keep
1363 # trying until the attempts run out
1364 self.logger.debug(
1365 "Automatic group re-join attempt %d/%d for %s: no playing group found",
1366 attempt,
1367 max_attempts,
1368 self.display_name,
1369 )
1370 continue
1371 # When the sync membership survived the stream loss (a static group,
1372 # where membership is configuration), only the running session needs
1373 # healing; a group command would no-op on the existing membership.
1374 heal_session = (
1375 target.stream.session
1376 if self.player_id in target.group_members and target.stream is not None
1377 else None
1378 )
1379 try:
1380 if heal_session is not None:
1381 await heal_session.add_client(self)
1382 else:
1383 await self.mass.players.cmd_group(self.player_id, target.player_id)
1384 except Exception as err:
1385 self.logger.warning(
1386 "Automatic re-join of %s to group of %s failed (attempt %d/%d): %s",
1387 self.display_name,
1388 target.display_name,
1389 attempt,
1390 max_attempts,
1391 err,
1392 )
1393 continue
1394 # A failed late-join is swallowed inside the grouping path (the player
1395 # then holds group membership without a live stream), so verify the
1396 # session actually carries this player before declaring success.
1397 if (
1398 self.stream
1399 and self.stream.running
1400 and self.stream.session
1401 and self in self.stream.session.sync_clients
1402 ):
1403 self.logger.info(
1404 "Automatically re-joined %s to the group of %s after stream loss",
1405 self.display_name,
1406 target.display_name,
1407 )
1408 return
1409 self.logger.warning(
1410 "Automatic re-join of %s did not produce a running stream (attempt %d/%d)",
1411 self.display_name,
1412 attempt,
1413 max_attempts,
1414 )
1415 if heal_session is None:
1416 # undo the group membership this attempt created so a retry (or
1417 # a manual regroup) starts from a clean join
1418 await self.mass.players.cmd_ungroup(self.player_id)
1419 self.logger.warning(
1420 "Giving up on automatic group re-join for %s after %d attempt(s); "
1421 "the player stays idle",
1422 self.display_name,
1423 max_attempts,
1424 )
1425
1426 def _resolve_rejoin_target(self, candidate_ids: list[str]) -> AirPlayPlayer | None:
1427 """Resolve which player now carries the group's actively playing session."""
1428 for candidate_id in candidate_ids:
1429 candidate = self.mass.players.get_player(candidate_id)
1430 if candidate is None or candidate is self:
1431 continue
1432 if not isinstance(candidate, AirPlayPlayer):
1433 continue
1434 if candidate.synced_to:
1435 # the candidate was absorbed into another group since the loss
1436 # (user intent): never follow the old group's players elsewhere.
1437 # A leadership transfer inside the original group is still found:
1438 # the promoted member is itself one of the candidates.
1439 continue
1440 if not candidate.available:
1441 continue
1442 # only a PLAYING session can absorb a late joiner: a parked (paused)
1443 # session has no live timeline to anchor against
1444 if candidate.playback_state != PlaybackState.PLAYING:
1445 continue
1446 if not (candidate.stream and candidate.stream.running and candidate.stream.session):
1447 continue
1448 return candidate
1449 return None
1450
1451 def _store_device_password(self, password: str) -> None:
1452 """
1453 Persist a device password so every later stream can authenticate with it.
1454
1455 :param password: The plaintext password entered by the user.
1456 """
1457 self.mass.config.set_raw_player_config_value(
1458 self.player_id, CONF_PASSWORD, self.mass.config.encrypt_string(password)
1459 )
1460 # a freshly entered password deserves a clean slate: the reject marker
1461 # would otherwise keep the player in "needs setup" until the next connect
1462 self.set_password_invalid(False)
1463
1464
1465class GenericAirPlayPlayer(AirPlayPlayer):
1466 """AirPlay protocol endpoint without independent device control."""
1467
1468 _attr_type = PlayerType.PROTOCOL
1469