diff options
Diffstat (limited to 'smart_home')
| -rw-r--r-- | smart_home/chromecasts.py | 29 | ||||
| -rw-r--r-- | smart_home/device.py | 10 | ||||
| -rw-r--r-- | smart_home/lights.py | 39 | ||||
| -rw-r--r-- | smart_home/outlets.py | 71 | ||||
| -rw-r--r-- | smart_home/registry.py | 18 |
5 files changed, 89 insertions, 78 deletions
diff --git a/smart_home/chromecasts.py b/smart_home/chromecasts.py index a5db86f..bd2a80c 100644 --- a/smart_home/chromecasts.py +++ b/smart_home/chromecasts.py @@ -6,17 +6,18 @@ 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 = [] + ccasts: List[Any] = [] 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() - 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 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: - 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 @@ -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})" ) - diff --git a/smart_home/device.py b/smart_home/device.py index 9675b7c..02717a3 100644 --- a/smart_home/device.py +++ b/smart_home/device.py @@ -8,17 +8,17 @@ import arper class Device(object): def __init__( - self, - name: str, - mac: str, - keywords: Optional[List[str]], + self, + name: str, + mac: str, + keywords: Optional[List[str]], ): self.name = name self.mac = mac self.keywords = keywords self.arper = arper.Arper() if keywords is not None: - self.kws = keywords.split() + self.kws = keywords else: self.kws = [] diff --git a/smart_home/lights.py b/smart_home/lights.py index 64f2105..240e7da 100644 --- a/smart_home/lights.py +++ b/smart_home/lights.py @@ -2,7 +2,6 @@ """Utilities for dealing with the smart lights.""" -from abc import abstractmethod import datetime import json import logging @@ -10,10 +9,11 @@ import os import re import subprocess import sys +from abc import abstractmethod from typing import Any, Dict, List, Optional, Tuple -from overrides import overrides import tinytuya as tt +from overrides import overrides import ansi import argparse_utils @@ -21,8 +21,8 @@ import arper import config import logging_utils import smart_home.device as dev -from google_assistant import ask_google, GoogleResponse -from decorator_utils import timeout, memoized +from decorator_utils import memoized, timeout +from google_assistant import GoogleResponse, ask_google logger = logging.getLogger(__name__) @@ -39,9 +39,7 @@ args.add_argument( ) -@timeout( - 5.0, use_signals=False, error_message="Timed out waiting for tplink.py" -) +@timeout(5.0, use_signals=False, error_message="Timed out waiting for tplink.py") def tplink_light_command(command: str) -> bool: result = os.system(command) signal = result & 0xFF @@ -69,9 +67,9 @@ class BaseLight(dev.Device): def parse_color_string(color: str) -> Optional[Tuple[int, int, int]]: m = re.match( 'r#?([0-9a-fA-F][0-9a-fA-F])([0-9a-fA-F][0-9a-fA-F])([0-9a-fA-F][0-9a-fA-F])', - color + color, ) - if m is not None and len(m.group) == 3: + if m is not None and len(m.groups()) == 3: red = int(m.group(0), 16) green = int(m.group(1), 16) blue = int(m.group(2), 16) @@ -147,7 +145,9 @@ class GoogleLight(BaseLight): r = ask_google(f"is {self.goog_name()} on?") if not r.success: return False - return 'is on' in r.audio_transcription + if r.audio_transcription is not None: + return 'is on' in r.audio_transcription + raise Exception("Can't reach Google?!") @overrides def is_off(self) -> bool: @@ -163,11 +163,12 @@ class GoogleLight(BaseLight): # the bookcase one is set to 40% bright txt = r.audio_transcription - m = re.search(r"(\d+)% bright", txt) - if m is not None: - return int(m.group(1)) - if "is off" in txt: - return 0 + if txt is not None: + m = re.search(r"(\d+)% bright", txt) + if m is not None: + return int(m.group(1)) + if "is off" in txt: + return 0 return None @overrides @@ -301,9 +302,7 @@ class TPLinkLight(BaseLight): def get_children(self) -> List[str]: return self.children - def command( - self, cmd: str, child: str = None, extra_args: str = None - ) -> bool: + def command(self, cmd: str, child: str = None, extra_args: str = None) -> bool: cmd = self.get_cmdline(child) + f"-c {cmd}" if extra_args is not None: cmd += f" {extra_args}" @@ -333,9 +332,7 @@ class TPLinkLight(BaseLight): def make_color(self, color: str) -> bool: raise NotImplementedError - @timeout( - 10.0, use_signals=False, error_message="Timed out waiting for tplink.py" - ) + @timeout(10.0, use_signals=False, error_message="Timed out waiting for tplink.py") def get_info(self) -> Optional[Dict]: cmd = self.get_cmdline() + "-c info" out = subprocess.getoutput(cmd) diff --git a/smart_home/outlets.py b/smart_home/outlets.py index c079cfd..d4a4886 100644 --- a/smart_home/outlets.py +++ b/smart_home/outlets.py @@ -2,7 +2,6 @@ """Utilities for dealing with the smart outlets.""" -from abc import abstractmethod import asyncio import atexit import datetime @@ -12,10 +11,12 @@ import os import re import subprocess import sys +from abc import abstractmethod from typing import Any, Dict, List, Optional from meross_iot.http_api import MerossHttpClient from meross_iot.manager import MerossManager +from overrides import overrides import argparse_utils import config @@ -23,8 +24,8 @@ import decorator_utils import logging_utils import scott_secrets import smart_home.device as dev -from google_assistant import ask_google, GoogleResponse -from decorator_utils import timeout, memoized +from decorator_utils import memoized, timeout +from google_assistant import GoogleResponse, ask_google logger = logging.getLogger(__name__) @@ -41,9 +42,7 @@ parser.add_argument( ) -@timeout( - 5.0, use_signals=False, error_message="Timed out waiting for tplink.py" -) +@timeout(5.0, use_signals=False, error_message="Timed out waiting for tplink.py") def tplink_outlet_command(command: str) -> bool: result = os.system(command) signal = result & 0xFF @@ -66,7 +65,6 @@ def tplink_outlet_command(command: str) -> bool: class BaseOutlet(dev.Device): def __init__(self, name: str, mac: str, keywords: str = "") -> None: super().__init__(name.strip(), mac.strip(), keywords) - self.info = None @abstractmethod def turn_on(self) -> bool: @@ -105,27 +103,29 @@ class TPLinkOutlet(BaseOutlet): ) return cmd - def command(self, cmd: str, extra_args: str = None) -> bool: + def command(self, cmd: str, extra_args: str = None, **kwargs) -> bool: cmd = self.get_cmdline() + f"-c {cmd}" if extra_args is not None: cmd += f" {extra_args}" return tplink_outlet_command(cmd) + @overrides def turn_on(self) -> bool: return self.command('on') + @overrides def turn_off(self) -> bool: return self.command('off') + @overrides def is_on(self) -> bool: return self.get_on_duration_seconds() > 0 + @overrides def is_off(self) -> bool: return not self.is_on() - @timeout( - 10.0, use_signals=False, error_message="Timed out waiting for tplink.py" - ) + @timeout(10.0, use_signals=False, error_message="Timed out waiting for tplink.py") def get_info(self) -> Optional[Dict]: cmd = self.get_cmdline() + "-c info" out = subprocess.getoutput(cmd) @@ -161,7 +161,7 @@ class TPLinkOutletWithChildren(TPLinkOutlet): for child in self.info["children"]: self.children.append(child["id"]) - # override + @overrides def get_cmdline(self, child: Optional[str] = None) -> str: cmd = ( f"{config.config['smart_outlets_tplink_location']} -m {self.mac} " @@ -171,10 +171,9 @@ class TPLinkOutletWithChildren(TPLinkOutlet): cmd += f"-x {child} " return cmd - # override - def command( - self, cmd: str, child: str = None, extra_args: str = None - ) -> bool: + @overrides + def command(self, cmd: str, extra_args: str = None, **kwargs) -> bool: + child: Optional[str] = kwargs.get('child', None) cmd = self.get_cmdline(child) + f"-c {cmd}" if extra_args is not None: cmd += f" {extra_args}" @@ -184,9 +183,11 @@ class TPLinkOutletWithChildren(TPLinkOutlet): def get_children(self) -> List[str]: return self.children + @overrides def turn_on(self, child: str = None) -> bool: return self.command("on", child) + @overrides def turn_off(self, child: str = None) -> bool: return self.command("off", child) @@ -218,22 +219,28 @@ class GoogleOutlet(BaseOutlet): def parse_google_response(response: GoogleResponse) -> bool: return response.success + @overrides def turn_on(self) -> bool: return GoogleOutlet.parse_google_response( - ask_google('turn {self.goog_name()} on') + ask_google(f'turn {self.goog_name()} on') ) + @overrides def turn_off(self) -> bool: return GoogleOutlet.parse_google_response( - ask_google('turn {self.goog_name()} off') + ask_google(f'turn {self.goog_name()} off') ) + @overrides def is_on(self) -> bool: r = ask_google(f'is {self.goog_name()} on?') if not r.success: return False - return 'is on' in r.audio_transcription + if r.audio_transcription is not None: + return 'is on' in r.audio_transcription + raise Exception('Can\'t talk to Google right now!?') + @overrides def is_off(self) -> bool: return not self.is_on() @@ -247,10 +254,13 @@ class MerossWrapper(object): this class. """ + def __init__(self): self.loop = asyncio.get_event_loop() self.email = os.environ.get('MEROSS_EMAIL') or scott_secrets.MEROSS_EMAIL - self.password = os.environ.get('MEROSS_PASSWORD') or scott_secrets.MEROSS_PASSWORD + self.password = ( + os.environ.get('MEROSS_PASSWORD') or scott_secrets.MEROSS_PASSWORD + ) self.devices = self.loop.run_until_complete(self.find_meross_devices()) atexit.register(self.loop.close) @@ -282,8 +292,8 @@ class MerossWrapper(object): class MerossOutlet(BaseOutlet): def __init__(self, name: str, mac: str, keywords: str = '') -> None: super().__init__(name, mac, keywords) - self.meross_wrapper = None - self.device = None + self.meross_wrapper: Optional[MerossWrapper] = None + self.device: Optional[Any] = None def lazy_initialize_device(self): """If we make too many calls to Meross they will block us; only talk @@ -294,23 +304,28 @@ class MerossOutlet(BaseOutlet): if self.device is None: raise Exception(f'{self.name} is not a known Meross device?!') + @overrides def turn_on(self) -> bool: self.lazy_initialize_device() - self.meross_wrapper.loop.run_until_complete( - self.device.async_turn_on() - ) + assert self.meross_wrapper + assert self.device + self.meross_wrapper.loop.run_until_complete(self.device.async_turn_on()) return True + @overrides def turn_off(self) -> bool: self.lazy_initialize_device() - self.meross_wrapper.loop.run_until_complete( - self.device.async_turn_off() - ) + assert self.meross_wrapper + assert self.device + self.meross_wrapper.loop.run_until_complete(self.device.async_turn_off()) return True + @overrides def is_on(self) -> bool: self.lazy_initialize_device() + assert self.device return self.device.is_on() + @overrides def is_off(self) -> bool: return not self.is_on() diff --git a/smart_home/registry.py b/smart_home/registry.py index 7349081..16e18ba 100644 --- a/smart_home/registry.py +++ b/smart_home/registry.py @@ -8,15 +8,15 @@ import argparse_utils import config import file_utils import logical_search -import smart_home.device as device import smart_home.cameras as cameras import smart_home.chromecasts as chromecasts +import smart_home.device as device import smart_home.lights as lights import smart_home.outlets as outlets args = config.add_commandline_args( f"Smart Home Registry ({__file__})", - "Args related to the smart home configuration registry." + "Args related to the smart home configuration registry.", ) args.add_argument( '--smart_home_registry_file_location', @@ -32,9 +32,9 @@ logger = logging.getLogger(__file__) class SmartHomeRegistry(object): def __init__( - self, - registry_file: Optional[str] = None, - filters: List[str] = ['smart'], + self, + registry_file: Optional[str] = None, + filters: List[str] = ['smart'], ) -> None: self._macs_by_name = {} self._keywords_by_name = {} @@ -44,13 +44,11 @@ class SmartHomeRegistry(object): # Read the disk config file... if registry_file is None: - registry_file = config.config[ - 'smart_home_registry_file_location' - ] + registry_file = config.config['smart_home_registry_file_location'] assert file_utils.does_file_exist(registry_file) logger.debug(f'Reading {registry_file}') - with open(registry_file, "r") as f: - contents = f.readlines() + with open(registry_file, "r") as rf: + contents = rf.readlines() # Parse the contents... for line in contents: |