smart_home/tplink_utils.py

   1 #!/usr/bin/env python3
   2
   3 """Wrapper functions for calling tplink.py"""
   4
   5 import json
   6 import logging
   7 import os
   8 import re
   9 import subprocess
  10 import sys
  11 from typing import Dict, Optional
  12
  13 import logging_utils
  14 from decorator_utils import timeout
  15
  16 logger = logging.getLogger(__name__)
  17
  18
  19 @timeout(10.0, use_signals=False, error_message="Timed out waiting for tplink.py")
  20 def tplink_command(command: str) -> bool:
  21     result = os.system(command)
  22     signal = result & 0xFF
  23     if signal != 0:
  24         msg = f'{command} died with signal {signal}'
  25         logger.warning(msg)
  26         logging_utils.hlog(msg)
  27         return False
  28     else:
  29         exit_value = result >> 8
  30         if exit_value != 0:
  31             msg = f'{command} failed, exited {exit_value}'
  32             logger.warning(msg)
  33             logging_utils.hlog(msg)
  34             return False
  35     logger.debug('%s succeeded.', command)
  36     return True
  37
  38
  39 @timeout(10.0, use_signals=False, error_message="Timed out waiting for tplink.py")
  40 def tplink_get_info(cmd: str) -> Optional[Dict]:
  41     logger.debug('Getting tplink device status via "%s"', cmd)
  42     try:
  43         out = subprocess.getoutput(cmd)
  44         logger.debug('RAW OUT> %s', out)
  45         out = re.sub("Sent:.*\n", "", out)
  46         out = re.sub("Received: *", "", out)
  47         info = json.loads(out)["system"]["get_sysinfo"]
  48         logger.debug("%s", json.dumps(info, indent=4, sort_keys=True))
  49         return info
  50     except Exception as e:
  51         logger.exception(e)
  52         print(out, file=sys.stderr)
  53         return None