From 1220ffcf56e968de31b62f6d0c5af250fed9028e Mon Sep 17 00:00:00 2001 From: Scott Gasch Date: Thu, 17 Feb 2022 11:52:14 -0800 Subject: [PATCH] Cleanup. --- smart_home/device.py | 8 ++++ smart_home/lights.py | 72 +++++++++++------------------------ smart_home/outlets.py | 78 +++++++++++--------------------------- smart_home/tplink_utils.py | 53 ++++++++++++++++++++++++++ 4 files changed, 107 insertions(+), 104 deletions(-) create mode 100644 smart_home/tplink_utils.py diff --git a/smart_home/device.py b/smart_home/device.py index 06187d7..82f1fa6 100644 --- a/smart_home/device.py +++ b/smart_home/device.py @@ -1,5 +1,9 @@ #!/usr/bin/env python3 +"""Most basic definition of a smart device: it must have a name and a +MAC address and may have some optional keywords. All devices have +these whether they are lights, outlets, thermostats, etc...""" + import re from typing import List, Optional @@ -7,6 +11,10 @@ import arper class Device(object): + """Most basic definition of a smart device: it must have a name and a + MAC address and may have some optional keywords. All devices have + these whether they are lights, outlets, thermostats, etc...""" + def __init__( self, name: str, diff --git a/smart_home/lights.py b/smart_home/lights.py index 908fe8d..fd9a091 100644 --- a/smart_home/lights.py +++ b/smart_home/lights.py @@ -3,12 +3,8 @@ """Utilities for dealing with the smart lights.""" import datetime -import json import logging -import os import re -import subprocess -import sys from abc import abstractmethod from typing import Any, Dict, List, Optional, Tuple @@ -19,9 +15,9 @@ import ansi import argparse_utils import arper import config -import logging_utils import smart_home.device as dev -from decorator_utils import memoized, timeout +import smart_home.tplink_utils as tplink +from decorator_utils import memoized from google_assistant import GoogleResponse, ask_google logger = logging.getLogger(__name__) @@ -39,26 +35,6 @@ args.add_argument( ) -@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 - if signal != 0: - msg = f'{command} died with signal {signal}' - logger.warning(msg) - logging_utils.hlog(msg) - return False - else: - exit_value = result >> 8 - if exit_value != 0: - msg = f'{command} failed, exited {exit_value}' - logger.warning(msg) - logging_utils.hlog(msg) - return False - logger.debug('%s succeeded.', command) - return True - - class BaseLight(dev.Device): """A base class representing a smart light.""" @@ -307,14 +283,20 @@ class TPLinkLight(BaseLight): if extra_args is not None: cmd += f" {extra_args}" logger.debug('About to execute: %s', cmd) - return tplink_light_command(cmd) + return tplink.tplink_command(cmd) @overrides - def turn_on(self, child: str = None) -> bool: - return self.command("on", child) + def turn_on(self) -> bool: + return self.command("on", None) @overrides - def turn_off(self, child: str = None) -> bool: + def turn_off(self) -> bool: + return self.command("off", None) + + def turn_on_child(self, child: str = None) -> bool: + return self.command("on", child) + + def turn_off_child(self, child: str = None) -> bool: return self.command("off", child) @overrides @@ -332,31 +314,22 @@ 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") def get_info(self) -> Optional[Dict]: cmd = self.get_cmdline() + "-c info" - logger.debug('Getting status of %s via "%s"...', self.mac, cmd) - out = subprocess.getoutput(cmd) - logger.debug('RAW OUT> %s', out) - out = re.sub("Sent:.*\n", "", out) - out = re.sub("Received: *", "", out) - try: - self.info = json.loads(out)["system"]["get_sysinfo"] - logger.debug("%s", json.dumps(self.info, indent=4, sort_keys=True)) + self.info = tplink.tplink_get_info(cmd) + if self.info is not None: self.info_ts = datetime.datetime.now() - return self.info - except Exception as e: - logger.exception(e) - print(out, file=sys.stderr) - self.info = None + else: self.info_ts = None - return None + return self.info @overrides def status(self) -> str: ret = '' - for k, v in self.get_info().items(): - ret += f'{k} = {v}\n' + info = self.get_info() + if info is not None: + for k, v in info: + ret += f'{k} = {v}\n' return ret def get_on_duration_seconds(self, child: str = None) -> int: @@ -388,9 +361,10 @@ class TPLinkLight(BaseLight): return False cmd = ( self.get_cmdline() - + f'-j \'{{"smartlife.iot.dimmer":{{"set_brightness":{{"brightness":{level} }} }} }}\'' + + '-j \'{{"smartlife.iot.dimmer":{{"set_brightness":{{"brightness":{%d} }} }} }}\'' + % level ) - return tplink_light_command(cmd) + return tplink.tplink_command(cmd) # class GoogleLightGroup(GoogleLight): diff --git a/smart_home/outlets.py b/smart_home/outlets.py index 60b98a6..77e9c2d 100644 --- a/smart_home/outlets.py +++ b/smart_home/outlets.py @@ -5,12 +5,8 @@ import asyncio import atexit import datetime -import json import logging import os -import re -import subprocess -import sys from abc import abstractmethod from typing import Any, Dict, List, Optional @@ -21,10 +17,10 @@ from overrides import overrides import argparse_utils import config import decorator_utils -import logging_utils import scott_secrets import smart_home.device as dev -from decorator_utils import memoized, timeout +import smart_home.tplink_utils as tplink +from decorator_utils import memoized from google_assistant import GoogleResponse, ask_google logger = logging.getLogger(__name__) @@ -42,26 +38,6 @@ parser.add_argument( ) -@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 - if signal != 0: - msg = f'{command} died with signal {signal}' - logger.warning(msg) - logging_utils.hlog(msg) - return False - else: - exit_value = result >> 8 - if exit_value != 0: - msg = f'{command} failed, exited {exit_value}' - logger.warning(msg) - logging_utils.hlog(msg) - return False - logger.debug('%s succeeded.', command) - return True - - class BaseOutlet(dev.Device): """An abstract base class for smart outlets.""" @@ -111,7 +87,7 @@ class TPLinkOutlet(BaseOutlet): cmd = self.get_cmdline() + f"-c {cmd}" if extra_args is not None: cmd += f" {extra_args}" - return tplink_outlet_command(cmd) + return tplink.tplink_command(cmd) @overrides def turn_on(self) -> bool: @@ -129,22 +105,14 @@ class TPLinkOutlet(BaseOutlet): def is_off(self) -> bool: return not self.is_on() - @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) - out = re.sub("Sent:.*\n", "", out) - out = re.sub("Received: *", "", out) - try: - self.info = json.loads(out)["system"]["get_sysinfo"] + self.info = tplink.tplink_get_info(cmd) + if self.info is not None: self.info_ts = datetime.datetime.now() - return self.info - except Exception as e: - logger.exception(e) - print(out, file=sys.stderr) - self.info = None + else: self.info_ts = None - return None + return self.info def get_on_duration_seconds(self) -> int: self.info = self.get_info() @@ -169,12 +137,8 @@ class TPLinkOutletWithChildren(TPLinkOutlet): for child in self.info["children"]: self.children.append(child["id"]) - @overrides - def get_cmdline(self, child: Optional[str] = None) -> str: - cmd = ( - f"{config.config['smart_outlets_tplink_location']} -m {self.mac} " - f"--no_logging_console " - ) + def get_cmdline_with_child(self, child: Optional[str] = None) -> str: + cmd = super().get_cmdline() if child is not None: cmd += f"-x {child} " return cmd @@ -182,30 +146,34 @@ class TPLinkOutletWithChildren(TPLinkOutlet): @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}" + cmd = self.get_cmdline_with_child(child) + f"-c {cmd}" if extra_args is not None: cmd += f" {extra_args}" logger.debug('About to execute: %s', cmd) - return tplink_outlet_command(cmd) + return tplink.tplink_command(cmd) def get_children(self) -> List[str]: return self.children @overrides - def turn_on(self, child: str = None) -> bool: - return self.command("on", child) + def turn_on(self) -> bool: + return self.command("on", None) @overrides - def turn_off(self, child: str = None) -> bool: + def turn_off(self) -> bool: + return self.command("off", None) + + def turn_on_child(self, child: str = None) -> bool: + return self.command("on", child) + + def turn_off_child(self, child: str = None) -> bool: return self.command("off", child) - def get_on_duration_seconds(self, child: str = None) -> int: - self.info = self.get_info() + def get_child_on_duration_seconds(self, child: str = None) -> int: if child is None: - if self.info is None: - return 0 - return int(self.info.get("on_time", "0")) + return super().get_on_duration_seconds() else: + self.info = self.get_info() if self.info is None: return 0 for chi in self.info.get("children", {}): diff --git a/smart_home/tplink_utils.py b/smart_home/tplink_utils.py new file mode 100644 index 0000000..91cdd75 --- /dev/null +++ b/smart_home/tplink_utils.py @@ -0,0 +1,53 @@ +#!/usr/bin/env python3 + +"""Wrapper functions for calling tplink.py""" + +import json +import logging +import os +import re +import subprocess +import sys +from typing import Dict, Optional + +import logging_utils +from decorator_utils import timeout + +logger = logging.getLogger(__name__) + + +@timeout(10.0, use_signals=False, error_message="Timed out waiting for tplink.py") +def tplink_command(command: str) -> bool: + result = os.system(command) + signal = result & 0xFF + if signal != 0: + msg = f'{command} died with signal {signal}' + logger.warning(msg) + logging_utils.hlog(msg) + return False + else: + exit_value = result >> 8 + if exit_value != 0: + msg = f'{command} failed, exited {exit_value}' + logger.warning(msg) + logging_utils.hlog(msg) + return False + logger.debug('%s succeeded.', command) + return True + + +@timeout(10.0, use_signals=False, error_message="Timed out waiting for tplink.py") +def tplink_get_info(cmd: str) -> Optional[Dict]: + logger.debug('Getting tplink device status via "%s"', cmd) + try: + out = subprocess.getoutput(cmd) + logger.debug('RAW OUT> %s', out) + out = re.sub("Sent:.*\n", "", out) + out = re.sub("Received: *", "", out) + info = json.loads(out)["system"]["get_sysinfo"] + logger.debug("%s", json.dumps(info, indent=4, sort_keys=True)) + return info + except Exception as e: + logger.exception(e) + print(out, file=sys.stderr) + return None -- 2.45.2