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