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_wrapper(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         ip = self.get_ip()
 110         if ip is not None:
 111             self.info = tplink.tplink_get_info(ip)
 112             if self.info is not None:
 113                 self.info_ts = datetime.datetime.now()
 114             else:
 115                 self.info_ts = None
 116             return self.info
 117         return None
 118
 119     def get_on_duration_seconds(self) -> int:
 120         self.info = self.get_info()
 121         if self.info is None:
 122             return 0
 123         return int(self.info.get("on_time", "0"))
 124
 125
 126 class TPLinkOutletWithChildren(TPLinkOutlet):
 127     """A TPLink outlet where the top and bottom plus are individually
 128     controllable."""
 129
 130     def __init__(self, name: str, mac: str, keywords: str = "") -> None:
 131         super().__init__(name, mac, keywords)
 132         self.children: List[str] = []
 133         self.info: Optional[Dict] = None
 134         self.info_ts: Optional[datetime.datetime] = None
 135         assert self.keywords is not None
 136         assert "children" in self.keywords
 137         self.info = self.get_info()
 138         if self.info is not None:
 139             for child in self.info["children"]:
 140                 self.children.append(child["id"])
 141
 142     def get_cmdline_with_child(self, child: Optional[str] = None) -> str:
 143         cmd = super().get_cmdline()
 144         if child is not None:
 145             cmd += f"-x {child} "
 146         return cmd
 147
 148     @overrides
 149     def command(self, cmd: str, extra_args: str = None, **kwargs) -> bool:
 150         child: Optional[str] = kwargs.get('child', None)
 151         cmd = self.get_cmdline_with_child(child) + f"-c {cmd}"
 152         if extra_args is not None:
 153             cmd += f" {extra_args}"
 154         logger.debug('About to execute: %s', cmd)
 155         return tplink.tplink_command_wrapper(cmd)
 156
 157     def get_children(self) -> List[str]:
 158         return self.children
 159
 160     @overrides
 161     def turn_on(self) -> bool:
 162         return self.command("on", None)
 163
 164     @overrides
 165     def turn_off(self) -> bool:
 166         return self.command("off", None)
 167
 168     def turn_on_child(self, child: str = None) -> bool:
 169         return self.command("on", child)
 170
 171     def turn_off_child(self, child: str = None) -> bool:
 172         return self.command("off", child)
 173
 174     def get_child_on_duration_seconds(self, child: str = None) -> int:
 175         if child is None:
 176             return super().get_on_duration_seconds()
 177         else:
 178             self.info = self.get_info()
 179             if self.info is None:
 180                 return 0
 181             for chi in self.info.get("children", {}):
 182                 if chi["id"] == child:
 183                     return int(chi.get("on_time", "0"))
 184         return 0
 185
 186
 187 class GoogleOutlet(BaseOutlet):
 188     """A smart outlet controlled via Google Assistant."""
 189
 190     def __init__(self, name: str, mac: str, keywords: str = "") -> None:
 191         super().__init__(name.strip(), mac.strip(), keywords)
 192         self.info = None
 193
 194     def goog_name(self) -> str:
 195         name = self.get_name()
 196         return name.replace('_', ' ')
 197
 198     @staticmethod
 199     def parse_google_response(response: GoogleResponse) -> bool:
 200         return response.success
 201
 202     @overrides
 203     def turn_on(self) -> bool:
 204         return GoogleOutlet.parse_google_response(ask_google(f'turn {self.goog_name()} on'))
 205
 206     @overrides
 207     def turn_off(self) -> bool:
 208         return GoogleOutlet.parse_google_response(ask_google(f'turn {self.goog_name()} off'))
 209
 210     @overrides
 211     def is_on(self) -> bool:
 212         r = ask_google(f'is {self.goog_name()} on?')
 213         if not r.success:
 214             return False
 215         if r.audio_transcription is not None:
 216             return 'is on' in r.audio_transcription
 217         raise Exception('Can\'t talk to Google right now!?')
 218
 219     @overrides
 220     def is_off(self) -> bool:
 221         return not self.is_on()
 222
 223
 224 @decorator_utils.singleton
 225 class MerossWrapper(object):
 226     """Global singleton helper class for MerossOutlets.  Note that
 227     instantiating this class causes HTTP traffic with an external
 228     Meross server.  Meross blocks customers who hit their servers too
 229     aggressively so MerossOutlet is lazy about creating instances of
 230     this class.
 231
 232     """
 233
 234     def __init__(self):
 235         self.loop = asyncio.get_event_loop()
 236         self.email = os.environ.get('MEROSS_EMAIL') or scott_secrets.MEROSS_EMAIL
 237         self.password = os.environ.get('MEROSS_PASSWORD') or scott_secrets.MEROSS_PASSWORD
 238         self.devices = self.loop.run_until_complete(self.find_meross_devices())
 239         atexit.register(self.loop.close)
 240
 241     async def find_meross_devices(self) -> List[Any]:
 242         http_api_client = await MerossHttpClient.async_from_user_password(
 243             email=self.email, password=self.password
 244         )
 245
 246         # Setup and start the device manager
 247         manager = MerossManager(http_client=http_api_client)
 248         await manager.async_init()
 249
 250         # Discover devices
 251         await manager.async_device_discovery()
 252         devices = manager.find_devices()
 253         for device in devices:
 254             await device.async_update()
 255         return devices
 256
 257     def get_meross_device_by_name(self, name: str) -> Optional[Any]:
 258         name = name.lower()
 259         name = name.replace('_', ' ')
 260         for device in self.devices:
 261             if device.name.lower() == name:
 262                 return device
 263         return None
 264
 265
 266 class MerossOutlet(BaseOutlet):
 267     """A Meross smart outlet class."""
 268
 269     def __init__(self, name: str, mac: str, keywords: str = '') -> None:
 270         super().__init__(name, mac, keywords)
 271         self.meross_wrapper: Optional[MerossWrapper] = None
 272         self.device: Optional[Any] = None
 273
 274     def lazy_initialize_device(self):
 275         """If we make too many calls to Meross they will block us; only talk
 276         to them when someone actually wants to control a device."""
 277         if self.meross_wrapper is None:
 278             self.meross_wrapper = MerossWrapper()
 279             self.device = self.meross_wrapper.get_meross_device_by_name(self.name)
 280             if self.device is None:
 281                 raise Exception(f'{self.name} is not a known Meross device?!')
 282
 283     @overrides
 284     def turn_on(self) -> bool:
 285         self.lazy_initialize_device()
 286         assert self.meross_wrapper is not None
 287         assert self.device is not None
 288         self.meross_wrapper.loop.run_until_complete(self.device.async_turn_on())
 289         return True
 290
 291     @overrides
 292     def turn_off(self) -> bool:
 293         self.lazy_initialize_device()
 294         assert self.meross_wrapper is not None
 295         assert self.device is not None
 296         self.meross_wrapper.loop.run_until_complete(self.device.async_turn_off())
 297         return True
 298
 299     @overrides
 300     def is_on(self) -> bool:
 301         self.lazy_initialize_device()
 302         assert self.device is not None
 303         return self.device.is_on()
 304
 305     @overrides
 306     def is_off(self) -> bool:
 307         return not self.is_on()