#!/usr/bin/env python3
+# © Copyright 2021-2022, Scott Gasch
+
"""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
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__)
)
-@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."""
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_wrapper(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
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_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
@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:
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_wrapper(cmd)
# class GoogleLightGroup(GoogleLight):