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