music-assistant-server

1"""Dashboard casting support for the Chromecast provider."""
2
3from __future__ import annotations
4
5import functools
6from typing import TYPE_CHECKING
7from uuid import UUID
8
9import pychromecast
10from music_assistant_models.dashboard import DashboardDevice
11from music_assistant_models.enums import DashboardType
12from music_assistant_models.errors import PlayerUnavailableError
13from pychromecast.const import CAST_TYPE_CHROMECAST
14
15from .helpers import send_hide_dashboard, send_show_dashboard
16from .player import ChromecastPlayer
17
18if TYPE_CHECKING:
19    from collections.abc import Callable
20
21    from pychromecast.models import CastInfo
22
23    from .provider import ChromecastProvider
24
25# Seconds to wait for an on-demand dashboard cast connection before giving up.
26DASHBOARD_CONNECT_TIMEOUT = 10.0
27
28# Dashboard types the chromecast receiver app has routes for.
29SUPPORTED_DASHBOARD_TYPES = {DashboardType.PARTY, DashboardType.NOW_PLAYING}
30
31
32class ChromecastDashboards:
33    """Registers video-capable Cast devices as dashboard endpoints and casts to them."""
34
35    def __init__(self, provider: ChromecastProvider) -> None:
36        """
37        Initialize dashboard handling for a Chromecast provider.
38
39        :param provider: The Chromecast provider owning the discovered Cast devices.
40        """
41        self.provider = provider
42        self.mass = provider.mass
43        self.logger = provider.logger.getChild("dashboard")
44        self._unregister_callbacks: dict[str, Callable[[], None]] = {}
45        # Cast connections opened on-demand for dashboard casting (not registered players)
46        self._dashboard_connections: dict[str, pychromecast.Chromecast] = {}
47        self._unloaded = False
48
49    def register(self, uuid: UUID, cast_info: CastInfo) -> None:
50        """
51        Register (or refresh) a discovered Cast device as a dashboard endpoint.
52
53        Non video-capable devices (audio speakers/groups) are ignored. A no-op
54        once the provider has unloaded, closing a race with late discovery callbacks.
55
56        :param uuid: Cast device uuid, as reported by discovery.
57        :param cast_info: Discovery info for the Cast device.
58        """
59        if self._unloaded or cast_info.cast_type != CAST_TYPE_CHROMECAST:
60            return
61        self._register_device(str(uuid), cast_info)
62
63    def unregister(self, uuid: UUID) -> None:
64        """
65        Unregister a Cast device that is no longer discovered.
66
67        :param uuid: Cast device uuid, as reported by discovery.
68        """
69        device_id = str(uuid)
70        if unregister_callback := self._unregister_callbacks.pop(device_id, None):
71            unregister_callback()
72        if chromecast := self._dashboard_connections.pop(device_id, None):
73            # non-blocking: close the socket, the daemon thread exits on its own
74            chromecast.disconnect(0)
75
76    async def unload(self) -> None:
77        """Unregister all dashboard endpoints and disconnect cached on-demand connections."""
78        self._unloaded = True
79        for unregister_callback in list(self._unregister_callbacks.values()):
80            unregister_callback()
81        self._unregister_callbacks.clear()
82
83        dashboard_connections = list(self._dashboard_connections.values())
84        self._dashboard_connections.clear()
85        for chromecast in dashboard_connections:
86            if self.mass.closing:
87                # Non-blocking disconnect: close socket, don't wait for thread.
88                # Socket threads are daemon threads and die on process exit.
89                chromecast.disconnect(0)
90            else:
91                await self.mass.loop.run_in_executor(None, chromecast.disconnect, 10)
92
93    def _register_device(self, device_id: str, cast_info: CastInfo) -> None:
94        """Build a DashboardDevice for device_id and (re-)register it with the controller."""
95        device = DashboardDevice(
96            dashboard_id=f"chromecast_{device_id}",
97            name=cast_info.friendly_name or device_id,
98            supported_types=SUPPORTED_DASHBOARD_TYPES,
99            provider_domain_hint=self.provider.domain,
100        )
101        self._unregister_callbacks[device_id] = self.mass.dashboard.register_dashboard_handler(
102            device,
103            functools.partial(self._on_show, device_id),
104            functools.partial(self._on_hide, device_id),
105        )
106
107    async def _on_show(
108        self, device_id: str, dashboard: DashboardType, player_id: str | None
109    ) -> None:
110        """
111        Show a Music Assistant dashboard on a Cast display device.
112
113        :param device_id: Cast device uuid (as string) to show the dashboard on.
114        :param dashboard: Dashboard to show.
115        :param player_id: Player to show, when dashboard is NOW_PLAYING.
116        """
117        url = await self.mass.dashboard.resolve_dashboard_url(dashboard, player_id)
118        chromecast = await self._get_or_create_chromecast(device_id)
119        try:
120            await self.mass.loop.run_in_executor(None, send_show_dashboard, chromecast, url)
121        except TimeoutError as err:
122            # the helper's message already carries device name + failure reason
123            raise PlayerUnavailableError(
124                str(err),
125                translation_key="show_dashboard_failed",
126                translation_owner=self.provider.translation_owner,
127                translation_args=[chromecast.name],
128            ) from err
129
130    async def _on_hide(self, device_id: str) -> None:
131        """
132        Hide a Music Assistant dashboard from a Cast display device.
133
134        :param device_id: Cast device uuid (as string) to hide the dashboard from.
135        """
136        chromecast = self._get_existing_chromecast(device_id)
137        if chromecast is None:
138            # nothing connected to this device: it can't be showing a dashboard
139            return
140
141        hidden = await self.mass.loop.run_in_executor(None, send_hide_dashboard, chromecast)
142        if not hidden:
143            self.logger.debug("No dashboard was showing on %s", chromecast.name)
144
145        # only tear down a connection we opened on-demand; never an active player's own cc
146        on_demand = self._dashboard_connections.pop(device_id, None)
147        if on_demand is not None:
148            await self.mass.loop.run_in_executor(None, on_demand.disconnect, 10)
149
150    async def _get_or_create_chromecast(self, device_id: str) -> pychromecast.Chromecast:
151        """Resolve a device_id to a connected Chromecast, reusing an existing connection."""
152        castplayer = self.mass.players.get_player(device_id)
153        if isinstance(castplayer, ChromecastPlayer) and castplayer.cc.socket_client.is_connected:
154            return castplayer.cc
155
156        if (chromecast := self._dashboard_connections.get(device_id)) is not None:
157            if chromecast.socket_client.is_connected:
158                return chromecast
159            del self._dashboard_connections[device_id]
160
161        assert self.provider.browser is not None  # for type checking
162        try:
163            disc_info = self.provider.browser.devices[UUID(device_id)]
164        except (KeyError, ValueError) as err:
165            msg = f"Unknown Cast device: {device_id}"
166            raise PlayerUnavailableError(msg) from err
167
168        def _connect() -> pychromecast.Chromecast:
169            """Create the Chromecast connection and wait for it to come up (blocking)."""
170            chromecast = pychromecast.get_chromecast_from_cast_info(
171                disc_info, self.mass.discovery.aiozc.zeroconf
172            )
173            chromecast.wait(timeout=DASHBOARD_CONNECT_TIMEOUT)
174            if not chromecast.socket_client.is_connected:
175                chromecast.disconnect(0)
176                msg = f"Timed out connecting to Cast device: {disc_info.friendly_name}"
177                raise PlayerUnavailableError(msg)
178            return chromecast
179
180        chromecast = await self.mass.loop.run_in_executor(None, _connect)
181        self._dashboard_connections[device_id] = chromecast
182        return chromecast
183
184    def _get_existing_chromecast(self, device_id: str) -> pychromecast.Chromecast | None:
185        """Return an already-connected Chromecast for device_id, without opening a new one."""
186        castplayer = self.mass.players.get_player(device_id)
187        if isinstance(castplayer, ChromecastPlayer) and castplayer.cc.socket_client.is_connected:
188            return castplayer.cc
189
190        if (chromecast := self._dashboard_connections.get(device_id)) is not None:
191            if chromecast.socket_client.is_connected:
192                return chromecast
193            del self._dashboard_connections[device_id]
194
195        return None
196