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