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