From: Scott Gasch <scott@gasch.org>
Date: Sat, 26 Feb 2022 19:15:31 +0000 (-0800)
Subject: More work on tplink utils to avoid having to call tplink.py.
X-Git-Url: https://wannabe.guru.org/gitweb/?a=commitdiff_plain;h=4863b00af3b2ddcfaa2e5f47fd6dd7e95c1eb6c5;p=python_utils.git

More work on tplink utils to avoid having to call tplink.py.
---

diff --git a/smart_home/lights.py b/smart_home/lights.py
index 096e0de..9a5a231 100644
--- a/smart_home/lights.py
+++ b/smart_home/lights.py
@@ -315,13 +315,15 @@ class TPLinkLight(BaseLight):
         raise NotImplementedError
 
     def get_info(self) -> Optional[Dict]:
-        cmd = self.get_cmdline() + "-c info"
-        self.info = tplink.tplink_get_info(cmd)
-        if self.info is not None:
-            self.info_ts = datetime.datetime.now()
-        else:
-            self.info_ts = None
-        return self.info
+        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
+        return None
 
     @overrides
     def status(self) -> str:
diff --git a/smart_home/outlets.py b/smart_home/outlets.py
index 77e9c2d..b9bfe22 100644
--- a/smart_home/outlets.py
+++ b/smart_home/outlets.py
@@ -106,13 +106,15 @@ class TPLinkOutlet(BaseOutlet):
         return not self.is_on()
 
     def get_info(self) -> Optional[Dict]:
-        cmd = self.get_cmdline() + "-c info"
-        self.info = tplink.tplink_get_info(cmd)
-        if self.info is not None:
-            self.info_ts = datetime.datetime.now()
-        else:
-            self.info_ts = None
-        return self.info
+        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
+        return None
 
     def get_on_duration_seconds(self) -> int:
         self.info = self.get_info()
diff --git a/smart_home/tplink_utils.py b/smart_home/tplink_utils.py
index 9b8eb6d..053ce33 100644
--- a/smart_home/tplink_utils.py
+++ b/smart_home/tplink_utils.py
@@ -24,10 +24,7 @@ limitations under the License.
 import json
 import logging
 import os
-import re
 import socket
-import subprocess
-import sys
 import time
 from struct import pack
 from typing import Dict, List, Optional, Tuple
@@ -76,21 +73,14 @@ def tplink_command(command: str) -> bool:
     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"]
+def tplink_get_info(ip: str, port: int = 9999) -> Optional[Dict]:
+    success, response = communicate_with_device(ip, port, commands['info'], quiet=True)
+    if success:
+        assert len(response) == 1
+        info = json.loads(response[0])["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
+    return None
 
 
 def encrypt(string: str) -> bytes: