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