X-Git-Url: https://wannabe.guru.org/gitweb/?a=blobdiff_plain;f=smart_home%2Flights.py;h=ee23fb08f4e60ef4664338cde7e53bdefd340544;hb=532df2c5b57c7517dfb3dddd8c1358fbadf8baf3;hp=80bfffa9db61b1639d5ff19aa9a12df387bdb90d;hpb=713a609bd19d491de03debf8a4a6ddf2540b13dc;p=python_utils.git diff --git a/smart_home/lights.py b/smart_home/lights.py index 80bfffa..ee23fb0 100644 --- a/smart_home/lights.py +++ b/smart_home/lights.py @@ -1,14 +1,12 @@ #!/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 @@ -19,9 +17,9 @@ import ansi 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__) @@ -39,27 +37,9 @@ args.add_argument( ) -@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(f'{command} succeeded.') - return True - - class BaseLight(dev.Device): + """A base class representing a smart light.""" + def __init__(self, name: str, mac: str, keywords: str = "") -> None: super().__init__(name.strip(), mac.strip(), keywords) @@ -111,8 +91,7 @@ class BaseLight(dev.Device): class GoogleLight(BaseLight): - def __init__(self, name: str, mac: str, keywords: str = "") -> None: - super().__init__(name, mac, keywords) + """A smart light controlled by talking to Google.""" def goog_name(self) -> str: name = self.get_name() @@ -187,6 +166,8 @@ class GoogleLight(BaseLight): class TuyaLight(BaseLight): + """A Tuya smart light.""" + ids_by_mac = { '68:C6:3A:DE:1A:94': '8844664268c63ade1a94', '68:C6:3A:DE:27:1A': '8844664268c63ade271a', @@ -251,14 +232,14 @@ class TuyaLight(BaseLight): @overrides def set_dimmer_level(self, level: int) -> bool: - logger.debug(f'Setting brightness to {level}') + logger.debug('Setting brightness to %d', level) self.bulb.set_brightness(level) return True @overrides def make_color(self, color: str) -> bool: rgb = BaseLight.parse_color_string(color) - logger.debug(f'Light color: {color} -> {rgb}') + logger.debug('Light color: %s -> %s', color, rgb) if rgb is not None: self.bulb.set_colour(rgb[0], rgb[1], rgb[2]) return True @@ -266,16 +247,19 @@ class TuyaLight(BaseLight): class TPLinkLight(BaseLight): + """A TPLink smart light.""" + 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 - if "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"]) + if self.keywords is not None: + if "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"]) @memoized def get_tplink_name(self) -> Optional[str]: @@ -300,15 +284,21 @@ class TPLinkLight(BaseLight): cmd = self.get_cmdline(child) + f"-c {cmd}" if extra_args is not None: cmd += f" {extra_args}" - logger.debug(f'About to execute {cmd}') - return tplink_light_command(cmd) + logger.debug('About to execute: %s', 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 @@ -326,30 +316,24 @@ class TPLinkLight(BaseLight): 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" - out = subprocess.getoutput(cmd) - logger.debug(f'RAW OUT> {out}') - out = re.sub("Sent:.*\n", "", out) - out = re.sub("Received: *", "", out) - try: - self.info = json.loads(out)["system"]["get_sysinfo"] - logger.debug(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: @@ -381,9 +365,9 @@ class TPLinkLight(BaseLight): 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):