music-assistant-server

1"""Chromecast Player Provider implementation."""
2
3from __future__ import annotations
4
5import asyncio
6import contextlib
7import logging
8import threading
9import time
10from typing import TYPE_CHECKING, cast
11from uuid import UUID
12
13import pychromecast
14from pychromecast.controllers.multizone import MultizoneManager
15from pychromecast.discovery import CastBrowser, SimpleCastListener
16
17from music_assistant.constants import (
18    CONF_ENABLED,
19    CONF_ENTRY_MANUAL_DISCOVERY_IPS,
20    CONF_LOG_LEVEL,
21    VERBOSE_LOG_LEVEL,
22)
23from music_assistant.helpers.json import SerializableType
24from music_assistant.models.player_provider import PlayerProvider
25
26from .constants import MULTICHANNEL_RECHECK_INTERVAL
27from .dashboard import ChromecastDashboards
28from .helpers import ChromecastInfo, without_ipv6_host_services
29from .player import ChromecastPlayer
30from .sendspin_bridge import SendspinBridgeManager
31
32if TYPE_CHECKING:
33    from music_assistant_models.config_entries import ConfigEntry, ProviderConfig
34    from music_assistant_models.enums import ProviderFeature
35    from music_assistant_models.provider import ProviderManifest
36    from pychromecast.models import CastInfo
37
38    from music_assistant.mass import MusicAssistant
39
40
41class ChromecastProvider(PlayerProvider):
42    """Player provider for Chromecast based players."""
43
44    mz_mgr: MultizoneManager | None = None
45    browser: CastBrowser | None = None
46    _discover_lock: threading.Lock
47
48    def __init__(
49        self,
50        mass: MusicAssistant,
51        manifest: ProviderManifest,
52        config: ProviderConfig,
53        supported_features: set[ProviderFeature],
54    ) -> None:
55        """Handle async initialization of the provider."""
56        super().__init__(mass, manifest, config, supported_features)
57        self._discover_lock = threading.Lock()
58        self._pending_discoveries: set[str] = set()
59        self.mz_mgr = MultizoneManager()
60        # Handle config option for manual IP's. Read a default: at construction the config
61        # carries only the server defaults + stored raw values (typed option entries are
62        # applied right after by the config controller).
63        manual_ip_config = cast(
64            "list[str]", config.get_value(CONF_ENTRY_MANUAL_DISCOVERY_IPS.key) or []
65        )
66        self.browser = CastBrowser(
67            SimpleCastListener(
68                add_callback=self._on_chromecast_discovered,
69                remove_callback=self._on_chromecast_removed,
70                update_callback=self._on_chromecast_discovered,
71            ),
72            self.mass.discovery.aiozc.zeroconf,
73            known_hosts=manual_ip_config,
74        )
75        self._discovery_running = False
76        self.bridge_manager = SendspinBridgeManager(self)
77        self.dashboards = ChromecastDashboards(self)
78        self._set_pychromecast_log_level()
79
80    async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
81        """Return Config entries to configure this provider."""
82        return (CONF_ENTRY_MANUAL_DISCOVERY_IPS,)
83
84    async def discover_players(self) -> None:
85        """Discover Cast players on the network."""
86        if self._discovery_running:
87            return
88        self._discovery_running = True
89        assert self.browser is not None  # for type checking
90        await self.mass.loop.run_in_executor(None, self.browser.start_discovery)
91
92    async def unload(self, is_removed: bool = False) -> None:
93        """Handle close/cleanup of the provider."""
94        await self.dashboards.unload()
95
96        # Stop all Sendspin bridges and remove listeners
97        await self.bridge_manager.close()
98
99        # Suppress pychromecast's noisy ERROR logs during disconnect
100        # (PyChromecastStopped race in socket thread is unavoidable)
101        logging.getLogger("pychromecast").setLevel(logging.CRITICAL)
102
103        # Stop discovery first to prevent new callbacks during disconnect
104        if self.browser:
105
106            def stop_discovery() -> None:
107                """Stop the chromecast discovery threads."""
108                assert self.browser is not None  # for type checking
109                if self.browser._zc_browser:
110                    with contextlib.suppress(RuntimeError):
111                        self.browser._zc_browser.cancel()
112
113                self.browser.host_browser.stop.set()
114                self.browser.host_browser.join()
115
116                self._discovery_running = False
117
118            await self.mass.loop.run_in_executor(None, stop_discovery)
119
120    async def update_config(self, config: ProviderConfig, changed_keys: set[str]) -> None:
121        """Handle logic when the config is updated."""
122        await super().update_config(config, changed_keys)
123        # a log level(-only) change does not reload the provider,
124        # so realign pychromecast's logger here
125        if f"values/{CONF_LOG_LEVEL}" in changed_keys:
126            self._set_pychromecast_log_level()
127
128    async def get_diagnostics(self) -> dict[str, SerializableType]:
129        """Return diagnostics info for this provider to include in diagnostics reports."""
130        cast_players = [player for player in self.players if isinstance(player, ChromecastPlayer)]
131        models: dict[str, int] = {}
132        for cast_player in cast_players:
133            model_name = cast_player.cast_info.model_name
134            models[model_name] = models.get(model_name, 0) + 1
135        return {
136            "discovery_running": self._discovery_running,
137            "pending_discoveries": len(self._pending_discoveries),
138            "cast_groups": sum(
139                cast_player.cast_info.is_audio_group for cast_player in cast_players
140            ),
141            "cast_speakers": sum(
142                not cast_player.cast_info.is_audio_group for cast_player in cast_players
143            ),
144            "models": models,
145        }
146
147    def _set_pychromecast_log_level(self) -> None:
148        """Align pychromecast's log level with the provider's log level."""
149        # pychromecast is very chatty at debug level (it logs every socket
150        # message of each cast connection), so only pass through its debug
151        # logging when verbose logging is enabled
152        if self.logger.isEnabledFor(VERBOSE_LOG_LEVEL):
153            logging.getLogger("pychromecast").setLevel(logging.DEBUG)
154        else:
155            logging.getLogger("pychromecast").setLevel(self.logger.level + 10)
156
157    ### Discovery callbacks
158
159    def _on_chromecast_discovered(self, uuid: UUID, _: str) -> None:
160        """
161        Handle Chromecast discovered callback.
162
163        NOTE: Called from pychromecast's discovery thread, NOT async friendly!
164        """
165        if self.mass.closing:
166            return
167
168        assert self.browser is not None  # for type checking
169
170        # Quick lookup and existing-player update under lock (fast path)
171        with self._discover_lock:
172            # filter here so the socket client and the eureka_info HTTP calls below
173            # only receive addresses pychromecast can connect to
174            disc_info: CastInfo = without_ipv6_host_services(self.browser.devices[uuid])
175
176            if disc_info.uuid is None:
177                self.logger.error("Discovered chromecast without uuid %s", disc_info)  # type: ignore[unreachable]
178                return
179            disc_uuid: UUID = disc_info.uuid
180
181            player_id = str(disc_uuid)
182
183            # (Re-)register as a dashboard endpoint; includes devices disabled as a
184            # player, since dashboard casting targets a display, not a MA player.
185            self.mass.loop.call_soon_threadsafe(self.dashboards.register, disc_uuid, disc_info)
186
187            # If player already registered, just update cast info (fast path)
188            castplayer = self.mass.players.get_player(player_id)
189            if castplayer:
190                assert isinstance(castplayer, ChromecastPlayer)  # for type checking
191                castplayer.cast_info.update(disc_info)
192                socket_client = castplayer.cc.socket_client
193                if socket_client.services != disc_info.services:
194                    socket_client.services.clear()
195                    socket_client.services.update(disc_info.services)
196                self.mass.loop.call_soon_threadsafe(castplayer.update_state)
197                # An unavailable player may be a passive multichannel endpoint that
198                # slipped past the discovery filter (e.g. incomplete multizone info
199                # while the stereo pair was rebooting). Re-evaluate and remove it
200                # if it is now positively identified as such.
201                if self._should_recheck_multichannel_child(castplayer, player_id):
202                    self._pending_discoveries.add(player_id)
203                    try:
204                        asyncio.run_coroutine_threadsafe(
205                            self._recheck_multichannel_child(player_id, disc_info),
206                            loop=self.mass.loop,
207                        )
208                    except RuntimeError:
209                        # event loop already closed (shutdown): release the marker
210                        self._pending_discoveries.discard(player_id)
211                return
212
213            # Prevent duplicate discovery while async setup is in progress
214            if player_id in self._pending_discoveries:
215                return
216            self._pending_discoveries.add(player_id)
217
218        # Blocking work outside the lock to avoid blocking the discovery thread
219        # for other devices while HTTP calls are in progress.
220        # On success, _create_and_register_player clears _pending_discoveries.
221        # On early return or exception, we clean it up here via finally.
222        scheduled = False
223        try:
224            if not self.mass.config.get_raw_player_config_value(player_id, CONF_ENABLED, True):
225                self.logger.debug("Ignoring disabled player: %s", player_id)
226                return
227
228            self.logger.debug("Discovered new chromecast %s", disc_info)
229
230            cast_info = ChromecastInfo.from_cast_info(disc_info)
231            cast_info.fill_out_missing_chromecast_info(self.mass.discovery.aiozc.zeroconf)
232            if cast_info.is_dynamic_group:
233                self.logger.debug("Discovered a dynamic cast group which will be ignored.")
234                return
235            if cast_info.is_multichannel_child:
236                self.logger.debug(
237                    "Discovered a passive (multichannel) endpoint which will be ignored."
238                )
239                return
240            # create new Chromecast instance
241            chromecast = pychromecast.get_chromecast_from_cast_info(
242                disc_info,
243                self.mass.discovery.aiozc.zeroconf,
244            )
245            # create and register the new ChromeCastPlayer
246            asyncio.run_coroutine_threadsafe(
247                self._create_and_register_player(player_id, cast_info, chromecast),
248                loop=self.mass.loop,
249            )
250            scheduled = True
251        finally:
252            if not scheduled:
253                self._pending_discoveries.discard(player_id)
254
255    async def _create_and_register_player(
256        self, player_id: str, cast_info: ChromecastInfo, chromecast: pychromecast.Chromecast
257    ) -> None:
258        """Create and register a new ChromecastPlayer."""
259        try:
260            castplayer = ChromecastPlayer(
261                self, player_id, cast_info=cast_info, chromecast=chromecast
262            )
263            await castplayer.async_setup()
264            await self.mass.players.register_or_update(castplayer)
265            # Set up Sendspin bridge
266            await self.bridge_manager.evaluate_bridge(castplayer)
267        finally:
268            self._pending_discoveries.discard(player_id)
269
270    def _on_chromecast_removed(
271        self,
272        uuid: UUID,
273        service: str,
274        cast_info: CastInfo,
275    ) -> None:
276        """Handle zeroconf discovery of a removed Chromecast."""
277        player_id = str(uuid)
278        self.logger.debug("Chromecast removed: %s - %s", cast_info.friendly_name, player_id)
279        # we ignore this for the player itself, as the Chromecast socket client handles that,
280        # but the dashboard registration has no such fallback and must be dropped explicitly
281        self.mass.loop.call_soon_threadsafe(self.dashboards.unregister, uuid)
282
283    def _should_recheck_multichannel_child(
284        self, castplayer: ChromecastPlayer, player_id: str
285    ) -> bool:
286        """Return whether an unavailable player should be re-evaluated as a multichannel child."""
287        if player_id in self._pending_discoveries:
288            return False
289        if castplayer.available or castplayer.cast_info.is_audio_group:
290            return False
291        return (
292            time.monotonic() - castplayer.last_multichannel_check
293        ) > MULTICHANNEL_RECHECK_INTERVAL
294
295    async def _recheck_multichannel_child(self, player_id: str, disc_info: CastInfo) -> None:
296        """Re-evaluate a player and remove it if it is a passive multichannel endpoint."""
297        try:
298            castplayer = self.mass.players.get_player(player_id)
299            if not isinstance(castplayer, ChromecastPlayer):
300                return
301            castplayer.last_multichannel_check = time.monotonic()
302            cast_info = ChromecastInfo.from_cast_info(disc_info)
303            await asyncio.to_thread(
304                cast_info.fill_out_missing_chromecast_info, self.mass.discovery.aiozc.zeroconf
305            )
306            if cast_info.is_multichannel_child and self.mass.players.get_player(player_id):
307                self.logger.info(
308                    "Removing %s as it is now identified as a passive (multichannel) endpoint",
309                    castplayer.cast_info.friendly_name,
310                )
311                await self.mass.players.unregister(player_id, permanent=True)
312        except Exception:
313            self.logger.debug("Multichannel re-check failed for %s", player_id, exc_info=True)
314        finally:
315            self._pending_discoveries.discard(player_id)
316