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