/
/
/
1"""Core controller that casts Music Assistant dashboards to display devices."""
2
3from __future__ import annotations
4
5from collections.abc import Awaitable, Callable
6from dataclasses import dataclass
7from typing import TYPE_CHECKING
8from urllib.parse import urlencode
9
10from music_assistant_models.auth import Scope
11from music_assistant_models.dashboard import DashboardDevice, DashboardSession
12from music_assistant_models.enums import DashboardType, EventType
13from music_assistant_models.errors import (
14 ActionUnavailable,
15 InsufficientPermissions,
16 InvalidCommand,
17 MusicAssistantError,
18)
19
20from music_assistant.controllers.webserver.helpers.auth_middleware import (
21 get_current_client_id,
22 get_current_user,
23 has_scope,
24)
25from music_assistant.helpers.api import api_command
26from music_assistant.helpers.guest_access import get_or_create_guest_user
27from music_assistant.models.core_controller import CoreController
28
29if TYPE_CHECKING:
30 from music_assistant.mass import MusicAssistant
31
32DASHBOARD_VIEWER_USERNAME = "dashboard_viewer"
33DASHBOARD_VIEWER_DISPLAY_NAME = "Dashboard Viewer"
34DASHBOARD_CODE_EXPIRY_HOURS = 1
35APP_MA_HOST = "https://app.music-assistant.io"
36# every real dashboard type, i.e. what a registration supports when not given explicitly
37ALL_DASHBOARD_TYPES = frozenset(t for t in DashboardType if t != DashboardType.UNKNOWN)
38
39# the frontend's router leaves these literal in a query value; escaped, it never matches
40ROUTE_SAFE_CHARS = ":!'()*@,;$/"
41
42
43@dataclass
44class _RegisteredDashboard:
45 """A dashboard endpoint tracked by the controller, either API- or callback-registered."""
46
47 device: DashboardDevice
48 # callbacks set for in-server registrations (e.g. chromecast); API registrations use events
49 on_show: Callable[[DashboardType, str | None], Awaitable[None]] | None = None
50 on_hide: Callable[[], Awaitable[None]] | None = None
51 client_id: str | None = None # websocket connection that owns an API registration
52
53
54class DashboardController(CoreController):
55 """Casts Music Assistant dashboards (e.g. Party mode) to display devices."""
56
57 domain: str = "dashboard"
58
59 def __init__(self, mass: MusicAssistant) -> None:
60 """Initialize the DashboardController."""
61 super().__init__(mass)
62 self.manifest.name = "Dashboard"
63 self.manifest.description = "Casts Music Assistant dashboards to display devices."
64 self._dashboards: dict[str, _RegisteredDashboard] = {}
65 self._sessions: dict[str, DashboardSession] = {}
66
67 @api_command("dashboard/register")
68 async def register_dashboard(
69 self,
70 dashboard_id: str,
71 name: str,
72 supported_types: set[DashboardType] | None = None,
73 provider_domain_hint: str | None = None,
74 ) -> None:
75 """
76 Register the calling client as a dashboard endpoint.
77
78 :param dashboard_id: Unique id chosen by the registering client.
79 :param name: Display name for the dashboard endpoint.
80 :param supported_types: Dashboard types this endpoint can show, defaults to all
81 when omitted; an explicitly empty set is rejected.
82 :param provider_domain_hint: Optional provider domain used to resolve the endpoint's icon.
83 :raises InvalidCommand: If not called from a websocket client, if supported_types
84 is an empty set, or if dashboard_id is already registered by another owner.
85 """
86 client_id = get_current_client_id()
87 if client_id is None:
88 msg = "Registering a dashboard is only available for websocket clients"
89 raise InvalidCommand(msg)
90 if supported_types is not None and not supported_types:
91 msg = "supported_types cannot be an empty set"
92 raise InvalidCommand(msg)
93
94 existing = self._dashboards.get(dashboard_id)
95 if existing is not None and existing.client_id != client_id:
96 msg = f"Dashboard {dashboard_id} is already registered by another owner"
97 raise InvalidCommand(msg)
98
99 device = DashboardDevice(
100 dashboard_id=dashboard_id,
101 name=name,
102 supported_types=set(supported_types)
103 if supported_types is not None
104 else set(ALL_DASHBOARD_TYPES),
105 provider_domain_hint=provider_domain_hint,
106 )
107 self._dashboards[dashboard_id] = _RegisteredDashboard(device=device, client_id=client_id)
108 # don't spam subscribers when a re-registration changed nothing
109 if existing is None or existing.device != device:
110 self._signal_dashboards_updated()
111
112 @api_command("dashboard/unregister")
113 async def unregister_dashboard(self, dashboard_id: str) -> None:
114 """
115 Unregister a dashboard endpoint, dropping any active session for it.
116
117 :param dashboard_id: Id of a registered dashboard endpoint.
118 :raises InvalidCommand: If not called from a websocket client, or if dashboard_id
119 is registered by another owner.
120 """
121 client_id = get_current_client_id()
122 if client_id is None:
123 msg = "Unregistering a dashboard is only available for websocket clients"
124 raise InvalidCommand(msg)
125
126 existing = self._dashboards.get(dashboard_id)
127 if existing is None:
128 return
129 if existing.client_id != client_id:
130 msg = f"Dashboard {dashboard_id} is registered by another owner"
131 raise InvalidCommand(msg)
132
133 del self._dashboards[dashboard_id]
134 self._signal_dashboards_updated()
135 if self._sessions.pop(dashboard_id, None) is not None:
136 self._signal_sessions_updated()
137
138 @api_command("dashboard/dashboards")
139 async def get_dashboards(self, dashboard: DashboardType | None = None) -> list[DashboardDevice]:
140 """
141 Return all registered dashboard endpoints.
142
143 :param dashboard: When given, only return endpoints that support this dashboard type.
144 """
145 devices = [registration.device for registration in self._dashboards.values()]
146 if dashboard is None:
147 return devices
148 return [device for device in devices if dashboard in device.supported_types]
149
150 @api_command("dashboard/sessions")
151 async def get_dashboard_sessions(self) -> list[DashboardSession]:
152 """Return all active dashboard cast sessions."""
153 return list(self._sessions.values())
154
155 @api_command("dashboard/show", required_scope=Scope.USERS_INVITE)
156 async def show_dashboard(
157 self,
158 dashboard_id: str,
159 dashboard: DashboardType,
160 player_id: str | None = None,
161 ) -> None:
162 """
163 Show a Music Assistant dashboard on a registered dashboard endpoint.
164
165 :param dashboard_id: Id of a registered dashboard endpoint, as returned
166 by `dashboard/dashboards`.
167 :param dashboard: Dashboard to show.
168 :param player_id: Player to show, required when dashboard is NOW_PLAYING.
169 :raises InvalidCommand: If dashboard_id is unknown, doesn't support this dashboard,
170 or the dashboard/player_id combination isn't routable.
171 """
172 registration = self._dashboards.get(dashboard_id)
173 if registration is None:
174 msg = f"Unknown dashboard: {dashboard_id}"
175 raise InvalidCommand(msg)
176 if dashboard not in registration.device.supported_types:
177 msg = f"Dashboard {dashboard_id} does not support {dashboard}"
178 raise InvalidCommand(msg)
179 # validate intent up-front so both branches below reject it identically
180 self._dashboard_route(dashboard, player_id)
181
182 session = DashboardSession(
183 dashboard_id=dashboard_id,
184 name=registration.device.name,
185 dashboard=dashboard,
186 player_id=player_id,
187 )
188
189 if registration.on_show is not None:
190 # the consumer resolves its own url if needed; raises before showing on failure
191 await registration.on_show(dashboard, player_id)
192 else:
193 # API registration: url-based clients resolve their own url via `dashboard/get_url`
194 self.mass.signal_event(EventType.DASHBOARD_SHOW, object_id=dashboard_id, data=session)
195
196 self._sessions[dashboard_id] = session
197 self._signal_sessions_updated()
198
199 @api_command("dashboard/hide", required_scope=Scope.USERS_INVITE)
200 async def hide_dashboard(self, dashboard_id: str) -> None:
201 """
202 Hide a Music Assistant dashboard from a registered dashboard endpoint.
203
204 :param dashboard_id: Id of a registered dashboard endpoint, as returned
205 by `dashboard/dashboards`.
206 """
207 registration = self._dashboards.get(dashboard_id)
208 if registration is not None:
209 if registration.on_hide is not None:
210 try:
211 await registration.on_hide()
212 except MusicAssistantError:
213 # the endpoint may simply not have been showing a dashboard: not fatal
214 self.logger.debug(
215 "Dashboard %s could not hide its dashboard", dashboard_id, exc_info=True
216 )
217 else:
218 self.mass.signal_event(EventType.DASHBOARD_HIDE, object_id=dashboard_id)
219
220 self._sessions.pop(dashboard_id, None)
221 self._signal_sessions_updated()
222
223 @api_command("dashboard/get_url")
224 async def get_url_for_dashboard(
225 self, dashboard: DashboardType, player_id: str | None = None, prefer_local: bool = False
226 ) -> str:
227 """
228 Return a fully-qualified dashboard URL for a client to load itself.
229
230 :param dashboard: Dashboard to load.
231 :param player_id: Player to show, required when dashboard is NOW_PLAYING.
232 :param prefer_local: Return the plain local base url form (native LAN viewers
233 that are not bound by the https/remote-access requirement).
234 :raises InsufficientPermissions: If the caller has neither the required scope
235 nor a matching active session of its own.
236 """
237 if not self._can_resolve_url_for_caller(dashboard, player_id):
238 msg = "Insufficient permissions to resolve a dashboard url"
239 raise InsufficientPermissions(msg)
240 return await self.resolve_dashboard_url(dashboard, player_id, prefer_local=prefer_local)
241
242 def register_dashboard_handler(
243 self,
244 device: DashboardDevice,
245 on_show: Callable[[DashboardType, str | None], Awaitable[None]],
246 on_hide: Callable[[], Awaitable[None]],
247 ) -> Callable[[], None]:
248 """
249 Register an in-server dashboard endpoint (e.g. chromecast) with show/hide callbacks.
250
251 :param device: Metadata for the dashboard endpoint being registered.
252 :param on_show: Called with (dashboard, player_id) to show a dashboard.
253 :param on_hide: Called to hide whatever dashboard is currently showing.
254 :return: Callable that unregisters the endpoint and drops its active session.
255 """
256 dashboard_id = device.dashboard_id
257 existing = self._dashboards.get(dashboard_id)
258 self._dashboards[dashboard_id] = _RegisteredDashboard(
259 device=device, on_show=on_show, on_hide=on_hide
260 )
261 # don't spam subscribers when a re-registration changed nothing
262 if existing is None or existing.device != device:
263 self._signal_dashboards_updated()
264
265 def unregister() -> None:
266 self._dashboards.pop(dashboard_id, None)
267 self._signal_dashboards_updated()
268 if self._sessions.pop(dashboard_id, None) is not None:
269 self._signal_sessions_updated()
270
271 return unregister
272
273 def handle_client_disconnected(self, client_id: str) -> None:
274 """Drop all dashboard registrations (and sessions) owned by a disconnected client."""
275 stale_ids = [
276 dashboard_id
277 for dashboard_id, registration in self._dashboards.items()
278 if registration.client_id == client_id
279 ]
280 if not stale_ids:
281 return
282
283 sessions_changed = False
284 for dashboard_id in stale_ids:
285 del self._dashboards[dashboard_id]
286 if self._sessions.pop(dashboard_id, None) is not None:
287 sessions_changed = True
288
289 self._signal_dashboards_updated()
290 if sessions_changed:
291 self._signal_sessions_updated()
292
293 def end_session(self, dashboard_id: str, reason: str) -> None:
294 """
295 End the active session for an endpoint that stopped showing its dashboard.
296
297 :param dashboard_id: Id of a registered dashboard endpoint.
298 :param reason: Human-readable cause, logged as a warning.
299 """
300 session = self._sessions.pop(dashboard_id, None)
301 if session is None:
302 return
303 self.logger.warning("Dashboard session on %s ended: %s", session.name, reason)
304 self._signal_sessions_updated()
305
306 async def resolve_dashboard_url(
307 self, dashboard: DashboardType, player_id: str | None, *, prefer_local: bool = False
308 ) -> str:
309 """
310 Build the fully-qualified URL a dashboard endpoint should load to show a dashboard.
311
312 By default an externally-reachable https base url (reverse-proxied server, same
313 origin) is preferred over remote access (the app.music-assistant.io signaling portal),
314 as required by cast receivers. With ``prefer_local`` the server's own base url is
315 always returned, plain http included: native apps on the LAN are not bound by the cast
316 receiver's https requirement. In-server consumers (e.g. the chromecast provider) call
317 this to resolve the url themselves.
318
319 :param dashboard: Dashboard to show.
320 :param player_id: Player to show, required when dashboard is NOW_PLAYING.
321 :param prefer_local: Return the plain local base url instead of the https/remote form.
322 :raises ActionUnavailable: If neither an https base url nor remote access is configured
323 (never raised when ``prefer_local`` is set).
324 """
325 route = self._dashboard_route(dashboard, player_id)
326 base_url = self.mass.webserver.base_url
327 if prefer_local:
328 # native LAN apps talk straight to this server, no https/remote gate needed
329 query = {"dashboard": await self._get_dashboard_code(), "path": route}
330 return f"{base_url}?{urlencode(query)}"
331 remote_access = self.mass.webserver.remote_access
332 use_https_base = base_url.startswith("https://")
333 if not use_https_base and not (remote_access.is_enabled and remote_access.remote_id):
334 msg = "Remote access or an https base url is required to cast dashboards"
335 raise ActionUnavailable(
336 msg,
337 translation_key="remote_access_required",
338 translation_owner=self.translation_owner,
339 )
340
341 dashboard_code = await self._get_dashboard_code()
342 if use_https_base:
343 # same origin: the receiver talks straight to this server, no remote_id needed
344 query = {"dashboard": dashboard_code, "path": route}
345 return f"{base_url}?{urlencode(query)}"
346
347 query = {"remote_id": remote_access.remote_id, "dashboard": dashboard_code, "path": route}
348 channel = self._frontend_channel()
349 return f"{APP_MA_HOST}/{channel}/?{urlencode(query)}"
350
351 def _can_resolve_url_for_caller(self, dashboard: DashboardType, player_id: str | None) -> bool:
352 """
353 Return whether the current caller may resolve a dashboard url for itself.
354
355 :param dashboard: Dashboard the caller wants a url for.
356 :param player_id: Player the caller wants a url for, when dashboard is NOW_PLAYING.
357 """
358 user = get_current_user()
359 if user is not None and has_scope(user, Scope.USERS_INVITE):
360 return True
361
362 client_id = get_current_client_id()
363 if client_id is None:
364 return False
365 for dashboard_id, registration in self._dashboards.items():
366 if registration.client_id != client_id:
367 continue
368 session = self._sessions.get(dashboard_id)
369 if session is None or session.dashboard != dashboard:
370 continue
371 if dashboard == DashboardType.NOW_PLAYING and session.player_id != player_id:
372 continue
373 return True
374 return False
375
376 async def _get_dashboard_code(self) -> str:
377 """Mint a fresh one-time code a cast receiver can exchange for a viewer token."""
378 # exchanged viewer tokens are fixed-lifetime guest tokens; a re-cast mints a fresh code
379 user = await get_or_create_guest_user(
380 self.mass, DASHBOARD_VIEWER_USERNAME, DASHBOARD_VIEWER_DISPLAY_NAME
381 )
382 code, _expires_at = await self.mass.webserver.auth.generate_join_code(
383 user,
384 expires_in_hours=DASHBOARD_CODE_EXPIRY_HOURS,
385 max_uses=1,
386 device_name="Dashboard Receiver",
387 )
388 return code
389
390 def _dashboard_route(self, dashboard: DashboardType, player_id: str | None) -> str:
391 """
392 Map a dashboard type to its frontend route.
393
394 :param dashboard: Dashboard to show.
395 :param player_id: Player to show, required when dashboard is NOW_PLAYING.
396 :raises InvalidCommand: If dashboard is NOW_PLAYING without a player_id, or unsupported.
397 """
398 if dashboard == DashboardType.PARTY:
399 return "/party"
400 if dashboard == DashboardType.NOW_PLAYING:
401 if not player_id:
402 msg = "player_id is required to show the now_playing dashboard"
403 raise InvalidCommand(msg)
404 return f"/now-playing?{urlencode({'player': player_id}, safe=ROUTE_SAFE_CHARS)}"
405 if dashboard == DashboardType.MUSIC_QUIZ:
406 # the viewer-only kiosk view: the host page needs USERS_INVITE, which a
407 # dashboard viewer never has
408 return "/music-quiz/dashboard"
409 msg = f"Unsupported dashboard type: {dashboard}"
410 raise InvalidCommand(msg)
411
412 def _frontend_channel(self) -> str:
413 """Derive the app.music-assistant.io frontend channel from the server version."""
414 version = self.mass.version
415 if version == "0.0.0" or ".dev" in version:
416 return "nightly"
417 if "b" in version or "rc" in version:
418 return "beta"
419 return "stable"
420
421 def _signal_dashboards_updated(self) -> None:
422 """Signal the current list of registered dashboard endpoints to subscribers."""
423 self.mass.signal_event(
424 EventType.DASHBOARDS_UPDATED,
425 data=[registration.device for registration in self._dashboards.values()],
426 )
427
428 def _signal_sessions_updated(self) -> None:
429 """Signal the current list of dashboard cast sessions to subscribers."""
430 self.mass.signal_event(
431 EventType.DASHBOARD_SESSIONS_UPDATED, data=list(self._sessions.values())
432 )
433