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
 189 class GoogleOutlet(BaseOutlet):
 190     """A smart outlet controlled via Google Assistant."""
 191
 192     def __init__(self, name: str, mac: str, keywords: str = "") -> None:
 193         super().__init__(name.strip(), mac.strip(), keywords)
 194         self.info = None
 195
 196     def goog_name(self) -> str:
 197         name = self.get_name()
 198         return name.replace('_', ' ')
 199
 200     @staticmethod
 201     def parse_google_response(response: GoogleResponse) -> bool:
 202         return response.success
 203
 204     @overrides
 205     def turn_on(self) -> bool:
 206         return GoogleOutlet.parse_google_response(ask_google(f'turn {self.goog_name()} on'))
 207
 208     @overrides
 209     def turn_off(self) -> bool:
 210         return GoogleOutlet.parse_google_response(ask_google(f'turn {self.goog_name()} off'))
 211
 212     @overrides
 213     def is_on(self) -> bool:
 214         r = ask_google(f'is {self.goog_name()} on?')
 215         if not r.success:
 216             return False
 217         if r.audio_transcription is not None:
 218             return 'is on' in r.audio_transcription
 219         raise Exception('Can\'t talk to Google right now!?')
 220
 221     @overrides
 222     def is_off(self) -> bool:
 223         return not self.is_on()
 224
 225
 226 @decorator_utils.singleton
 227 class MerossWrapper(object):
 228     """Global singleton helper class for MerossOutlets.  Note that
 229     instantiating this class causes HTTP traffic with an external
 230     Meross server.  Meross blocks customers who hit their servers too
 231     aggressively so MerossOutlet is lazy about creating instances of
 232     this class.
 233
 234     """
 235
 236     def __init__(self):
 237         self.loop = asyncio.get_event_loop()
 238         self.email = os.environ.get('MEROSS_EMAIL') or scott_secrets.MEROSS_EMAIL
 239         self.password = os.environ.get('MEROSS_PASSWORD') or scott_secrets.MEROSS_PASSWORD
 240         self.devices = self.loop.run_until_complete(self.find_meross_devices())
 241         atexit.register(self.loop.close)
 242
 243     async def find_meross_devices(self) -> List[Any]:
 244         http_api_client = await MerossHttpClient.async_from_user_password(
 245             email=self.email, password=self.password
 246         )
 247
 248         # Setup and start the device manager
 249         manager = MerossManager(http_client=http_api_client)
 250         await manager.async_init()
 251
 252         # Discover devices
 253         await manager.async_device_discovery()
 254         devices = manager.find_devices()
 255         for device in devices:
 256             await device.async_update()
 257         return devices
 258
 259     def get_meross_device_by_name(self, name: str) -> Optional[Any]:
 260         name = name.lower()
 261         name = name.replace('_', ' ')
 262         for device in self.devices:
 263             if device.name.lower() == name:
 264                 return device
 265         return None
 266
 267
 268 class MerossOutlet(BaseOutlet):
 269     """A Meross smart outlet class."""
 270
 271     def __init__(self, name: str, mac: str, keywords: str = '') -> None:
 272         super().__init__(name, mac, keywords)
 273         self.meross_wrapper: Optional[MerossWrapper] = None
 274         self.device: Optional[Any] = None
 275
 276     def lazy_initialize_device(self):
 277         """If we make too many calls to Meross they will block us; only talk
 278         to them when someone actually wants to control a device."""
 279         if self.meross_wrapper is None:
 280             self.meross_wrapper = MerossWrapper()
 281             self.device = self.meross_wrapper.get_meross_device_by_name(self.name)
 282             if self.device is None:
 283                 raise Exception(f'{self.name} is not a known Meross device?!')
 284
 285     @overrides
 286     def turn_on(self) -> bool:
 287         self.lazy_initialize_device()
 288         assert self.meross_wrapper is not None
 289         assert self.device is not None
 290         self.meross_wrapper.loop.run_until_complete(self.device.async_turn_on())
 291         return True
 292
 293     @overrides
 294     def turn_off(self) -> bool:
 295         self.lazy_initialize_device()
 296         assert self.meross_wrapper is not None
 297         assert self.device is not None
 298         self.meross_wrapper.loop.run_until_complete(self.device.async_turn_off())
 299         return True
 300
 301     @overrides
 302     def is_on(self) -> bool:
 303         self.lazy_initialize_device()
 304         assert self.device is not None
 305         return self.device.is_on()
 306
 307     @overrides
 308     def is_off(self) -> bool:
 309         return not self.is_on()