smart_home/chromecasts.py

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