More type annotations.

[python_utils.git] / smart_home / outlets.py

diff --git a/smart_home/outlets.py b/smart_home/outlets.py
index 81e10a71fa0cd4960c2e5bf1a03789c0663375e3..c079cfd09d9cbbfbb8595379770ebc1c8d7497ae 100644 (file)
--- a/smart_home/outlets.py
+++ b/smart_home/outlets.py
@@ -3,6 +3,8 @@
 """Utilities for dealing with the smart outlets."""
 
 from abc import abstractmethod
+import asyncio
+import atexit
 import datetime
 import json
 import logging
@@ -10,11 +12,16 @@ import os
 import re
 import subprocess
 import sys
-from typing import Any, Dict, List, Optional, Set
+from typing import Any, Dict, List, Optional
+
+from meross_iot.http_api import MerossHttpClient
+from meross_iot.manager import MerossManager
 
 import argparse_utils
 import config
+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
@@ -41,14 +48,16 @@ def tplink_outlet_command(command: str) -> bool:
     result = os.system(command)
     signal = result & 0xFF
     if signal != 0:
-        logger.warning(f'{command} died with signal {signal}')
-        logging_utils.hlog("%s died with signal %d" % (command, signal))
+        msg = f'{command} died with signal {signal}'
+        logger.warning(msg)
+        logging_utils.hlog(msg)
         return False
     else:
         exit_value = result >> 8
         if exit_value != 0:
-            logger.warning(f'{command} failed, exited {exit_value}')
-            logging_utils.hlog("%s failed, exit %d" % (command, exit_value))
+            msg = f'{command} failed, exited {exit_value}'
+            logger.warning(msg)
+            logging_utils.hlog(msg)
             return False
     logger.debug(f'{command} succeeded.')
     return True
@@ -227,3 +236,81 @@ class GoogleOutlet(BaseOutlet):
 
     def is_off(self) -> bool:
         return not self.is_on()
+
+
+@decorator_utils.singleton
+class MerossWrapper(object):
+    """Global singleton helper class for MerossOutlets.  Note that
+    instantiating this class causes HTTP traffic with an external
+    Meross server.  Meross blocks customers who hit their servers too
+    aggressively so MerossOutlet is lazy about creating instances of
+    this class.
+
+    """
+    def __init__(self):
+        self.loop = asyncio.get_event_loop()
+        self.email = os.environ.get('MEROSS_EMAIL') or scott_secrets.MEROSS_EMAIL
+        self.password = os.environ.get('MEROSS_PASSWORD') or scott_secrets.MEROSS_PASSWORD
+        self.devices = self.loop.run_until_complete(self.find_meross_devices())
+        atexit.register(self.loop.close)
+
+    async def find_meross_devices(self) -> List[Any]:
+        http_api_client = await MerossHttpClient.async_from_user_password(
+            email=self.email, password=self.password
+        )
+
+        # Setup and start the device manager
+        manager = MerossManager(http_client=http_api_client)
+        await manager.async_init()
+
+        # Discover devices
+        await manager.async_device_discovery()
+        devices = manager.find_devices()
+        for device in devices:
+            await device.async_update()
+        return devices
+
+    def get_meross_device_by_name(self, name: str) -> Optional[Any]:
+        name = name.lower()
+        name = name.replace('_', ' ')
+        for device in self.devices:
+            if device.name.lower() == name:
+                return device
+        return None
+
+
+class MerossOutlet(BaseOutlet):
+    def __init__(self, name: str, mac: str, keywords: str = '') -> None:
+        super().__init__(name, mac, keywords)
+        self.meross_wrapper = None
+        self.device = None
+
+    def lazy_initialize_device(self):
+        """If we make too many calls to Meross they will block us; only talk
+        to them when someone actually wants to control a device."""
+        if self.meross_wrapper is None:
+            self.meross_wrapper = MerossWrapper()
+            self.device = self.meross_wrapper.get_meross_device_by_name(self.name)
+            if self.device is None:
+                raise Exception(f'{self.name} is not a known Meross device?!')
+
+    def turn_on(self) -> bool:
+        self.lazy_initialize_device()
+        self.meross_wrapper.loop.run_until_complete(
+            self.device.async_turn_on()
+        )
+        return True
+
+    def turn_off(self) -> bool:
+        self.lazy_initialize_device()
+        self.meross_wrapper.loop.run_until_complete(
+            self.device.async_turn_off()
+        )
+        return True
+
+    def is_on(self) -> bool:
+        self.lazy_initialize_device()
+        return self.device.is_on()
+
+    def is_off(self) -> bool:
+        return not self.is_on()