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
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(media, session_pcm_format, self.player_id)
588
589            # setup StreamSession for player (and its sync childs if any)
590            provider = cast("AirPlayProvider", self.provider)
591            stream_session = AirPlayStreamSession(
592                provider,
593                sync_clients,
594                session_pcm_format,
595                media,
596            )
597            await stream_session.start(audio_source)
598            self._transitioning = False
599
600    async def play_announcement(
601        self, announcement: PlayerMedia, volume_level: int | None = None
602    ) -> None:
603        """
604        Play an announcement natively: mixed over live playback, or as its own session.
605
606        :param announcement: Details of the announcement that needs to be played.
607        :param volume_level: Optional volume level for the announcement.
608        """
609        # The lock windows live inside the orchestration: the dispatch decision
610        # and session mutations hold self._lock like play_media does, while the
611        # multi-second clip waits run outside it (see announce.py).
612        await announce.play_announcement(self, announcement, volume_level)
613
614    async def volume_set(self, volume_level: int) -> None:
615        """Send VOLUME_SET command to given player."""
616        # Record before sending: the connect-time volume push reads this attribute,
617        # so a send that suspends first would let that push send the stale level.
618        self._attr_volume_level = volume_level
619        if self.stream and self.stream.running and self.volume_muted is not True:
620            await self.stream.send_cli_command(f"VOLUME={volume_level}")
621        self.update_state()
622        # store last state in playerconfig
623        self.mass.config.set_raw_player_config_value(
624            self.player_id, CONF_STORED_VOLUME, volume_level
625        )
626
627    async def volume_mute(self, muted: bool) -> None:
628        """Handle VOLUME_MUTE command on the player."""
629        self._attr_volume_muted = muted
630        if self.stream and self.stream.running:
631            volume = 0 if muted else (self.volume_level or 0)
632            await self.stream.send_cli_command(f"VOLUME={volume}")
633        self.update_state()
634
635    async def set_members(
636        self,
637        player_ids_to_add: list[str] | None = None,
638        player_ids_to_remove: list[str] | None = None,
639    ) -> None:
640        """Handle SET_MEMBERS command on the player."""
641        async with self._lock:
642            if self.synced_to:
643                # this should not happen, but guard anyways
644                raise RuntimeError("Player is synced, cannot set members")
645            if not player_ids_to_add and not player_ids_to_remove:
646                # nothing to do
647                return
648
649            stream_session = (
650                self.stream.session
651                if self.stream and self.stream.running and self.stream.session
652                else None
653            )
654            # handle removals first
655            if player_ids_to_remove:
656                if self.player_id in player_ids_to_remove:
657                    # Callers only ask for this leader alone or for the whole group at once.
658                    # A partial self+subset removal would need the other requested members
659                    # released here as well, instead of returning right after the leader.
660                    remaining_members = [
661                        member_id
662                        for member_id in self._attr_group_members
663                        if member_id != self.player_id and member_id not in player_ids_to_remove
664                    ]
665                    if stream_session and remaining_members:
666                        # Members stay behind: remove only this leader client,
667                        # the session continues for the remaining players
668                        await stream_session.remove_client(self, reason="leader removed from group")
669                    elif stream_session:
670                        # The whole group is being removed, tear the session down
671                        await stream_session.stop()
672                    self._attr_group_members = []
673                    self.update_state()
674                    return
675
676                for child_player in self._get_sync_clients():
677                    if child_player.player_id in player_ids_to_remove:
678                        # update group_members first to prevent race conditions
679                        # where a concurrent play_media could re-include this player
680                        if child_player.player_id in self._attr_group_members:
681                            self._attr_group_members.remove(child_player.player_id)
682                        if stream_session:
683                            await stream_session.remove_client(
684                                child_player, reason="child removed from group"
685                            )
686                        elif child_player.stream and child_player.stream.running:
687                            # leader's stream is no longer running but child still has
688                            # an active stream - stop it directly
689                            await child_player.stream.stop(force=True)
690
691                # If group leader is left alone after removals, clear the group_members list
692                if (
693                    self._attr_group_members
694                    and len(self._attr_group_members) == 1
695                    and self.player_id in self._attr_group_members
696                ):
697                    self._attr_group_members = []
698
699            # handle additions
700            for player_id in player_ids_to_add or []:
701                if player_id == self.player_id or player_id in self.group_members:
702                    # nothing to do: player is already part of the group
703                    continue
704                child_player_to_add: AirPlayPlayer | None = cast(
705                    "AirPlayPlayer | None", self.mass.players.get_player(player_id)
706                )
707                if not child_player_to_add:
708                    # should not happen, but guard against it
709                    continue
710
711                # ensure the child does not have an existing stream session active
712                if child_player_to_add := cast(
713                    "AirPlayPlayer | None", self.mass.players.get_player(player_id)
714                ):
715                    if (
716                        child_player_to_add.playback_state == PlaybackState.PAUSED
717                        and child_player_to_add.stream
718                    ):
719                        # Stop the paused stream to avoid a deadlock situation
720                        await child_player_to_add.stream.stop()
721                    if (
722                        child_player_to_add.stream
723                        and child_player_to_add.stream.running
724                        and child_player_to_add.stream.session
725                        and child_player_to_add.stream.session != stream_session
726                    ):
727                        await child_player_to_add.stream.session.remove_client(
728                            child_player_to_add, reason="moving to different session"
729                        )
730
731                # add new child to the existing stream (RAOP or AirPlay2) session (if any)
732                self._attr_group_members.append(player_id)
733                if stream_session and child_player_to_add is not None:
734                    # Skip add_client if the player is already streaming in this session
735                    # (e.g. after a dynamic leader switch where the stream continues)
736                    if child_player_to_add not in stream_session.sync_clients:
737                        await stream_session.add_client(child_player_to_add)
738                elif self.active_output_protocol not in (None, "native"):
739                    # Members can only be attached to this player's own stream session, which
740                    # does not exist while it renders through one of its output protocols.
741                    self.logger.warning(
742                        "%s joined the group of %s while that player renders through another "
743                        "output protocol: there is no stream session to join, so it stays silent",
744                        child_player_to_add.display_name if child_player_to_add else player_id,
745                        self.display_name,
746                    )
747
748            # Ensure group leader includes itself in group_members when it has members
749            # This is required for the synced_to property to work correctly
750            if self._attr_group_members and self.player_id not in self._attr_group_members:
751                self._attr_group_members.insert(0, self.player_id)
752
753            # always update the state after modifying group members
754            self.update_state()
755
756    def update_volume_from_device(self, volume: int) -> None:
757        """Update volume from device feedback."""
758        ignore_volume_report = (
759            self.config.get_value(CONF_IGNORE_VOLUME)
760            or self.device_info.manufacturer.lower() == "apple"
761        )
762
763        if ignore_volume_report:
764            return
765
766        cur_volume = self.volume_level or 0
767        if abs(cur_volume - volume) > 1 or (time.time() - self.last_command_sent) > 3:
768            self.mass.create_task(self.volume_set(volume))
769        else:
770            self._attr_volume_level = volume
771            self.mass.config.set_raw_player_config_value(self.player_id, CONF_STORED_VOLUME, volume)
772            self.update_state()
773
774    def set_discovery_info(self, discovery_info: AsyncServiceInfo, display_name: str) -> None:
775        """Set/update the discovery info for the player."""
776        self._attr_name = display_name
777        if discovery_info.type == AIRPLAY_DISCOVERY_TYPE:
778            self.airplay_discovery_info = discovery_info
779        elif discovery_info.type == RAOP_DISCOVERY_TYPE:
780            self.raop_discovery_info = discovery_info
781        else:  # guard
782            return
783        cur_address = self.address
784        prefer_ipv6 = ":" in str(self.mass.streams.publish_ip)
785        new_address = get_primary_ip_address_from_zeroconf(discovery_info, prefer_ipv6=prefer_ipv6)
786        if new_address is None:
787            # should always be set, but guard against None
788            return
789        if cur_address != new_address:
790            # Ignore mDNS updates that replace a routable address with a Docker bridge one.
791            try:
792                if (
793                    cur_address
794                    and ipaddress.ip_address(new_address) in _DOCKER_SUBNET
795                    and ipaddress.ip_address(cur_address) not in _DOCKER_SUBNET
796                ):
797                    self.logger.warning(
798                        "Ignoring mDNS update from %s to Docker address %s",
799                        cur_address,
800                        new_address,
801                    )
802                    self.update_state()
803                    return
804            except ValueError:
805                pass
806            self.logger.debug("Address updated from %s to %s", cur_address, new_address)
807            self._attr_device_info.add_identifier(IdentifierType.IP_ADDRESS, new_address)
808            self.address = new_address
809        self.update_state()
810
811    def set_state_from_stream(
812        self,
813        state: PlaybackState | None = None,
814        elapsed_time: float | None = None,
815        stream: AirPlayStream | None = None,
816    ) -> None:
817        """
818        Set the playback state from stream (RAOP or AirPlay2).
819
820        :param state: New playback state (or None to keep current).
821        :param elapsed_time: New elapsed time (or None to keep current).
822        :param stream: The stream instance sending this update (for validation).
823        """
824        # Ignore state updates from old/stale streams
825        if stream is not None and stream != self.stream:
826            return
827        # The stream reclaims the device: an external (Companion-observed)
828        # source snapshot can leak in during a brief stream-restart window and
829        # would otherwise stick, freezing the UI on a stale "external source"
830        # view while we stream. While MA streams, the stream is the sole
831        # authority on this player's state.
832        active_source = getattr(self, "_attr_active_source", None)
833        if active_source is not None and active_source in getattr(self, "_external_source_ids", ()):
834            media = getattr(self, "_attr_current_media", None)
835            if media is not None and media.source_id == active_source:
836                self._attr_current_media = None
837            self._attr_active_source = None
838        if state is not None:
839            self._attr_playback_state = state
840        if elapsed_time is not None:
841            self._attr_elapsed_time = elapsed_time
842            self._attr_elapsed_time_last_updated = time.time()
843        self.update_state()
844
845    def get_stream_pcm_format(self, session_pcm_format: AudioFormat) -> AudioFormat:
846        """
847        Return the PCM format to feed this player's cliairplay process.
848
849        :param session_pcm_format: The PCM format of the (shared) stream session.
850        """
851        if not self.hires_playback_enabled:
852            return AIRPLAY_PCM_FORMAT
853        # 24-bit: the binary expects raw s32le input on stdin (--bitdepth 24)
854        # and truncates to 24-bit ALAC internally.
855        supported_rates = {sample_rate for sample_rate, _ in self.supported_sample_rates}
856        sample_rate = (
857            session_pcm_format.sample_rate
858            if session_pcm_format.sample_rate in supported_rates
859            else AIRPLAY_PCM_FORMAT.sample_rate
860        )
861        return AudioFormat(
862            content_type=ContentType.PCM_S32LE,
863            sample_rate=sample_rate,
864            bit_depth=24,
865        )
866
867    @property
868    def owns_volume(self) -> bool:
869        """
870        Return True if this output is the resolved owner of its own volume.
871
872        AirPlay volume is the receiver's own volume: setting it writes through to the
873        device and persists there after the session ends. It may therefore only be set
874        when no other control owns the volume of this output.
875        """
876        if not (parent_id := self.protocol_parent_id):
877            # a standalone AirPlay player has no other interface to defer to
878            return True
879        if not (parent_player := self.mass.players.get_player(parent_id)):
880            return True
881        return self._control_routes_to_self(parent_player.volume_control_for_output(self.player_id))
882
883    def release_foreign_mute_latch(self) -> None:
884        """Clear our mute latch when another control owns the mute of this output."""
885        if not self._attr_volume_muted:
886            # nothing latched, so nothing that could silence this stream
887            return
888        if not (parent_id := self.protocol_parent_id):
889            return
890        if not (parent_player := self.mass.players.get_player(parent_id)):
891            return
892        if self._control_routes_to_self(parent_player.mute_control_for_output(self.player_id)):
893            # our own mute, applied through the parent
894            return
895        # The mute belongs to a control that does not own this output (a sibling interface,
896        # the receiver itself, or nothing at all). Our mute is a latch that only an explicit
897        # unmute clears, so leaving it set would report a mute we do not own and turn the
898        # next volume command into a silent one.
899        self._attr_volume_muted = False
900        self.update_state()
901
902    async def on_config_updated(self) -> None:
903        """Handle logic when the player config is updated."""
904        await super().on_config_updated()
905        prov = cast("AirPlayProvider", self.provider)
906        await prov.bridge_manager.evaluate_bridge(self)
907
908    async def on_unload(self) -> None:
909        """Handle logic when the player is unloaded from the Player controller."""
910        await super().on_unload()
911        self.cancel_group_rejoin()
912        if self.stream:
913            # remove this player from the stream session if it is running
914            if self.stream.running and self.stream.session:
915                await self.stream.session.remove_client(self, reason="player unloaded")
916            self.stream = None
917
918    def schedule_group_rejoin(self, candidate_ids: list[str]) -> None:
919        """
920        Schedule a bounded automatic re-join of this player to its still-active group.
921
922        Used when this player's stream process died unexpectedly while it was part
923        of a playing sync group (e.g. the device rode out a network blackout): the
924        player is re-added to the group's live session through the regular
925        late-join path after a short backoff. Any user action on the player (or it
926        joining a session by other means) cancels the re-join; when the group is
927        no longer playing, its membership was changed meanwhile or the device is
928        offline, the re-join is abandoned and the player simply stays idle.
929
930        :param candidate_ids: Player ids that led or shared the group at the
931            moment the stream was lost, used to resolve the re-join target (the
932            leadership may transfer while the backoff runs).
933        """
934        self.cancel_group_rejoin()
935        self.logger.info(
936            "Scheduling automatic re-join of %s to its group after unexpected stream loss",
937            self.display_name,
938        )
939        self._rejoin_task = self.mass.create_task(self._group_rejoin_attempts(candidate_ids))
940
941    def cancel_group_rejoin(self) -> None:
942        """Cancel any pending automatic group re-join attempts for this player."""
943        rejoin_task = self._rejoin_task
944        self._rejoin_task = None
945        # never self-cancel: the re-join attempt itself flows through the same
946        # session (re)start paths that call this to clear stale schedules
947        if rejoin_task and not rejoin_task.done() and rejoin_task is not asyncio.current_task():
948            rejoin_task.cancel()
949
950    def on_player_media_updated(self) -> None:
951        """Handle callback when the current media of the player is updated."""
952        if not self.stream or not self.stream.running:
953            return
954        metadata = self.state.current_media
955        if not metadata:
956            return
957        progress = int(metadata.corrected_elapsed_time or 0)
958        self.mass.create_task(self.stream.send_metadata(progress, metadata))
959
960    def _control_routes_to_self(self, control: str) -> bool:
961        """Return True if the given (resolved) control routes to this player."""
962        if control == self.player_id:
963            return True
964        # bridge players riding on this player (e.g. Sendspin-over-AirPlay) forward to us
965        if control_player := self.mass.players.get_player(control):
966            return control_player.underlying_player_id == self.player_id
967        return False
968
969    def _get_flags(self) -> int:
970        # Flags are either present via "sf" or "flags". Taken from pyatv.protocols.airplay.utils.
971        # We combine flags from both RAOP and AirPlay discovery services because
972        # LEGACY_PAIRING_BIT (0x200) is typically only in the RAOP service sf field
973        # (e.g. Apple TV HD), while PIN_REQUIRED (0x8) may only appear in the AirPlay
974        # service sf/flags field. Using only one source misses the pairing requirement.
975        flags = 0
976        for discovery_info in filter(None, [self.raop_discovery_info, self.airplay_discovery_info]):
977            raw = (
978                discovery_info.properties.get(b"sf")
979                or discovery_info.properties.get(b"flags")
980                or b"0x0"
981            )
982            with contextlib.suppress(ValueError, TypeError):
983                flags |= int(raw, 16)
984        return flags
985
986    def _requires_pin_pairing(self) -> bool:
987        """
988        Check if this device requires pairing.
989
990        Adapted from pyatv.protocols.airplay.utils.get_pairing_requirement.
991        """
992        return bool(self._get_flags() & (LEGACY_PAIRING_BIT | PIN_REQUIRED))
993
994    def _get_credentials_key(self, protocol: StreamingProtocol) -> str:
995        """Get the config key for credentials for given protocol."""
996        if protocol == StreamingProtocol.RAOP:
997            return CONF_RAOP_CREDENTIALS
998        return CONF_AIRPLAY_CREDENTIALS
999
1000    @property
1001    def _advertised_features(self) -> str | None:
1002        """Return the AirPlay features bitmask the device advertises via mDNS."""
1003        # Prefer the _airplay service's ``features``, falling back to the _raop
1004        # service's ``ft`` when the former is absent (some devices only populate one).
1005        features: str | None = None
1006        if self.airplay_discovery_info:
1007            features = self.airplay_discovery_info.decoded_properties.get(
1008                "features"
1009            ) or self.airplay_discovery_info.decoded_properties.get("ft")
1010        if not features and self.raop_discovery_info:
1011            features = self.raop_discovery_info.decoded_properties.get("ft")
1012        return features
1013
1014    @property
1015    def _is_airplay2_capable(self) -> bool:
1016        """
1017        Return whether this device can stream over AirPlay 2.
1018
1019        Mirrors the feature-bit test the cliairplay binary uses for its own route
1020        selection: a device is AirPlay 2 capable when it exposes the _airplay
1021        service and either advertises the AirPlay 2 feature bits or offers no RAOP
1022        fallback at all (i.e. it is a pure AirPlay 2 receiver).
1023        """
1024        if not self.airplay_discovery_info:
1025            return False
1026        return supports_airplay2(self._advertised_features) or not self.raop_discovery_info
1027
1028    async def _run_streaming_pairing(
1029        self, session: SetupSession, collected: dict[str, ConfigValueType]
1030    ) -> None:
1031        """
1032        Pair the streaming protocol (RAOP or AirPlay 2) and collect the device password.
1033
1034        The two are evaluated independently: a device that is already paired can
1035        still be missing its password (or have had it rejected), which is exactly
1036        the state a receiver ends up in when it gains password protection after
1037        it was set up.
1038
1039        :param session: The setup flow session used to interact with the user.
1040        :param collected: The values collected so far; updated in place.
1041        """
1042        password_collected = await self._run_protocol_pairing(session, collected)
1043        if not password_collected and self.needs_password_setup:
1044            await self._ask_device_password(session)
1045
1046    async def _run_protocol_pairing(
1047        self, session: SetupSession, collected: dict[str, ConfigValueType]
1048    ) -> bool:
1049        """
1050        Pair the streaming protocol (RAOP or AirPlay 2), unless already paired.
1051
1052        When the device requires pairing this runs it, re-offering it as a skippable
1053        step when credentials are already stored (so a re-launched flow can replace a
1054        stale pairing). When the device requires no pairing, any leftover credentials
1055        are cleared: they would keep forcing the pair-verify route, which some
1056        receivers (e.g. HomePods after their password was removed) accept while
1057        refusing to actually output audio. The obtained credentials are added to
1058        ``collected`` under the protocol-specific key.
1059
1060        :param session: The setup flow session used to interact with the user.
1061        :param collected: The values collected so far; updated in place.
1062        :return: Whether the device password was collected as part of the pairing.
1063        """
1064        pin_pairing = self._requires_pin_pairing()
1065        # a password only replaces PIN pairing on the native AirPlay 2 flow
1066        password_pairing = self.password_required and self.protocol == StreamingProtocol.AIRPLAY2
1067        if not (pin_pairing or password_pairing):
1068            for cred_key in (CONF_AIRPLAY_CREDENTIALS, CONF_RAOP_CREDENTIALS):
1069                if self.get_setup_value(cred_key) is not None:
1070                    collected[cred_key] = None
1071            return False
1072        already_paired = bool(
1073            self.get_setup_value(CONF_AIRPLAY_CREDENTIALS)
1074            or self.get_setup_value(CONF_RAOP_CREDENTIALS)
1075        )
1076        if already_paired and not await self._offer_optional_pairing(
1077            session, "streaming_repair_offer"
1078        ):
1079            return False
1080
1081        protocol = self.protocol
1082        cred_key = self._get_credentials_key(protocol)
1083        if pin_pairing:
1084            step_id, field_key, field_type = "pair_pin", CONF_PAIRING_PIN, ConfigEntryType.STRING
1085        else:
1086            step_id, field_key, field_type = (
1087                "pair_password",
1088                CONF_PAIRING_PASSWORD,
1089                ConfigEntryType.SECURE_STRING,
1090            )
1091
1092        errors: dict[str, str] | None = None
1093        while True:
1094            # Each attempt uses a fresh session: finish_pairing() closes the live
1095            # subprocess/session on completion, so a rejected PIN needs a new one
1096            # (and the device re-shows its PIN).
1097            pairing = await self._prepare_streaming_pairing(protocol, pin_pairing=pin_pairing)
1098            try:
1099                values = await session.form(
1100                    [
1101                        ConfigEntry(
1102                            key=field_key,
1103                            type=field_type,
1104                            required=True,
1105                            category="protocol_generic",
1106                        )
1107                    ],
1108                    step_id=step_id,
1109                    errors=errors,
1110                )
1111                entered_value = str(values[field_key])
1112                credentials = await pairing.finish_pairing(pin=entered_value)
1113            except PlayerCommandFailed as err:
1114                # leave a default-level trace: the flow swallows the error into
1115                # the re-served form, which support logs otherwise never show
1116                self.logger.warning("Pairing with %s failed: %s", self.display_name, err)
1117                errors = {"base": err.translation_key or str(err)}
1118                continue
1119            finally:
1120                # tears down the subprocess on retry, success and abort (cancellation)
1121                await pairing.close()
1122            collected[cred_key] = credentials
1123            if password_pairing:
1124                # The device password authenticates every later stream too (the
1125                # binary's transient leg), so keep it next to the credentials
1126                # instead of discarding it with the setup form.
1127                self._store_device_password(entered_value)
1128            return password_pairing
1129
1130    async def _ask_device_password(self, session: SetupSession) -> None:
1131        """
1132        Ask for the device password and store it, without attempting any pairing.
1133
1134        Covers the devices that have no pairing to do: a legacy RAOP receiver, and
1135        an already paired device whose password is missing or was rejected. There
1136        is no live session to validate the entry against, so a wrong password only
1137        surfaces on the next connect - which marks the player as needing setup again.
1138
1139        :param session: The setup flow session used to interact with the user.
1140        """
1141        values = await session.form(
1142            [
1143                ConfigEntry(
1144                    key=CONF_PAIRING_PASSWORD,
1145                    type=ConfigEntryType.SECURE_STRING,
1146                    required=True,
1147                    category="protocol_generic",
1148                )
1149            ],
1150            step_id="pair_password",
1151        )
1152        self._store_device_password(str(values[CONF_PAIRING_PASSWORD]))
1153
1154    async def _offer_optional_pairing(self, session: SetupSession, step_id: str) -> bool:
1155        """
1156        Ask whether to run the offered (optional) pairing now.
1157
1158        :param session: The setup flow session used to interact with the user.
1159        :param step_id: The (i18n) step id describing the offered pairing.
1160        """
1161        values = await session.form(
1162            [
1163                ConfigEntry(
1164                    key=CONF_PAIR_NOW,
1165                    type=ConfigEntryType.BOOLEAN,
1166                    default_value=False,
1167                    category="protocol_generic",
1168                )
1169            ],
1170            step_id=step_id,
1171        )
1172        return bool(values[CONF_PAIR_NOW])
1173
1174    async def _prepare_streaming_pairing(
1175        self, protocol: StreamingProtocol, *, pin_pairing: bool
1176    ) -> AirPlayPairing:
1177        """
1178        Build and start a streaming pairing session (the device shows its PIN).
1179
1180        A failure here cannot be recovered by re-prompting the user, so it aborts the
1181        flow; a partially started session is torn down first.
1182
1183        :param protocol: The streaming protocol to pair (RAOP or AirPlay 2).
1184        :param pin_pairing: Whether the device shows a PIN the user must enter.
1185        """
1186        pairing: AirPlayPairing | None = None
1187        started = False
1188        try:
1189            pairing = self._build_streaming_pairing(protocol)
1190            await pairing.start_pairing_session()
1191            if pin_pairing:
1192                await pairing.start_pin_pairing()
1193            started = True
1194        except Exception as err:
1195            # a failure starting the session (device unreachable, binary/system
1196            # issue, ...) cannot be fixed by re-prompting, so abort with a clear
1197            # reason instead of letting it surface as a generic internal error
1198            self.logger.warning("Could not start AirPlay pairing session: %s", err)
1199            raise AbortFlow("pairing_failed") from err
1200        finally:
1201            if not started and pairing is not None:
1202                await pairing.close()
1203        assert pairing is not None  # reached only when started, i.e. a live session
1204        return pairing
1205
1206    def _build_streaming_pairing(self, protocol: StreamingProtocol) -> AirPlayPairing:
1207        """
1208        Build an AirPlayPairing for the given streaming protocol.
1209
1210        :param protocol: The streaming protocol to pair (RAOP or AirPlay 2).
1211        """
1212        from .pairing import AirPlayPairing  # noqa: PLC0415
1213
1214        # For Apple devices pairing always happens on the AirPlay port (7000) even
1215        # when streaming will use RAOP; the RAOP port (5000) is only for streaming.
1216        port: int | None = None
1217        if self.airplay_discovery_info:
1218            port = self.airplay_discovery_info.port or 7000
1219        elif self.raop_discovery_info:
1220            port = self.raop_discovery_info.port or 5000
1221        provider = cast("AirPlayProvider", self.provider)
1222        device_id = provider.dacp_id
1223        pairing_address = self.address
1224        if protocol == StreamingProtocol.AIRPLAY2 and not isinstance(
1225            ipaddress.ip_address(pairing_address), ipaddress.IPv4Address
1226        ):
1227            if self.airplay_discovery_info:
1228                discovered_address = get_primary_ip_address_from_zeroconf(
1229                    self.airplay_discovery_info
1230                )
1231                if discovered_address and isinstance(
1232                    ipaddress.ip_address(discovered_address), ipaddress.IPv4Address
1233                ):
1234                    pairing_address = discovered_address
1235            if not isinstance(ipaddress.ip_address(pairing_address), ipaddress.IPv4Address):
1236                raise PlayerCommandFailed("AirPlay pairing requires an IPv4 device address")
1237        return AirPlayPairing(
1238            address=pairing_address,
1239            name=self.display_name,
1240            protocol=protocol,
1241            logger=self.logger,
1242            port=port,
1243            device_id=device_id,
1244        )
1245
1246    async def _get_session_pcm_format(
1247        self, sync_clients: list[AirPlayPlayer], media: PlayerMedia
1248    ) -> AudioFormat:
1249        """
1250        Select the shared PCM format for a new stream session.
1251
1252        :param sync_clients: All players that will take part in the session.
1253        :param media: The media that is about to be played.
1254        """
1255        queue = self.mass.player_queues.get(media.source_id) if media.source_id else None
1256        queue_item = (
1257            self.mass.player_queues.get_item(media.source_id, media.queue_item_id)
1258            if media.source_id and media.queue_item_id
1259            else None
1260        )
1261        streamdetails = queue_item.streamdetails if queue_item else None
1262        crossfade_enabled = bool(
1263            queue
1264            and media.media_type == MediaType.TRACK
1265            and self.mass.streams.get_crossfade_mode(queue) != CrossfadeMode.DISABLED
1266        )
1267        return await self.mass.streams.audio.select_flow_pcm_format(
1268            self,
1269            start_streamdetails=streamdetails,
1270            crossfade_enabled=crossfade_enabled,
1271            overlay_active=bool(queue and overlay_active(queue)),
1272            fallback_sample_rate=AIRPLAY_PCM_FORMAT.sample_rate,
1273            output_players=sync_clients,
1274        )
1275
1276    def _get_sync_clients(self) -> list[AirPlayPlayer]:
1277        """Get all sync clients for a player."""
1278        sync_clients: list[AirPlayPlayer] = []
1279        # we need to return the player itself too
1280        group_child_ids = {self.player_id}
1281        group_child_ids.update(self.group_members)
1282        for child_id in group_child_ids:
1283            if client := cast("AirPlayPlayer | None", self.mass.players.get_player(child_id)):
1284                sync_clients.append(client)
1285        return sync_clients
1286
1287    async def _group_rejoin_attempts(self, candidate_ids: list[str]) -> None:
1288        """Re-join this player to its group's live session after a bounded backoff."""
1289        max_attempts = len(AIRPLAY_REJOIN_ATTEMPT_DELAYS)
1290        for attempt, delay in enumerate(AIRPLAY_REJOIN_ATTEMPT_DELAYS, start=1):
1291            await asyncio.sleep(delay)
1292            if (
1293                self.group_members
1294                or (self.stream and self.stream.running)
1295                or self.playback_state != PlaybackState.IDLE
1296                # synced into a group outside the original one = deliberate regroup.
1297                # Still pointing at an original candidate is fine: a static group
1298                # keeps the sync membership while only the session lost this player.
1299                or (self.synced_to and self.synced_to not in candidate_ids)
1300            ):
1301                # the player was grouped or repurposed by other means meanwhile
1302                self.logger.debug(
1303                    "Automatic group re-join for %s cancelled: player is active again",
1304                    self.display_name,
1305                )
1306                return
1307            if not self.available:
1308                # the device is offline: an attempt cannot succeed and the user
1309                # may well have switched it off on purpose
1310                self.logger.debug(
1311                    "Automatic group re-join for %s cancelled: player is unavailable",
1312                    self.display_name,
1313                )
1314                return
1315            target = self._resolve_rejoin_target(candidate_ids)
1316            if target is None:
1317                # the group may be between sessions (e.g. a track change); keep
1318                # trying until the attempts run out
1319                self.logger.debug(
1320                    "Automatic group re-join attempt %d/%d for %s: no playing group found",
1321                    attempt,
1322                    max_attempts,
1323                    self.display_name,
1324                )
1325                continue
1326            # When the sync membership survived the stream loss (a static group,
1327            # where membership is configuration), only the running session needs
1328            # healing; a group command would no-op on the existing membership.
1329            heal_session = (
1330                target.stream.session
1331                if self.player_id in target.group_members and target.stream is not None
1332                else None
1333            )
1334            try:
1335                if heal_session is not None:
1336                    await heal_session.add_client(self)
1337                else:
1338                    await self.mass.players.cmd_group(self.player_id, target.player_id)
1339            except Exception as err:
1340                self.logger.warning(
1341                    "Automatic re-join of %s to group of %s failed (attempt %d/%d): %s",
1342                    self.display_name,
1343                    target.display_name,
1344                    attempt,
1345                    max_attempts,
1346                    err,
1347                )
1348                continue
1349            # A failed late-join is swallowed inside the grouping path (the player
1350            # then holds group membership without a live stream), so verify the
1351            # session actually carries this player before declaring success.
1352            if (
1353                self.stream
1354                and self.stream.running
1355                and self.stream.session
1356                and self in self.stream.session.sync_clients
1357            ):
1358                self.logger.info(
1359                    "Automatically re-joined %s to the group of %s after stream loss",
1360                    self.display_name,
1361                    target.display_name,
1362                )
1363                return
1364            self.logger.warning(
1365                "Automatic re-join of %s did not produce a running stream (attempt %d/%d)",
1366                self.display_name,
1367                attempt,
1368                max_attempts,
1369            )
1370            if heal_session is None:
1371                # undo the group membership this attempt created so a retry (or
1372                # a manual regroup) starts from a clean join
1373                await self.mass.players.cmd_ungroup(self.player_id)
1374        self.logger.warning(
1375            "Giving up on automatic group re-join for %s after %d attempt(s); "
1376            "the player stays idle",
1377            self.display_name,
1378            max_attempts,
1379        )
1380
1381    def _resolve_rejoin_target(self, candidate_ids: list[str]) -> AirPlayPlayer | None:
1382        """Resolve which player now carries the group's actively playing session."""
1383        for candidate_id in candidate_ids:
1384            candidate = self.mass.players.get_player(candidate_id)
1385            if candidate is None or candidate is self:
1386                continue
1387            if not isinstance(candidate, AirPlayPlayer):
1388                continue
1389            if candidate.synced_to:
1390                # the candidate was absorbed into another group since the loss
1391                # (user intent): never follow the old group's players elsewhere.
1392                # A leadership transfer inside the original group is still found:
1393                # the promoted member is itself one of the candidates.
1394                continue
1395            if not candidate.available:
1396                continue
1397            # only a PLAYING session can absorb a late joiner: a parked (paused)
1398            # session has no live timeline to anchor against
1399            if candidate.playback_state != PlaybackState.PLAYING:
1400                continue
1401            if not (candidate.stream and candidate.stream.running and candidate.stream.session):
1402                continue
1403            return candidate
1404        return None
1405
1406    def _store_device_password(self, password: str) -> None:
1407        """
1408        Persist a device password so every later stream can authenticate with it.
1409
1410        :param password: The plaintext password entered by the user.
1411        """
1412        self.mass.config.set_raw_player_config_value(
1413            self.player_id, CONF_PASSWORD, self.mass.config.encrypt_string(password)
1414        )
1415        # a freshly entered password deserves a clean slate: the reject marker
1416        # would otherwise keep the player in "needs setup" until the next connect
1417        self.set_password_invalid(False)
1418
1419
1420class GenericAirPlayPlayer(AirPlayPlayer):
1421    """AirPlay protocol endpoint without independent device control."""
1422
1423    _attr_type = PlayerType.PROTOCOL
1424