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