X-Git-Url: https://wannabe.guru.org/gitweb/?a=blobdiff_plain;f=smart_home%2Fchromecasts.py;h=408ccf0e12ff3b1702ef7cd4b63d42c357fac7db;hb=532df2c5b57c7517dfb3dddd8c1358fbadf8baf3;hp=a5db86f3f003395d61ee2436688e44eaaa55d963;hpb=4e73c61749f94e49d022be886014a02e6c3fe79c;p=python_utils.git diff --git a/smart_home/chromecasts.py b/smart_home/chromecasts.py index a5db86f..408ccf0 100644 --- a/smart_home/chromecasts.py +++ b/smart_home/chromecasts.py @@ -1,22 +1,27 @@ #!/usr/bin/env python3 +# © Copyright 2021-2022, Scott Gasch + """Utilities for dealing with the webcams.""" import atexit import datetime import logging import threading +from typing import Any, List import pychromecast -from decorator_utils import memoized import smart_home.device as dev +from decorator_utils import memoized logger = logging.getLogger(__name__) class BaseChromecast(dev.Device): - ccasts = [] + """A base class to represent a Google Chromecase device.""" + + ccasts: List[Any] = [] refresh_ts = None browser = None lock = threading.Lock() @@ -25,27 +30,26 @@ class BaseChromecast(dev.Device): super().__init__(name.strip(), mac.strip(), keywords) ip = self.get_ip() now = datetime.datetime.now() - with BaseChromecast.lock as l: + with BaseChromecast.lock: 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() - 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: - if ( - cc.cast_info.host == ip - and cc.cast_info.cast_type != 'group' - ): - logger.debug(f'Found chromecast at {ip}: {cc}') + if cc.cast_info.host == ip and cc.cast_info.cast_type != 'group': + logger.debug('Found chromecast at %s: %s', ip, cc) self.cast = cc self.cast.wait(timeout=1.0) if self.cast is None: @@ -116,4 +120,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})" ) -