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