music-assistant-server

1"""Control-capable player implementation for the AirPlay provider."""
2
3from __future__ import annotations
4
5import asyncio
6import contextlib
7import time
8from collections.abc import Awaitable, Callable
9from ipaddress import AddressValueError, IPv4Address
10from typing import TYPE_CHECKING, Final
11
12import pyatv
13from music_assistant_models.config_entries import ConfigEntry, ConfigValueType
14from music_assistant_models.enums import (
15    ConfigEntryType,
16    ImageType,
17    MediaType,
18    PlaybackState,
19    PlayerFeature,
20    PlayerType,
21)
22from music_assistant_models.errors import MediaNotFoundError, PlayerCommandFailed
23from music_assistant_models.media_items import MediaItemImage
24from music_assistant_models.player import PlayerSource
25from pyatv import exceptions as pyatv_exceptions
26from pyatv.conf import AppleTV as AppleTVConfig
27from pyatv.conf import ManualService
28from pyatv.const import (
29    DeviceState,
30    FeatureName,
31    FeatureState,
32    PairingRequirement,
33    PowerState,
34    Protocol,
35)
36from pyatv.const import (
37    MediaType as PyatvMediaType,
38)
39from pyatv.interface import (
40    AppleTV,
41    AudioListener,
42    DeviceListener,
43    OutputDevice,
44    PairingHandler,
45    Playing,
46    PowerListener,
47    PushListener,
48)
49from pyatv.settings import MrpTunnel
50from pyatv.storage.memory_storage import MemoryStorage
51
52from music_assistant.models.player import PlayerMedia
53from music_assistant.models.setup_flow import AbortFlow
54
55from .constants import (
56    CONF_COMPANION_CREDENTIALS,
57    CONF_COMPANION_PAIRING_PIN,
58    CONF_MRP_CREDENTIALS,
59    CONF_MRP_PAIRING_PIN,
60    CONF_NATIVE_MRP_CREDENTIALS,
61    CONF_STORED_VOLUME,
62    EXTERNAL_ARTWORK_PATH_PREFIX,
63    FALLBACK_VOLUME,
64)
65from .helpers import (
66    get_decoded_property,
67    supports_companion_pairing,
68    supports_mrp_service,
69    supports_mrp_tunnel,
70    supports_transient_mrp,
71)
72from .player import AirPlayPlayer
73
74if TYPE_CHECKING:
75    from zeroconf.asyncio import AsyncServiceInfo
76
77    from music_assistant.models.setup_flow import SetupSession
78
79    from .provider import AirPlayProvider
80
81_CONTROL_RECONNECT_DELAY: Final[float] = 30.0
82_WAKE_TIMEOUT: Final[float] = 10.0
83
84# mDNS TXT keys whose values change with transient playback, session or group
85# state - most notably `flags`, which a receiver toggles while it is receiving a
86# stream. They never affect how the control connection is established, so a
87# change must not force a reconnect: doing so made Apple TVs tear down and
88# re-establish the control channel on every stream start and stop, each time
89# surfacing the on-screen pairing code. Compared case-insensitively (RFC 6763).
90_VOLATILE_DISCOVERY_KEYS: Final = frozenset({"flags", "gcgl", "gid", "igl", "gpn", "pgcgl"})
91
92_CONNECTION_ERRORS = (
93    pyatv_exceptions.AuthenticationError,
94    pyatv_exceptions.BackOffError,
95    pyatv_exceptions.ConnectionFailedError,
96    pyatv_exceptions.DeviceIdMissingError,
97    pyatv_exceptions.InvalidConfigError,
98    pyatv_exceptions.InvalidCredentialsError,
99    pyatv_exceptions.InvalidResponseError,
100    pyatv_exceptions.NoCredentialsError,
101    pyatv_exceptions.NoServiceError,
102    pyatv_exceptions.OperationTimeoutError,
103    pyatv_exceptions.ProtocolError,
104    OSError,
105    TimeoutError,
106    ValueError,
107)
108_COMMAND_ERRORS = (
109    pyatv_exceptions.AuthenticationError,
110    pyatv_exceptions.BlockedStateError,
111    pyatv_exceptions.CommandError,
112    pyatv_exceptions.ConnectionLostError,
113    pyatv_exceptions.InvalidStateError,
114    pyatv_exceptions.NotSupportedError,
115    pyatv_exceptions.OperationTimeoutError,
116    pyatv_exceptions.ProtocolError,
117    OSError,
118    TimeoutError,
119)
120
121
122class AirPlayControlPlayer(AirPlayPlayer):
123    """AirPlay player with independent device monitoring and control."""
124
125    _attr_type = PlayerType.PLAYER
126
127    def __init__(  # noqa: PLR0913
128        self,
129        provider: AirPlayProvider,
130        player_id: str,
131        raop_discovery_info: AsyncServiceInfo | None,
132        airplay_discovery_info: AsyncServiceInfo | None,
133        companion_discovery_info: AsyncServiceInfo | None,
134        mrp_discovery_info: AsyncServiceInfo | None,
135        address: str,
136        display_name: str,
137        manufacturer: str,
138        model: str,
139        initial_volume: int,
140    ) -> None:
141        """Initialize a control-capable AirPlay player."""
142        self.companion_discovery_info = companion_discovery_info
143        self.mrp_discovery_info = mrp_discovery_info
144        self._companion_device: AppleTV | None = None
145        self._mrp_device: AppleTV | None = None
146        super().__init__(
147            provider=provider,
148            player_id=player_id,
149            raop_discovery_info=raop_discovery_info,
150            airplay_discovery_info=airplay_discovery_info,
151            address=address,
152            display_name=display_name,
153            manufacturer=manufacturer,
154            model=model,
155            initial_volume=initial_volume,
156        )
157        self._companion_listener: _AirPlayStateListener | None = None
158        self._mrp_state_listener: _AirPlayStateListener | None = None
159        self._mrp_push_listener: _AirPlayPushListener | None = None
160        self._connection_task: asyncio.Task[None] | None = None
161        self._connection_lock = asyncio.Lock()
162        self._power_on_event = asyncio.Event()
163        self._volume_before_mute: int | None = None
164        self._disconnecting = False
165        self._restart_connections = False
166        self._unloading = False
167        # invoked (if set) whenever the Companion connection comes up or goes down,
168        # so an observer (e.g. the dashboard adapter) can re-evaluate its state
169        self.on_companion_state_change: Callable[[], None] | None = None
170
171    @property
172    def companion_pairing_supported(self) -> bool:
173        """Return whether this device advertises Companion PIN pairing."""
174        return supports_companion_pairing(self.companion_discovery_info)
175
176    @property
177    def mrp_pairing_supported(self) -> bool:
178        """Return whether MRP playback monitoring can be paired."""
179        endpoint = self._mrp_endpoint
180        if endpoint is None:
181            return False
182        discovery_info, protocol = endpoint
183        if protocol == Protocol.MRP:
184            allow_pairing = get_decoded_property(discovery_info, "AllowPairing") or "no"
185            return allow_pairing.lower() == "yes"
186        return bool(
187            not self._uses_transient_mrp
188            and protocol == Protocol.AirPlay
189            and self._is_airplay2_capable
190            and discovery_info.decoded_properties.get("acl", "0") != "1"
191        )
192
193    @property
194    def supported_features(self) -> set[PlayerFeature]:
195        """Return the supported features of this controlled device."""
196        features = {*super().supported_features}
197        if self._device_for_feature(FeatureName.Next) or self._device_for_feature(
198            FeatureName.Previous
199        ):
200            features.add(PlayerFeature.NEXT_PREVIOUS)
201        # POWER is advertised only when it can actually be served: a connected
202        # control channel exposing power commands, or stored Companion
203        # credentials (so the feature does not flap while (re)connecting).
204        if (
205            self.get_setup_value(CONF_COMPANION_CREDENTIALS)
206            or self._device_for_power_feature(FeatureName.TurnOn)
207            or self._device_for_power_feature(FeatureName.TurnOff)
208        ):
209            features.add(PlayerFeature.POWER)
210        if not self._stream_active and not self._device_for_feature(FeatureName.SetVolume):
211            features.discard(PlayerFeature.VOLUME_MUTE)
212        return features
213
214    @property
215    def companion_connected(self) -> bool:
216        """Return whether the Companion control channel is currently connected."""
217        return self._companion_device is not None
218
219    async def get_config_entries(self) -> list[ConfigEntry]:
220        """Return player configuration entries."""
221        # Companion/MRP pairing is handled by the interactive setup flow
222        # (run_setup_flow) and stored in setup_data, no longer as config entries.
223        return await super().get_config_entries()
224
225    async def run_setup_flow(self, session: SetupSession) -> None:
226        """
227        Run the interactive setup flow for this controlled AirPlay player.
228
229        The streaming pairing (if any) is the required "device code" that gates
230        ``needs_setup``; the optional Companion (remote control) and MRP (playback
231        monitoring) pairings are offered afterwards as sequential, skippable steps.
232        Re-launching from the player settings re-offers every pairing, so a stored
233        pairing can be redone (replaced) when it went stale.
234
235        :param session: The setup flow session used to interact with the user.
236        """
237        collected: dict[str, ConfigValueType] = {}
238        await self._run_streaming_pairing(session, collected)
239        await self._run_companion_pairing(session, collected)
240        await self._run_mrp_pairing(session, collected)
241        await session.finish(collected)
242        if collected.keys() & {
243            CONF_COMPANION_CREDENTIALS,
244            CONF_MRP_CREDENTIALS,
245            CONF_NATIVE_MRP_CREDENTIALS,
246        }:
247            # bring the freshly paired control channel up right away
248            self._schedule_connection(force=True)
249
250    async def power(self, powered: bool) -> None:
251        """Turn the controlled device on or off."""
252        feature = FeatureName.TurnOn if powered else FeatureName.TurnOff
253        device = self._device_for_power_feature(feature)
254        if device is None:
255            raise PlayerCommandFailed(f"Power control is unavailable for {self.display_name}")
256        if powered:
257            self._power_on_event.clear()
258            await self._run_control_command(device.power.turn_on(), "turn on")
259            await self._wait_for_wake()
260        else:
261            await self._run_control_command(device.power.turn_off(), "turn off")
262
263    async def play(self) -> None:
264        """Resume Music Assistant or external playback."""
265        await self._wake_for_playback()
266        if self._stream_active:
267            await super().play()
268            return
269        device = self._device_for_feature(FeatureName.Play)
270        if device is None:
271            device = self._device_for_feature(FeatureName.PlayPause)
272            if device is None:
273                raise PlayerCommandFailed(f"Play control is unavailable for {self.display_name}")
274            await self._run_control_command(device.remote_control.play_pause(), "play")
275            return
276        await self._run_control_command(device.remote_control.play(), "play")
277
278    async def pause(self) -> None:
279        """Pause Music Assistant or external playback."""
280        if self._stream_active:
281            await super().pause()
282            return
283        device = self._device_for_feature(FeatureName.Pause)
284        if device is None:
285            device = self._device_for_feature(FeatureName.PlayPause)
286            if device is None:
287                raise PlayerCommandFailed(f"Pause control is unavailable for {self.display_name}")
288            await self._run_control_command(device.remote_control.play_pause(), "pause")
289            return
290        await self._run_control_command(device.remote_control.pause(), "pause")
291
292    async def stop(self) -> None:
293        """Stop Music Assistant playback, or return the device to its home screen."""
294        if self._stream_active:
295            await super().stop()
296            return
297        # For external playback there is no real "stop"; returning to the home
298        # screen backgrounds the current app, which is the closest equivalent.
299        device = self._device_for_feature(FeatureName.Home)
300        if device is not None:
301            await self._run_control_command(device.remote_control.home(), "stop")
302            return
303        device = self._device_for_feature(FeatureName.Stop)
304        if device is not None:
305            await self._run_control_command(device.remote_control.stop(), "stop")
306            return
307        device = self._device_for_feature(FeatureName.Pause)
308        if device is not None:
309            await self._run_control_command(device.remote_control.pause(), "stop")
310            return
311        raise PlayerCommandFailed(f"Stop control is unavailable for {self.display_name}")
312
313    async def play_media(self, media: PlayerMedia) -> None:
314        """Wake the controlled device and start Music Assistant playback."""
315        await self._wake_for_playback()
316        await super().play_media(media)
317
318    async def volume_set(self, volume_level: int) -> None:
319        """Set stream or native device volume."""
320        if self._stream_active:
321            await super().volume_set(volume_level)
322            return
323        device = self._device_for_feature(FeatureName.SetVolume)
324        if device is None:
325            await super().volume_set(volume_level)
326            return
327        await self._run_volume_command(device.audio.set_volume(volume_level), "set volume")
328        self._handle_volume_update("command", volume_level)
329
330    async def volume_mute(self, muted: bool) -> None:
331        """Mute an active stream or native device volume."""
332        if self._stream_active:
333            await super().volume_mute(muted)
334            return
335        device = self._device_for_feature(FeatureName.SetVolume)
336        if device is None:
337            raise PlayerCommandFailed(f"Mute control is unavailable for {self.display_name}")
338        if muted:
339            if self.volume_muted:
340                return
341            if self.volume_level and self.volume_level > 0:
342                self._volume_before_mute = self.volume_level
343            await self._run_volume_command(device.audio.set_volume(0), "mute")
344            self._handle_volume_update("command", 0)
345            return
346        if not self.volume_muted:
347            return
348        volume = self._volume_before_mute or self.volume_level or FALLBACK_VOLUME
349        await self._run_volume_command(device.audio.set_volume(volume), "unmute")
350        self._handle_volume_update("command", volume)
351
352    async def next_track(self) -> None:
353        """Skip to the next item in external playback."""
354        device = self._device_for_feature(FeatureName.Next)
355        if device is None:
356            raise PlayerCommandFailed(f"Next control is unavailable for {self.display_name}")
357        await self._run_control_command(device.remote_control.next(), "skip to next")
358
359    async def previous_track(self) -> None:
360        """Return to the previous item in external playback."""
361        device = self._device_for_feature(FeatureName.Previous)
362        if device is None:
363            raise PlayerCommandFailed(f"Previous control is unavailable for {self.display_name}")
364        await self._run_control_command(device.remote_control.previous(), "skip to previous")
365
366    async def wake(self) -> None:
367        """Wake the device from sleep when it exposes power control."""
368        await self._wake_for_playback()
369
370    async def async_list_installed_app_ids(self) -> set[str] | None:
371        """
372        Return the bundle ids of the apps installed on the device.
373
374        Uses the Companion app-listing feature. Returns None when the app list cannot be
375        retrieved (Companion channel down or the query failed), so a caller can tell
376        "unknown" apart from "installed, but not this app".
377        """
378        device = self._device_for_feature(FeatureName.AppList)
379        if device is None:
380            return None
381        try:
382            apps = await device.apps.app_list()
383        except _COMMAND_ERRORS as err:
384            self.logger.debug("Unable to list installed apps for %s: %s", self.name, err)
385            return None
386        return {app.identifier for app in apps}
387
388    async def async_launch_app(self, bundle_id_or_url: str) -> None:
389        """
390        Launch an app (bundle id) or custom URL on the device over Companion.
391
392        :param bundle_id_or_url: A bundle id or a URL-scheme value to launch.
393        :raises PlayerCommandFailed: If app launching is unavailable or the launch fails.
394        """
395        device = self._device_for_feature(FeatureName.LaunchApp)
396        if device is None:
397            raise PlayerCommandFailed(f"App launching is unavailable for {self.display_name}")
398        await self._run_control_command(device.apps.launch_app(bundle_id_or_url), "launch app")
399
400    async def async_get_external_artwork(self, artwork_id: str) -> bytes:
401        """
402        Return artwork bytes for the current externally playing item.
403
404        :param artwork_id: Identifier of the artwork requested by the image proxy.
405        :raises MediaNotFoundError: If the artwork is stale or unavailable.
406        """
407        device = self._mrp_device
408        if (
409            device is None
410            or self._stream_active
411            or not self._feature_available(device, FeatureName.Artwork)
412            or device.metadata.artwork_id != artwork_id
413        ):
414            raise MediaNotFoundError("External AirPlay artwork is unavailable")
415        try:
416            artwork = await device.metadata.artwork()
417        except _COMMAND_ERRORS as err:
418            raise MediaNotFoundError("Unable to retrieve external AirPlay artwork") from err
419        if (
420            artwork is None
421            or not artwork.bytes
422            or self._mrp_device is not device
423            or self._stream_active
424            or device.metadata.artwork_id != artwork_id
425        ):
426            raise MediaNotFoundError("External AirPlay artwork is no longer current")
427        return artwork.bytes
428
429    def set_discovery_info(self, discovery_info: AsyncServiceInfo, display_name: str) -> None:
430        """Update AirPlay discovery data and reconnect device control if needed."""
431        previous_signature = self._service_signature(self.airplay_discovery_info)
432        previous_address = self.address
433        super().set_discovery_info(discovery_info, display_name)
434        if (
435            previous_signature != self._service_signature(self.airplay_discovery_info)
436            or previous_address != self.address
437        ):
438            self._schedule_connection(force=True)
439
440    async def set_companion_discovery_info(self, discovery_info: AsyncServiceInfo | None) -> None:
441        """Update Companion discovery data and reconnect the control channel."""
442        if self._service_signature(self.companion_discovery_info) == self._service_signature(
443            discovery_info
444        ):
445            return
446        self.companion_discovery_info = discovery_info
447        self.update_state()
448        self._schedule_connection(force=True)
449
450    async def set_mrp_discovery_info(self, discovery_info: AsyncServiceInfo | None) -> None:
451        """Update native MRP discovery data and reconnect playback monitoring."""
452        if self._service_signature(self.mrp_discovery_info) == self._service_signature(
453            discovery_info
454        ):
455            return
456        self.mrp_discovery_info = discovery_info
457        self.update_state()
458        self._schedule_connection(force=True)
459
460    async def on_config_updated(self) -> None:
461        """Reconnect control services when player configuration changes."""
462        await super().on_config_updated()
463        self._schedule_connection(force=True)
464
465    async def on_unload(self) -> None:
466        """Close control connections and pairing resources."""
467        self._unloading = True
468        if self._connection_task and not self._connection_task.done():
469            self._connection_task.cancel()
470            with contextlib.suppress(asyncio.CancelledError):
471                await self._connection_task
472        await self._disconnect_control_services()
473        await super().on_unload()
474
475    def _schedule_connection(self, *, force: bool = False) -> None:
476        """Start or restart the Apple service connection loop."""
477        if self._unloading:
478            return
479        if self._connection_task and not self._connection_task.done():
480            if not force:
481                return
482            self._connection_task.cancel()
483        self._restart_connections = self._restart_connections or force
484        self._connection_task = self.mass.create_task(
485            self._connection_loop,
486            task_id=f"airplay_apple_control_{self.player_id}",
487            abort_existing=force,
488        )
489
490    async def _connection_loop(self) -> None:
491        """Connect control services and retry transient failures."""
492        retry = True
493        while retry and not self._unloading:
494            retry = await self._connect_control_services()
495            if retry:
496                await asyncio.sleep(_CONTROL_RECONNECT_DELAY)
497
498    async def _connect_control_services(self) -> bool:
499        """Connect Companion and MRP independently."""
500        async with self._connection_lock:
501            if self._restart_connections:
502                self._restart_connections = False
503                await self._disconnect_control_services()
504            companion_retry, mrp_retry = await asyncio.gather(
505                self._connect_companion(),
506                self._connect_mrp(),
507            )
508            return companion_retry or mrp_retry
509
510    async def _connect_companion(self) -> bool:
511        """Connect the Companion control channel."""
512        if self._companion_device is not None:
513            return False
514        credentials = self.get_setup_value(CONF_COMPANION_CREDENTIALS)
515        if not credentials or not self.companion_discovery_info:
516            return False
517        config = self._build_config(
518            self.companion_discovery_info,
519            Protocol.Companion,
520            str(credentials),
521            PairingRequirement.Mandatory,
522        )
523        if config is None:
524            return False
525        try:
526            device = await pyatv.connect(config, self.mass.loop)
527        except pyatv_exceptions.AuthenticationError, pyatv_exceptions.InvalidCredentialsError:
528            self.logger.warning(
529                "Stored Companion credentials are no longer valid for %s",
530                self.display_name,
531            )
532            self._clear_stored_credentials(CONF_COMPANION_CREDENTIALS)
533            return False
534        except _CONNECTION_ERRORS as err:
535            self.logger.debug("Unable to connect Companion control for %s: %s", self.name, err)
536            return True
537
538        self._companion_device = device
539        listener = _AirPlayStateListener(self, device, "companion")
540        self._companion_listener = listener
541        device.listener = listener
542        device.power.listener = listener
543        device.audio.listener = listener
544        self._apply_initial_device_state(device, "companion")
545        self.update_state()
546        self.logger.debug("Connected Companion control for %s", self.display_name)
547        self._notify_companion_state_change()
548        return False
549
550    async def _connect_mrp(self) -> bool:
551        """Connect MRP playback monitoring."""
552        if self._mrp_device is not None:
553            return False
554        endpoint = self._mrp_endpoint
555        if endpoint is None:
556            return False
557        discovery_info, protocol = endpoint
558        credentials = self._mrp_credentials
559        if protocol == Protocol.AirPlay and credentials is None and not self._uses_transient_mrp:
560            return False
561        config = self._build_config(
562            discovery_info,
563            protocol,
564            str(credentials) if credentials else None,
565            PairingRequirement.NotNeeded,
566        )
567        if config is None:
568            return False
569        storage: MemoryStorage | None = None
570        if protocol == Protocol.AirPlay:
571            storage = MemoryStorage()
572            settings = await storage.get_settings(config)
573            settings.protocols.airplay.mrp_tunnel = MrpTunnel.Force
574        try:
575            device = await pyatv.connect(config, self.mass.loop, storage=storage)
576        except (
577            pyatv_exceptions.AuthenticationError,
578            pyatv_exceptions.InvalidCredentialsError,
579        ) as err:
580            self.logger.warning(
581                "Unable to authenticate playback monitoring for %s: %s", self.name, err
582            )
583            if credentials:
584                self._clear_stored_credentials(self._mrp_credentials_key)
585            return False
586        except _CONNECTION_ERRORS as err:
587            self.logger.debug("Unable to connect playback monitoring for %s: %s", self.name, err)
588            return True
589
590        if not self._feature_available(device, FeatureName.PushUpdates):
591            device.close()
592            self.logger.debug("Playback monitoring is not supported by %s", self.name)
593            return False
594
595        self._mrp_device = device
596        state_listener = _AirPlayStateListener(self, device, "mrp")
597        push_listener = _AirPlayPushListener(self, device)
598        self._mrp_state_listener = state_listener
599        self._mrp_push_listener = push_listener
600        device.listener = state_listener
601        device.power.listener = state_listener
602        device.audio.listener = state_listener
603        device.push_updater.listener = push_listener
604        device.push_updater.start()
605        self._apply_initial_device_state(device, "mrp")
606        try:
607            self._handle_playing_update(await device.metadata.playing())
608        except _CONNECTION_ERRORS as err:
609            self.logger.debug("Unable to read initial playback state for %s: %s", self.name, err)
610        self.logger.debug("Connected MRP playback monitoring for %s", self.display_name)
611        self.update_state()
612        return False
613
614    async def _disconnect_control_services(self) -> None:
615        """Close all active pyatv connections."""
616        self._disconnecting = True
617        companion_device = self._companion_device
618        mrp_device = self._mrp_device
619        self._companion_device = None
620        self._mrp_device = None
621        self._companion_listener = None
622        self._mrp_state_listener = None
623        self._mrp_push_listener = None
624        if mrp_device:
625            with contextlib.suppress(pyatv_exceptions.NotSupportedError):
626                if mrp_device.push_updater.active:
627                    mrp_device.push_updater.stop()
628            mrp_device.close()
629        if companion_device:
630            companion_device.close()
631        self._disconnecting = False
632        if companion_device is not None:
633            self._notify_companion_state_change()
634
635    def _build_config(
636        self,
637        info: AsyncServiceInfo,
638        protocol: Protocol,
639        credentials: str | None,
640        pairing_requirement: PairingRequirement,
641    ) -> AppleTVConfig | None:
642        """Build a pyatv configuration from an existing mDNS record."""
643        address = self._control_address(info)
644        if address is None:
645            self.logger.debug("Device control requires an IPv4 address for %s", self.name)
646            return None
647        if info.port is None:
648            self.logger.debug("Device control service has no port for %s", self.name)
649            return None
650        properties = {
651            key: value for key, value in info.decoded_properties.items() if value is not None
652        }
653        config = AppleTVConfig(address, self.display_name)
654        config.add_service(
655            ManualService(
656                self.player_id,
657                protocol,
658                info.port,
659                properties,
660                credentials=credentials,
661                pairing_requirement=pairing_requirement,
662            )
663        )
664        return config
665
666    def _control_address(self, info: AsyncServiceInfo) -> IPv4Address | None:
667        """Return an IPv4 address suitable for pyatv."""
668        for address in info.parsed_addresses():
669            try:
670                return IPv4Address(address)
671            except AddressValueError:
672                continue
673        try:
674            return IPv4Address(self.address)
675        except AddressValueError:
676            return None
677
678    @property
679    def _stream_active(self) -> bool:
680        """Return whether Music Assistant is actively streaming to this device."""
681        active = bool((stream := getattr(self, "stream", None)) and stream.running)
682        if active:
683            self._stream_last_active = time.monotonic()
684        return active
685
686    @property
687    def _external_state_blocked(self) -> bool:
688        """
689        Return whether externally observed playback state must be ignored.
690
691        While Music Assistant streams to this device, the stream is the SOLE
692        authority on player state. The check extends a grace period past a
693        stream's end because a warm-to-cold fallback briefly tears the stream
694        down mid-playback: a Companion/MRP update slipping through that window
695        applies the device's view of our own dying session as an "external
696        source", freezing the UI on a stale snapshot.
697        """
698        if self._stream_active:
699            return True
700        return time.monotonic() - getattr(self, "_stream_last_active", 0.0) < 15.0
701
702    @property
703    def _mrp_endpoint(self) -> tuple[AsyncServiceInfo, Protocol] | None:
704        """Return the preferred MRP endpoint and transport protocol."""
705        if supports_mrp_service(self.mrp_discovery_info):
706            assert self.mrp_discovery_info is not None
707            return self.mrp_discovery_info, Protocol.MRP
708        if supports_mrp_tunnel(self.airplay_discovery_info):
709            assert self.airplay_discovery_info is not None
710            return self.airplay_discovery_info, Protocol.AirPlay
711        return None
712
713    @property
714    def _mrp_credentials_key(self) -> str:
715        """Return the credential key for the active MRP transport."""
716        endpoint = self._mrp_endpoint
717        if endpoint is not None and endpoint[1] == Protocol.MRP:
718            return CONF_NATIVE_MRP_CREDENTIALS
719        return CONF_MRP_CREDENTIALS
720
721    @property
722    def _mrp_credentials(self) -> str | None:
723        """Return credentials for the active MRP transport."""
724        credentials = self.get_setup_value(self._mrp_credentials_key)
725        return str(credentials) if credentials else None
726
727    @property
728    def _uses_transient_mrp(self) -> bool:
729        """Return whether playback monitoring uses transient AirPlay credentials."""
730        # Mirrors pyatv's device rules: Apple TVs only accept real (paired) HAP
731        # credentials on the MRP tunnel and answer a transient pair-setup by
732        # showing the on-screen AirPlay pairing dialog, so they must never take
733        # this path - not even while the Companion record is still undiscovered.
734        # HomePods (and tunnel-capable third-party receivers) accept the
735        # transient handshake silently.
736        if self._is_apple_tv_device:
737            return False
738        return supports_transient_mrp(self.airplay_discovery_info)
739
740    @property
741    def _is_apple_tv_device(self) -> bool:
742        """Return whether the underlying device identifies itself as an Apple TV."""
743        if self.airplay_discovery_info and (
744            model := get_decoded_property(self.airplay_discovery_info, "model")
745        ):
746            return model.startswith("AppleTV")
747        return "apple tv" in self.device_info.model.lower()
748
749    def _device_for_feature(self, feature: FeatureName) -> AppleTV | None:
750        """Return the preferred connected device facade for a feature."""
751        for device in (self._companion_device, self._mrp_device):
752            if device and self._feature_available(device, feature):
753                return device
754        return None
755
756    def _device_for_power_feature(self, feature: FeatureName) -> AppleTV | None:
757        """
758        Return a connected device facade that can genuinely serve a power command.
759
760        pyatv reports the power commands as available on every MRP connection, but a
761        transient tunnel - the only control channel current HomePod firmware offers -
762        cannot act on them, and derives its power state from ``logicalDeviceCount``,
763        which does not count AirPlay audio sessions. Trusting it would leave a playing
764        HomePod stranded as "off".
765
766        :param feature: The power feature to look for.
767        """
768        if self._companion_device and self._feature_available(self._companion_device, feature):
769            return self._companion_device
770        if (
771            self._mrp_device
772            and not self._uses_transient_mrp
773            and self._feature_available(self._mrp_device, feature)
774        ):
775            return self._mrp_device
776        return None
777
778    @staticmethod
779    def _feature_available(device: AppleTV, feature: FeatureName) -> bool:
780        """Return whether pyatv currently exposes a feature."""
781        return device.features.in_state(FeatureState.Available, feature)
782
783    async def _wake_for_playback(self) -> None:
784        """Wake the device before starting or resuming playback."""
785        if self.powered is True:
786            return
787        device = self._device_for_power_feature(FeatureName.TurnOn)
788        if device is None:
789            return
790        self._power_on_event.clear()
791        await self._run_control_command(device.power.turn_on(), "wake")
792        await self._wait_for_wake()
793
794    async def _wait_for_wake(self) -> None:
795        """Wait briefly for a pushed powered-on state."""
796        if self.powered is True:
797            return
798        try:
799            await asyncio.wait_for(self._power_on_event.wait(), _WAKE_TIMEOUT)
800        except TimeoutError:
801            self.logger.debug("No power-state confirmation received from %s", self.display_name)
802
803    async def _run_control_command(self, command: Awaitable[None], description: str) -> None:
804        """Run a pyatv command and expose failures as player command errors."""
805        try:
806            await command
807        except _COMMAND_ERRORS as err:
808            raise PlayerCommandFailed(
809                f"Unable to {description} {self.display_name}: {err}"
810            ) from err
811
812    async def _run_volume_command(self, command: Awaitable[None], description: str) -> None:
813        """Run a native volume command, tolerating a missing confirmation event."""
814        # pyatv waits (up to 5s) for a pushed volume confirmation after a Companion
815        # volume command. Apple TVs that pass volume through to an HDMI-CEC amplifier
816        # apply the change but never emit that event, so the call times out even
817        # though it succeeded. Treat the timeout as success and let the caller apply
818        # the requested level; genuine command failures still surface.
819        try:
820            await command
821        except TimeoutError:
822            self.logger.debug(
823                "No volume confirmation from %s; assuming the change was applied",
824                self.display_name,
825            )
826        except _COMMAND_ERRORS as err:
827            raise PlayerCommandFailed(
828                f"Unable to {description} {self.display_name}: {err}"
829            ) from err
830
831    async def _run_companion_pairing(
832        self, session: SetupSession, collected: dict[str, ConfigValueType]
833    ) -> None:
834        """
835        Offer optional Companion (remote control) pairing, when supported.
836
837        Shows a skippable choice; on "set up now" it drives the PIN pairing and adds
838        the resulting credentials to ``collected`` (replacing any stored ones).
839
840        :param session: The setup flow session used to interact with the user.
841        :param collected: The values collected so far; updated in place.
842        """
843        if not self.companion_pairing_supported:
844            return
845        if not await self._offer_optional_pairing(session, "companion_offer"):
846            return
847        errors: dict[str, str] | None = None
848        while True:
849            pairing = await self._begin_pyatv_pairing(
850                self.companion_discovery_info, Protocol.Companion
851            )
852            try:
853                values = await session.form(
854                    [
855                        ConfigEntry(
856                            key=CONF_COMPANION_PAIRING_PIN,
857                            type=ConfigEntryType.STRING,
858                            required=True,
859                            category="protocol_generic",
860                        )
861                    ],
862                    step_id="pair_companion",
863                    errors=errors,
864                )
865                credentials = await self._finish_pyatv_pairing(
866                    pairing, str(values[CONF_COMPANION_PAIRING_PIN])
867                )
868            except PlayerCommandFailed as err:
869                errors = {"base": err.translation_key or str(err)}
870                continue
871            finally:
872                await pairing.close()
873            collected[CONF_COMPANION_CREDENTIALS] = credentials
874            return
875
876    async def _run_mrp_pairing(
877        self, session: SetupSession, collected: dict[str, ConfigValueType]
878    ) -> None:
879        """
880        Offer optional MRP (playback monitoring) pairing, when supported.
881
882        :param session: The setup flow session used to interact with the user.
883        :param collected: The values collected so far; updated in place.
884        """
885        if not self.mrp_pairing_supported:
886            return
887        endpoint = self._mrp_endpoint
888        if endpoint is None:
889            return
890        discovery_info, protocol = endpoint
891        cred_key = self._mrp_credentials_key
892        if not await self._offer_optional_pairing(session, "mrp_offer"):
893            return
894        errors: dict[str, str] | None = None
895        while True:
896            pairing = await self._begin_pyatv_pairing(discovery_info, protocol)
897            try:
898                values = await session.form(
899                    [
900                        ConfigEntry(
901                            key=CONF_MRP_PAIRING_PIN,
902                            type=ConfigEntryType.STRING,
903                            required=True,
904                            category="protocol_generic",
905                        )
906                    ],
907                    step_id="pair_mrp",
908                    errors=errors,
909                )
910                credentials = await self._finish_pyatv_pairing(
911                    pairing, str(values[CONF_MRP_PAIRING_PIN])
912                )
913            except PlayerCommandFailed as err:
914                errors = {"base": err.translation_key or str(err)}
915                continue
916            finally:
917                await pairing.close()
918            collected[cred_key] = credentials
919            return
920
921    async def _begin_pyatv_pairing(
922        self, discovery_info: AsyncServiceInfo | None, protocol: Protocol
923    ) -> PairingHandler:
924        """
925        Build a pyatv config and begin a Companion/MRP pairing (the device shows its PIN).
926
927        A failure here cannot be recovered by re-prompting, so it aborts the flow; a
928        partially started session is torn down first.
929
930        :param discovery_info: The mDNS record of the service to pair.
931        :param protocol: The pyatv protocol to pair (Companion or the MRP transport).
932        """
933        pairing_requirement = (
934            PairingRequirement.Optional
935            if protocol == Protocol.MRP
936            else PairingRequirement.Mandatory
937        )
938        config = (
939            self._build_config(discovery_info, protocol, None, pairing_requirement)
940            if discovery_info is not None
941            else None
942        )
943        if config is None:
944            raise AbortFlow("pairing_failed")
945        pairing: PairingHandler | None = None
946        started = False
947        try:
948            pairing = await pyatv.pair(config, protocol, self.mass.loop, name="Music Assistant")
949            await pairing.begin()
950            started = True
951        except Exception as err:
952            # any failure starting the pairing (device unreachable, pyatv/system
953            # issue, ...) is unrecoverable here; abort with a clear reason rather
954            # than letting it surface as a generic internal error
955            self.logger.warning("Could not start Apple TV pairing: %s", err)
956            raise AbortFlow("pairing_failed") from err
957        finally:
958            if not started and pairing is not None:
959                await pairing.close()
960        assert pairing is not None  # reached only when started, i.e. a live session
961        return pairing
962
963    async def _finish_pyatv_pairing(self, pairing: PairingHandler, pin: str) -> str:
964        """
965        Submit the PIN and return the credentials from a Companion/MRP pairing session.
966
967        :param pairing: The active pyatv pairing session.
968        :param pin: The PIN the user entered.
969        """
970        try:
971            pin_code = int(pin)
972        except (TypeError, ValueError) as err:
973            raise PlayerCommandFailed(
974                "Enter the numeric PIN shown on the device",
975                translation_key="invalid_pin",
976                translation_owner=self.translation_owner,
977            ) from err
978        try:
979            pairing.pin(pin_code)
980            await pairing.finish()
981        except (pyatv_exceptions.PairingError, *_CONNECTION_ERRORS) as err:
982            raise PlayerCommandFailed(
983                f"Unable to finish pairing for {self.display_name}: {err}"
984            ) from err
985        credentials = pairing.service.credentials
986        if not pairing.has_paired or not credentials:
987            raise PlayerCommandFailed(
988                "Pairing did not complete", translation_key="authentication_failed"
989            )
990        return str(credentials)
991
992    def _apply_initial_device_state(self, device: AppleTV, source: str) -> None:
993        """Apply power and volume snapshots exposed after connection."""
994        if self._feature_available(device, FeatureName.PowerState):
995            self._handle_power_update(source, device.power.power_state)
996        if self._feature_available(device, FeatureName.Volume):
997            self._handle_volume_update(source, device.audio.volume)
998
999    def _handle_power_update(self, source: str, power_state: PowerState) -> None:
1000        """Apply a pushed pyatv power state."""
1001        if source == "mrp" and self._companion_device is not None:
1002            return
1003        if power_state == PowerState.On:
1004            self._attr_powered = True
1005            self._power_on_event.set()
1006        elif power_state == PowerState.Off:
1007            # A device streaming from Music Assistant is not powered off, whatever the
1008            # control channel claims: MRP derives the state from logicalDeviceCount,
1009            # which does not count AirPlay audio sessions, and a Companion SystemStatus
1010            # can briefly report sleep mid-stream. Acting on it would strand the player
1011            # as "off" and trip the auto-ungroup in the players controller.
1012            if self._stream_active:
1013                return
1014            self._attr_powered = False
1015            self._power_on_event.clear()
1016            self._attr_playback_state = PlaybackState.IDLE
1017            self._attr_active_source = None
1018            self._attr_current_media = None
1019        else:
1020            self._attr_powered = None
1021        self.update_state()
1022
1023    def _handle_volume_update(self, source: str, volume: float) -> None:
1024        """Apply a pushed pyatv volume level."""
1025        # MRP volume is ignored when Companion owns volume, and always on a transient
1026        # tunnel: that only exposes the speaker's own volume, which Music Assistant
1027        # never drives. Worse, a 0 from there latches a mute, and a muted player
1028        # skips the stream volume command entirely - so every later volume change
1029        # silently stops reaching the device.
1030        if source == "mrp" and (self._companion_device is not None or self._uses_transient_mrp):
1031            return
1032        # While Music Assistant streams, volume_set drives the stream volume and
1033        # deliberately leaves the native device volume alone. The two are separate
1034        # knobs on a different scale, so a report about the native one must not
1035        # overwrite (or persist) the level the user just set on the stream.
1036        if source != "command" and self._stream_active:
1037            return
1038        volume_level = max(0, min(100, round(volume)))
1039        if volume_level == 0:
1040            if self._volume_before_mute is None and self._attr_volume_level:
1041                self._volume_before_mute = self._attr_volume_level
1042            mute_changed = self._attr_volume_muted is not True
1043            self._attr_volume_muted = True
1044            if mute_changed:
1045                self.update_state()
1046            return
1047        mute_changed = self._attr_volume_muted is not False
1048        self._attr_volume_muted = False
1049        self._volume_before_mute = None
1050        self._update_native_volume(volume_level, state_changed=mute_changed)
1051
1052    def _update_native_volume(self, volume: int, *, state_changed: bool = False) -> None:
1053        """Update and persist a volume reported by device control."""
1054        volume = max(0, min(100, volume))
1055        if self._attr_volume_level == volume:
1056            if state_changed:
1057                self.update_state()
1058            return
1059        self._attr_volume_level = volume
1060        self.mass.config.set_raw_player_config_value(
1061            self.player_id,
1062            CONF_STORED_VOLUME,
1063            volume,
1064        )
1065        self.update_state()
1066
1067    def _clear_stored_credentials(self, credentials_key: str) -> None:
1068        """Clear credentials that the receiver rejected."""
1069        self._update_setup_data(credentials_key, None)
1070        self.update_state()
1071
1072    def _handle_playing_update(self, playing: Playing) -> None:
1073        """Apply external playback state received over the MRP tunnel."""
1074        if self._external_state_blocked:
1075            return
1076        app = self._mrp_device.metadata.app if self._mrp_device else None
1077        playback_state = {
1078            DeviceState.Playing: PlaybackState.PLAYING,
1079            DeviceState.Loading: PlaybackState.PLAYING,
1080            DeviceState.Seeking: PlaybackState.PLAYING,
1081            DeviceState.Paused: PlaybackState.PAUSED,
1082        }.get(playing.device_state, PlaybackState.IDLE)
1083        # Many tvOS apps (e.g. Netflix) report Idle rather than Paused when
1084        # paused. While the same app stays the active source, keep it paused
1085        # instead of going idle so transport controls resume the app itself
1086        # rather than falling back to the Music Assistant queue.
1087        if (
1088            playback_state == PlaybackState.IDLE
1089            and app is not None
1090            and self._attr_active_source == app.identifier
1091        ):
1092            playback_state = PlaybackState.PAUSED
1093        self._attr_playback_state = playback_state
1094        if playback_state == PlaybackState.IDLE:
1095            self._attr_active_source = None
1096            self._attr_current_media = None
1097            self.update_state()
1098            return
1099
1100        source_id = app.identifier if app else "airplay_control"
1101        source_name = (app.name or app.identifier) if app else "AirPlay device"
1102        self._attr_active_source = source_id
1103        self._ensure_source(source_id, source_name)
1104        self._attr_elapsed_time = float(playing.position or 0)
1105        self._attr_elapsed_time_last_updated = time.time()
1106        image_url: str | None = None
1107        if (
1108            self._mrp_device
1109            and self._feature_available(self._mrp_device, FeatureName.Artwork)
1110            and isinstance(artwork_id := self._mrp_device.metadata.artwork_id, str)
1111            and artwork_id
1112        ):
1113            image_url = self._get_external_artwork_url(artwork_id)
1114        media_type = (
1115            MediaType.TRACK if playing.media_type == PyatvMediaType.Music else MediaType.UNKNOWN
1116        )
1117        self._attr_current_media = PlayerMedia(
1118            uri=playing.content_identifier or f"apple-device://{self.player_id}/{playing.hash}",
1119            media_type=media_type,
1120            title=playing.title or source_name,
1121            artist=playing.artist,
1122            album=playing.album,
1123            image_url=image_url,
1124            duration=playing.total_time,
1125            source_id=source_id,
1126            elapsed_time=playing.position,
1127            elapsed_time_last_updated=self._attr_elapsed_time_last_updated,
1128        )
1129        self.update_state()
1130
1131    def _ensure_source(self, source_id: str, source_name: str) -> None:
1132        """Add a passive source reported by MRP playback monitoring."""
1133        # Track external ids so the stream can reclaim the device state from a
1134        # leaked external snapshot (see AirPlayPlayer.set_state_from_stream).
1135        if not hasattr(self, "_external_source_ids"):
1136            self._external_source_ids: set[str] = set()
1137        self._external_source_ids.add(source_id)
1138        can_play_pause = bool(
1139            self._device_for_feature(FeatureName.Play)
1140            or self._device_for_feature(FeatureName.Pause)
1141        )
1142        can_next_previous = bool(
1143            self._device_for_feature(FeatureName.Next)
1144            or self._device_for_feature(FeatureName.Previous)
1145        )
1146        for source in self._attr_source_list:
1147            if source.id != source_id:
1148                continue
1149            source.name = source_name
1150            source.can_play_pause = can_play_pause
1151            source.can_next_previous = can_next_previous
1152            return
1153        self._attr_source_list.append(
1154            PlayerSource(
1155                id=source_id,
1156                name=source_name,
1157                passive=True,
1158                can_play_pause=can_play_pause,
1159                can_seek=False,
1160                can_next_previous=can_next_previous,
1161            )
1162        )
1163
1164    def _get_external_artwork_url(self, artwork_id: str) -> str:
1165        """Return the image-proxy URL for external MRP artwork."""
1166        image = MediaItemImage(
1167            type=ImageType.THUMB,
1168            path=f"{EXTERNAL_ARTWORK_PATH_PREFIX}/{self.player_id}/{artwork_id}",
1169            provider=self.provider_id,
1170            remotely_accessible=False,
1171        )
1172        return self.mass.metadata.get_image_url(image)
1173
1174    def _handle_connection_closed(
1175        self,
1176        source: str,
1177        device: AppleTV,
1178        exception: Exception | None = None,
1179    ) -> None:
1180        """Handle a pyatv connection closing."""
1181        companion_closed = False
1182        if source == "companion" and self._companion_device is device:
1183            self._companion_device = None
1184            self._companion_listener = None
1185            companion_closed = True
1186        elif source == "mrp" and self._mrp_device is device:
1187            self._mrp_device = None
1188            self._mrp_state_listener = None
1189            self._mrp_push_listener = None
1190            self._clear_external_state()
1191        else:
1192            return
1193        if exception:
1194            self.logger.debug("Apple %s connection lost for %s: %s", source, self.name, exception)
1195        if companion_closed:
1196            self._notify_companion_state_change()
1197        if not self._disconnecting and not self._unloading:
1198            self._schedule_connection()
1199
1200    def _handle_push_error(self, device: AppleTV, exception: Exception) -> None:
1201        """Restart MRP playback monitoring after a push update error."""
1202        if self._mrp_device is not device:
1203            return
1204        self.logger.debug("MRP playback updates failed for %s: %s", self.name, exception)
1205        device.push_updater.stop()
1206        self._mrp_device = None
1207        self._mrp_state_listener = None
1208        self._mrp_push_listener = None
1209        device.close()
1210        self._clear_external_state()
1211        self._schedule_connection()
1212
1213    def _clear_external_state(self) -> None:
1214        """Drop the playback snapshot observed over a closed MRP connection."""
1215        if self._external_state_blocked or self._attr_active_source is None:
1216            return
1217        # Without a live connection the snapshot can no longer be updated, and the
1218        # last one is typically an app held at paused: leaving it in place keeps
1219        # transport commands aimed at that app instead of the Music Assistant queue.
1220        self.mark_external_source_ended()
1221        self.update_state()
1222
1223    def _notify_companion_state_change(self) -> None:
1224        """Notify a wired-up observer that the Companion connection state changed."""
1225        if self.on_companion_state_change is not None:
1226            self.on_companion_state_change()
1227
1228    @staticmethod
1229    def _service_signature(info: AsyncServiceInfo | None) -> tuple[object, ...] | None:
1230        """Return fields that require a pyatv reconnection when changed."""
1231        if info is None:
1232            return None
1233        # TXT keys are case-insensitive (RFC 6763); casefold them so a re-cased
1234        # key is never mistaken for a connection-relevant change.
1235        stable_properties = tuple(
1236            sorted(
1237                (key.casefold(), value)
1238                for key, value in info.decoded_properties.items()
1239                if key.casefold() not in _VOLATILE_DISCOVERY_KEYS
1240            )
1241        )
1242        return (
1243            info.name,
1244            info.port,
1245            tuple(info.addresses),
1246            stable_properties,
1247        )
1248
1249
1250class _AirPlayStateListener(DeviceListener, PowerListener, AudioListener):
1251    """Forward pyatv device, power, and volume events to a controlled player."""
1252
1253    def __init__(self, player: AirPlayControlPlayer, device: AppleTV, source: str) -> None:
1254        """Initialize a listener for one pyatv connection."""
1255        self._player = player
1256        self._device = device
1257        self._source = source
1258
1259    def connection_lost(self, exception: Exception) -> None:
1260        """Handle an unexpected pyatv disconnect."""
1261        self._player._handle_connection_closed(self._source, self._device, exception)
1262
1263    def connection_closed(self) -> None:
1264        """Handle a closed pyatv connection."""
1265        self._player._handle_connection_closed(self._source, self._device)
1266
1267    def powerstate_update(self, old_state: PowerState, new_state: PowerState) -> None:
1268        """Forward a power-state update."""
1269        self._player._handle_power_update(self._source, new_state)
1270
1271    def volume_update(self, old_level: float, new_level: float) -> None:
1272        """Forward a volume update."""
1273        self._player._handle_volume_update(self._source, new_level)
1274
1275    def volume_device_update(
1276        self,
1277        output_device: OutputDevice,
1278        old_level: float,
1279        new_level: float,
1280    ) -> None:
1281        """Ignore volume updates for secondary output devices."""
1282
1283    def outputdevices_update(
1284        self,
1285        old_devices: list[OutputDevice],
1286        new_devices: list[OutputDevice],
1287    ) -> None:
1288        """Ignore output-device membership updates."""
1289
1290
1291class _AirPlayPushListener(PushListener):
1292    """Forward MRP now-playing updates to a controlled player."""
1293
1294    def __init__(self, player: AirPlayControlPlayer, device: AppleTV) -> None:
1295        """Initialize an MRP push listener."""
1296        self._player = player
1297        self._device = device
1298
1299    def playstatus_update(self, updater: object, playstatus: Playing) -> None:
1300        """Forward an external playback update."""
1301        self._player._handle_playing_update(playstatus)
1302
1303    def playstatus_error(self, updater: object, exception: Exception) -> None:
1304        """Handle an MRP push update failure."""
1305        self._player._handle_push_error(self._device, exception)
1306