music-assistant-server

1"""Helpers to deal with Cast devices."""
2
3from __future__ import annotations
4
5import threading
6import time
7import urllib.error
8from dataclasses import asdict, dataclass
9from typing import TYPE_CHECKING, Any, cast
10from uuid import UUID
11
12from pychromecast import dial
13from pychromecast.const import CAST_TYPE_GROUP
14
15from music_assistant.constants import VERBOSE_LOG_LEVEL
16
17from .constants import DASHBOARD_NAMESPACE, MASS_APP_ID
18
19if TYPE_CHECKING:
20    from pychromecast import Chromecast
21    from pychromecast.controllers.media import MediaStatus, MediaStatusListener
22    from pychromecast.controllers.multizone import MultizoneManager, MultiZoneManagerListener
23    from pychromecast.controllers.receiver import CastStatus
24    from pychromecast.controllers.receiver import CastStatusListener as ReceiverStatusListener
25    from pychromecast.models import CastInfo, HostServiceInfo, MDNSServiceInfo
26    from pychromecast.socket_client import ConnectionStatus, ConnectionStatusListener
27    from zeroconf import Zeroconf
28
29    from .player import ChromecastPlayer
30
31DEFAULT_PORT = 8009
32DASHBOARD_NAMESPACE_POLL_INTERVAL = 0.1
33
34
35def send_show_dashboard(
36    chromecast: Chromecast,
37    url: str,
38    timeout: float = 30.0,
39) -> None:
40    """
41    Launch the MA cast receiver app and send it a show_dashboard message.
42
43    Blocking call, run from an executor.
44
45    :param chromecast: Connected Chromecast to show the dashboard on.
46    :param url: Fully-qualified dashboard URL for the receiver to load.
47    :param timeout: Seconds to wait for the app launch and the dashboard namespace.
48    :raises TimeoutError: If the receiver app did not launch (in time), or the
49        dashboard namespace never became available.
50    """
51    launched = threading.Event()
52    launch_success = False
53
54    def _on_launched(success: bool, _response: dict[str, Any] | None) -> None:
55        nonlocal launch_success
56        launch_success = success
57        launched.set()
58
59    deadline = time.monotonic() + timeout
60    chromecast.socket_client.receiver_controller.launch_app(
61        MASS_APP_ID, callback_function=_on_launched
62    )
63    if not launched.wait(timeout):
64        msg = f"Timed out launching app on {chromecast.name}"
65        raise TimeoutError(msg)
66    if not launch_success:
67        msg = f"Launching app on {chromecast.name} failed"
68        raise TimeoutError(msg)
69
70    # tiny race: the namespace only appears once the socket client has processed
71    # the same receiver status that completes the launch callback
72    while DASHBOARD_NAMESPACE not in chromecast.socket_client.app_namespaces:
73        if time.monotonic() >= deadline:
74            msg = f"Timed out waiting for the dashboard namespace on {chromecast.name}"
75            raise TimeoutError(msg)
76        time.sleep(DASHBOARD_NAMESPACE_POLL_INTERVAL)
77
78    chromecast.socket_client.send_app_message(
79        DASHBOARD_NAMESPACE, {"type": "show_dashboard", "url": url}
80    )
81
82
83def send_hide_dashboard(chromecast: Chromecast) -> bool:
84    """
85    Send a hide_dashboard message to an already-running MA receiver app.
86
87    Blocking call, run from an executor. Does not launch the app: if the
88    receiver isn't already showing a dashboard, there is nothing to hide.
89
90    :param chromecast: Connected Chromecast to hide the dashboard on.
91    :return: Whether a hide_dashboard message was sent.
92    """
93    if (
94        chromecast.app_id != MASS_APP_ID
95        or DASHBOARD_NAMESPACE not in chromecast.socket_client.app_namespaces
96    ):
97        return False
98
99    chromecast.socket_client.send_app_message(DASHBOARD_NAMESPACE, {"type": "hide_dashboard"})
100    return True
101
102
103@dataclass
104class ChromecastInfo:
105    """
106    Class to hold all data about a chromecast for creating connections.
107
108    This also has the same attributes as the mDNS fields by zeroconf.
109    """
110
111    services: set[HostServiceInfo | MDNSServiceInfo]
112    uuid: UUID
113    model_name: str
114    friendly_name: str
115    host: str
116    port: int
117    cast_type: str | None = None
118    manufacturer: str | None = None
119    is_dynamic_group: bool | None = None
120    is_multichannel_group: bool = False  # group created for e.g. stereo pair
121    is_multichannel_child: bool = False  # speaker that is part of multichannel setup
122    mac_address: str | None = None  # MAC address from eureka_info API
123
124    @property
125    def is_audio_group(self) -> bool:
126        """Return if the cast is an audio group."""
127        return self.cast_type == CAST_TYPE_GROUP
128
129    @classmethod
130    def from_cast_info(cls, cast_info: CastInfo) -> ChromecastInfo:
131        """Instantiate ChromecastInfo from CastInfo."""
132        return cls(**asdict(cast_info))
133
134    def update(self, cast_info: CastInfo) -> None:
135        """Update ChromecastInfo from CastInfo."""
136        for key, value in asdict(cast_info).items():
137            if not value:
138                continue
139            setattr(self, key, value)
140
141    def fill_out_missing_chromecast_info(self, zconf: Zeroconf) -> None:
142        """
143        Return a new ChromecastInfo object with missing attributes filled in.
144
145        Uses blocking HTTP / HTTPS.
146        """
147        if self.cast_type is None or self.manufacturer is None:
148            # Manufacturer and cast type is not available in mDNS data,
149            # get it over HTTP
150            cast_info = dial.get_cast_type(
151                cast("CastInfo", self),
152                zconf=zconf,
153            )
154            self.cast_type = cast_info.cast_type
155            self.manufacturer = cast_info.manufacturer
156
157        # Fill out missing group information via HTTP API.
158        dynamic_groups, multichannel_groups = get_multizone_info(self.services, zconf)
159        self.is_dynamic_group = self.uuid in dynamic_groups
160        if self.uuid in multichannel_groups:
161            self.is_multichannel_group = True
162        elif (
163            multichannel_groups
164            # Prevent a multichannel group being marked as a multichannel child
165            # if not in UUID list
166            and self.cast_type != "group"
167            and self.model_name != "Google Cast Group"
168        ):
169            self.is_multichannel_child = True
170
171        # Get MAC address for device matching (not available for groups)
172        if self.mac_address is None and self.cast_type != "group":
173            self.mac_address = get_mac_address(self.services, zconf)
174
175
176def get_multizone_info(
177    services: set[HostServiceInfo | MDNSServiceInfo],
178    zconf: Zeroconf,
179    timeout: int = 30,
180) -> tuple[set[UUID], set[UUID]]:
181    """Get multizone info from eureka endpoint."""
182    dynamic_groups: set[UUID] = set()
183    multichannel_groups: set[UUID] = set()
184    try:
185        _, status = dial._get_status(
186            services,
187            zconf,
188            "/setup/eureka_info?params=multizone",
189            True,
190            timeout,
191            None,
192        )
193        if "multizone" in status and "dynamic_groups" in status["multizone"]:
194            for group in status["multizone"]["dynamic_groups"]:
195                if udn := group.get("uuid"):
196                    uuid = UUID(udn.replace("-", ""))
197                    dynamic_groups.add(uuid)
198
199        if "multizone" in status and "groups" in status["multizone"]:
200            for group in status["multizone"]["groups"]:
201                if "multichannel_group" not in group:
202                    continue
203                if group["multichannel_group"] and (udn := group.get("uuid")):
204                    uuid = UUID(udn.replace("-", ""))
205                    # new firmware drops cast_port and renames elected_leader to leader
206                    is_leader = (
207                        group.get("elected_leader") == "self" or group.get("leader") == "self"
208                    )
209                    if group.get("cast_port") or not is_leader:
210                        multichannel_groups.add(uuid)
211    except urllib.error.HTTPError, urllib.error.URLError, OSError, KeyError, ValueError:
212        pass
213    return (dynamic_groups, multichannel_groups)
214
215
216def get_mac_address(
217    services: set[HostServiceInfo | MDNSServiceInfo], zconf: Zeroconf, timeout: int = 10
218) -> str | None:
219    """
220    Get MAC address from Chromecast eureka_info API.
221
222    :param services: Set of zeroconf service info.
223    :param zconf: Zeroconf instance.
224    :param timeout: Request timeout in seconds.
225    :return: MAC address string or None if not available.
226    """
227    try:
228        _, status = dial._get_status(
229            services,
230            zconf,
231            "/setup/eureka_info?options=detail",
232            True,
233            timeout,
234            None,
235        )
236        if mac_address := status.get("mac_address"):
237            # Normalize to uppercase with colons
238            mac = mac_address.upper().replace("-", ":")
239            # Ensure proper format
240            if ":" not in mac and len(mac) == 12:
241                mac = ":".join(mac[i : i + 2] for i in range(0, 12, 2))
242            return str(mac)
243    except urllib.error.HTTPError, urllib.error.URLError, OSError, KeyError, ValueError:
244        pass
245    return None
246
247
248class CastStatusListener:
249    """
250    Helper class to handle pychromecast status callbacks.
251
252    Necessary because a CastDevice entity can create a new socket client
253    and therefore callbacks from multiple chromecast connections can
254    potentially arrive. This class allows invalidating past chromecast objects.
255    """
256
257    def __init__(
258        self,
259        castplayer: ChromecastPlayer,
260        mz_mgr: MultizoneManager,
261        mz_only: bool = False,
262    ) -> None:
263        """Initialize the status listener."""
264        self.castplayer = castplayer
265        self._uuid = castplayer.cc.uuid
266        self._valid = True
267        self._mz_mgr = mz_mgr
268        if self.castplayer.cast_info.is_audio_group:
269            self._mz_mgr.add_multizone(castplayer.cc)
270        if mz_only:
271            return
272        castplayer.cc.register_status_listener(cast("ReceiverStatusListener", self))
273        castplayer.cc.socket_client.media_controller.register_status_listener(
274            cast("MediaStatusListener", self)
275        )
276        castplayer.cc.register_connection_listener(cast("ConnectionStatusListener", self))
277        if not self.castplayer.cast_info.is_audio_group:
278            self._mz_mgr.register_listener(
279                castplayer.cc.uuid, cast("MultiZoneManagerListener", self)
280            )
281
282    def new_cast_status(self, status: CastStatus) -> None:
283        """Handle updated CastStatus."""
284        if not self._valid:
285            return
286        self.castplayer.on_new_cast_status(status)
287
288    def new_media_status(self, status: MediaStatus) -> None:
289        """Handle updated MediaStatus."""
290        if not self._valid:
291            return
292        self.castplayer.on_new_media_status(status)
293
294    def new_connection_status(self, status: ConnectionStatus) -> None:
295        """Handle updated ConnectionStatus."""
296        if not self._valid:
297            return
298        self.castplayer.on_new_connection_status(status)
299
300    def added_to_multizone(self, group_uuid: str) -> None:
301        """Handle the cast added to a group."""
302        self.castplayer.logger.debug(
303            "%s is added to multizone: %s", self.castplayer.display_name, group_uuid
304        )
305        player_status = self.castplayer.cc.status
306        if player_status is None:
307            return
308
309        self.new_cast_status(player_status)
310
311    def removed_from_multizone(self, group_uuid: str) -> None:
312        """Handle the cast removed from a group."""
313        if not self._valid:
314            return
315        if group_uuid == self.castplayer.active_source:
316            mass = self.castplayer.mass
317            mass.loop.call_soon_threadsafe(self.castplayer.update_state)
318        self.castplayer.logger.debug(
319            "%s is removed from multizone: %s", self.castplayer.display_name, group_uuid
320        )
321        player_status = self.castplayer.cc.status
322        if player_status is None:
323            return
324
325        self.new_cast_status(player_status)
326
327    def multizone_new_cast_status(self, group_uuid: str, cast_status: CastStatus) -> None:
328        """Handle reception of a new CastStatus for a group."""
329        mass = self.castplayer.mass
330        if group_player := mass.players.get_player(group_uuid):
331            if TYPE_CHECKING:
332                assert isinstance(group_player, ChromecastPlayer)
333            if group_player.cc.media_controller.is_active:
334                self.castplayer.active_cast_group = group_uuid
335            elif group_uuid == self.castplayer.active_cast_group:
336                self.castplayer.active_cast_group = None
337
338        self.castplayer.logger.log(
339            VERBOSE_LOG_LEVEL,
340            "%s got new cast status for group: %s",
341            self.castplayer.display_name,
342            group_uuid,
343        )
344        player_status = self.castplayer.cc.status
345        if player_status is None:
346            return
347
348        self.new_cast_status(player_status)
349
350    def multizone_new_media_status(self, group_uuid: str, media_status: MediaStatus) -> None:
351        """Handle reception of a new MediaStatus for a group."""
352        if not self._valid:
353            return
354        self.castplayer.logger.log(
355            VERBOSE_LOG_LEVEL,
356            "%s got new media_status for group: %s",
357            self.castplayer.display_name,
358            group_uuid,
359        )
360        self.castplayer.on_new_media_status(media_status)
361
362    def load_media_failed(self, queue_item_id: int, error_code: int) -> None:
363        """Call when media failed to load."""
364        self.castplayer.logger.warning(
365            "Load media failed: %s - error code: %s", queue_item_id, error_code
366        )
367
368    def invalidate(self) -> None:
369        """
370        Invalidate this status listener.
371
372        All following callbacks won't be forwarded.
373        """
374        if self.castplayer.cast_info.is_audio_group:
375            self._mz_mgr.remove_multizone(self._uuid)
376        else:
377            self._mz_mgr.deregister_listener(self._uuid, cast("MultiZoneManagerListener", self))
378        self._valid = False
379