1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
|
#!/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):
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
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(f'Found chromecast at {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})"
)
|
