/
/
/
1"""Dashboard casting support for the Chromecast provider."""
2
3from __future__ import annotations
4
5import functools
6from dataclasses import dataclass
7from typing import TYPE_CHECKING, cast
8from uuid import UUID
9
10import pychromecast
11from music_assistant_models.dashboard import DashboardDevice
12from music_assistant_models.enums import DashboardType
13from music_assistant_models.errors import PlayerUnavailableError
14from pychromecast.const import CAST_TYPE_CHROMECAST
15from pychromecast.socket_client import CONNECTION_STATUS_CONNECTED, CONNECTION_STATUS_LOST
16
17from .constants import MASS_APP_ID
18from .helpers import send_hide_dashboard, send_show_dashboard
19from .player import ChromecastPlayer
20
21if TYPE_CHECKING:
22 import asyncio
23 from collections.abc import Callable
24
25 from pychromecast.controllers.receiver import CastStatus
26 from pychromecast.controllers.receiver import CastStatusListener as ReceiverStatusListener
27 from pychromecast.models import CastInfo
28 from pychromecast.socket_client import ConnectionStatus, ConnectionStatusListener
29
30 from .provider import ChromecastProvider
31
32# Seconds to wait for an on-demand dashboard cast connection before giving up.
33DASHBOARD_CONNECT_TIMEOUT = 10.0
34
35# Seconds a lost connection is allowed to recover before we consider the dashboard gone.
36DASHBOARD_CONNECTION_LOSS_GRACE = 60.0
37
38# Dashboard types the chromecast receiver app has routes for.
39SUPPORTED_DASHBOARD_TYPES = {
40 DashboardType.PARTY,
41 DashboardType.NOW_PLAYING,
42 DashboardType.MUSIC_QUIZ,
43}
44
45
46@dataclass
47class _ActiveDashboardCast:
48 """Tracks a dashboard currently showing on a Cast device, to detect it dying externally."""
49
50 cast_session_id: str | None
51 listener: _DashboardCastListener
52 loss_timer: asyncio.TimerHandle | None = None
53
54
55class ChromecastDashboards:
56 """Registers video-capable Cast devices as dashboard endpoints and casts to them."""
57
58 def __init__(self, provider: ChromecastProvider) -> None:
59 """
60 Initialize dashboard handling for a Chromecast provider.
61
62 :param provider: The Chromecast provider owning the discovered Cast devices.
63 """
64 self.provider = provider
65 self.mass = provider.mass
66 self.logger = provider.logger.getChild("dashboard")
67 self._unregister_callbacks: dict[str, Callable[[], None]] = {}
68 # Cast connections opened on-demand for dashboard casting (not registered players)
69 self._dashboard_connections: dict[str, pychromecast.Chromecast] = {}
70 self._active_casts: dict[str, _ActiveDashboardCast] = {}
71 self._unloaded = False
72
73 def register(self, uuid: UUID, cast_info: CastInfo) -> None:
74 """
75 Register (or refresh) a discovered Cast device as a dashboard endpoint.
76
77 Non video-capable devices (audio speakers/groups) are ignored. A no-op
78 once the provider has unloaded, closing a race with late discovery callbacks.
79
80 :param uuid: Cast device uuid, as reported by discovery.
81 :param cast_info: Discovery info for the Cast device.
82 """
83 if self._unloaded or cast_info.cast_type != CAST_TYPE_CHROMECAST:
84 return
85 self._register_device(str(uuid), cast_info)
86
87 def unregister(self, uuid: UUID) -> None:
88 """
89 Unregister a Cast device that is no longer discovered.
90
91 :param uuid: Cast device uuid, as reported by discovery.
92 """
93 device_id = str(uuid)
94 if unregister_callback := self._unregister_callbacks.pop(device_id, None):
95 unregister_callback()
96 if chromecast := self._dashboard_connections.pop(device_id, None):
97 # non-blocking: close the socket, the daemon thread exits on its own
98 chromecast.disconnect(0)
99 self._drop_active_cast(device_id)
100
101 async def unload(self) -> None:
102 """Unregister all dashboard endpoints and disconnect cached on-demand connections."""
103 self._unloaded = True
104 for unregister_callback in list(self._unregister_callbacks.values()):
105 unregister_callback()
106 self._unregister_callbacks.clear()
107
108 for device_id in list(self._active_casts):
109 self._drop_active_cast(device_id)
110
111 dashboard_connections = list(self._dashboard_connections.values())
112 self._dashboard_connections.clear()
113 for chromecast in dashboard_connections:
114 if self.mass.closing:
115 # Non-blocking disconnect: close socket, don't wait for thread.
116 # Socket threads are daemon threads and die on process exit.
117 chromecast.disconnect(0)
118 else:
119 await self.mass.loop.run_in_executor(None, chromecast.disconnect, 10)
120
121 def _register_device(self, device_id: str, cast_info: CastInfo) -> None:
122 """Build a DashboardDevice for device_id and (re-)register it with the controller."""
123 device = DashboardDevice(
124 dashboard_id=f"chromecast_{device_id}",
125 name=cast_info.friendly_name or device_id,
126 supported_types=SUPPORTED_DASHBOARD_TYPES,
127 provider_domain_hint=self.provider.domain,
128 )
129 self._unregister_callbacks[device_id] = self.mass.dashboard.register_dashboard_handler(
130 device,
131 functools.partial(self._on_show, device_id),
132 functools.partial(self._on_hide, device_id),
133 )
134
135 async def _on_show(
136 self, device_id: str, dashboard: DashboardType, player_id: str | None
137 ) -> None:
138 """
139 Show a Music Assistant dashboard on a Cast display device.
140
141 :param device_id: Cast device uuid (as string) to show the dashboard on.
142 :param dashboard: Dashboard to show.
143 :param player_id: Player to show, when dashboard is NOW_PLAYING.
144 """
145 player = self.mass.players.get_player(device_id)
146 castplayer = player if isinstance(player, ChromecastPlayer) else None
147 force_launch = False
148 if castplayer is not None:
149 # an earlier stop on the player may still have a release of the receiver
150 # app pending, which would close the dashboard we are about to show
151 castplayer.cancel_pending_app_quit()
152 # a release that is already on the wire will close the app anyway, so the
153 # receiver's 'already running' short-circuit has to be bypassed
154 force_launch = castplayer.app_quit_sent
155 url = await self.mass.dashboard.resolve_dashboard_url(dashboard, player_id)
156 chromecast = await self._get_or_create_chromecast(device_id)
157 try:
158 await self.mass.loop.run_in_executor(
159 None,
160 functools.partial(send_show_dashboard, chromecast, url, force_launch=force_launch),
161 )
162 except TimeoutError as err:
163 # the helper's message already carries device name + failure reason
164 raise PlayerUnavailableError(
165 str(err),
166 translation_key="show_dashboard_failed",
167 translation_owner=self.provider.translation_owner,
168 translation_args=[chromecast.name],
169 ) from err
170
171 if force_launch and castplayer is not None:
172 # this launch replaced the session the recorded release was closing
173 castplayer.app_quit_sent = False
174
175 self._drop_active_cast(device_id)
176 session_id = chromecast.status.session_id if chromecast.status else None
177 listener = _DashboardCastListener(self, device_id)
178 self._active_casts[device_id] = _ActiveDashboardCast(
179 cast_session_id=session_id, listener=listener
180 )
181 chromecast.register_status_listener(cast("ReceiverStatusListener", listener))
182 chromecast.register_connection_listener(cast("ConnectionStatusListener", listener))
183
184 async def _on_hide(self, device_id: str) -> None:
185 """
186 Hide a Music Assistant dashboard from a Cast display device.
187
188 :param device_id: Cast device uuid (as string) to hide the dashboard from.
189 """
190 self._drop_active_cast(device_id)
191
192 chromecast = self._get_existing_chromecast(device_id)
193 if chromecast is None:
194 # nothing connected to this device: it can't be showing a dashboard
195 return
196
197 hidden = await self.mass.loop.run_in_executor(None, send_hide_dashboard, chromecast)
198 if not hidden:
199 self.logger.debug("No dashboard was showing on %s", chromecast.name)
200
201 # only tear down a connection we opened on-demand; never an active player's own cc
202 on_demand = self._dashboard_connections.pop(device_id, None)
203 if on_demand is not None:
204 await self.mass.loop.run_in_executor(None, on_demand.disconnect, 10)
205
206 async def _get_or_create_chromecast(self, device_id: str) -> pychromecast.Chromecast:
207 """Resolve a device_id to a connected Chromecast, reusing an existing connection."""
208 castplayer = self.mass.players.get_player(device_id)
209 if isinstance(castplayer, ChromecastPlayer) and castplayer.cc.socket_client.is_connected:
210 return castplayer.cc
211
212 if (chromecast := self._dashboard_connections.get(device_id)) is not None:
213 if chromecast.socket_client.is_connected:
214 return chromecast
215 del self._dashboard_connections[device_id]
216
217 assert self.provider.browser is not None # for type checking
218 try:
219 disc_info = self.provider.browser.devices[UUID(device_id)]
220 except (KeyError, ValueError) as err:
221 msg = f"Unknown Cast device: {device_id}"
222 raise PlayerUnavailableError(msg) from err
223
224 def _connect() -> pychromecast.Chromecast:
225 """Create the Chromecast connection and wait for it to come up (blocking)."""
226 chromecast = pychromecast.get_chromecast_from_cast_info(
227 disc_info, self.mass.discovery.aiozc.zeroconf
228 )
229 chromecast.wait(timeout=DASHBOARD_CONNECT_TIMEOUT)
230 if not chromecast.socket_client.is_connected:
231 chromecast.disconnect(0)
232 msg = f"Timed out connecting to Cast device: {disc_info.friendly_name}"
233 raise PlayerUnavailableError(msg)
234 return chromecast
235
236 chromecast = await self.mass.loop.run_in_executor(None, _connect)
237 self._dashboard_connections[device_id] = chromecast
238 return chromecast
239
240 def _get_existing_chromecast(self, device_id: str) -> pychromecast.Chromecast | None:
241 """Return an already-connected Chromecast for device_id, without opening a new one."""
242 castplayer = self.mass.players.get_player(device_id)
243 if isinstance(castplayer, ChromecastPlayer) and castplayer.cc.socket_client.is_connected:
244 return castplayer.cc
245
246 if (chromecast := self._dashboard_connections.get(device_id)) is not None:
247 if chromecast.socket_client.is_connected:
248 return chromecast
249 del self._dashboard_connections[device_id]
250
251 return None
252
253 def _handle_cast_status(self, device_id: str, status: CastStatus) -> None:
254 """Detect the dashboard receiver app being replaced or restarted, on the event loop."""
255 entry = self._active_casts.get(device_id)
256 if entry is None:
257 return
258 if status.app_id != MASS_APP_ID or status.session_id != entry.cast_session_id:
259 reason = f"the receiver app was closed (active app: {status.display_name or status.app_id or 'none'})"
260 self._session_lost(device_id, reason)
261
262 def _handle_connection_status(self, device_id: str, status: ConnectionStatus) -> None:
263 """Arm/disarm the connection-loss grace timer, on the event loop."""
264 entry = self._active_casts.get(device_id)
265 if entry is None:
266 return
267 if status.status == CONNECTION_STATUS_LOST:
268 if entry.loss_timer is None:
269 entry.loss_timer = self.mass.loop.call_later(
270 DASHBOARD_CONNECTION_LOSS_GRACE,
271 self._session_lost,
272 device_id,
273 "connection to the device was lost",
274 )
275 elif status.status == CONNECTION_STATUS_CONNECTED:
276 # the receiver status that follows reconnection re-verifies the app itself
277 self._cancel_loss_timer(device_id)
278
279 def _session_lost(self, device_id: str, reason: str) -> None:
280 """Report a dashboard session as lost and drop the on-demand connection, if any."""
281 if device_id not in self._active_casts:
282 return
283 self._drop_active_cast(device_id)
284 self.mass.dashboard.end_session(f"chromecast_{device_id}", reason)
285
286 if chromecast := self._dashboard_connections.pop(device_id, None):
287 # non-blocking: close the socket, the daemon thread exits on its own
288 chromecast.disconnect(0)
289
290 def _cancel_loss_timer(self, device_id: str) -> None:
291 """Cancel the pending connection-loss timer for device_id, if any."""
292 entry = self._active_casts.get(device_id)
293 if entry is not None and entry.loss_timer is not None:
294 entry.loss_timer.cancel()
295 entry.loss_timer = None
296
297 def _drop_active_cast(self, device_id: str) -> None:
298 """Stop tracking device_id: cancel its loss timer and invalidate its listener."""
299 entry = self._active_casts.pop(device_id, None)
300 if entry is None:
301 return
302 if entry.loss_timer is not None:
303 entry.loss_timer.cancel()
304 entry.listener.invalidate()
305
306
307class _DashboardCastListener:
308 """
309 Watches a Cast device to detect its dashboard receiver dying externally.
310
311 pychromecast has no API to unregister a listener from a Chromecast object,
312 so a superseded listener is invalidated instead: it stays registered but
313 its callbacks become no-ops.
314 """
315
316 def __init__(self, dashboards: ChromecastDashboards, device_id: str) -> None:
317 """Bind this listener to the dashboards owner and the Cast device it watches."""
318 self.dashboards = dashboards
319 self.device_id = device_id
320 self._valid = True
321
322 def invalidate(self) -> None:
323 """Stop forwarding callbacks from this listener."""
324 self._valid = False
325
326 def new_cast_status(self, status: CastStatus) -> None:
327 """Handle updated CastStatus (called from the pychromecast socket thread)."""
328 if not self._valid or self.dashboards.mass.closing:
329 return
330 self.dashboards.mass.loop.call_soon_threadsafe(
331 self.dashboards._handle_cast_status, self.device_id, status
332 )
333
334 def new_connection_status(self, status: ConnectionStatus) -> None:
335 """Handle updated ConnectionStatus (called from the pychromecast socket thread)."""
336 if not self._valid or self.dashboards.mass.closing:
337 return
338 self.dashboards.mass.loop.call_soon_threadsafe(
339 self.dashboards._handle_connection_status, self.device_id, status
340 )
341