116 lines
3.7 KiB
Python
116 lines
3.7 KiB
Python
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):
|
|
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() |