music-assistant-server

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