"""Utilities for dealing with the smart outlets."""
from abc import abstractmethod
+import asyncio
+import atexit
import datetime
import json
import logging
import re
import subprocess
import sys
-from typing import Dict, List, Optional
+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
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()