homeassistant-config/custom_components/httpsispmctl/switch.py

import logging
import voluptuous as vol
from collections import defaultdict
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers import aiohttp_client
from homeassistant.const import CONF_HOST, CONF_PORT, CONF_NAME
from homeassistant.components.switch import SwitchDevice, PLATFORM_SCHEMA
from aiohttp import ClientConnectionError


DOMAIN = 'httpsispmctl'

CONF_NUM_PLUGS = "num_power_plugs"
_LOGGER = logging.getLogger(__name__)

PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
    vol.Required(CONF_HOST): cv.string,
    vol.Required(CONF_NAME): cv.string,
    vol.Optional(CONF_PORT, default=2638): cv.port,
    vol.Optional(CONF_NUM_PLUGS, default=4): int,
})


async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
    switches = []
    session = aiohttp_client.async_get_clientsession(hass)

    switches = []
    for i in range(config[CONF_NUM_PLUGS]):
        switches.append(HttpSispmctlSwitch(config[CONF_NAME], i+1)) 
    host_obj = SispmctlHost(session, config[CONF_HOST], config[CONF_PORT], switches)
    for s in switches:
        s.set_host_object(host_obj)

    async_add_entities(switches)


class SispmctlHost:
    def __init__(self, session, hostname, port, switch_objects):
        self._session = session
        self.hostname = hostname
        self.port = port
        self._switch_objects = switch_objects

    async def switch(self, number, on=True):
        cmd = "on" if on else "off"
        try:
            url = f"http://{self.hostname}:{self.port}/{cmd}{number}.html"
            async with self._session.get(url, timeout=3) as resp:
                if resp.status == 200:
                    result = await resp.json()
                    #await self.notify_switches(result)
                else:
                    await self.set_switches_unreachable()
        except ClientConnectionError:
            await self.set_switches_unreachable()

    async def update_state(self):
        pass
        #url = f"http://{self.hostname}:{self.port}"
        #async with self._session.get(url, timeout=3) as resp:
        #    if resp.status == 200:
        #        result = await resp.json()
        #        await self.notify_switches(result)
        #    else:
        #        await self.set_switches_unreachable()

    async def notify_switches(self, response):
        for i, switch in enumerate(self._switch_objects):
            state = bool(response[f"{i+1}"])
            await switch.set_state(on=state, reachable=True)

    async def set_switches_unreachable(self):
        for switch in self._switch_objects:
            await switch.set_state(on=None, reachable=False)


class HttpSispmctlSwitch(SwitchDevice):
    def __init__(self, basename, number):
        self._basename = basename
        self._number = number
        self._available = True
        self._on = None
        self._host = None
    
    def set_host_object(self, host):
        self._host = host
        assert self._host

    async def set_state(self, on, reachable):
        self._on = on
        await self.async_update_ha_state()

    @property
    def name(self):
        return f"{self._basename}{self._number}"
    
    @property
    def available(self) -> bool:
        return self._available
    
    @property
    def is_on(self):
        return self._on
    
    async def async_turn_on(self, **kwargs):
        assert self._host, "Call set_host_object first"
        self._on = True
        await self._host.switch(self._number, on=True)

    async def async_turn_off(self, **kwargs):
        assert self._host, "Call set_host_object first"
        self._on = False
        await self._host.switch(self._number, on=False)

    async def async_update(self):
        await self._host.update_state()