/
/
/
1"""Helpers to deal with Cast devices."""
2
3from __future__ import annotations
4
5import urllib.error
6from dataclasses import asdict, dataclass
7from typing import TYPE_CHECKING
8from uuid import UUID
9
10from pychromecast import dial
11from pychromecast.const import CAST_TYPE_GROUP
12
13from music_assistant.constants import VERBOSE_LOG_LEVEL
14
15if TYPE_CHECKING:
16 from pychromecast.controllers.media import MediaStatus
17 from pychromecast.controllers.multizone import MultizoneManager
18 from pychromecast.controllers.receiver import CastStatus
19 from pychromecast.models import CastInfo
20 from pychromecast.socket_client import ConnectionStatus
21 from zeroconf import ServiceInfo, Zeroconf
22
23 from .player import ChromecastPlayer
24
25DEFAULT_PORT = 8009
26
27
28@dataclass
29class ChromecastInfo:
30 """Class to hold all data about a chromecast for creating connections.
31
32 This also has the same attributes as the mDNS fields by zeroconf.
33 """
34
35 services: set
36 uuid: UUID
37 model_name: str
38 friendly_name: str
39 host: str
40 port: int
41 cast_type: str | None = None
42 manufacturer: str | None = None
43 is_dynamic_group: bool | None = None
44 is_multichannel_group: bool = False # group created for e.g. stereo pair
45 is_multichannel_child: bool = False # speaker that is part of multichannel setup
46
47 @property
48 def is_audio_group(self) -> bool:
49 """Return if the cast is an audio group."""
50 return self.cast_type == CAST_TYPE_GROUP
51
52 @classmethod
53 def from_cast_info(cls, cast_info: CastInfo) -> ChromecastInfo:
54 """Instantiate ChromecastInfo from CastInfo."""
55 return cls(**asdict(cast_info))
56
57 def update(self, cast_info: CastInfo) -> None:
58 """Update ChromecastInfo from CastInfo."""
59 for key, value in asdict(cast_info).items():
60 if not value:
61 continue
62 setattr(self, key, value)
63
64 def fill_out_missing_chromecast_info(self, zconf: Zeroconf) -> None:
65 """
66 Return a new ChromecastInfo object with missing attributes filled in.
67
68 Uses blocking HTTP / HTTPS.
69 """
70 if self.cast_type is None or self.manufacturer is None:
71 # Manufacturer and cast type is not available in mDNS data,
72 # get it over HTTP
73 cast_info = dial.get_cast_type(
74 self,
75 zconf=zconf,
76 )
77 self.cast_type = cast_info.cast_type
78 self.manufacturer = cast_info.manufacturer
79
80 # Fill out missing group information via HTTP API.
81 dynamic_groups, multichannel_groups = get_multizone_info(self.services, zconf)
82 self.is_dynamic_group = self.uuid in dynamic_groups
83 if self.uuid in multichannel_groups:
84 self.is_multichannel_group = True
85 elif (
86 multichannel_groups
87 # Prevent a multichannel group being marked as a multichannel child
88 # if not in UUID list
89 and self.cast_type != "group"
90 and self.model_name != "Google Cast Group"
91 ):
92 self.is_multichannel_child = True
93
94
95def get_multizone_info(services: list[ServiceInfo], zconf: Zeroconf, timeout=30):
96 """Get multizone info from eureka endpoint."""
97 dynamic_groups: set[str] = set()
98 multichannel_groups: set[str] = set()
99 try:
100 _, status = dial._get_status(
101 services,
102 zconf,
103 "/setup/eureka_info?params=multizone",
104 True,
105 timeout,
106 None,
107 )
108 if "multizone" in status and "dynamic_groups" in status["multizone"]:
109 for group in status["multizone"]["dynamic_groups"]:
110 if udn := group.get("uuid"):
111 uuid = UUID(udn.replace("-", ""))
112 dynamic_groups.add(uuid)
113
114 if "multizone" in status and "groups" in status["multizone"]:
115 for group in status["multizone"]["groups"]:
116 if "multichannel_group" not in group:
117 continue
118 if group["multichannel_group"] and (udn := group.get("uuid")):
119 uuid = UUID(udn.replace("-", ""))
120 multichannel_groups.add(uuid)
121 except (urllib.error.HTTPError, urllib.error.URLError, OSError, KeyError, ValueError):
122 pass
123 return (dynamic_groups, multichannel_groups)
124
125
126class CastStatusListener:
127 """
128 Helper class to handle pychromecast status callbacks.
129
130 Necessary because a CastDevice entity can create a new socket client
131 and therefore callbacks from multiple chromecast connections can
132 potentially arrive. This class allows invalidating past chromecast objects.
133 """
134
135 def __init__(
136 self,
137 castplayer: ChromecastPlayer,
138 mz_mgr: MultizoneManager,
139 mz_only=False,
140 ) -> None:
141 """Initialize the status listener."""
142 self.castplayer = castplayer
143 self._uuid = castplayer.cc.uuid
144 self._valid = True
145 self._mz_mgr = mz_mgr
146 if self.castplayer.cast_info.is_audio_group:
147 self._mz_mgr.add_multizone(castplayer.cc)
148 if mz_only:
149 return
150 castplayer.cc.register_status_listener(self)
151 castplayer.cc.socket_client.media_controller.register_status_listener(self)
152 castplayer.cc.register_connection_listener(self)
153 if not self.castplayer.cast_info.is_audio_group:
154 self._mz_mgr.register_listener(castplayer.cc.uuid, self)
155
156 def new_cast_status(self, status: CastStatus) -> None:
157 """Handle updated CastStatus."""
158 if not self._valid:
159 return
160 self.castplayer.on_new_cast_status(status)
161
162 def new_media_status(self, status: MediaStatus) -> None:
163 """Handle updated MediaStatus."""
164 if not self._valid:
165 return
166 self.castplayer.on_new_media_status(status)
167
168 def new_connection_status(self, status: ConnectionStatus) -> None:
169 """Handle updated ConnectionStatus."""
170 if not self._valid:
171 return
172 self.castplayer.on_new_connection_status(status)
173
174 def added_to_multizone(self, group_uuid) -> None:
175 """Handle the cast added to a group."""
176 self.castplayer.logger.debug(
177 "%s is added to multizone: %s", self.castplayer.display_name, group_uuid
178 )
179 self.new_cast_status(self.castplayer.cc.status)
180
181 def removed_from_multizone(self, group_uuid) -> None:
182 """Handle the cast removed from a group."""
183 if not self._valid:
184 return
185 if group_uuid == self.castplayer.active_source:
186 mass = self.castplayer.mass
187 mass.loop.call_soon_threadsafe(self.castplayer.update_state)
188 self.castplayer.logger.debug(
189 "%s is removed from multizone: %s", self.castplayer.display_name, group_uuid
190 )
191 self.new_cast_status(self.castplayer.cc.status)
192
193 def multizone_new_cast_status(self, group_uuid, cast_status) -> None:
194 """Handle reception of a new CastStatus for a group."""
195 mass = self.castplayer.mass
196 if group_player := mass.players.get(group_uuid):
197 if TYPE_CHECKING:
198 assert isinstance(group_player, ChromecastPlayer)
199 if group_player.cc.media_controller.is_active:
200 self.castplayer.active_cast_group = group_uuid
201 elif group_uuid == self.castplayer.active_cast_group:
202 self.castplayer.active_cast_group = None
203
204 self.castplayer.logger.log(
205 VERBOSE_LOG_LEVEL,
206 "%s got new cast status for group: %s",
207 self.castplayer.display_name,
208 group_uuid,
209 )
210 self.new_cast_status(self.castplayer.cc.status)
211
212 def multizone_new_media_status(self, group_uuid, media_status) -> None:
213 """Handle reception of a new MediaStatus for a group."""
214 if not self._valid:
215 return
216 self.castplayer.logger.log(
217 VERBOSE_LOG_LEVEL,
218 "%s got new media_status for group: %s",
219 self.castplayer.display_name,
220 group_uuid,
221 )
222 self.castplayer.on_new_media_status(media_status)
223
224 def load_media_failed(self, queue_item_id, error_code) -> None:
225 """Call when media failed to load."""
226 self.castplayer.logger.warning(
227 "Load media failed: %s - error code: %s", queue_item_id, error_code
228 )
229
230 def invalidate(self) -> None:
231 """
232 Invalidate this status listener.
233
234 All following callbacks won't be forwarded.
235 """
236 if self.castplayer.cast_info.is_audio_group:
237 self._mz_mgr.remove_multizone(self._uuid)
238 else:
239 self._mz_mgr.deregister_listener(self._uuid, self)
240 self._valid = False
241