Since this thing is on the innerwebs I suppose it should have a

[python_utils.git] / smart_home / outlets.py

diff --git a/smart_home/outlets.py b/smart_home/outlets.py
index 60b98a63534e6897f6b448edaedebe9bb9c638b1..fcc3c4eb8f7fa74d428d31c7ebe4814c5cb11cbd 100644 (file)
--- 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,26 +40,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 +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:
@@ -129,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()
@@ -169,12 +141,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 +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('About to execute: %s', cmd)
-        return tplink_outlet_command(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", {}):