projects
/
python_utils.git
/ blobdiff
commit
grep
author
committer
pickaxe
?
search:
re
summary
|
shortlog
|
log
|
commit
|
commitdiff
|
tree
raw
|
inline
| side by side
Let's be explicit with asserts; there was a bug in histogram
[python_utils.git]
/
smart_home
/
chromecasts.py
diff --git
a/smart_home/chromecasts.py
b/smart_home/chromecasts.py
index a5db86f3f003395d61ee2436688e44eaaa55d963..bec8461a4d764ce97a812f1af08f4a5087fd8869 100644
(file)
--- a/
smart_home/chromecasts.py
+++ b/
smart_home/chromecasts.py
@@
-6,17
+6,18
@@
import atexit
import datetime
import logging
import threading
import datetime
import logging
import threading
+from typing import Any, List
import pychromecast
import pychromecast
-from decorator_utils import memoized
import smart_home.device as dev
import smart_home.device as dev
+from decorator_utils import memoized
logger = logging.getLogger(__name__)
class BaseChromecast(dev.Device):
logger = logging.getLogger(__name__)
class BaseChromecast(dev.Device):
- ccasts = []
+ ccasts
: List[Any]
= []
refresh_ts = None
browser = None
lock = threading.Lock()
refresh_ts = None
browser = None
lock = threading.Lock()
@@
-25,31
+26,32
@@
class BaseChromecast(dev.Device):
super().__init__(name.strip(), mac.strip(), keywords)
ip = self.get_ip()
now = datetime.datetime.now()
super().__init__(name.strip(), mac.strip(), keywords)
ip = self.get_ip()
now = datetime.datetime.now()
- with BaseChromecast.lock
as l
:
+ with BaseChromecast.lock:
if (
if (
-
BaseChromecast.refresh_ts is None
-
or (now - BaseChromecast.refresh_ts).total_seconds() > 60
+ 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()
):
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
- )
+ (
+ BaseChromecast.ccasts,
+ BaseChromecast.browser,
+ ) = pychromecast.get_chromecasts(timeout=15.0)
+ assert BaseChromecast.browser is not None
atexit.register(BaseChromecast.browser.stop_discovery)
BaseChromecast.refresh_ts = now
self.cast = None
for cc in BaseChromecast.ccasts:
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'
- ):
+ 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:
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?')
+ 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
def is_idle(self):
return self.cast.is_idle
@@
-116,4
+118,3
@@
class BaseChromecast(dev.Device):
f"Chromecast({self.cast.socket_client.host!r}, port={self.cast.socket_client.port!r}, "
f"device={self.cast.cast_info.friendly_name!r})"
)
f"Chromecast({self.cast.socket_client.host!r}, port={self.cast.socket_client.port!r}, "
f"device={self.cast.cast_info.friendly_name!r})"
)
-