/
/
/
1"""Dashboard support for the AirPlay provider (Apple TV via the tvOS app)."""
2
3from __future__ import annotations
4
5import functools
6from typing import TYPE_CHECKING
7from urllib.parse import urlencode, urlsplit
8
9from music_assistant_models.dashboard import DashboardDevice
10from music_assistant_models.enums import DashboardType
11from music_assistant_models.errors import PlayerCommandFailed, PlayerUnavailableError
12
13from .constants import TVOS_APP_BUNDLE_ID
14from .control_player import AirPlayControlPlayer
15from .helpers import is_apple_tv
16
17if TYPE_CHECKING:
18 from collections.abc import Callable
19
20 from .provider import AirPlayProvider
21
22# Version of the launch contract this adapter emits (see tvos/docs/launch-contract.md).
23LAUNCH_CONTRACT_VERSION = "1"
24# Dashboard types the tvOS app has native views for.
25SUPPORTED_DASHBOARD_TYPES = {
26 DashboardType.NOW_PLAYING,
27 DashboardType.PARTY,
28 DashboardType.MUSIC_QUIZ,
29}
30
31
32class AirPlayDashboards:
33 """Registers eligible Apple TVs as dashboard endpoints and launches the tvOS app on them."""
34
35 def __init__(self, provider: AirPlayProvider) -> None:
36 """
37 Initialize dashboard handling for an AirPlay provider.
38
39 :param provider: The AirPlay provider owning the discovered Apple devices.
40 """
41 self.provider = provider
42 self.mass = provider.mass
43 self.logger = provider.logger.getChild("dashboard")
44 self._unregister_callbacks: dict[str, Callable[[], None]] = {}
45 # installed-app bundle ids per player: frozenset when known, None on fetch
46 # failure, absent until first fetched
47 self._installed_apps: dict[str, frozenset[str] | None] = {}
48 # last-seen Companion connection state, to detect (re)connect edges
49 self._companion_connected: dict[str, bool] = {}
50 self._unloaded = False
51
52 def setup_player(self, player: AirPlayControlPlayer) -> None:
53 """
54 Start tracking a control player for dashboard eligibility.
55
56 :param player: The control-capable AirPlay player to track.
57 """
58 if self._unloaded:
59 return
60 player.on_companion_state_change = functools.partial(self.reconcile, player.player_id)
61 self.reconcile(player.player_id)
62
63 def reconcile(self, player_id: str) -> None:
64 """
65 Re-evaluate a player's dashboard eligibility in the background.
66
67 :param player_id: The player to re-evaluate.
68 """
69 if self._unloaded:
70 return
71 self.mass.create_task(
72 self._async_reconcile,
73 player_id,
74 task_id=f"airplay_dashboard_reconcile_{player_id}",
75 abort_existing=True,
76 )
77
78 def unregister(self, player_id: str) -> None:
79 """
80 Drop a player's dashboard registration and cached state.
81
82 :param player_id: The player whose registration to drop.
83 """
84 self._companion_connected.pop(player_id, None)
85 self._installed_apps.pop(player_id, None)
86 self._unregister_endpoint(player_id)
87
88 async def unload(self) -> None:
89 """Unregister all dashboard endpoints owned by this provider."""
90 self._unloaded = True
91 for unregister_callback in list(self._unregister_callbacks.values()):
92 unregister_callback()
93 self._unregister_callbacks.clear()
94 self._installed_apps.clear()
95 self._companion_connected.clear()
96
97 async def _async_reconcile(self, player_id: str) -> None:
98 """Evaluate a player's eligibility and (un)register it accordingly."""
99 if self._unloaded:
100 return
101 player = self.mass.players.get_player(player_id)
102 if not isinstance(player, AirPlayControlPlayer):
103 self.unregister(player_id)
104 return
105 # a Companion (re)connect re-checks whether the tvOS app is installed
106 connected = player.companion_connected
107 if connected != self._companion_connected.get(player_id, False):
108 self._companion_connected[player_id] = connected
109 self._installed_apps.pop(player_id, None)
110 if await self._is_eligible(player):
111 self._register(player)
112 else:
113 self._unregister_endpoint(player_id)
114
115 async def _is_eligible(self, player: AirPlayControlPlayer) -> bool:
116 """Return whether a control player should be exposed as a dashboard endpoint."""
117 if not player.available or not player.enabled:
118 return False
119 if not is_apple_tv(player.device_info.manufacturer, player.device_info.model):
120 return False
121 if not player.companion_connected:
122 return False
123 app_ids = await self._installed_app_ids(player)
124 return app_ids is not None and TVOS_APP_BUNDLE_ID in app_ids
125
126 async def _installed_app_ids(self, player: AirPlayControlPlayer) -> frozenset[str] | None:
127 """Return the cached installed-app bundle ids, fetching them once if still unknown."""
128 if player.player_id not in self._installed_apps:
129 app_ids = await player.async_list_installed_app_ids()
130 self._installed_apps[player.player_id] = (
131 frozenset(app_ids) if app_ids is not None else None
132 )
133 return self._installed_apps[player.player_id]
134
135 def _register(self, player: AirPlayControlPlayer) -> None:
136 """Register (or refresh) a player as a dashboard endpoint with the controller."""
137 # a reconcile awaiting the app list may resolve after unload(); never resurrect
138 if self._unloaded:
139 return
140 device = DashboardDevice(
141 dashboard_id=self._dashboard_id(player.player_id),
142 name=player.display_name,
143 supported_types=set(SUPPORTED_DASHBOARD_TYPES),
144 provider_domain_hint=self.provider.domain,
145 )
146 self._unregister_callbacks[player.player_id] = (
147 self.mass.dashboard.register_dashboard_handler(
148 device,
149 functools.partial(self._on_show, player.player_id),
150 functools.partial(self._on_hide, player.player_id),
151 )
152 )
153
154 def _unregister_endpoint(self, player_id: str) -> None:
155 """Unregister a player's dashboard endpoint if it is registered."""
156 if unregister_callback := self._unregister_callbacks.pop(player_id, None):
157 unregister_callback()
158
159 async def _on_show(
160 self, player_id: str, dashboard: DashboardType, target_player_id: str | None
161 ) -> None:
162 """
163 Launch the tvOS app on an Apple TV to show a dashboard.
164
165 :param player_id: The Apple TV endpoint to show the dashboard on.
166 :param dashboard: Dashboard to show.
167 :param target_player_id: Player to show, required when the dashboard is NOW_PLAYING.
168 :raises PlayerUnavailableError: If the Apple TV is gone or the launch fails.
169 """
170 player = self.mass.players.get_player(player_id)
171 if not isinstance(player, AirPlayControlPlayer):
172 raise PlayerUnavailableError(f"Apple TV {player_id} is no longer available")
173 target_url = await self.mass.dashboard.resolve_dashboard_url(
174 dashboard, target_player_id, prefer_local=True
175 )
176 dashboard_id = self._dashboard_id(player_id)
177 launch_uri = self._build_launch_uri(dashboard, target_url, dashboard_id)
178 # never log the full target url or the one-time viewer code embedded in it
179 self.logger.debug(
180 "Showing %s dashboard %s on %s (target origin %s)",
181 dashboard.value,
182 dashboard_id,
183 player.display_name,
184 self._target_origin(target_url),
185 )
186 try:
187 await player.wake()
188 await player.async_launch_app(launch_uri)
189 except PlayerCommandFailed as err:
190 raise PlayerUnavailableError(
191 f"Unable to show the dashboard on {player.display_name}",
192 translation_key="show_dashboard_failed",
193 translation_owner=self.provider.translation_owner,
194 translation_args=[player.display_name],
195 ) from err
196
197 async def _on_hide(self, player_id: str) -> None:
198 """
199 Handle a hide request for an Apple TV endpoint.
200
201 :param player_id: The Apple TV endpoint the dashboard was showing on.
202 """
203 # hide is session-driven: the app returns to idle when its session disappears.
204 # pyatv has no foreground-app control, so there is nothing to do on the device.
205 self.logger.debug("Hide requested for dashboard %s (session-driven)", player_id)
206
207 def _build_launch_uri(
208 self, dashboard: DashboardType, target_url: str, dashboard_id: str
209 ) -> str:
210 """Build the versioned tvOS launch URI wrapping the resolved dashboard url."""
211 query = urlencode(
212 {
213 "v": LAUNCH_CONTRACT_VERSION,
214 "type": dashboard.value,
215 "target": target_url,
216 "dashboard_id": dashboard_id,
217 }
218 )
219 return f"musicassistant://dashboard/show?{query}"
220
221 @staticmethod
222 def _dashboard_id(player_id: str) -> str:
223 """Return the dashboard endpoint id for an AirPlay player."""
224 return f"airplay_{player_id}"
225
226 @staticmethod
227 def _target_origin(url: str) -> str:
228 """Return only the scheme://host[:port] of a url, for safe logging."""
229 parts = urlsplit(url)
230 if parts.scheme and parts.netloc:
231 return f"{parts.scheme}://{parts.netloc}"
232 return "unknown"
233