music-assistant-server

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