smart_home/outlets.py

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