X-Git-Url: https://wannabe.guru.org/gitweb/?a=blobdiff_plain;f=smart_home%2Foutlets.py;h=fcc3c4eb8f7fa74d428d31c7ebe4814c5cb11cbd;hb=532df2c5b57c7517dfb3dddd8c1358fbadf8baf3;hp=500ea05372dd200444ba3268427b1d6f814850c9;hpb=713a609bd19d491de03debf8a4a6ddf2540b13dc;p=python_utils.git diff --git a/smart_home/outlets.py b/smart_home/outlets.py index 500ea05..fcc3c4e 100644 --- a/smart_home/outlets.py +++ b/smart_home/outlets.py @@ -1,16 +1,14 @@ #!/usr/bin/env python3 +# © Copyright 2021-2022, Scott Gasch + """Utilities for dealing with the smart outlets.""" 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 +19,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,27 +40,9 @@ 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(f'{command} succeeded.') - return True - - class BaseOutlet(dev.Device): + """An abstract base class for smart outlets.""" + def __init__(self, name: str, mac: str, keywords: str = "") -> None: super().__init__(name.strip(), mac.strip(), keywords) @@ -84,6 +64,8 @@ class BaseOutlet(dev.Device): class TPLinkOutlet(BaseOutlet): + """A TPLink smart outlet.""" + def __init__(self, name: str, mac: str, keywords: str = '') -> None: super().__init__(name, mac, keywords) self.info: Optional[Dict] = None @@ -107,7 +89,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_wrapper(cmd) @overrides def turn_on(self) -> bool: @@ -125,22 +107,16 @@ 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_ts = datetime.datetime.now() + ip = self.get_ip() + if ip is not None: + self.info = tplink.tplink_get_info(ip) + if self.info is not None: + self.info_ts = datetime.datetime.now() + else: + self.info_ts = None return self.info - except Exception as e: - logger.exception(e) - print(out, file=sys.stderr) - self.info = None - self.info_ts = None - return None + return None def get_on_duration_seconds(self) -> int: self.info = self.get_info() @@ -150,23 +126,23 @@ class TPLinkOutlet(BaseOutlet): class TPLinkOutletWithChildren(TPLinkOutlet): + """A TPLink outlet where the top and bottom plus are individually + controllable.""" + def __init__(self, name: str, mac: str, keywords: str = "") -> None: super().__init__(name, mac, keywords) self.children: List[str] = [] self.info: Optional[Dict] = None self.info_ts: Optional[datetime.datetime] = None + assert self.keywords is not None assert "children" in self.keywords self.info = self.get_info() if self.info is not None: 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 @@ -174,30 +150,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(f'About to execute {cmd}') - return tplink_outlet_command(cmd) + logger.debug('About to execute: %s', cmd) + return tplink.tplink_command_wrapper(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", {}): @@ -207,6 +187,8 @@ class TPLinkOutletWithChildren(TPLinkOutlet): class GoogleOutlet(BaseOutlet): + """A smart outlet controlled via Google Assistant.""" + def __init__(self, name: str, mac: str, keywords: str = "") -> None: super().__init__(name.strip(), mac.strip(), keywords) self.info = None @@ -284,6 +266,8 @@ class MerossWrapper(object): class MerossOutlet(BaseOutlet): + """A Meross smart outlet class.""" + def __init__(self, name: str, mac: str, keywords: str = '') -> None: super().__init__(name, mac, keywords) self.meross_wrapper: Optional[MerossWrapper] = None