#!/usr/bin/env python3 """Utilities for dealing with the webcams.""" import atexit import datetime import logging import threading from typing import Any, List import pychromecast import smart_home.device as dev from decorator_utils import memoized logger = logging.getLogger(__name__) class BaseChromecast(dev.Device): """A base class to represent a Google Chromecase device.""" ccasts: List[Any] = [] refresh_ts = None browser = None lock = threading.Lock() def __init__(self, name: str, mac: str, keywords: str = "") -> None: super().__init__(name.strip(), mac.strip(), keywords) ip = self.get_ip() now = datetime.datetime.now() with BaseChromecast.lock: if ( BaseChromecast.refresh_ts is None or (now - BaseChromecast.refresh_ts).total_seconds() > 60 ): logger.debug('Refreshing the shared chromecast info list') if BaseChromecast.browser is not None: BaseChromecast.browser.stop_discovery() ( BaseChromecast.ccasts, BaseChromecast.browser, ) = pychromecast.get_chromecasts(timeout=15.0) assert BaseChromecast.browser is not None atexit.register(BaseChromecast.browser.stop_discovery) BaseChromecast.refresh_ts = now self.cast = None for cc in BaseChromecast.ccasts: if cc.cast_info.host == ip and cc.cast_info.cast_type != 'group': logger.debug('Found chromecast at %s: %s', ip, cc) self.cast = cc self.cast.wait(timeout=1.0) if self.cast is None: raise Exception(f'Can\'t find ccast device at {ip}, is that really a ccast device?') def is_idle(self): return self.cast.is_idle @memoized def get_uuid(self): return self.cast.uuid @memoized def get_friendly_name(self): return self.cast.name def get_uri(self): return self.cast.url @memoized def get_model_name(self): return self.cast.model_name @memoized def get_cast_type(self): return self.cast.cast_type @memoized def app_id(self): return self.cast.app_id def get_app_display_name(self): return self.cast.app_display_name def get_media_controller(self): return self.cast.media_controller def status(self): if self.is_idle(): return 'idle' app = self.get_app_display_name() mc = self.get_media_controller() status = mc.status return f'{app} / {status.title}' def start_app(self, app_id, force_launch=False): """Start an app on the Chromecast.""" self.cast.start_app(app_id, force_launch) def quit_app(self): """Tells the Chromecast to quit current app_id.""" self.cast.quit_app() def volume_up(self, delta=0.1): """Increment volume by 0.1 (or delta) unless it is already maxed. Returns the new volume. """ return self.cast.volume_up(delta) def volume_down(self, delta=0.1): """Decrement the volume by 0.1 (or delta) unless it is already 0. Returns the new volume. """ return self.cast.volume_down(delta) def __repr__(self): return ( f"Chromecast({self.cast.socket_client.host!r}, port={self.cast.socket_client.port!r}, " f"device={self.cast.cast_info.friendly_name!r})" )