Velbus auto-discovery (#13742)

* remove velbus fan and light platforms

these platforms should not be there since they can be created with template components based on switch platform

* use latest version of python-velbus which supports auto-discovery of modules

* fix linting errors

* fix linting errors

* fix linting errors

* address review comments from @MartinHjelmare

* update based on automatic feedback

* fix linting errors

* update dependency

* syntax corrections

* fix lint warning

* split out common functionality in VelbusEntity
use sync methods for loading platforms
support unique_ids so that entities are registred in entity registry

* fix linting errors

* fix linting errors

* fix linting errors

* integrate review comments (common functionality in VelbusEntity class)

* rename DOMAIN import to VELBUS_DOMAIN

* revert change created by requirements script

* regen
This commit is contained in:
Thomas Delaet 2018-08-05 11:47:17 +03:00 committed by Martin Hjelmare
parent 5e1836f3a2
commit 8a81ee3b4f
6 changed files with 100 additions and 456 deletions

View file

@ -4,93 +4,34 @@ Support for Velbus Binary Sensors.
For more details about this platform, please refer to the documentation at
https://home-assistant.io/components/binary_sensor.velbus/
"""
import asyncio
import logging
import voluptuous as vol
from homeassistant.const import CONF_NAME, CONF_DEVICES
from homeassistant.components.binary_sensor import BinarySensorDevice
from homeassistant.components.binary_sensor import PLATFORM_SCHEMA
from homeassistant.components.velbus import DOMAIN
import homeassistant.helpers.config_validation as cv
DEPENDENCIES = ['velbus']
from homeassistant.components.velbus import (
DOMAIN as VELBUS_DOMAIN, VelbusEntity)
_LOGGER = logging.getLogger(__name__)
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
vol.Required(CONF_DEVICES): vol.All(cv.ensure_list, [
{
vol.Required('module'): cv.positive_int,
vol.Required('channel'): cv.positive_int,
vol.Required(CONF_NAME): cv.string,
vol.Optional('is_pushbutton'): cv.boolean
}
])
})
DEPENDENCIES = ['velbus']
def setup_platform(hass, config, add_devices, discovery_info=None):
async def async_setup_platform(hass, config, async_add_devices,
discovery_info=None):
"""Set up Velbus binary sensors."""
velbus = hass.data[DOMAIN]
add_devices(VelbusBinarySensor(sensor, velbus)
for sensor in config[CONF_DEVICES])
if discovery_info is None:
return
sensors = []
for sensor in discovery_info:
module = hass.data[VELBUS_DOMAIN].get_module(sensor[0])
channel = sensor[1]
sensors.append(VelbusBinarySensor(module, channel))
async_add_devices(sensors)
class VelbusBinarySensor(BinarySensorDevice):
class VelbusBinarySensor(VelbusEntity, BinarySensorDevice):
"""Representation of a Velbus Binary Sensor."""
def __init__(self, binary_sensor, velbus):
"""Initialize a Velbus light."""
self._velbus = velbus
self._name = binary_sensor[CONF_NAME]
self._module = binary_sensor['module']
self._channel = binary_sensor['channel']
self._is_pushbutton = 'is_pushbutton' in binary_sensor \
and binary_sensor['is_pushbutton']
self._state = False
@asyncio.coroutine
def async_added_to_hass(self):
"""Add listener for Velbus messages on bus."""
yield from self.hass.async_add_job(
self._velbus.subscribe, self._on_message)
def _on_message(self, message):
import velbus
if isinstance(message, velbus.PushButtonStatusMessage):
if message.address == self._module and \
self._channel in message.get_channels():
if self._is_pushbutton:
if self._channel in message.closed:
self._toggle()
else:
pass
else:
self._toggle()
def _toggle(self):
if self._state is True:
self._state = False
else:
self._state = True
self.schedule_update_ha_state()
@property
def should_poll(self):
"""No polling needed."""
return False
@property
def name(self):
"""Return the display name of this sensor."""
return self._name
@property
def is_on(self):
"""Return true if the sensor is on."""
return self._state
return self._module.is_closed(self._channel)

View file

@ -1,187 +0,0 @@
"""
Support for Velbus platform.
For more details about this platform, please refer to the documentation at
https://home-assistant.io/components/fan.velbus/
"""
import asyncio
import logging
import voluptuous as vol
from homeassistant.components.fan import (
SPEED_LOW, SPEED_MEDIUM, SPEED_HIGH, FanEntity, SUPPORT_SET_SPEED,
PLATFORM_SCHEMA)
from homeassistant.components.velbus import DOMAIN
from homeassistant.const import CONF_NAME, CONF_DEVICES, STATE_OFF
import homeassistant.helpers.config_validation as cv
DEPENDENCIES = ['velbus']
_LOGGER = logging.getLogger(__name__)
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
vol.Required(CONF_DEVICES): vol.All(cv.ensure_list, [
{
vol.Required('module'): cv.positive_int,
vol.Required('channel_low'): cv.positive_int,
vol.Required('channel_medium'): cv.positive_int,
vol.Required('channel_high'): cv.positive_int,
vol.Required(CONF_NAME): cv.string,
}
])
})
def setup_platform(hass, config, add_devices, discovery_info=None):
"""Set up Fans."""
velbus = hass.data[DOMAIN]
add_devices(VelbusFan(fan, velbus) for fan in config[CONF_DEVICES])
class VelbusFan(FanEntity):
"""Representation of a Velbus Fan."""
def __init__(self, fan, velbus):
"""Initialize a Velbus light."""
self._velbus = velbus
self._name = fan[CONF_NAME]
self._module = fan['module']
self._channel_low = fan['channel_low']
self._channel_medium = fan['channel_medium']
self._channel_high = fan['channel_high']
self._channels = [self._channel_low, self._channel_medium,
self._channel_high]
self._channels_state = [False, False, False]
self._speed = STATE_OFF
@asyncio.coroutine
def async_added_to_hass(self):
"""Add listener for Velbus messages on bus."""
def _init_velbus():
"""Initialize Velbus on startup."""
self._velbus.subscribe(self._on_message)
self.get_status()
yield from self.hass.async_add_job(_init_velbus)
def _on_message(self, message):
import velbus
if isinstance(message, velbus.RelayStatusMessage) and \
message.address == self._module and \
message.channel in self._channels:
if message.channel == self._channel_low:
self._channels_state[0] = message.is_on()
elif message.channel == self._channel_medium:
self._channels_state[1] = message.is_on()
elif message.channel == self._channel_high:
self._channels_state[2] = message.is_on()
self._calculate_speed()
self.schedule_update_ha_state()
def _calculate_speed(self):
if self._is_off():
self._speed = STATE_OFF
elif self._is_low():
self._speed = SPEED_LOW
elif self._is_medium():
self._speed = SPEED_MEDIUM
elif self._is_high():
self._speed = SPEED_HIGH
def _is_off(self):
return self._channels_state[0] is False and \
self._channels_state[1] is False and \
self._channels_state[2] is False
def _is_low(self):
return self._channels_state[0] is True and \
self._channels_state[1] is False and \
self._channels_state[2] is False
def _is_medium(self):
return self._channels_state[0] is True and \
self._channels_state[1] is True and \
self._channels_state[2] is False
def _is_high(self):
return self._channels_state[0] is True and \
self._channels_state[1] is False and \
self._channels_state[2] is True
@property
def name(self):
"""Return the display name of this light."""
return self._name
@property
def should_poll(self):
"""Disable polling."""
return False
@property
def speed(self):
"""Return the current speed."""
return self._speed
@property
def speed_list(self):
"""Get the list of available speeds."""
return [STATE_OFF, SPEED_LOW, SPEED_MEDIUM, SPEED_HIGH]
def turn_on(self, speed=None, **kwargs):
"""Turn on the entity."""
if speed is None:
speed = SPEED_MEDIUM
self.set_speed(speed)
def turn_off(self, **kwargs):
"""Turn off the entity."""
self.set_speed(STATE_OFF)
def set_speed(self, speed):
"""Set the speed of the fan."""
channels_off = []
channels_on = []
if speed == STATE_OFF:
channels_off = self._channels
elif speed == SPEED_LOW:
channels_off = [self._channel_medium, self._channel_high]
channels_on = [self._channel_low]
elif speed == SPEED_MEDIUM:
channels_off = [self._channel_high]
channels_on = [self._channel_low, self._channel_medium]
elif speed == SPEED_HIGH:
channels_off = [self._channel_medium]
channels_on = [self._channel_low, self._channel_high]
for channel in channels_off:
self._relay_off(channel)
for channel in channels_on:
self._relay_on(channel)
self.schedule_update_ha_state()
def _relay_on(self, channel):
import velbus
message = velbus.SwitchRelayOnMessage()
message.set_defaults(self._module)
message.relay_channels = [channel]
self._velbus.send(message)
def _relay_off(self, channel):
import velbus
message = velbus.SwitchRelayOffMessage()
message.set_defaults(self._module)
message.relay_channels = [channel]
self._velbus.send(message)
def get_status(self):
"""Retrieve current status."""
import velbus
message = velbus.ModuleStatusRequestMessage()
message.set_defaults(self._module)
message.channels = self._channels
self._velbus.send(message)
@property
def supported_features(self):
"""Flag supported features."""
return SUPPORT_SET_SPEED

View file

@ -1,104 +0,0 @@
"""
Support for Velbus lights.
For more details about this platform, please refer to the documentation at
https://home-assistant.io/components/light.velbus/
"""
import asyncio
import logging
import voluptuous as vol
from homeassistant.const import CONF_NAME, CONF_DEVICES
from homeassistant.components.light import Light, PLATFORM_SCHEMA
from homeassistant.components.velbus import DOMAIN
import homeassistant.helpers.config_validation as cv
DEPENDENCIES = ['velbus']
_LOGGER = logging.getLogger(__name__)
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
vol.Required(CONF_DEVICES): vol.All(cv.ensure_list, [
{
vol.Required('module'): cv.positive_int,
vol.Required('channel'): cv.positive_int,
vol.Required(CONF_NAME): cv.string
}
])
})
def setup_platform(hass, config, add_devices, discovery_info=None):
"""Set up Lights."""
velbus = hass.data[DOMAIN]
add_devices(VelbusLight(light, velbus) for light in config[CONF_DEVICES])
class VelbusLight(Light):
"""Representation of a Velbus Light."""
def __init__(self, light, velbus):
"""Initialize a Velbus light."""
self._velbus = velbus
self._name = light[CONF_NAME]
self._module = light['module']
self._channel = light['channel']
self._state = False
@asyncio.coroutine
def async_added_to_hass(self):
"""Add listener for Velbus messages on bus."""
def _init_velbus():
"""Initialize Velbus on startup."""
self._velbus.subscribe(self._on_message)
self.get_status()
yield from self.hass.async_add_job(_init_velbus)
def _on_message(self, message):
import velbus
if isinstance(message, velbus.RelayStatusMessage) and \
message.address == self._module and \
message.channel == self._channel:
self._state = message.is_on()
self.schedule_update_ha_state()
@property
def name(self):
"""Return the display name of this light."""
return self._name
@property
def should_poll(self):
"""Disable polling."""
return False
@property
def is_on(self):
"""Return true if the light is on."""
return self._state
def turn_on(self, **kwargs):
"""Instruct the light to turn on."""
import velbus
message = velbus.SwitchRelayOnMessage()
message.set_defaults(self._module)
message.relay_channels = [self._channel]
self._velbus.send(message)
def turn_off(self, **kwargs):
"""Instruct the light to turn off."""
import velbus
message = velbus.SwitchRelayOffMessage()
message.set_defaults(self._module)
message.relay_channels = [self._channel]
self._velbus.send(message)
def get_status(self):
"""Retrieve current status."""
import velbus
message = velbus.ModuleStatusRequestMessage()
message.set_defaults(self._module)
message.channels = [self._channel]
self._velbus.send(message)

View file

@ -4,108 +4,42 @@ Support for Velbus switches.
For more details about this platform, please refer to the documentation at
https://home-assistant.io/components/switch.velbus/
"""
import asyncio
import logging
import voluptuous as vol
from homeassistant.const import CONF_NAME, CONF_DEVICES
from homeassistant.components.switch import SwitchDevice, PLATFORM_SCHEMA
from homeassistant.components.velbus import DOMAIN
import homeassistant.helpers.config_validation as cv
from homeassistant.components.switch import SwitchDevice
from homeassistant.components.velbus import (
DOMAIN as VELBUS_DOMAIN, VelbusEntity)
_LOGGER = logging.getLogger(__name__)
SWITCH_SCHEMA = {
vol.Required('module'): cv.positive_int,
vol.Required('channel'): cv.positive_int,
vol.Required(CONF_NAME): cv.string
}
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
vol.Required(CONF_DEVICES):
vol.All(cv.ensure_list, [SWITCH_SCHEMA])
})
DEPENDENCIES = ['velbus']
def setup_platform(hass, config, add_devices, discovery_info=None):
"""Set up the Switch."""
velbus = hass.data[DOMAIN]
devices = []
for switch in config[CONF_DEVICES]:
devices.append(VelbusSwitch(switch, velbus))
add_devices(devices)
return True
async def async_setup_platform(hass, config, async_add_devices,
discovery_info=None):
"""Set up the Velbus Switch platform."""
if discovery_info is None:
return
switches = []
for switch in discovery_info:
module = hass.data[VELBUS_DOMAIN].get_module(switch[0])
channel = switch[1]
switches.append(VelbusSwitch(module, channel))
async_add_devices(switches)
class VelbusSwitch(SwitchDevice):
class VelbusSwitch(VelbusEntity, SwitchDevice):
"""Representation of a switch."""
def __init__(self, switch, velbus):
"""Initialize a Velbus switch."""
self._velbus = velbus
self._name = switch[CONF_NAME]
self._module = switch['module']
self._channel = switch['channel']
self._state = False
@asyncio.coroutine
def async_added_to_hass(self):
"""Add listener for Velbus messages on bus."""
def _init_velbus():
"""Initialize Velbus on startup."""
self._velbus.subscribe(self._on_message)
self.get_status()
yield from self.hass.async_add_job(_init_velbus)
def _on_message(self, message):
import velbus
if isinstance(message, velbus.RelayStatusMessage) and \
message.address == self._module and \
message.channel == self._channel:
self._state = message.is_on()
self.schedule_update_ha_state()
@property
def name(self):
"""Return the display name of this switch."""
return self._name
@property
def should_poll(self):
"""Disable polling."""
return False
@property
def is_on(self):
"""Return true if the switch is on."""
return self._state
return self._module.is_on(self._channel)
def turn_on(self, **kwargs):
"""Instruct the switch to turn on."""
import velbus
message = velbus.SwitchRelayOnMessage()
message.set_defaults(self._module)
message.relay_channels = [self._channel]
self._velbus.send(message)
self._module.turn_on(self._channel)
def turn_off(self, **kwargs):
"""Instruct the switch to turn off."""
import velbus
message = velbus.SwitchRelayOffMessage()
message.set_defaults(self._module)
message.relay_channels = [self._channel]
self._velbus.send(message)
def get_status(self):
"""Retrieve current status."""
import velbus
message = velbus.ModuleStatusRequestMessage()
message.set_defaults(self._module)
message.channels = [self._channel]
self._velbus.send(message)
self._module.turn_off(self._channel)

View file

@ -9,8 +9,10 @@ import voluptuous as vol
import homeassistant.helpers.config_validation as cv
from homeassistant.const import EVENT_HOMEASSISTANT_STOP, CONF_PORT
from homeassistant.helpers.discovery import load_platform
from homeassistant.helpers.entity import Entity
REQUIREMENTS = ['python-velbus==2.0.11']
REQUIREMENTS = ['python-velbus==2.0.17']
_LOGGER = logging.getLogger(__name__)
@ -26,18 +28,76 @@ CONFIG_SCHEMA = vol.Schema({
}, extra=vol.ALLOW_EXTRA)
def setup(hass, config):
async def async_setup(hass, config):
"""Set up the Velbus platform."""
import velbus
port = config[DOMAIN].get(CONF_PORT)
connection = velbus.VelbusUSBConnection(port)
controller = velbus.Controller(connection)
controller = velbus.Controller(port)
hass.data[DOMAIN] = controller
def stop_velbus(event):
"""Disconnect from serial port."""
_LOGGER.debug("Shutting down ")
connection.stop()
controller.stop()
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, stop_velbus)
def callback():
modules = controller.get_modules()
discovery_info = {
'switch': [],
'binary_sensor': []
}
for module in modules:
for channel in range(1, module.number_of_channels() + 1):
for category in discovery_info:
if category in module.get_categories(channel):
discovery_info[category].append((
module.get_module_address(),
channel
))
load_platform(hass, 'switch', DOMAIN,
discovery_info['switch'], config)
load_platform(hass, 'binary_sensor', DOMAIN,
discovery_info['binary_sensor'], config)
controller.scan(callback)
hass.bus.listen_once(EVENT_HOMEASSISTANT_STOP, stop_velbus)
return True
class VelbusEntity(Entity):
"""Representation of a Velbus entity."""
def __init__(self, module, channel):
"""Initialize a Velbus entity."""
self._module = module
self._channel = channel
@property
def unique_id(self):
"""Get unique ID."""
serial = 0
if self._module.serial == 0:
serial = self._module.get_module_address()
else:
serial = self._module.serial
return "{}-{}".format(serial, self._channel)
@property
def name(self):
"""Return the display name of this entity."""
return self._module.get_name(self._channel)
@property
def should_poll(self):
"""Disable polling."""
return False
async def async_added_to_hass(self):
"""Add listener for state changes."""
self._module.on_status_update(self._channel, self._on_update)
def _on_update(self, state):
self.schedule_update_ha_state()

View file

@ -1117,7 +1117,7 @@ python-telegram-bot==10.1.0
python-twitch==1.3.0
# homeassistant.components.velbus
python-velbus==2.0.11
python-velbus==2.0.17
# homeassistant.components.media_player.vlc
python-vlc==1.1.2