jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-27 07:07:28 +00:00
Code Issues Packages Projects Releases Wiki Activity

Migrate to voluptuous (#3290)

Browse Source
This commit is contained in:
Fabian Affolter 2016-09-11 09:22:08 +02:00 committed by GitHub
parent 78313c793c
commit f341974b8b
1 changed files with 59 additions and 41 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

88
homeassistant/components/switch/netio.py
View File

@ -7,59 +7,75 @@ https://home-assistant.io/components/switch.netio/
import logging import logging
from collections import namedtuple from collections import namedtuple
from datetime import timedelta from datetime import timedelta
import voluptuous as vol
from homeassistant import util from homeassistant import util
from homeassistant.components.http import HomeAssistantView from homeassistant.components.http import HomeAssistantView
from homeassistant.const import CONF_HOST, CONF_PORT, CONF_USERNAME, \ from homeassistant.const import (
CONF_PASSWORD, EVENT_HOMEASSISTANT_STOP, STATE_ON CONF_HOST, CONF_PORT, CONF_USERNAME, CONF_PASSWORD,
from homeassistant.helpers import validate_config EVENT_HOMEASSISTANT_STOP, STATE_ON)
from homeassistant.components.switch import SwitchDevice from homeassistant.components.switch import (SwitchDevice, PLATFORM_SCHEMA)
import homeassistant.helpers.config_validation as cv
REQUIREMENTS = ['pynetio==0.1.6']
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
DEPENDENCIES = ['http'] ATTR_CURRENT_POWER_MWH = 'current_power_mwh'
REQUIREMENTS = ['pynetio==0.1.6'] ATTR_CURRENT_POWER_W = 'current_power_w'
DEFAULT_USERNAME = 'admin' ATTR_START_DATE = 'start_date'
ATTR_TODAY_MWH = 'today_mwh'
ATTR_TOTAL_CONSUMPTION_KWH = 'total_energy_kwh'
CONF_OUTLETS = 'outlets'
DEFAULT_PORT = 1234 DEFAULT_PORT = 1234
URL_API_NETIO_EP = "/api/netio/<host>" DEFAULT_USERNAME = 'admin'
DEPENDENCIES = ['http']
CONF_OUTLETS = "outlets"
REQ_CONF = [CONF_HOST, CONF_OUTLETS]
ATTR_TODAY_MWH = "today_mwh"
ATTR_TOTAL_CONSUMPTION_KWH = "total_energy_kwh"
ATTR_CURRENT_POWER_MWH = "current_power_mwh"
ATTR_CURRENT_POWER_W = "current_power_w"
Device = namedtuple('device', ['netio', 'entities']) Device = namedtuple('device', ['netio', 'entities'])
DEVICES = {} DEVICES = {}
ATTR_START_DATE = 'start_date'
MIN_TIME_BETWEEN_SCANS = timedelta(seconds=10) MIN_TIME_BETWEEN_SCANS = timedelta(seconds=10)
REQ_CONF = [CONF_HOST, CONF_OUTLETS]
def setup_platform(hass, config, add_devices_callback, discovery_info=None): URL_API_NETIO_EP = '/api/netio/<host>'
"""Configure the netio platform."""
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
vol.Required(CONF_HOST): cv.string,
vol.Required(CONF_PORT, default=DEFAULT_PORT): cv.port,
vol.Required(CONF_USERNAME, default=DEFAULT_USERNAME): cv.string,
vol.Required(CONF_PASSWORD): cv.string,
vol.Optional(CONF_OUTLETS): {cv.string: cv.string},
})
def setup_platform(hass, config, add_devices, discovery_info=None):
"""Configure the Netio platform."""
from pynetio import Netio from pynetio import Netio
if validate_config({"conf": config}, {"conf": [CONF_OUTLETS, host = config.get(CONF_HOST)
CONF_HOST]}, _LOGGER): username = config.get(CONF_USERNAME)
password = config.get(CONF_PASSWORD)
port = config.get(CONF_PORT)
if len(DEVICES) == 0: if len(DEVICES) == 0:
hass.wsgi.register_view(NetioApiView) hass.wsgi.register_view(NetioApiView)
dev = Netio(config[CONF_HOST], dev = Netio(host, port, username, password)
config.get(CONF_PORT, DEFAULT_PORT),
config.get(CONF_USERNAME, DEFAULT_USERNAME),
config.get(CONF_PASSWORD, DEFAULT_USERNAME))
DEVICES[config[CONF_HOST]] = Device(dev, []) DEVICES[host] = Device(dev, [])
# Throttle the update for all NetioSwitches of one Netio # Throttle the update for all NetioSwitches of one Netio
dev.update = util.Throttle(MIN_TIME_BETWEEN_SCANS)(dev.update) dev.update = util.Throttle(MIN_TIME_BETWEEN_SCANS)(dev.update)
for key in config[CONF_OUTLETS]: for key in config[CONF_OUTLETS]:
switch = NetioSwitch(DEVICES[config[CONF_HOST]].netio, key, switch = NetioSwitch(
config[CONF_OUTLETS][key]) DEVICES[host].netio, key, config[CONF_OUTLETS][key])
DEVICES[config[CONF_HOST]].entities.append(switch) DEVICES[host].entities.append(switch)
add_devices_callback(DEVICES[config[CONF_HOST]].entities) add_devices(DEVICES[host].entities)
hass.bus.listen_once(EVENT_HOMEASSISTANT_STOP, dispose) hass.bus.listen_once(EVENT_HOMEASSISTANT_STOP, dispose)
return True return True
@ -75,7 +91,7 @@ class NetioApiView(HomeAssistantView):
"""WSGI handler class.""" """WSGI handler class."""
url = URL_API_NETIO_EP url = URL_API_NETIO_EP
name = "api:netio" name = 'api:netio'
def get(self, request, host): def get(self, request, host):
"""Request handler.""" """Request handler."""
@ -135,7 +151,7 @@ class NetioSwitch(SwitchDevice):
def _set(self, value): def _set(self, value):
val = list('uuuu') val = list('uuuu')
val[self.outlet - 1] = "1" if value else "0" val[self.outlet - 1] = '1' if value else '0'
self.netio.get('port list %s' % ''.join(val)) self.netio.get('port list %s' % ''.join(val))
self.netio.states[self.outlet - 1] = value self.netio.states[self.outlet - 1] = value
self.update_ha_state() self.update_ha_state()
@ -146,15 +162,17 @@ class NetioSwitch(SwitchDevice):
return self.netio.states[self.outlet - 1] return self.netio.states[self.outlet - 1]
def update(self): def update(self):
"""Called by HA.""" """Called by Home Assistant."""
self.netio.update() self.netio.update()
@property @property
def state_attributes(self): def state_attributes(self):
"""Return optional state attributes.""" """Return optional state attributes."""
return {ATTR_CURRENT_POWER_W: self.current_power_w, return {
ATTR_CURRENT_POWER_W: self.current_power_w,
ATTR_TOTAL_CONSUMPTION_KWH: self.cumulated_consumption_kwh, ATTR_TOTAL_CONSUMPTION_KWH: self.cumulated_consumption_kwh,
ATTR_START_DATE: self.start_date.split('|')[0]} ATTR_START_DATE: self.start_date.split('|')[0]
}
@property @property
def current_power_w(self): def current_power_w(self):