path: root/smart_home/outlets.py

#!/usr/bin/env python3

# © Copyright 2021-2022, Scott Gasch

"""Utilities for dealing with the smart outlets."""

import asyncio
import atexit
import datetime
import logging
import os
from abc import abstractmethod
from typing import Any, Dict, List, Optional

from meross_iot.http_api import MerossHttpClient
from meross_iot.manager import MerossManager
from overrides import overrides

import argparse_utils
import config
import decorator_utils
import scott_secrets
import smart_home.device as dev
import smart_home.tplink_utils as tplink
from decorator_utils import memoized
from google_assistant import GoogleResponse, ask_google

logger = logging.getLogger(__name__)

parser = config.add_commandline_args(
    f"Outlet Utils ({__file__})",
    "Args related to smart outlets.",
)
parser.add_argument(
    '--smart_outlets_tplink_location',
    default='/home/scott/bin/tplink.py',
    metavar='FILENAME',
    help='The location of the tplink.py helper',
    type=argparse_utils.valid_filename,
)


class BaseOutlet(dev.Device):
    """An abstract base class for smart outlets."""

    def __init__(self, name: str, mac: str, keywords: str = "") -> None:
        super().__init__(name.strip(), mac.strip(), keywords)

    @abstractmethod
    def turn_on(self) -> bool:
        pass

    @abstractmethod
    def turn_off(self) -> bool:
        pass

    @abstractmethod
    def is_on(self) -> bool:
        pass

    @abstractmethod
    def is_off(self) -> bool:
        pass


class TPLinkOutlet(BaseOutlet):
    """A TPLink smart outlet."""

    def __init__(self, name: str, mac: str, keywords: str = '') -> None:
        super().__init__(name, mac, keywords)
        self.info: Optional[Dict] = None
        self.info_ts: Optional[datetime.datetime] = None

    @memoized
    def get_tplink_name(self) -> Optional[str]:
        self.info = self.get_info()
        if self.info is not None:
            return self.info["alias"]
        return None

    def get_cmdline(self) -> str:
        cmd = (
            f"{config.config['smart_outlets_tplink_location']} -m {self.mac} "
            f"--no_logging_console "
        )
        return cmd

    def command(self, cmd: str, extra_args: str = None, **kwargs) -> bool:
        cmd = self.get_cmdline() + f"-c {cmd}"
        if extra_args is not None:
            cmd += f" {extra_args}"
        return tplink.tplink_command_wrapper(cmd)

    @overrides
    def turn_on(self) -> bool:
        return self.command('on')

    @overrides
    def turn_off(self) -> bool:
        return self.command('off')

    @overrides
    def is_on(self) -> bool:
        return self.get_on_duration_seconds() > 0

    @overrides
    def is_off(self) -> bool:
        return not self.is_on()

    def get_info(self) -> Optional[Dict]:
        ip = self.get_ip()
        if ip is not None:
            self.info = tplink.tplink_get_info(ip)
            if self.info is not None:
                self.info_ts = datetime.datetime.now()
            else:
                self.info_ts = None
            return self.info
        return None

    def get_on_duration_seconds(self) -> int:
        self.info = self.get_info()
        if self.info is None:
            return 0
        return int(self.info.get("on_time", "0"))


class TPLinkOutletWithChildren(TPLinkOutlet):
    """A TPLink outlet where the top and bottom plus are individually
    controllable."""

    def __init__(self, name: str, mac: str, keywords: str = "") -> None:
        super().__init__(name, mac, keywords)
        self.children: List[str] = []
        self.info: Optional[Dict] = None
        self.info_ts: Optional[datetime.datetime] = None
        assert self.keywords is not None
        assert "children" in self.keywords
        self.info = self.get_info()
        if self.info is not None:
            for child in self.info["children"]:
                self.children.append(child["id"])

    def get_cmdline_with_child(self, child: Optional[str] = None) -> str:
        cmd = super().get_cmdline()
        if child is not None:
            cmd += f"-x {child} "
        return cmd

    @overrides
    def command(self, cmd: str, extra_args: str = None, **kwargs) -> bool:
        child: Optional[str] = kwargs.get('child', None)
        cmd = self.get_cmdline_with_child(child) + f"-c {cmd}"
        if extra_args is not None:
            cmd += f" {extra_args}"
        logger.debug('About to execute: %s', cmd)
        return tplink.tplink_command_wrapper(cmd)

    def get_children(self) -> List[str]:
        return self.children

    @overrides
    def turn_on(self) -> bool:
        return self.command("on", None)

    @overrides
    def turn_off(self) -> bool:
        return self.command("off", None)

    def turn_on_child(self, child: str = None) -> bool:
        return self.command("on", child)

    def turn_off_child(self, child: str = None) -> bool:
        return self.command("off", child)

    def get_child_on_duration_seconds(self, child: str = None) -> int:
        if child is None:
            return super().get_on_duration_seconds()
        else:
            self.info = self.get_info()
            if self.info is None:
                return 0
            for chi in self.info.get("children", {}):
                if chi["id"] == child:
                    return int(chi.get("on_time", "0"))
        return 0


class GoogleOutlet(BaseOutlet):
    """A smart outlet controlled via Google Assistant."""

    def __init__(self, name: str, mac: str, keywords: str = "") -> None:
        super().__init__(name.strip(), mac.strip(), keywords)
        self.info = None

    def goog_name(self) -> str:
        name = self.get_name()
        return name.replace('_', ' ')

    @staticmethod
    def parse_google_response(response: GoogleResponse) -> bool:
        return response.success

    @overrides
    def turn_on(self) -> bool:
        return GoogleOutlet.parse_google_response(ask_google(f'turn {self.goog_name()} on'))

    @overrides
    def turn_off(self) -> bool:
        return GoogleOutlet.parse_google_response(ask_google(f'turn {self.goog_name()} off'))

    @overrides
    def is_on(self) -> bool:
        r = ask_google(f'is {self.goog_name()} on?')
        if not r.success:
            return False
        if r.audio_transcription is not None:
            return 'is on' in r.audio_transcription
        raise Exception('Can\'t talk to Google right now!?')

    @overrides
    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):
    """A Meross smart outlet class."""

    def __init__(self, name: str, mac: str, keywords: str = '') -> None:
        super().__init__(name, mac, keywords)
        self.meross_wrapper: Optional[MerossWrapper] = None
        self.device: Optional[Any] = 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?!')

    @overrides
    def turn_on(self) -> bool:
        self.lazy_initialize_device()
        assert self.meross_wrapper is not None
        assert self.device is not None
        self.meross_wrapper.loop.run_until_complete(self.device.async_turn_on())
        return True

    @overrides
    def turn_off(self) -> bool:
        self.lazy_initialize_device()
        assert self.meross_wrapper is not None
        assert self.device is not None
        self.meross_wrapper.loop.run_until_complete(self.device.async_turn_off())
        return True

    @overrides
    def is_on(self) -> bool:
        self.lazy_initialize_device()
        assert self.device is not None
        return self.device.is_on()

    @overrides
    def is_off(self) -> bool:
        return not self.is_on()