"""Helpers to deal with Cast devices.""" from __future__ import annotations from typing import Optional import attr from pychromecast import dial from pychromecast.const import CAST_TYPE_GROUP from pychromecast.models import CastInfo @attr.s(slots=True, frozen=True) class ChromecastInfo: """Class to hold all data about a chromecast for creating connections. This also has the same attributes as the mDNS fields by zeroconf. """ cast_info: CastInfo = attr.ib() is_dynamic_group = attr.ib(type=Optional[bool], default=None) @property def friendly_name(self) -> str: """Return the UUID.""" return self.cast_info.friendly_name @property def is_audio_group(self) -> bool: """Return if the cast is an audio group.""" return self.cast_info.cast_type == CAST_TYPE_GROUP @property def uuid(self) -> bool: """Return the UUID.""" return self.cast_info.uuid def fill_out_missing_chromecast_info(self) -> ChromecastInfo: """Return a new ChromecastInfo object with missing attributes filled in. Uses blocking HTTP / HTTPS. """ if not self.is_audio_group or self.is_dynamic_group is not None: # We have all information, no need to check HTTP API. return self # Fill out missing group information via HTTP API. is_dynamic_group = False http_group_status = None http_group_status = dial.get_multizone_status( None, services=self.cast_info.services, zconf=ChromeCastZeroconf.get_zeroconf(), ) if http_group_status is not None: is_dynamic_group = any( g.uuid == self.cast_info.uuid for g in http_group_status.dynamic_groups ) return ChromecastInfo( cast_info=self.cast_info, is_dynamic_group=is_dynamic_group, ) class ChromeCastZeroconf: """Class to hold a zeroconf instance.""" __zconf = None @classmethod def set_zeroconf(cls, zconf): """Set zeroconf.""" cls.__zconf = zconf @classmethod def get_zeroconf(cls): """Get zeroconf.""" return cls.__zconf class CastStatusListener: """Helper class to handle pychromecast status callbacks. Necessary because a CastDevice entity can create a new socket client and therefore callbacks from multiple chromecast connections can potentially arrive. This class allows invalidating past chromecast objects. """ def __init__(self, cast_device, chromecast, mz_mgr, mz_only=False): """Initialize the status listener.""" self._cast_device = cast_device self._uuid = chromecast.uuid self._valid = True self._mz_mgr = mz_mgr if cast_device._cast_info.is_audio_group: self._mz_mgr.add_multizone(chromecast) if mz_only: return chromecast.register_status_listener(self) chromecast.socket_client.media_controller.register_status_listener(self) chromecast.register_connection_listener(self) if not cast_device._cast_info.is_audio_group: self._mz_mgr.register_listener(chromecast.uuid, self) def new_cast_status(self, cast_status): """Handle reception of a new CastStatus.""" if self._valid: self._cast_device.new_cast_status(cast_status) def new_media_status(self, media_status): """Handle reception of a new MediaStatus.""" if self._valid: self._cast_device.new_media_status(media_status) def new_connection_status(self, connection_status): """Handle reception of a new ConnectionStatus.""" if self._valid: self._cast_device.new_connection_status(connection_status) @staticmethod def added_to_multizone(group_uuid): """Handle the cast added to a group.""" def removed_from_multizone(self, group_uuid): """Handle the cast removed from a group.""" if self._valid: self._cast_device.multizone_new_media_status(group_uuid, None) def multizone_new_cast_status(self, group_uuid, cast_status): """Handle reception of a new CastStatus for a group.""" def multizone_new_media_status(self, group_uuid, media_status): """Handle reception of a new MediaStatus for a group.""" if self._valid: self._cast_device.multizone_new_media_status(group_uuid, media_status) def invalidate(self): """Invalidate this status listener. All following callbacks won't be forwarded. """ # pylint: disable=protected-access if self._cast_device._cast_info.is_audio_group: self._mz_mgr.remove_multizone(self._uuid) else: self._mz_mgr.deregister_listener(self._uuid, self) self._valid = False