music-assistant-server

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        url = await self.mass.dashboard.resolve_dashboard_url(dashboard, player_id)
146        chromecast = await self._get_or_create_chromecast(device_id)
147        try:
148            await self.mass.loop.run_in_executor(None, send_show_dashboard, chromecast, url)
149        except TimeoutError as err:
150            # the helper's message already carries device name + failure reason
151            raise PlayerUnavailableError(
152                str(err),
153                translation_key="show_dashboard_failed",
154                translation_owner=self.provider.translation_owner,
155                translation_args=[chromecast.name],
156            ) from err
157
158        self._drop_active_cast(device_id)
159        session_id = chromecast.status.session_id if chromecast.status else None
160        listener = _DashboardCastListener(self, device_id)
161        self._active_casts[device_id] = _ActiveDashboardCast(
162            cast_session_id=session_id, listener=listener
163        )
164        chromecast.register_status_listener(cast("ReceiverStatusListener", listener))
165        chromecast.register_connection_listener(cast("ConnectionStatusListener", listener))
166
167    async def _on_hide(self, device_id: str) -> None:
168        """
169        Hide a Music Assistant dashboard from a Cast display device.
170
171        :param device_id: Cast device uuid (as string) to hide the dashboard from.
172        """
173        self._drop_active_cast(device_id)
174
175        chromecast = self._get_existing_chromecast(device_id)
176        if chromecast is None:
177            # nothing connected to this device: it can't be showing a dashboard
178            return
179
180        hidden = await self.mass.loop.run_in_executor(None, send_hide_dashboard, chromecast)
181        if not hidden:
182            self.logger.debug("No dashboard was showing on %s", chromecast.name)
183
184        # only tear down a connection we opened on-demand; never an active player's own cc
185        on_demand = self._dashboard_connections.pop(device_id, None)
186        if on_demand is not None:
187            await self.mass.loop.run_in_executor(None, on_demand.disconnect, 10)
188
189    async def _get_or_create_chromecast(self, device_id: str) -> pychromecast.Chromecast:
190        """Resolve a device_id to a connected Chromecast, reusing an existing connection."""
191        castplayer = self.mass.players.get_player(device_id)
192        if isinstance(castplayer, ChromecastPlayer) and castplayer.cc.socket_client.is_connected:
193            return castplayer.cc
194
195        if (chromecast := self._dashboard_connections.get(device_id)) is not None:
196            if chromecast.socket_client.is_connected:
197                return chromecast
198            del self._dashboard_connections[device_id]
199
200        assert self.provider.browser is not None  # for type checking
201        try:
202            disc_info = self.provider.browser.devices[UUID(device_id)]
203        except (KeyError, ValueError) as err:
204            msg = f"Unknown Cast device: {device_id}"
205            raise PlayerUnavailableError(msg) from err
206
207        def _connect() -> pychromecast.Chromecast:
208            """Create the Chromecast connection and wait for it to come up (blocking)."""
209            chromecast = pychromecast.get_chromecast_from_cast_info(
210                disc_info, self.mass.discovery.aiozc.zeroconf
211            )
212            chromecast.wait(timeout=DASHBOARD_CONNECT_TIMEOUT)
213            if not chromecast.socket_client.is_connected:
214                chromecast.disconnect(0)
215                msg = f"Timed out connecting to Cast device: {disc_info.friendly_name}"
216                raise PlayerUnavailableError(msg)
217            return chromecast
218
219        chromecast = await self.mass.loop.run_in_executor(None, _connect)
220        self._dashboard_connections[device_id] = chromecast
221        return chromecast
222
223    def _get_existing_chromecast(self, device_id: str) -> pychromecast.Chromecast | None:
224        """Return an already-connected Chromecast for device_id, without opening a new one."""
225        castplayer = self.mass.players.get_player(device_id)
226        if isinstance(castplayer, ChromecastPlayer) and castplayer.cc.socket_client.is_connected:
227            return castplayer.cc
228
229        if (chromecast := self._dashboard_connections.get(device_id)) is not None:
230            if chromecast.socket_client.is_connected:
231                return chromecast
232            del self._dashboard_connections[device_id]
233
234        return None
235
236    def _handle_cast_status(self, device_id: str, status: CastStatus) -> None:
237        """Detect the dashboard receiver app being replaced or restarted, on the event loop."""
238        entry = self._active_casts.get(device_id)
239        if entry is None:
240            return
241        if status.app_id != MASS_APP_ID or status.session_id != entry.cast_session_id:
242            reason = f"the receiver app was closed (active app: {status.display_name or status.app_id or 'none'})"
243            self._session_lost(device_id, reason)
244
245    def _handle_connection_status(self, device_id: str, status: ConnectionStatus) -> None:
246        """Arm/disarm the connection-loss grace timer, on the event loop."""
247        entry = self._active_casts.get(device_id)
248        if entry is None:
249            return
250        if status.status == CONNECTION_STATUS_LOST:
251            if entry.loss_timer is None:
252                entry.loss_timer = self.mass.loop.call_later(
253                    DASHBOARD_CONNECTION_LOSS_GRACE,
254                    self._session_lost,
255                    device_id,
256                    "connection to the device was lost",
257                )
258        elif status.status == CONNECTION_STATUS_CONNECTED:
259            # the receiver status that follows reconnection re-verifies the app itself
260            self._cancel_loss_timer(device_id)
261
262    def _session_lost(self, device_id: str, reason: str) -> None:
263        """Report a dashboard session as lost and drop the on-demand connection, if any."""
264        if device_id not in self._active_casts:
265            return
266        self._drop_active_cast(device_id)
267        self.mass.dashboard.end_session(f"chromecast_{device_id}", reason)
268
269        if chromecast := self._dashboard_connections.pop(device_id, None):
270            # non-blocking: close the socket, the daemon thread exits on its own
271            chromecast.disconnect(0)
272
273    def _cancel_loss_timer(self, device_id: str) -> None:
274        """Cancel the pending connection-loss timer for device_id, if any."""
275        entry = self._active_casts.get(device_id)
276        if entry is not None and entry.loss_timer is not None:
277            entry.loss_timer.cancel()
278            entry.loss_timer = None
279
280    def _drop_active_cast(self, device_id: str) -> None:
281        """Stop tracking device_id: cancel its loss timer and invalidate its listener."""
282        entry = self._active_casts.pop(device_id, None)
283        if entry is None:
284            return
285        if entry.loss_timer is not None:
286            entry.loss_timer.cancel()
287        entry.listener.invalidate()
288
289
290class _DashboardCastListener:
291    """
292    Watches a Cast device to detect its dashboard receiver dying externally.
293
294    pychromecast has no API to unregister a listener from a Chromecast object,
295    so a superseded listener is invalidated instead: it stays registered but
296    its callbacks become no-ops.
297    """
298
299    def __init__(self, dashboards: ChromecastDashboards, device_id: str) -> None:
300        """Bind this listener to the dashboards owner and the Cast device it watches."""
301        self.dashboards = dashboards
302        self.device_id = device_id
303        self._valid = True
304
305    def invalidate(self) -> None:
306        """Stop forwarding callbacks from this listener."""
307        self._valid = False
308
309    def new_cast_status(self, status: CastStatus) -> None:
310        """Handle updated CastStatus (called from the pychromecast socket thread)."""
311        if not self._valid or self.dashboards.mass.closing:
312            return
313        self.dashboards.mass.loop.call_soon_threadsafe(
314            self.dashboards._handle_cast_status, self.device_id, status
315        )
316
317    def new_connection_status(self, status: ConnectionStatus) -> None:
318        """Handle updated ConnectionStatus (called from the pychromecast socket thread)."""
319        if not self._valid or self.dashboards.mass.closing:
320            return
321        self.dashboards.mass.loop.call_soon_threadsafe(
322            self.dashboards._handle_connection_status, self.device_id, status
323        )
324