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
764        """
765        if (
766            self.protocol_parent_id
767            and (parent_player := self.mass.players.get_player(self.protocol_parent_id))
768            and parent_player.state.volume_level is not None
769        ):
770            if self._has_native_protocol_parent:
771                # Native parent volume is on the receiver/amplifier scale.
772                # Keep the AirPlay child volume learned from DACP feedback instead.
773                return
774            if parent_player.state.volume_level == 0:
775                # A parent volume of 0 usually means the (idle) sibling interface
776                # feeding the parent doesn't know the real device volume, e.g. the
777                # cast side of the same device reports 0 while in standby. Adopting
778                # it would start the stream hard muted, so keep our own last known
779                # volume instead.
780                return
781            if self._attr_volume_level == parent_player.state.volume_level:
782                return
783            self._attr_volume_level = parent_player.state.volume_level
784            self.mass.config.set_raw_player_config_value(
785                self.player_id, CONF_STORED_VOLUME, self._attr_volume_level
786            )
787            self.update_state()
788
789    async def on_config_updated(self) -> None:
790        """Handle logic when the player config is updated."""
791        await super().on_config_updated()
792        prov = cast("AirPlayProvider", self.provider)
793        await prov.bridge_manager.evaluate_bridge(self)
794
795    async def on_unload(self) -> None:
796        """Handle logic when the player is unloaded from the Player controller."""
797        await super().on_unload()
798        self.cancel_group_rejoin()
799        if self.stream:
800            # remove this player from the stream session if it is running
801            if self.stream.running and self.stream.session:
802                await self.stream.session.remove_client(self, reason="player unloaded")
803            self.stream = None
804
805    def schedule_group_rejoin(self, candidate_ids: list[str]) -> None:
806        """
807        Schedule a bounded automatic re-join of this player to its still-active group.
808
809        Used when this player's stream process died unexpectedly while it was part
810        of a playing sync group (e.g. the device rode out a network blackout): the
811        player is re-added to the group's live session through the regular
812        late-join path after a short backoff. Any user action on the player (or it
813        joining a session by other means) cancels the re-join; when the group is
814        no longer playing, its membership was changed meanwhile or the device is
815        offline, the re-join is abandoned and the player simply stays idle.
816
817        :param candidate_ids: Player ids that led or shared the group at the
818            moment the stream was lost, used to resolve the re-join target (the
819            leadership may transfer while the backoff runs).
820        """
821        self.cancel_group_rejoin()
822        self.logger.info(
823            "Scheduling automatic re-join of %s to its group after unexpected stream loss",
824            self.display_name,
825        )
826        self._rejoin_task = self.mass.create_task(self._group_rejoin_attempts(candidate_ids))
827
828    def cancel_group_rejoin(self) -> None:
829        """Cancel any pending automatic group re-join attempts for this player."""
830        rejoin_task = self._rejoin_task
831        self._rejoin_task = None
832        # never self-cancel: the re-join attempt itself flows through the same
833        # session (re)start paths that call this to clear stale schedules
834        if rejoin_task and not rejoin_task.done() and rejoin_task is not asyncio.current_task():
835            rejoin_task.cancel()
836
837    @property
838    def _has_native_protocol_parent(self) -> bool:
839        """Return True if this AirPlay protocol player is linked to a native parent."""
840        if not self.protocol_parent_id:
841            return False
842        parent_player = self.mass.players.get_player(self.protocol_parent_id)
843        return bool(parent_player and parent_player.volume_control == PLAYER_CONTROL_NATIVE)
844
845    def _get_flags(self) -> int:
846        # Flags are either present via "sf" or "flags". Taken from pyatv.protocols.airplay.utils.
847        # We combine flags from both RAOP and AirPlay discovery services because
848        # LEGACY_PAIRING_BIT (0x200) is typically only in the RAOP service sf field
849        # (e.g. Apple TV HD), while PIN_REQUIRED (0x8) may only appear in the AirPlay
850        # service sf/flags field. Using only one source misses the pairing requirement.
851        flags = 0
852        for discovery_info in filter(None, [self.raop_discovery_info, self.airplay_discovery_info]):
853            raw = (
854                discovery_info.properties.get(b"sf")
855                or discovery_info.properties.get(b"flags")
856                or b"0x0"
857            )
858            with contextlib.suppress(ValueError, TypeError):
859                flags |= int(raw, 16)
860        return flags
861
862    def _requires_pin_pairing(self) -> bool:
863        """
864        Check if this device requires pairing.
865
866        Adapted from pyatv.protocols.airplay.utils.get_pairing_requirement.
867        """
868        return bool(self._get_flags() & (LEGACY_PAIRING_BIT | PIN_REQUIRED))
869
870    def _get_credentials_key(self, protocol: StreamingProtocol) -> str:
871        """Get the config key for credentials for given protocol."""
872        if protocol == StreamingProtocol.RAOP:
873            return CONF_RAOP_CREDENTIALS
874        return CONF_AIRPLAY_CREDENTIALS
875
876    @property
877    def _advertised_features(self) -> str | None:
878        """Return the AirPlay features bitmask the device advertises via mDNS."""
879        # Prefer the _airplay service's ``features``, falling back to the _raop
880        # service's ``ft`` when the former is absent (some devices only populate one).
881        features: str | None = None
882        if self.airplay_discovery_info:
883            features = self.airplay_discovery_info.decoded_properties.get(
884                "features"
885            ) or self.airplay_discovery_info.decoded_properties.get("ft")
886        if not features and self.raop_discovery_info:
887            features = self.raop_discovery_info.decoded_properties.get("ft")
888        return features
889
890    @property
891    def _is_airplay2_capable(self) -> bool:
892        """
893        Return whether this device can stream over AirPlay 2.
894
895        Mirrors the feature-bit test the cliairplay binary uses for its own route
896        selection: a device is AirPlay 2 capable when it exposes the _airplay
897        service and either advertises the AirPlay 2 feature bits or offers no RAOP
898        fallback at all (i.e. it is a pure AirPlay 2 receiver).
899        """
900        if not self.airplay_discovery_info:
901            return False
902        return supports_airplay2(self._advertised_features) or not self.raop_discovery_info
903
904    @property
905    def _force_raop_available(self) -> bool:
906        """
907        Return whether the "force RAOP" escape hatch applies to this device.
908
909        Offered only for AirPlay-2-capable non-Apple receivers that also advertise
910        a RAOP service to fall back to. Genuine Apple devices are always AirPlay 2,
911        while RAOP-only and AirPlay-2-only devices have nothing to force.
912        """
913        return (
914            self._is_airplay2_capable
915            and self.raop_discovery_info is not None
916            and not is_apple_device(self.device_info.manufacturer, self.device_info.model)
917        )
918
919    @property
920    def _force_raop_active(self) -> bool:
921        """Return whether RAOP is being forced through the escape-hatch toggle."""
922        return self._force_raop_available and bool(self.config.get_value(CONF_FORCE_RAOP, False))
923
924    async def _run_streaming_pairing(
925        self, session: SetupSession, collected: dict[str, ConfigValueType]
926    ) -> None:
927        """
928        Pair the streaming protocol (RAOP or AirPlay 2) and collect the device password.
929
930        The two are evaluated independently: a device that is already paired can
931        still be missing its password (or have had it rejected), which is exactly
932        the state a receiver ends up in when it gains password protection after
933        it was set up.
934
935        :param session: The setup flow session used to interact with the user.
936        :param collected: The values collected so far; updated in place.
937        """
938        password_collected = await self._run_protocol_pairing(session, collected)
939        if not password_collected and self.needs_password_setup:
940            await self._ask_device_password(session)
941
942    async def _run_protocol_pairing(
943        self, session: SetupSession, collected: dict[str, ConfigValueType]
944    ) -> bool:
945        """
946        Pair the streaming protocol (RAOP or AirPlay 2), unless already paired.
947
948        When the device requires pairing this runs it, re-offering it as a skippable
949        step when credentials are already stored (so a re-launched flow can replace a
950        stale pairing). When the device requires no pairing, any leftover credentials
951        are cleared: they would keep forcing the pair-verify route, which some
952        receivers (e.g. HomePods after their password was removed) accept while
953        refusing to actually output audio. The obtained credentials are added to
954        ``collected`` under the protocol-specific key.
955
956        :param session: The setup flow session used to interact with the user.
957        :param collected: The values collected so far; updated in place.
958        :return: Whether the device password was collected as part of the pairing.
959        """
960        pin_pairing = self._requires_pin_pairing()
961        # a password only replaces PIN pairing on the native AirPlay 2 flow
962        password_pairing = self.password_required and self.protocol == StreamingProtocol.AIRPLAY2
963        if not (pin_pairing or password_pairing):
964            for cred_key in (CONF_AIRPLAY_CREDENTIALS, CONF_RAOP_CREDENTIALS):
965                if self.get_setup_value(cred_key) is not None:
966                    collected[cred_key] = None
967            return False
968        already_paired = bool(
969            self.get_setup_value(CONF_AIRPLAY_CREDENTIALS)
970            or self.get_setup_value(CONF_RAOP_CREDENTIALS)
971        )
972        if already_paired and not await self._offer_optional_pairing(
973            session, "streaming_repair_offer"
974        ):
975            return False
976
977        protocol = self.protocol
978        cred_key = self._get_credentials_key(protocol)
979        if pin_pairing:
980            step_id, field_key, field_type = "pair_pin", CONF_PAIRING_PIN, ConfigEntryType.STRING
981        else:
982            step_id, field_key, field_type = (
983                "pair_password",
984                CONF_PAIRING_PASSWORD,
985                ConfigEntryType.SECURE_STRING,
986            )
987
988        errors: dict[str, str] | None = None
989        while True:
990            # Each attempt uses a fresh session: finish_pairing() closes the live
991            # subprocess/session on completion, so a rejected PIN needs a new one
992            # (and the device re-shows its PIN).
993            pairing = await self._prepare_streaming_pairing(protocol, pin_pairing=pin_pairing)
994            try:
995                values = await session.form(
996                    [
997                        ConfigEntry(
998                            key=field_key,
999                            type=field_type,
1000                            required=True,
1001                            category="protocol_generic",
1002                        )
1003                    ],
1004                    step_id=step_id,
1005                    errors=errors,
1006                )
1007                entered_value = str(values[field_key])
1008                credentials = await pairing.finish_pairing(pin=entered_value)
1009            except PlayerCommandFailed as err:
1010                # leave a default-level trace: the flow swallows the error into
1011                # the re-served form, which support logs otherwise never show
1012                self.logger.warning("Pairing with %s failed: %s", self.display_name, err)
1013                errors = {"base": err.translation_key or str(err)}
1014                continue
1015            finally:
1016                # tears down the subprocess on retry, success and abort (cancellation)
1017                await pairing.close()
1018            collected[cred_key] = credentials
1019            if password_pairing:
1020                # The device password authenticates every later stream too (the
1021                # binary's transient leg), so keep it next to the credentials
1022                # instead of discarding it with the setup form.
1023                self._store_device_password(entered_value)
1024            return password_pairing
1025
1026    async def _ask_device_password(self, session: SetupSession) -> None:
1027        """
1028        Ask for the device password and store it, without attempting any pairing.
1029
1030        Covers the devices that have no pairing to do: a legacy RAOP receiver, and
1031        an already paired device whose password is missing or was rejected. There
1032        is no live session to validate the entry against, so a wrong password only
1033        surfaces on the next connect - which marks the player as needing setup again.
1034
1035        :param session: The setup flow session used to interact with the user.
1036        """
1037        values = await session.form(
1038            [
1039                ConfigEntry(
1040                    key=CONF_PAIRING_PASSWORD,
1041                    type=ConfigEntryType.SECURE_STRING,
1042                    required=True,
1043                    category="protocol_generic",
1044                )
1045            ],
1046            step_id="pair_password",
1047        )
1048        self._store_device_password(str(values[CONF_PAIRING_PASSWORD]))
1049
1050    async def _offer_optional_pairing(self, session: SetupSession, step_id: str) -> bool:
1051        """
1052        Ask whether to run the offered (optional) pairing now.
1053
1054        :param session: The setup flow session used to interact with the user.
1055        :param step_id: The (i18n) step id describing the offered pairing.
1056        """
1057        values = await session.form(
1058            [
1059                ConfigEntry(
1060                    key=CONF_PAIR_NOW,
1061                    type=ConfigEntryType.BOOLEAN,
1062                    default_value=False,
1063                    category="protocol_generic",
1064                )
1065            ],
1066            step_id=step_id,
1067        )
1068        return bool(values[CONF_PAIR_NOW])
1069
1070    async def _prepare_streaming_pairing(
1071        self, protocol: StreamingProtocol, *, pin_pairing: bool
1072    ) -> AirPlayPairing:
1073        """
1074        Build and start a streaming pairing session (the device shows its PIN).
1075
1076        A failure here cannot be recovered by re-prompting the user, so it aborts the
1077        flow; a partially started session is torn down first.
1078
1079        :param protocol: The streaming protocol to pair (RAOP or AirPlay 2).
1080        :param pin_pairing: Whether the device shows a PIN the user must enter.
1081        """
1082        pairing: AirPlayPairing | None = None
1083        started = False
1084        try:
1085            pairing = self._build_streaming_pairing(protocol)
1086            await pairing.start_pairing_session()
1087            if pin_pairing:
1088                await pairing.start_pin_pairing()
1089            started = True
1090        except Exception as err:
1091            # a failure starting the session (device unreachable, binary/system
1092            # issue, ...) cannot be fixed by re-prompting, so abort with a clear
1093            # reason instead of letting it surface as a generic internal error
1094            self.logger.warning("Could not start AirPlay pairing session: %s", err)
1095            raise AbortFlow("pairing_failed") from err
1096        finally:
1097            if not started and pairing is not None:
1098                await pairing.close()
1099        assert pairing is not None  # reached only when started, i.e. a live session
1100        return pairing
1101
1102    def _build_streaming_pairing(self, protocol: StreamingProtocol) -> AirPlayPairing:
1103        """
1104        Build an AirPlayPairing for the given streaming protocol.
1105
1106        :param protocol: The streaming protocol to pair (RAOP or AirPlay 2).
1107        """
1108        from .pairing import AirPlayPairing  # noqa: PLC0415
1109
1110        # For Apple devices pairing always happens on the AirPlay port (7000) even
1111        # when streaming will use RAOP; the RAOP port (5000) is only for streaming.
1112        port: int | None = None
1113        if self.airplay_discovery_info:
1114            port = self.airplay_discovery_info.port or 7000
1115        elif self.raop_discovery_info:
1116            port = self.raop_discovery_info.port or 5000
1117        provider = cast("AirPlayProvider", self.provider)
1118        device_id = provider.dacp_id
1119        pairing_address = self.address
1120        if protocol == StreamingProtocol.AIRPLAY2 and not isinstance(
1121            ipaddress.ip_address(pairing_address), ipaddress.IPv4Address
1122        ):
1123            if self.airplay_discovery_info:
1124                discovered_address = get_primary_ip_address_from_zeroconf(
1125                    self.airplay_discovery_info
1126                )
1127                if discovered_address and isinstance(
1128                    ipaddress.ip_address(discovered_address), ipaddress.IPv4Address
1129                ):
1130                    pairing_address = discovered_address
1131            if not isinstance(ipaddress.ip_address(pairing_address), ipaddress.IPv4Address):
1132                raise PlayerCommandFailed("AirPlay pairing requires an IPv4 device address")
1133        return AirPlayPairing(
1134            address=pairing_address,
1135            name=self.display_name,
1136            protocol=protocol,
1137            logger=self.logger,
1138            port=port,
1139            device_id=device_id,
1140        )
1141
1142    def _on_player_media_updated(self) -> None:
1143        """Handle callback when the current media of the player is updated."""
1144        if not self.stream or not self.stream.running:
1145            return
1146        metadata = self.state.current_media
1147        if not metadata:
1148            return
1149        progress = int(metadata.corrected_elapsed_time or 0)
1150        self.mass.create_task(self.stream.send_metadata(progress, metadata))
1151
1152    async def _get_session_pcm_format(
1153        self, sync_clients: list[AirPlayPlayer], media: PlayerMedia
1154    ) -> AudioFormat:
1155        """
1156        Select the shared PCM format for a new stream session.
1157
1158        :param sync_clients: All players that will take part in the session.
1159        :param media: The media that is about to be played.
1160        """
1161        queue = self.mass.player_queues.get(media.source_id) if media.source_id else None
1162        queue_item = (
1163            self.mass.player_queues.get_item(media.source_id, media.queue_item_id)
1164            if media.source_id and media.queue_item_id
1165            else None
1166        )
1167        streamdetails = queue_item.streamdetails if queue_item else None
1168        crossfade_enabled = bool(
1169            queue
1170            and media.media_type == MediaType.TRACK
1171            and self.mass.streams.get_crossfade_mode(queue) != CrossfadeMode.DISABLED
1172        )
1173        return await self.mass.streams.audio.select_flow_pcm_format(
1174            self,
1175            start_streamdetails=streamdetails,
1176            crossfade_enabled=crossfade_enabled,
1177            overlay_active=bool(queue and overlay_active(queue)),
1178            fallback_sample_rate=AIRPLAY_PCM_FORMAT.sample_rate,
1179            output_players=sync_clients,
1180        )
1181
1182    def _get_sync_clients(self) -> list[AirPlayPlayer]:
1183        """Get all sync clients for a player."""
1184        sync_clients: list[AirPlayPlayer] = []
1185        # we need to return the player itself too
1186        group_child_ids = {self.player_id}
1187        group_child_ids.update(self.group_members)
1188        for child_id in group_child_ids:
1189            if client := cast("AirPlayPlayer | None", self.mass.players.get_player(child_id)):
1190                sync_clients.append(client)
1191        return sync_clients
1192
1193    async def _group_rejoin_attempts(self, candidate_ids: list[str]) -> None:
1194        """Re-join this player to its group's live session after a bounded backoff."""
1195        max_attempts = len(AIRPLAY_REJOIN_ATTEMPT_DELAYS)
1196        for attempt, delay in enumerate(AIRPLAY_REJOIN_ATTEMPT_DELAYS, start=1):
1197            await asyncio.sleep(delay)
1198            if (
1199                self.group_members
1200                or (self.stream and self.stream.running)
1201                or self.playback_state != PlaybackState.IDLE
1202                # synced into a group outside the original one = deliberate regroup.
1203                # Still pointing at an original candidate is fine: a static group
1204                # keeps the sync membership while only the session lost this player.
1205                or (self.synced_to and self.synced_to not in candidate_ids)
1206            ):
1207                # the player was grouped or repurposed by other means meanwhile
1208                self.logger.debug(
1209                    "Automatic group re-join for %s cancelled: player is active again",
1210                    self.display_name,
1211                )
1212                return
1213            if not self.available:
1214                # the device is offline: an attempt cannot succeed and the user
1215                # may well have switched it off on purpose
1216                self.logger.debug(
1217                    "Automatic group re-join for %s cancelled: player is unavailable",
1218                    self.display_name,
1219                )
1220                return
1221            target = self._resolve_rejoin_target(candidate_ids)
1222            if target is None:
1223                # the group may be between sessions (e.g. a track change); keep
1224                # trying until the attempts run out
1225                self.logger.debug(
1226                    "Automatic group re-join attempt %d/%d for %s: no playing group found",
1227                    attempt,
1228                    max_attempts,
1229                    self.display_name,
1230                )
1231                continue
1232            # When the sync membership survived the stream loss (a static group,
1233            # where membership is configuration), only the running session needs
1234            # healing; a group command would no-op on the existing membership.
1235            heal_session = (
1236                target.stream.session
1237                if self.player_id in target.group_members and target.stream is not None
1238                else None
1239            )
1240            try:
1241                if heal_session is not None:
1242                    await heal_session.add_client(self)
1243                else:
1244                    await self.mass.players.cmd_group(self.player_id, target.player_id)
1245            except Exception as err:
1246                self.logger.warning(
1247                    "Automatic re-join of %s to group of %s failed (attempt %d/%d): %s",
1248                    self.display_name,
1249                    target.display_name,
1250                    attempt,
1251                    max_attempts,
1252                    err,
1253                )
1254                continue
1255            # A failed late-join is swallowed inside the grouping path (the player
1256            # then holds group membership without a live stream), so verify the
1257            # session actually carries this player before declaring success.
1258            if (
1259                self.stream
1260                and self.stream.running
1261                and self.stream.session
1262                and self in self.stream.session.sync_clients
1263            ):
1264                self.logger.info(
1265                    "Automatically re-joined %s to the group of %s after stream loss",
1266                    self.display_name,
1267                    target.display_name,
1268                )
1269                return
1270            self.logger.warning(
1271                "Automatic re-join of %s did not produce a running stream (attempt %d/%d)",
1272                self.display_name,
1273                attempt,
1274                max_attempts,
1275            )
1276            if heal_session is None:
1277                # undo the group membership this attempt created so a retry (or
1278                # a manual regroup) starts from a clean join
1279                await self.mass.players.cmd_ungroup(self.player_id)
1280        self.logger.warning(
1281            "Giving up on automatic group re-join for %s after %d attempt(s); "
1282            "the player stays idle",
1283            self.display_name,
1284            max_attempts,
1285        )
1286
1287    def _resolve_rejoin_target(self, candidate_ids: list[str]) -> AirPlayPlayer | None:
1288        """Resolve which player now carries the group's actively playing session."""
1289        for candidate_id in candidate_ids:
1290            candidate = self.mass.players.get_player(candidate_id)
1291            if candidate is None or candidate is self:
1292                continue
1293            if not isinstance(candidate, AirPlayPlayer):
1294                continue
1295            if candidate.synced_to:
1296                # the candidate was absorbed into another group since the loss
1297                # (user intent): never follow the old group's players elsewhere.
1298                # A leadership transfer inside the original group is still found:
1299                # the promoted member is itself one of the candidates.
1300                continue
1301            if not candidate.available:
1302                continue
1303            # only a PLAYING session can absorb a late joiner: a parked (paused)
1304            # session has no live timeline to anchor against
1305            if candidate.playback_state != PlaybackState.PLAYING:
1306                continue
1307            if not (candidate.stream and candidate.stream.running and candidate.stream.session):
1308                continue
1309            return candidate
1310        return None
1311
1312    def _store_device_password(self, password: str) -> None:
1313        """
1314        Persist a device password so every later stream can authenticate with it.
1315
1316        :param password: The plaintext password entered by the user.
1317        """
1318        self.mass.config.set_raw_player_config_value(
1319            self.player_id, CONF_PASSWORD, self.mass.config.encrypt_string(password)
1320        )
1321        # a freshly entered password deserves a clean slate: the reject marker
1322        # would otherwise keep the player in "needs setup" until the next connect
1323        self.set_password_invalid(False)
1324
1325
1326class GenericAirPlayPlayer(AirPlayPlayer):
1327    """AirPlay protocol endpoint without independent device control."""
1328
1329    _attr_type = PlayerType.PROTOCOL
1330