smart_home/outlets.py

   1 #!/usr/bin/env python3
   2
   3 """Utilities for dealing with the smart outlets."""
   4
   5 import asyncio
   6 import atexit
   7 import datetime
   8 import logging
   9 import os
  10 from abc import abstractmethod
  11 from typing import Any, Dict, List, Optional
  12
  13 from meross_iot.http_api import MerossHttpClient
  14 from meross_iot.manager import MerossManager
  15 from overrides import overrides
  16
  17 import argparse_utils
  18 import config
  19 import decorator_utils
  20 import scott_secrets
  21 import smart_home.device as dev
  22 import smart_home.tplink_utils as tplink
  23 from decorator_utils import memoized
  24 from google_assistant import GoogleResponse, ask_google
  25
  26 logger = logging.getLogger(__name__)
  27
  28 parser = config.add_commandline_args(
  29     f"Outlet Utils ({__file__})",
  30     "Args related to smart outlets.",
  31 )
  32 parser.add_argument(
  33     '--smart_outlets_tplink_location',
  34     default='/home/scott/bin/tplink.py',
  35     metavar='FILENAME',
  36     help='The location of the tplink.py helper',
  37     type=argparse_utils.valid_filename,
  38 )
  39
  40
  41 class BaseOutlet(dev.Device):
  42     """An abstract base class for smart outlets."""
  43
  44     def __init__(self, name: str, mac: str, keywords: str = "") -> None:
  45         super().__init__(name.strip(), mac.strip(), keywords)
  46
  47     @abstractmethod
  48     def turn_on(self) -> bool:
  49         pass
  50
  51     @abstractmethod
  52     def turn_off(self) -> bool:
  53         pass
  54
  55     @abstractmethod
  56     def is_on(self) -> bool:
  57         pass
  58
  59     @abstractmethod
  60     def is_off(self) -> bool:
  61         pass
  62
  63
  64 class TPLinkOutlet(BaseOutlet):
  65     """A TPLink smart outlet."""
  66
  67     def __init__(self, name: str, mac: str, keywords: str = '') -> None:
  68         super().__init__(name, mac, keywords)
  69         self.info: Optional[Dict] = None
  70         self.info_ts: Optional[datetime.datetime] = None
  71
  72     @memoized
  73     def get_tplink_name(self) -> Optional[str]:
  74         self.info = self.get_info()
  75         if self.info is not None:
  76             return self.info["alias"]
  77         return None
  78
  79     def get_cmdline(self) -> str:
  80         cmd = (
  81             f"{config.config['smart_outlets_tplink_location']} -m {self.mac} "
  82             f"--no_logging_console "
  83         )
  84         return cmd
  85
  86     def command(self, cmd: str, extra_args: str = None, **kwargs) -> bool:
  87         cmd = self.get_cmdline() + f"-c {cmd}"
  88         if extra_args is not None:
  89             cmd += f" {extra_args}"
  90         return tplink.tplink_command(cmd)
  91
  92     @overrides
  93     def turn_on(self) -> bool:
  94         return self.command('on')
  95
  96     @overrides
  97     def turn_off(self) -> bool:
  98         return self.command('off')
  99
 100     @overrides
 101     def is_on(self) -> bool:
 102         return self.get_on_duration_seconds() > 0
 103
 104     @overrides
 105     def is_off(self) -> bool:
 106         return not self.is_on()
 107
 108     def get_info(self) -> Optional[Dict]:
 109         cmd = self.get_cmdline() + "-c info"
 110         self.info = tplink.tplink_get_info(cmd)
 111         if self.info is not None:
 112             self.info_ts = datetime.datetime.now()
 113         else:
 114             self.info_ts = None
 115         return self.info
 116
 117     def get_on_duration_seconds(self) -> int:
 118         self.info = self.get_info()
 119         if self.info is None:
 120             return 0
 121         return int(self.info.get("on_time", "0"))
 122
 123
 124 class TPLinkOutletWithChildren(TPLinkOutlet):
 125     """A TPLink outlet where the top and bottom plus are individually
 126     controllable."""
 127
 128     def __init__(self, name: str, mac: str, keywords: str = "") -> None:
 129         super().__init__(name, mac, keywords)
 130         self.children: List[str] = []
 131         self.info: Optional[Dict] = None
 132         self.info_ts: Optional[datetime.datetime] = None
 133         assert self.keywords is not None
 134         assert "children" in self.keywords
 135         self.info = self.get_info()
 136         if self.info is not None:
 137             for child in self.info["children"]:
 138                 self.children.append(child["id"])
 139
 140     def get_cmdline_with_child(self, child: Optional[str] = None) -> str:
 141         cmd = super().get_cmdline()
 142         if child is not None:
 143             cmd += f"-x {child} "
 144         return cmd
 145
 146     @overrides
 147     def command(self, cmd: str, extra_args: str = None, **kwargs) -> bool:
 148         child: Optional[str] = kwargs.get('child', None)
 149         cmd = self.get_cmdline_with_child(child) + f"-c {cmd}"
 150         if extra_args is not None:
 151             cmd += f" {extra_args}"
 152         logger.debug('About to execute: %s', cmd)
 153         return tplink.tplink_command(cmd)
 154
 155     def get_children(self) -> List[str]:
 156         return self.children
 157
 158     @overrides
 159     def turn_on(self) -> bool:
 160         return self.command("on", None)
 161
 162     @overrides
 163     def turn_off(self) -> bool:
 164         return self.command("off", None)
 165
 166     def turn_on_child(self, child: str = None) -> bool:
 167         return self.command("on", child)
 168
 169     def turn_off_child(self, child: str = None) -> bool:
 170         return self.command("off", child)
 171
 172     def get_child_on_duration_seconds(self, child: str = None) -> int:
 173         if child is None:
 174             return super().get_on_duration_seconds()
 175         else:
 176             self.info = self.get_info()
 177             if self.info is None:
 178                 return 0
 179             for chi in self.info.get("children", {}):
 180                 if chi["id"] == child:
 181                     return int(chi.get("on_time", "0"))
 182         return 0
 183
 184
 185 class GoogleOutlet(BaseOutlet):
 186     """A smart outlet controlled via Google Assistant."""
 187
 188     def __init__(self, name: str, mac: str, keywords: str = "") -> None:
 189         super().__init__(name.strip(), mac.strip(), keywords)
 190         self.info = None
 191
 192     def goog_name(self) -> str:
 193         name = self.get_name()
 194         return name.replace('_', ' ')
 195
 196     @staticmethod
 197     def parse_google_response(response: GoogleResponse) -> bool:
 198         return response.success
 199
 200     @overrides
 201     def turn_on(self) -> bool:
 202         return GoogleOutlet.parse_google_response(ask_google(f'turn {self.goog_name()} on'))
 203
 204     @overrides
 205     def turn_off(self) -> bool:
 206         return GoogleOutlet.parse_google_response(ask_google(f'turn {self.goog_name()} off'))
 207
 208     @overrides
 209     def is_on(self) -> bool:
 210         r = ask_google(f'is {self.goog_name()} on?')
 211         if not r.success:
 212             return False
 213         if r.audio_transcription is not None:
 214             return 'is on' in r.audio_transcription
 215         raise Exception('Can\'t talk to Google right now!?')
 216
 217     @overrides
 218     def is_off(self) -> bool:
 219         return not self.is_on()
 220
 221
 222 @decorator_utils.singleton
 223 class MerossWrapper(object):
 224     """Global singleton helper class for MerossOutlets.  Note that
 225     instantiating this class causes HTTP traffic with an external
 226     Meross server.  Meross blocks customers who hit their servers too
 227     aggressively so MerossOutlet is lazy about creating instances of
 228     this class.
 229
 230     """
 231
 232     def __init__(self):
 233         self.loop = asyncio.get_event_loop()
 234         self.email = os.environ.get('MEROSS_EMAIL') or scott_secrets.MEROSS_EMAIL
 235         self.password = os.environ.get('MEROSS_PASSWORD') or scott_secrets.MEROSS_PASSWORD
 236         self.devices = self.loop.run_until_complete(self.find_meross_devices())
 237         atexit.register(self.loop.close)
 238
 239     async def find_meross_devices(self) -> List[Any]:
 240         http_api_client = await MerossHttpClient.async_from_user_password(
 241             email=self.email, password=self.password
 242         )
 243
 244         # Setup and start the device manager
 245         manager = MerossManager(http_client=http_api_client)
 246         await manager.async_init()
 247
 248         # Discover devices
 249         await manager.async_device_discovery()
 250         devices = manager.find_devices()
 251         for device in devices:
 252             await device.async_update()
 253         return devices
 254
 255     def get_meross_device_by_name(self, name: str) -> Optional[Any]:
 256         name = name.lower()
 257         name = name.replace('_', ' ')
 258         for device in self.devices:
 259             if device.name.lower() == name:
 260                 return device
 261         return None
 262
 263
 264 class MerossOutlet(BaseOutlet):
 265     """A Meross smart outlet class."""
 266
 267     def __init__(self, name: str, mac: str, keywords: str = '') -> None:
 268         super().__init__(name, mac, keywords)
 269         self.meross_wrapper: Optional[MerossWrapper] = None
 270         self.device: Optional[Any] = None
 271
 272     def lazy_initialize_device(self):
 273         """If we make too many calls to Meross they will block us; only talk
 274         to them when someone actually wants to control a device."""
 275         if self.meross_wrapper is None:
 276             self.meross_wrapper = MerossWrapper()
 277             self.device = self.meross_wrapper.get_meross_device_by_name(self.name)
 278             if self.device is None:
 279                 raise Exception(f'{self.name} is not a known Meross device?!')
 280
 281     @overrides
 282     def turn_on(self) -> bool:
 283         self.lazy_initialize_device()
 284         assert self.meross_wrapper is not None
 285         assert self.device is not None
 286         self.meross_wrapper.loop.run_until_complete(self.device.async_turn_on())
 287         return True
 288
 289     @overrides
 290     def turn_off(self) -> bool:
 291         self.lazy_initialize_device()
 292         assert self.meross_wrapper is not None
 293         assert self.device is not None
 294         self.meross_wrapper.loop.run_until_complete(self.device.async_turn_off())
 295         return True
 296
 297     @overrides
 298     def is_on(self) -> bool:
 299         self.lazy_initialize_device()
 300         assert self.device is not None
 301         return self.device.is_on()
 302
 303     @overrides
 304     def is_off(self) -> bool:
 305         return not self.is_on()