Hi,
I’m trying to develop a platform for INELS RF system with eLan gateway .
At this moment I have working prototype as custom component (code at the end of this post). However I’m not sure how to get it into shape ready for submitting = help, comments and tests are welcome.
Especially I’m not sure how to proceed with calls to the device. In doc there is a statement that this should be done in library. However creating full library (nI have never done that in python) just to preform and parse few xhttp calls semms to me a bit overkill.
Anyway here is my code:
Configuration
eLan:
url: “url of eLan”
Plaform part
“”"
Support for eLan devices manual discovery.
For more details about this component, please refer to the documentation at“”"
import asyncio
import loggingimport async_timeout
import voluptuous as vol#from homeassistant.components.discovery import SERVICE_ELAN
SERVICE_ELAN = ‘eLan’from homeassistant.helpers import discovery
from homeassistant.helpers.discovery import load_platform
from homeassistant.helpers import config_validation as cvfrom homeassistant.const import EVENT_HOMEASSISTANT_STOP
DOMAIN = ‘eLan’
#SUBSCRIPTION_REGISTRY = None # do budoucna zde bude ws link k eLanu abychom meli upozorneni na zmeny stavu
KNOWN_DEVICES =_LOGGER = logging.getLogger(name)
CONF_STATIC = ‘static’
CONFIG_SCHEMA = vol.Schema({
DOMAIN: vol.Schema({
vol.Required(‘url’): cv.string,
}),
}, extra=vol.ALLOW_EXTRA)@asyncio.coroutine
def async_setup(hass, config):
“”“Setup eLan component.”“”
conf = config.get(DOMAIN, {})
url = conf.get(‘url’)@asyncio.coroutine def eLan_discovered(service, info): """Run when a gateway is discovered.""" url = info['url'] yield from _setup_eLan(hass, config, url) discovery.async_listen(hass, SERVICE_ELAN, eLan_discovered) yield from eLan_discovered(None,{'url': url}) return True@asyncio.coroutine
def _setup_eLan(hass, hass_config, url):
“”“Call platform discovery for devices on particular eLan.”“”
hass.async_add_job(discovery.async_load_platform(
hass, ‘light’, DOMAIN, {‘url’: url}, hass_config))
hass.async_add_job(discovery.async_load_platform(
hass, ‘sensor’, DOMAIN, {‘url’: url}, hass_config))
return True
Light part:
“”"
Support for the eLan.
For more details about this platform, please refer to the documentation at“”"
import asyncio
import aiohttp
import logging
import voluptuous as volfrom homeassistant.core import callback
from homeassistant.const import ATTR_BATTERY_LEVEL
from homeassistant.components.light import (
ATTR_BRIGHTNESS, ATTR_RGB_COLOR,
SUPPORT_BRIGHTNESS, SUPPORT_RGB_COLOR, Light)
from homeassistant.components.light import PLATFORM_SCHEMA
from homeassistant.util import color as color_util
from homeassistant.helpers import config_validation as cv#REQUIREMENTS = [‘pyelan’]
_LOGGER = logging.getLogger(name)
#MIN_TIME_BETWEEN_SCANS = timedelta(seconds=10)
#MIN_TIME_BETWEEN_FORCED_SCANS = timedelta(milliseconds=100)Validation of the user’s configuration
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
vol.Required(‘url’): cv.string,
})@asyncio.coroutine
def async_setup_platform(hass, config, async_add_devices, discovery_info=None):
“”“Set up the eLan Light platform.”“”
if discovery_info is None:
returnurl = discovery_info['url'] session = aiohttp.ClientSession() resp = yield from session.get(url+'/api/devices', timeout=3) device_list = yield from resp.json() _LOGGER.info("eLan devices") _LOGGER.info(device_list) for device in device_list: resp = yield from session.get(device_list[device]['url'], timeout=3) info = yield from resp.json() _LOGGER.info("eLan device") _LOGGER.info(device) if info['device info']['type'] == 'light': _LOGGER.info("eLan Light to add") _LOGGER.info(device) async_add_devices([eLanLight(device_list[device]['url'], info)])session.close()
class eLanLight(Light):
“”“The platform class required by Home Assistant.”“”def __init__(self, light, info): """Initialize a Light.""" _LOGGER.info("eLan light initialisation") _LOGGER.info(info) self._light = light self._info = info self._name = info['device info']['label'] self._state = None self._brightness = None self._last_brightness = 100 self._dimmer = False self._available = True self._features = None self._rgb_color = None if info['primary actions'][0] == 'brightness': self._features = SUPPORT_BRIGHTNESS self._dimmer = True @property def device_state_attributes(self): """Return the devices' state attributes.""" attrs = {} return attrs @asyncio.coroutine def async_added_to_hass(self): """Start thread when added to hass.""" #self._async_start_observe() @property def available(self): """Return True if entity is available.""" return self._available @property def should_poll(self): """WS notification not implemented yet - polling is needed""" return True @property def supported_features(self): """Flag supported features.""" return self._features @property def name(self): """Return the display name of this light.""" return self._name @property def is_on(self): """Return true if light is on.""" return self._state @property def brightness(self): """Return the brightness of the light.""" return self._brightness @property def rgb_color(self): """RGB color of the light.""" return self._rgb_color @asyncio.coroutine def update(self): """Fetch new state data for this light. This is the only method that should fetch new data for Home Assistant. """ _LOGGER.info('eLan Light update') _LOGGER.info(self._light + '/state') session = aiohttp.ClientSession() resp = yield from session.get(self._light + '/state', timeout=3) state = yield from resp.json() _LOGGER.info(state) tmp = False if 'on' in state: tmp = state['on'] if 'brightness' in state: self._brightness = state['brightness'] if state['brightness'] > 0: tmp = True self._state = tmp @asyncio.coroutine def async_turn_off(self, **kwargs): """Instruct the light to turn off.""" _LOGGER.info('Turning off eLan light') _LOGGER.info(self._light) session = aiohttp.ClientSession() if self._dimmer: resp = yield from session.put(self._light, json={'brightness':0}) else: resp = yield from session.put(self._light, json={'on':False}) info = yield from resp.text() _LOGGER.info(info) @asyncio.coroutine def async_turn_on(self, **kwargs): """ Instruct the light to turn on. """ _LOGGER.info('Turning on eLan light') _LOGGER.info(self._light) if ATTR_BRIGHTNESS in kwargs: if kwargs[ATTR_BRIGHTNESS] > 0: self._last_brightness = kwargs[ATTR_BRIGHTNESS] session = aiohttp.ClientSession() if self._dimmer: if self._last_brightness is 0: self._last_brightness = 100 resp = yield from session.put(self._light, json={'brightness': self._last_brightness}) else: resp =yield from session.put(self._light, json={'on': True}) info = yield from resp.text() _LOGGER.info(info)
Sensor part:
“”"
Support for the eLan.
For more details about this platform, please refer to the documentation at“”"
import asyncio
import aiohttp
import logging
import voluptuous as volfrom homeassistant.core import callback
from homeassistant.const import TEMP_CELSIUS
from homeassistant.helpers.entity import Entity
from homeassistant.components.sensor import PLATFORM_SCHEMA
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.entity import Entity, async_generate_entity_id
from homeassistant.helpers.event import async_track_state_change
from homeassistant.exceptions import TemplateError#REQUIREMENTS = [‘pyelan’]
_LOGGER = logging.getLogger(name)
#MIN_TIME_BETWEEN_SCANS = timedelta(seconds=10)
#MIN_TIME_BETWEEN_FORCED_SCANS = timedelta(milliseconds=100)Validation of the user’s configuration
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
vol.Required(‘url’): cv.string,
})@asyncio.coroutine
def async_setup_platform(hass, config, async_add_devices, discovery_info=None):
“”“Set up the eLan thermostat platform.”“”
if discovery_info is None:
returnurl = discovery_info['url'] session = aiohttp.ClientSession() resp = yield from session.get(url+'/api/devices', timeout=3) device_list = yield from resp.json() _LOGGER.info("eLan devices") _LOGGER.info(device_list) for device in device_list: resp = yield from session.get(device_list[device]['url'], timeout=3) info = yield from resp.json() _LOGGER.info("eLan device") _LOGGER.info(device) if info['device info']['type'] == 'heating': _LOGGER.info("eLan Thermostat to add") _LOGGER.info(device) async_add_devices([eLanThermostat(device_list[device]['url'], info, 'temperature')]) async_add_devices([eLanThermostat(device_list[device]['url'], info, 'on')])session.close()
class eLanThermostat(Entity):
“”“The platform class required by Home Assistant.”“”def __init__(self, thermostat, info, var): """Initialize a thermostat.""" _LOGGER.info("eLan thermostat initialisation") _LOGGER.info(info) self._thermostat = thermostat self._var = 'temperature' self._info = info self._state = None self._name = info['device info']['label'] self._temperatureIN = None self._temperatureOUT = None self._on = None self._locked = None self._units = None if var is 'temperature': self._name = info['device info']['label'] + '-T' self._var = var self._units = TEMP_CELSIUS if var is 'temperature OUT': self._name = info['device info']['label'] + '-T OUT' self._var = var self._units = TEMP_CELSIUS if var is 'temperature IN': self._name = info['device info']['label'] + '-T IN' self._var = var self._units = TEMP_CELSIUS if var is 'on': self._name = info['device info']['label'] + '-ON' self._var = var self._available = True @asyncio.coroutine def async_added_to_hass(self): """Start thread when added to hass.""" #self._async_start_observe() @property def available(self): """Return True if entity is available.""" return self._available @property def should_poll(self): """WS notification not implemented yet - polling is needed""" return True @property def name(self): """Return the display name of this thermostat.""" return self._name @property def state(self): """Return the state of the sensor.""" return self._state @property def unit_of_measurement(self): """Return the unit of measurement.""" return self._units @asyncio.coroutine def update(self): """Fetch new state data for this thermostat. This is the only method that should fetch new data for Home Assistant. """ _LOGGER.info('eLan thermostat update') _LOGGER.info(self._thermostat + '/state') session = aiohttp.ClientSession() resp = yield from session.get(self._thermostat + '/state', timeout=3) state = yield from resp.json() _LOGGER.info(state) if 'temperature IN' in state: self._temperatureIN = state['temperature IN'] if 'temperature OUT' in state: self._temperatureOUT = state['temperature OUT'] if 'on' in state: self._on = state['on'] if 'locked' in state: self._locked = state['locked'] if self._var is 'temperature': if self._temperatureIN and (self._temperatureIN>-99): self._state = self._temperatureIN if self._temperatureOUT and (self._temperatureOUT>-99): self._state = self._temperatureOUT if self._var is 'temperature OUT': if self._temperatureOUT and (self._temperatureOUT>-99): self._state = self._temperatureOUT if self._var is 'temperature IN': if self._temperatureIN and (self._temperatureIN>-99): self._state = self._temperatureIN if self._var is 'on': self._state = self._on _LOGGER.info(self._state)