/
/
/
1"""Dashboard support for the Fully Kiosk Browser provider."""
2
3from __future__ import annotations
4
5import asyncio
6import functools
7from typing import TYPE_CHECKING
8
9from aiohttp import ClientError
10from fullykiosk.exceptions import FullyKioskError
11from music_assistant_models.dashboard import DashboardDevice
12from music_assistant_models.enums import DashboardType
13from music_assistant_models.errors import PlayerCommandFailed, PlayerUnavailableError
14
15from .player import FullyKioskPlayer
16
17if TYPE_CHECKING:
18 from collections.abc import Callable
19
20 from .provider import FullyKioskProvider
21
22# Fully Kiosk's webview can show any dashboard route, so all types are supported.
23SUPPORTED_DASHBOARD_TYPES = {
24 DashboardType.NOW_PLAYING,
25 DashboardType.PARTY,
26 DashboardType.MUSIC_QUIZ,
27}
28
29
30class FullyKioskDashboards:
31 """Registers connected Fully Kiosk devices as dashboard endpoints and drives their webview."""
32
33 def __init__(self, provider: FullyKioskProvider) -> None:
34 """
35 Initialize dashboard handling for a Fully Kiosk provider.
36
37 :param provider: The Fully Kiosk provider owning the managed devices.
38 """
39 self.provider = provider
40 self.mass = provider.mass
41 self.logger = provider.logger.getChild("dashboard")
42 self._unregister_callbacks: dict[str, Callable[[], None]] = {}
43 self._unloaded = False
44
45 def register(self, player: FullyKioskPlayer) -> None:
46 """
47 Register (or refresh) a connected player as a dashboard endpoint.
48
49 A no-op once the provider has unloaded or the player itself is gone,
50 closing a race with a late connect resolving after unload or removal.
51
52 :param player: The connected Fully Kiosk player to register.
53 """
54 if self._unloaded:
55 return
56 # a connect awaiting the device response may resolve after the player was
57 # removed; never resurrect an endpoint for it
58 if self.mass.players.get_player(player.player_id) is not player:
59 return
60 device = DashboardDevice(
61 dashboard_id=player.player_id,
62 name=player.display_name,
63 supported_types=set(SUPPORTED_DASHBOARD_TYPES),
64 provider_domain_hint=self.provider.domain,
65 )
66 self._unregister_callbacks[player.player_id] = (
67 self.mass.dashboard.register_dashboard_handler(
68 device,
69 functools.partial(self._on_show, player.player_id),
70 functools.partial(self._on_hide, player.player_id),
71 )
72 )
73
74 def unregister(self, player_id: str) -> None:
75 """
76 Drop a player's dashboard registration.
77
78 :param player_id: The player whose registration to drop.
79 """
80 if unregister_callback := self._unregister_callbacks.pop(player_id, None):
81 unregister_callback()
82
83 async def unload(self) -> None:
84 """Unregister all dashboard endpoints owned by this provider."""
85 self._unloaded = True
86 for unregister_callback in list(self._unregister_callbacks.values()):
87 unregister_callback()
88 self._unregister_callbacks.clear()
89
90 async def _on_show(
91 self, player_id: str, dashboard: DashboardType, target_player_id: str | None
92 ) -> None:
93 """Wake a Fully Kiosk device and load a Music Assistant dashboard in its webview."""
94 player = self.mass.players.get_player(player_id)
95 if not isinstance(player, FullyKioskPlayer) or player.fully_kiosk is None:
96 raise PlayerUnavailableError(f"Fully Kiosk device {player_id} is no longer available")
97 client = player.fully_kiosk
98 url = await self.mass.dashboard.resolve_dashboard_url(
99 dashboard, target_player_id, prefer_local=True
100 )
101 # never log the resolved url: it embeds a one-time viewer code
102 self.logger.debug("Showing %s dashboard on %s", dashboard.value, player.display_name)
103 try:
104 async with asyncio.timeout(15):
105 await client.screenOn()
106 await client.stopScreensaver()
107 await client.toForeground()
108 await client.loadUrl(url)
109 except FullyKioskError, ClientError, TimeoutError:
110 # never chain the caught error: its repr may embed the request url with the password
111 raise PlayerUnavailableError(
112 f"Unable to show the dashboard on {player.display_name}",
113 translation_key="show_dashboard_failed",
114 translation_owner=self.provider.translation_owner,
115 translation_args=[player.display_name],
116 ) from None
117
118 async def _on_hide(self, player_id: str) -> None:
119 """Return a Fully Kiosk device's webview to its configured start page."""
120 player = self.mass.players.get_player(player_id)
121 if not isinstance(player, FullyKioskPlayer) or player.fully_kiosk is None:
122 # nothing connected to this device: it can't be showing a dashboard
123 return
124 client = player.fully_kiosk
125 try:
126 async with asyncio.timeout(15):
127 await client.loadStartUrl()
128 except FullyKioskError, ClientError, TimeoutError:
129 # never chain the caught error: its repr may embed the request url with the password
130 raise PlayerCommandFailed(
131 f"Unable to hide the dashboard on {player.display_name}"
132 ) from None
133