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