X-Git-Url: https://wannabe.guru.org/gitweb/?a=blobdiff_plain;f=smart_home%2Foutlets.py;h=60b98a63534e6897f6b448edaedebe9bb9c638b1;hb=bc80a8e1bb262fd7898ae1e86b3c629dfacc3872;hp=6cc8d57ec065beabc5652c1015fa75df3307cd97;hpb=b454ad295eb3024a238d32bf2aef1ebc3c496b44;p=python_utils.git diff --git a/smart_home/outlets.py b/smart_home/outlets.py index 6cc8d57..60b98a6 100644 --- a/smart_home/outlets.py +++ b/smart_home/outlets.py @@ -2,7 +2,6 @@ """Utilities for dealing with the smart outlets.""" -from abc import abstractmethod import asyncio import atexit import datetime @@ -12,11 +11,12 @@ import os import re import subprocess import sys +from abc import abstractmethod from typing import Any, Dict, List, Optional -import warnings from meross_iot.http_api import MerossHttpClient from meross_iot.manager import MerossManager +from overrides import overrides import argparse_utils import config @@ -24,8 +24,8 @@ import decorator_utils import logging_utils import scott_secrets import smart_home.device as dev -from google_assistant import ask_google, GoogleResponse -from decorator_utils import timeout, memoized +from decorator_utils import memoized, timeout +from google_assistant import GoogleResponse, ask_google logger = logging.getLogger(__name__) @@ -42,16 +42,13 @@ parser.add_argument( ) -@timeout( - 5.0, use_signals=False, error_message="Timed out waiting for tplink.py" -) +@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) - warnings.warn(msg) logging_utils.hlog(msg) return False else: @@ -59,17 +56,17 @@ def tplink_outlet_command(command: str) -> bool: if exit_value != 0: msg = f'{command} failed, exited {exit_value}' logger.warning(msg) - warnings.warn(msg) logging_utils.hlog(msg) return False - logger.debug(f'{command} succeeded.') + logger.debug('%s succeeded.', command) return True class BaseOutlet(dev.Device): + """An abstract base class for smart outlets.""" + def __init__(self, name: str, mac: str, keywords: str = "") -> None: super().__init__(name.strip(), mac.strip(), keywords) - self.info = None @abstractmethod def turn_on(self) -> bool: @@ -89,6 +86,8 @@ class BaseOutlet(dev.Device): class TPLinkOutlet(BaseOutlet): + """A TPLink smart outlet.""" + def __init__(self, name: str, mac: str, keywords: str = '') -> None: super().__init__(name, mac, keywords) self.info: Optional[Dict] = None @@ -108,27 +107,29 @@ class TPLinkOutlet(BaseOutlet): ) return cmd - def command(self, cmd: str, extra_args: str = None) -> bool: + def command(self, cmd: str, extra_args: str = None, **kwargs) -> bool: cmd = self.get_cmdline() + f"-c {cmd}" if extra_args is not None: cmd += f" {extra_args}" return tplink_outlet_command(cmd) + @overrides def turn_on(self) -> bool: return self.command('on') + @overrides def turn_off(self) -> bool: return self.command('off') + @overrides def is_on(self) -> bool: return self.get_on_duration_seconds() > 0 + @overrides def is_off(self) -> bool: return not self.is_on() - @timeout( - 10.0, use_signals=False, error_message="Timed out waiting for tplink.py" - ) + @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) @@ -153,18 +154,22 @@ class TPLinkOutlet(BaseOutlet): class TPLinkOutletWithChildren(TPLinkOutlet): + """A TPLink outlet where the top and bottom plus are individually + controllable.""" + 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 + assert self.keywords is not None assert "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"]) - # override + @overrides def get_cmdline(self, child: Optional[str] = None) -> str: cmd = ( f"{config.config['smart_outlets_tplink_location']} -m {self.mac} " @@ -174,22 +179,23 @@ class TPLinkOutletWithChildren(TPLinkOutlet): cmd += f"-x {child} " return cmd - # override - def command( - self, cmd: str, child: str = None, extra_args: str = None - ) -> bool: + @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}" if extra_args is not None: cmd += f" {extra_args}" - logger.debug(f'About to execute {cmd}') + logger.debug('About to execute: %s', cmd) return tplink_outlet_command(cmd) def get_children(self) -> List[str]: return self.children + @overrides def turn_on(self, child: str = None) -> bool: return self.command("on", child) + @overrides def turn_off(self, child: str = None) -> bool: return self.command("off", child) @@ -209,6 +215,8 @@ class TPLinkOutletWithChildren(TPLinkOutlet): class GoogleOutlet(BaseOutlet): + """A smart outlet controlled via Google Assistant.""" + def __init__(self, name: str, mac: str, keywords: str = "") -> None: super().__init__(name.strip(), mac.strip(), keywords) self.info = None @@ -221,22 +229,24 @@ class GoogleOutlet(BaseOutlet): def parse_google_response(response: GoogleResponse) -> bool: return response.success + @overrides def turn_on(self) -> bool: - return GoogleOutlet.parse_google_response( - ask_google('turn {self.goog_name()} on') - ) + return GoogleOutlet.parse_google_response(ask_google(f'turn {self.goog_name()} on')) + @overrides def turn_off(self) -> bool: - return GoogleOutlet.parse_google_response( - ask_google('turn {self.goog_name()} off') - ) + return GoogleOutlet.parse_google_response(ask_google(f'turn {self.goog_name()} off')) + @overrides def is_on(self) -> bool: r = ask_google(f'is {self.goog_name()} on?') if not r.success: return False - return 'is on' in r.audio_transcription + if r.audio_transcription is not None: + return 'is on' in r.audio_transcription + raise Exception('Can\'t talk to Google right now!?') + @overrides def is_off(self) -> bool: return not self.is_on() @@ -250,6 +260,7 @@ class MerossWrapper(object): this class. """ + def __init__(self): self.loop = asyncio.get_event_loop() self.email = os.environ.get('MEROSS_EMAIL') or scott_secrets.MEROSS_EMAIL @@ -283,10 +294,12 @@ class MerossWrapper(object): class MerossOutlet(BaseOutlet): + """A Meross smart outlet class.""" + def __init__(self, name: str, mac: str, keywords: str = '') -> None: super().__init__(name, mac, keywords) - self.meross_wrapper = None - self.device = None + self.meross_wrapper: Optional[MerossWrapper] = None + self.device: Optional[Any] = None def lazy_initialize_device(self): """If we make too many calls to Meross they will block us; only talk @@ -297,23 +310,28 @@ class MerossOutlet(BaseOutlet): if self.device is None: raise Exception(f'{self.name} is not a known Meross device?!') + @overrides def turn_on(self) -> bool: self.lazy_initialize_device() - self.meross_wrapper.loop.run_until_complete( - self.device.async_turn_on() - ) + assert self.meross_wrapper is not None + assert self.device is not None + self.meross_wrapper.loop.run_until_complete(self.device.async_turn_on()) return True + @overrides def turn_off(self) -> bool: self.lazy_initialize_device() - self.meross_wrapper.loop.run_until_complete( - self.device.async_turn_off() - ) + assert self.meross_wrapper is not None + assert self.device is not None + self.meross_wrapper.loop.run_until_complete(self.device.async_turn_off()) return True + @overrides def is_on(self) -> bool: self.lazy_initialize_device() + assert self.device is not None return self.device.is_on() + @overrides def is_off(self) -> bool: return not self.is_on()