music-assistant-server

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    async def resolve_dashboard_url(
294        self, dashboard: DashboardType, player_id: str | None, *, prefer_local: bool = False
295    ) -> str:
296        """
297        Build the fully-qualified URL a dashboard endpoint should load to show a dashboard.
298
299        By default an externally-reachable https base url (reverse-proxied server, same
300        origin) is preferred over remote access (the app.music-assistant.io signaling portal),
301        as required by cast receivers. With ``prefer_local`` the server's own base url is
302        always returned, plain http included: native apps on the LAN are not bound by the cast
303        receiver's https requirement. In-server consumers (e.g. the chromecast provider) call
304        this to resolve the url themselves.
305
306        :param dashboard: Dashboard to show.
307        :param player_id: Player to show, required when dashboard is NOW_PLAYING.
308        :param prefer_local: Return the plain local base url instead of the https/remote form.
309        :raises ActionUnavailable: If neither an https base url nor remote access is configured
310            (never raised when ``prefer_local`` is set).
311        """
312        route = self._dashboard_route(dashboard, player_id)
313        base_url = self.mass.webserver.base_url
314        if prefer_local:
315            # native LAN apps talk straight to this server, no https/remote gate needed
316            query = {"dashboard": await self._get_dashboard_code(), "path": route}
317            return f"{base_url}?{urlencode(query)}"
318        remote_access = self.mass.webserver.remote_access
319        use_https_base = base_url.startswith("https://")
320        if not use_https_base and not (remote_access.is_enabled and remote_access.remote_id):
321            msg = "Remote access or an https base url is required to cast dashboards"
322            raise ActionUnavailable(
323                msg,
324                translation_key="remote_access_required",
325                translation_owner=self.translation_owner,
326            )
327
328        dashboard_code = await self._get_dashboard_code()
329        if use_https_base:
330            # same origin: the receiver talks straight to this server, no remote_id needed
331            query = {"dashboard": dashboard_code, "path": route}
332            return f"{base_url}?{urlencode(query)}"
333
334        query = {"remote_id": remote_access.remote_id, "dashboard": dashboard_code, "path": route}
335        channel = self._frontend_channel()
336        return f"{APP_MA_HOST}/{channel}/?{urlencode(query)}"
337
338    def _can_resolve_url_for_caller(self, dashboard: DashboardType, player_id: str | None) -> bool:
339        """
340        Return whether the current caller may resolve a dashboard url for itself.
341
342        :param dashboard: Dashboard the caller wants a url for.
343        :param player_id: Player the caller wants a url for, when dashboard is NOW_PLAYING.
344        """
345        user = get_current_user()
346        if user is not None and has_scope(user, Scope.USERS_INVITE):
347            return True
348
349        client_id = get_current_client_id()
350        if client_id is None:
351            return False
352        for dashboard_id, registration in self._dashboards.items():
353            if registration.client_id != client_id:
354                continue
355            session = self._sessions.get(dashboard_id)
356            if session is None or session.dashboard != dashboard:
357                continue
358            if dashboard == DashboardType.NOW_PLAYING and session.player_id != player_id:
359                continue
360            return True
361        return False
362
363    async def _get_dashboard_code(self) -> str:
364        """Mint a fresh one-time code a cast receiver can exchange for a viewer token."""
365        # exchanged viewer tokens are fixed-lifetime guest tokens; a re-cast mints a fresh code
366        user = await get_or_create_guest_user(
367            self.mass, DASHBOARD_VIEWER_USERNAME, DASHBOARD_VIEWER_DISPLAY_NAME
368        )
369        code, _expires_at = await self.mass.webserver.auth.generate_join_code(
370            user,
371            expires_in_hours=DASHBOARD_CODE_EXPIRY_HOURS,
372            max_uses=1,
373            device_name="Dashboard Receiver",
374        )
375        return code
376
377    def _dashboard_route(self, dashboard: DashboardType, player_id: str | None) -> str:
378        """
379        Map a dashboard type to its frontend route.
380
381        :param dashboard: Dashboard to show.
382        :param player_id: Player to show, required when dashboard is NOW_PLAYING.
383        :raises InvalidCommand: If dashboard is NOW_PLAYING without a player_id, or unsupported.
384        """
385        if dashboard == DashboardType.PARTY:
386            return "/party"
387        if dashboard == DashboardType.NOW_PLAYING:
388            if not player_id:
389                msg = "player_id is required to show the now_playing dashboard"
390                raise InvalidCommand(msg)
391            return f"/now-playing?{urlencode({'player': player_id}, safe=ROUTE_SAFE_CHARS)}"
392        if dashboard == DashboardType.MUSIC_QUIZ:
393            return "/music-quiz"
394        msg = f"Unsupported dashboard type: {dashboard}"
395        raise InvalidCommand(msg)
396
397    def _frontend_channel(self) -> str:
398        """Derive the app.music-assistant.io frontend channel from the server version."""
399        version = self.mass.version
400        if version == "0.0.0" or ".dev" in version:
401            return "nightly"
402        if "b" in version or "rc" in version:
403            return "beta"
404        return "stable"
405
406    def _signal_dashboards_updated(self) -> None:
407        """Signal the current list of registered dashboard endpoints to subscribers."""
408        self.mass.signal_event(
409            EventType.DASHBOARDS_UPDATED,
410            data=[registration.device for registration in self._dashboards.values()],
411        )
412
413    def _signal_sessions_updated(self) -> None:
414        """Signal the current list of dashboard cast sessions to subscribers."""
415        self.mass.signal_event(
416            EventType.DASHBOARD_SESSIONS_UPDATED, data=list(self._sessions.values())
417        )
418