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