Shelly-Plug-S API Example: The Squid game

shelly_plug_s.py

import logging
from base64 import b64encode
from json import loads as jsdecode
from urllib.parse import urlencode
from enum import Enum
from http.client import HTTPConnection
from time import sleep
from random import uniform
from typing import Union


PASSWORD='secret'
HOST='127.0.0.1'


class Method(str, Enum):
    GET = 'GET'
    POST = 'POST'


class ApiClient(HTTPConnection):
    def __init__(
            self,
            login: str,
            secret: str,
            host: str = '127.0.0.1',
            port: int = 80,
            base_url: str = '/relay/0',
            debuglevel: int = 0,
    ) -> None:
        super().__init__(host, port)
        self.logger = logging.getLogger('shelly.api.client')
        self.debuglevel = debuglevel
        self.base_url = base_url
        self.headers = {
            'Content-Type': 'application/x-www-form-urlencoded',
            # 'Content-Type': 'application/json',
            'Accept': 'application/json',
            'Authorization': 'Basic ' + b64encode(
                f"{login}:{secret}".encode()
            ).decode(),
        }

    def _call(
            self,
            url: Union[str, None] = None,
            body: Union[bytes, str, None] = None,
    ) -> dict:
        method = body and Method.POST or Method.GET
        self.request(method, url, body, self.headers)
        resp = self.getresponse()
        if not resp.status < 400:
            raise ValueError('Failed')
        return jsdecode(resp.read())

    def read_api(self) -> dict:
        return self._call(self.base_url)

    def is_on(self) -> bool:
        return self.read_api()['ison']

    def write_api(
            self,
            op: bool,
    ) -> dict:
        op = op and 'on' or 'off'
        self.logger.info(f"state={op}")
        return self._call(
            f"{self.base_url}?turn={op}",
            urlencode({'turn': op})
        )

    def switch(self) -> bool:
        return self.write_api(not self.is_on())['ison']

    def tick(self, min_sleep: float, max_sleep: float) -> float:
        self.switch()
        w = uniform(min_sleep, max_sleep)
        sleep(w)
        return w

    @classmethod
    def switch_loop(
            cls,
            login: str,
            secret: str,
            host: str,
            min_sleep: float = 0.1,
            max_sleep: float = 10.0,
    ) -> None:
        ua = ApiClient(login, secret, host)
        try:
            while 1:
                w = ua.tick(min_sleep, max_sleep)
                ua.logger.info(f"sleep={round(w, 2)}")
        except KeyboardInterrupt:
            ua.close()


if __name__ == '__main__':
    logging.basicConfig(level=logging.INFO)
    ApiClient.switch_loop(
        'admin',
        PASSWORD,
        HOST,
        0.1,
        5.0,
    )