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