smart_home/chromecasts.py

   1 #!/usr/bin/env python3
   2
   3 """Utilities for dealing with the webcams."""
   4
   5 import atexit
   6 import datetime
   7 import logging
   8 import threading
   9 from typing import Any, List
  10
  11 import pychromecast
  12
  13 import smart_home.device as dev
  14 from decorator_utils import memoized
  15
  16 logger = logging.getLogger(__name__)
  17
  18
  19 class BaseChromecast(dev.Device):
  20     """A base class to represent a Google Chromecase device."""
  21
  22     ccasts: List[Any] = []
  23     refresh_ts = None
  24     browser = None
  25     lock = threading.Lock()
  26
  27     def __init__(self, name: str, mac: str, keywords: str = "") -> None:
  28         super().__init__(name.strip(), mac.strip(), keywords)
  29         ip = self.get_ip()
  30         now = datetime.datetime.now()
  31         with BaseChromecast.lock:
  32             if (
  33                 BaseChromecast.refresh_ts is None
  34                 or (now - BaseChromecast.refresh_ts).total_seconds() > 60
  35             ):
  36                 logger.debug('Refreshing the shared chromecast info list')
  37                 if BaseChromecast.browser is not None:
  38                     BaseChromecast.browser.stop_discovery()
  39                 (
  40                     BaseChromecast.ccasts,
  41                     BaseChromecast.browser,
  42                 ) = pychromecast.get_chromecasts(timeout=15.0)
  43                 assert BaseChromecast.browser is not None
  44                 atexit.register(BaseChromecast.browser.stop_discovery)
  45                 BaseChromecast.refresh_ts = now
  46
  47         self.cast = None
  48         for cc in BaseChromecast.ccasts:
  49             if cc.cast_info.host == ip and cc.cast_info.cast_type != 'group':
  50                 logger.debug('Found chromecast at %s: %s', ip, cc)
  51                 self.cast = cc
  52                 self.cast.wait(timeout=1.0)
  53         if self.cast is None:
  54             raise Exception(f'Can\'t find ccast device at {ip}, is that really a ccast device?')
  55
  56     def is_idle(self):
  57         return self.cast.is_idle
  58
  59     @memoized
  60     def get_uuid(self):
  61         return self.cast.uuid
  62
  63     @memoized
  64     def get_friendly_name(self):
  65         return self.cast.name
  66
  67     def get_uri(self):
  68         return self.cast.url
  69
  70     @memoized
  71     def get_model_name(self):
  72         return self.cast.model_name
  73
  74     @memoized
  75     def get_cast_type(self):
  76         return self.cast.cast_type
  77
  78     @memoized
  79     def app_id(self):
  80         return self.cast.app_id
  81
  82     def get_app_display_name(self):
  83         return self.cast.app_display_name
  84
  85     def get_media_controller(self):
  86         return self.cast.media_controller
  87
  88     def status(self):
  89         if self.is_idle():
  90             return 'idle'
  91         app = self.get_app_display_name()
  92         mc = self.get_media_controller()
  93         status = mc.status
  94         return f'{app} / {status.title}'
  95
  96     def start_app(self, app_id, force_launch=False):
  97         """Start an app on the Chromecast."""
  98         self.cast.start_app(app_id, force_launch)
  99
 100     def quit_app(self):
 101         """Tells the Chromecast to quit current app_id."""
 102         self.cast.quit_app()
 103
 104     def volume_up(self, delta=0.1):
 105         """Increment volume by 0.1 (or delta) unless it is already maxed.
 106         Returns the new volume.
 107         """
 108         return self.cast.volume_up(delta)
 109
 110     def volume_down(self, delta=0.1):
 111         """Decrement the volume by 0.1 (or delta) unless it is already 0.
 112         Returns the new volume.
 113         """
 114         return self.cast.volume_down(delta)
 115
 116     def __repr__(self):
 117         return (
 118             f"Chromecast({self.cast.socket_client.host!r}, port={self.cast.socket_client.port!r}, "
 119             f"device={self.cast.cast_info.friendly_name!r})"
 120         )