music-assistant-server

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