Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Enable velbus config entries (#25308) 

* Initial work on config_flow

* Finish config flow

* Pylint checks, make sure the import only happens once

* Added support for unloading, small fixes

* Check in the hassfest output files

* Flake8 fixes

* pylint mistake after flake8 fixes

* Work on comments

* Abort the import if it is already imported

* More comments resolved

* Added testcases for velbus config flow

* Fix pylint and flake8

* Added connection test to the config flow

* More sugestions

* renamed the abort reason

* excluded all but the config_flow.py from the velbus component in coveragerc

* Rewrote testcases with a patched version of _test_connection

* Docstyle fixes

* Updated the velbus testcases

* just yield

* flake8 fixes
pull/25553/head
Maikel Punie 2019-07-29 09:21:26 +02:00 committed by Martin Hjelmare
parent 03052802a4
commit 1f9f201571
18 changed files with 407 additions and 81 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
.coveragerc
View File

@ -672,7 +672,13 @@ omit =
homeassistant/components/usps/*
homeassistant/components/vallox/*
homeassistant/components/vasttrafik/sensor.py
homeassistant/components/velbus/*
homeassistant/components/velbus/__init__.py
homeassistant/components/velbus/binary_sensor.py
homeassistant/components/velbus/climate.py
homeassistant/components/velbus/const.py
homeassistant/components/velbus/cover.py
homeassistant/components/velbus/sensor.py
homeassistant/components/velbus/switch.py
homeassistant/components/velux/*
homeassistant/components/venstar/climate.py
homeassistant/components/vera/*

1
CODEOWNERS
View File

@ -287,6 +287,7 @@ homeassistant/components/updater/* @home-assistant/core
homeassistant/components/upnp/* @robbiet480
homeassistant/components/uptimerobot/* @ludeeus
homeassistant/components/utility_meter/* @dgomes
homeassistant/components/velbus/* @ceral2nd
homeassistant/components/velux/* @Julius2342
homeassistant/components/version/* @fabaff
homeassistant/components/vesync/* @markperdue @webdjoe

21
homeassistant/components/velbus/.translations/en.json Normal file
View File

@ -0,0 +1,21 @@
{
"config": {
"title": "Velbus",
"step": {
"user": {
"title": "Define the velbus connection",
"data": {
"name": "The name for this velbus connection",
"port": "Connection string"
}
}
},
"error": {
"port_exists": "This port is already configured",
"connection_failed": "The velbus connection failed"
},
"abort": {
"port_exists": "This port is already configured"
}
}
}

107
homeassistant/components/velbus/__init__.py
View File

@ -1,16 +1,19 @@
"""Support for Velbus devices."""
import asyncio
import logging
import velbus
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.config_entries import SOURCE_IMPORT, ConfigEntry
from homeassistant.const import CONF_PORT, CONF_NAME
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.typing import HomeAssistantType
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
DOMAIN = 'velbus'
VELBUS_MESSAGE = 'velbus.message'
CONFIG_SCHEMA = vol.Schema({
@ -19,54 +22,69 @@ CONFIG_SCHEMA = vol.Schema({
})
}, extra=vol.ALLOW_EXTRA)
COMPONENT_TYPES = ['switch', 'sensor', 'binary_sensor', 'cover', 'climate']
async def async_setup(hass, config):
"""Set up the Velbus platform."""
import velbus
# Import from the configuration file if needed
if DOMAIN not in config:
return True
port = config[DOMAIN].get(CONF_PORT)
controller = velbus.Controller(port)
data = {}
hass.data[DOMAIN] = controller
if port:
data = {
CONF_PORT: port,
CONF_NAME: 'Velbus import'
}
def stop_velbus(event):
"""Disconnect from serial port."""
_LOGGER.debug("Shutting down ")
controller.stop()
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN,
context={'source': SOURCE_IMPORT},
data=data
))
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, stop_velbus)
return True
async def async_setup_entry(hass: HomeAssistantType, entry: ConfigEntry):
"""Establish connection with velbus."""
hass.data.setdefault(DOMAIN, {})
controller = velbus.Controller(entry.data[CONF_PORT])
def callback():
modules = controller.get_modules()
discovery_info = {
'cover': [],
'switch': [],
'binary_sensor': [],
'climate': [],
'sensor': []
'cntrl': controller
}
for category in COMPONENT_TYPES:
discovery_info[category] = []
for module in modules:
for channel in range(1, module.number_of_channels() + 1):
for category in discovery_info:
for category in COMPONENT_TYPES:
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, 'climate', DOMAIN,
discovery_info['climate'], config)
load_platform(hass, 'binary_sensor', DOMAIN,
discovery_info['binary_sensor'], config)
load_platform(hass, 'sensor', DOMAIN,
discovery_info['sensor'], config)
load_platform(hass, 'cover', DOMAIN,
discovery_info['cover'], config)
hass.data[DOMAIN][entry.entry_id] = discovery_info
for category in COMPONENT_TYPES:
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(
entry, category))
controller.scan(callback)
def syn_clock(self, service=None):
controller.sync_clock()
controller.scan(callback)
hass.services.async_register(
DOMAIN, 'sync_clock', syn_clock,
schema=vol.Schema({}))
@ -74,6 +92,19 @@ async def async_setup(hass, config):
return True
async def async_unload_entry(hass: HomeAssistantType, entry: ConfigEntry):
"""Remove the velbus connection."""
await asyncio.wait([
hass.config_entries.async_forward_entry_unload(entry, component)
for component in COMPONENT_TYPES
])
hass.data[DOMAIN][entry.entry_id]['cntrl'].stop()
hass.data[DOMAIN].pop(entry.entry_id)
if not hass.data[DOMAIN]:
hass.data.pop(DOMAIN)
return True
class VelbusEntity(Entity):
"""Representation of a Velbus entity."""
@ -108,3 +139,21 @@ class VelbusEntity(Entity):
def _on_update(self, state):
self.schedule_update_ha_state()
@property
def device_info(self):
"""Return the device info."""
return {
'identifiers': {
(DOMAIN, self._module.get_module_address(),
self._module.serial)
},
'name': "{} {}".format(
self._module.get_module_address(),
self._module.get_module_name()),
'manufacturer': 'Velleman',
'model': self._module.get_module_name(),
'sw_version': "{}.{}-{}".format(
self._module.memory_map_version, self._module.build_year,
self._module.build_week)
}

26
homeassistant/components/velbus/binary_sensor.py
View File

@ -3,22 +3,28 @@ import logging
from homeassistant.components.binary_sensor import BinarySensorDevice
from . import DOMAIN as VELBUS_DOMAIN, VelbusEntity
from .const import DOMAIN
from . import VelbusEntity
_LOGGER = logging.getLogger(__name__)
async def async_setup_platform(
hass, config, async_add_entities, discovery_info=None):
"""Set up Velbus binary sensors."""
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_entities(sensors)
"""Old way."""
pass
async def async_setup_entry(hass, entry, async_add_entities):
"""Set up Velbus binary sensor based on config_entry."""
cntrl = hass.data[DOMAIN][entry.entry_id]['cntrl']
modules_data = hass.data[DOMAIN][entry.entry_id]['binary_sensor']
entities = []
for address, channel in modules_data:
module = cntrl.get_module(address)
entities.append(
VelbusBinarySensor(module, channel))
async_add_entities(entities)
class VelbusBinarySensor(VelbusEntity, BinarySensorDevice):

24
homeassistant/components/velbus/climate.py
View File

@ -6,24 +6,28 @@ from homeassistant.components.climate.const import (
HVAC_MODE_HEAT, SUPPORT_TARGET_TEMPERATURE)
from homeassistant.const import ATTR_TEMPERATURE, TEMP_CELSIUS, TEMP_FAHRENHEIT
from . import DOMAIN as VELBUS_DOMAIN, VelbusEntity
from .const import DOMAIN
from . import VelbusEntity
_LOGGER = logging.getLogger(__name__)
async def async_setup_platform(
hass, config, async_add_entities, discovery_info=None):
"""Set up the Velbus thermostat platform."""
if discovery_info is None:
return
"""Set up Velbus binary sensors."""
pass
sensors = []
for sensor in discovery_info:
module = hass.data[VELBUS_DOMAIN].get_module(sensor[0])
channel = sensor[1]
sensors.append(VelbusClimate(module, channel))
async_add_entities(sensors)
async def async_setup_entry(hass, entry, async_add_entities):
"""Set up Velbus binary sensor based on config_entry."""
cntrl = hass.data[DOMAIN][entry.entry_id]['cntrl']
modules_data = hass.data[DOMAIN][entry.entry_id]['climate']
entities = []
for address, channel in modules_data:
module = cntrl.get_module(address)
entities.append(
VelbusClimate(module, channel))
async_add_entities(entities)
class VelbusClimate(VelbusEntity, ClimateDevice):

93
homeassistant/components/velbus/config_flow.py Normal file
View File

@ -0,0 +1,93 @@
"""Config flow for the Velbus platform."""
import velbus
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import CONF_PORT, CONF_NAME
from homeassistant.core import HomeAssistant, callback
from homeassistant.util import slugify
from .const import DOMAIN
@callback
def velbus_entries(hass: HomeAssistant):
"""Return connections for Velbus domain."""
return set((entry.data[CONF_PORT]) for
entry in hass.config_entries.async_entries(DOMAIN))
@config_entries.HANDLERS.register(DOMAIN)
class VelbusConfigFlow(config_entries.ConfigFlow):
"""Handle a config flow."""
VERSION = 1
CONNECTION_CLASS = config_entries.CONN_CLASS_LOCAL_PUSH
def __init__(self) -> None:
"""Initialize the velbus config flow."""
self._errors = {}
def _create_device(self, name: str, prt: str):
"""Create an antry async."""
return self.async_create_entry(
title=name,
data={
CONF_PORT: prt
}
)
def _test_connection(self, prt):
"""Try to connect to the velbus with the port specified."""
try:
controller = velbus.Controller(prt)
except Exception: # pylint: disable=broad-except
self._errors[CONF_PORT] = 'connection_failed'
return False
controller.stop()
return True
def _prt_in_configuration_exists(self, prt: str) -> bool:
"""Return True if port exists in configuration."""
if prt in velbus_entries(self.hass):
return True
return False
async def async_step_user(self, user_input=None):
"""Step when user intializes a integration."""
self._errors = {}
if user_input is not None:
name = slugify(user_input[CONF_NAME])
prt = user_input[CONF_PORT]
if not self._prt_in_configuration_exists(prt):
if self._test_connection(prt):
return self._create_device(name, prt)
else:
self._errors[CONF_PORT] = 'port_exists'
else:
user_input = {}
user_input[CONF_NAME] = ''
user_input[CONF_PORT] = ''
return self.async_show_form(
step_id='user',
data_schema=vol.Schema({
vol.Required(CONF_NAME,
default=user_input[CONF_NAME]): str,
vol.Required(CONF_PORT,
default=user_input[CONF_PORT]): str
}),
errors=self._errors
)
async def async_step_import(self, user_input=None):
"""Import a config entry."""
user_input[CONF_NAME] = 'Velbus Import'
prt = user_input[CONF_PORT]
if self._prt_in_configuration_exists(prt):
# if the velbus import is already in the config
# we should not proceed the import
return self.async_abort(
reason='port_exists'
)
return await self.async_step_user(user_input)

3
homeassistant/components/velbus/const.py Normal file
View File

@ -0,0 +1,3 @@
"""Const for Velbus."""
DOMAIN = "velbus"

26
homeassistant/components/velbus/cover.py
View File

@ -4,22 +4,28 @@ import logging
from homeassistant.components.cover import (
CoverDevice, SUPPORT_CLOSE, SUPPORT_OPEN, SUPPORT_STOP)
from . import DOMAIN as VELBUS_DOMAIN, VelbusEntity
from .const import DOMAIN
from . import VelbusEntity
_LOGGER = logging.getLogger(__name__)
async def async_setup_platform(
hass, config, async_add_entities, discovery_info=None):
"""Set up the Velbus xover platform."""
if discovery_info is None:
return
covers = []
for cover in discovery_info:
module = hass.data[VELBUS_DOMAIN].get_module(cover[0])
channel = cover[1]
covers.append(VelbusCover(module, channel))
async_add_entities(covers)
"""Set up Velbus covers."""
pass
async def async_setup_entry(hass, entry, async_add_entities):
"""Set up Velbus cover based on config_entry."""
cntrl = hass.data[DOMAIN][entry.entry_id]['cntrl']
modules_data = hass.data[DOMAIN][entry.entry_id]['cover']
entities = []
for address, channel in modules_data:
module = cntrl.get_module(address)
entities.append(
VelbusCover(module, channel))
async_add_entities(entities)
class VelbusCover(VelbusEntity, CoverDevice):

3
homeassistant/components/velbus/manifest.json
View File

@ -5,6 +5,7 @@
"requirements": [
"python-velbus==2.0.27"
],
"config_flow": true,
"dependencies": [],
"codeowners": []
"codeowners": ["@ceral2nd"]
}

26
homeassistant/components/velbus/sensor.py
View File

@ -1,22 +1,28 @@
"""Support for Velbus sensors."""
import logging
from . import DOMAIN as VELBUS_DOMAIN, VelbusEntity
from .const import DOMAIN
from . import VelbusEntity
_LOGGER = logging.getLogger(__name__)
async def async_setup_platform(
hass, config, async_add_entities, discovery_info=None):
"""Set up the Velbus temp sensor platform."""
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(VelbusSensor(module, channel))
async_add_entities(sensors)
"""Old way."""
pass
async def async_setup_entry(hass, entry, async_add_entities):
"""Set up Velbus sensor based on config_entry."""
cntrl = hass.data[DOMAIN][entry.entry_id]['cntrl']
modules_data = hass.data[DOMAIN][entry.entry_id]['sensor']
entities = []
for address, channel in modules_data:
module = cntrl.get_module(address)
entities.append(
VelbusSensor(module, channel))
async_add_entities(entities)
class VelbusSensor(VelbusEntity):

20
homeassistant/components/velbus/strings.json Normal file
View File

@ -0,0 +1,20 @@
{
"config": {
"title": "Velbus interface",
"step": {
"user": {
"title": "Define the velbus connection type",
"data": {
"name": "The name for this velbus connection",
"port": "Connection string"
}
}
},
"error": {
"port_exists": "This port is already configured",
"connection_failed": "The velbus connection failed"
},
"abort": {
"port_exists": "This port is already configured"
}
}

26
homeassistant/components/velbus/switch.py
View File

@ -3,22 +3,28 @@ import logging
from homeassistant.components.switch import SwitchDevice
from . import DOMAIN as VELBUS_DOMAIN, VelbusEntity
from . import VelbusEntity
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
async def async_setup_platform(
hass, config, async_add_entities, 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_entities(switches)
"""Old way."""
pass
async def async_setup_entry(hass, entry, async_add_entities):
"""Set up Velbus switch based on config_entry."""
cntrl = hass.data[DOMAIN][entry.entry_id]['cntrl']
modules_data = hass.data[DOMAIN][entry.entry_id]['switch']
entities = []
for address, channel in modules_data:
module = cntrl.get_module(address)
entities.append(
VelbusSwitch(module, channel))
async_add_entities(entities)
class VelbusSwitch(VelbusEntity, SwitchDevice):

1
homeassistant/generated/config_flows.py
View File

@ -56,6 +56,7 @@ FLOWS = [
"twilio",
"unifi",
"upnp",
"velbus",
"vesync",
"wemo",
"wwlln",

3
requirements_test_all.txt
View File

@ -312,6 +312,9 @@ python-forecastio==1.4.0
# homeassistant.components.nest
python-nest==4.1.0
# homeassistant.components.velbus
python-velbus==2.0.27
# homeassistant.components.awair
python_awair==0.0.4

1
script/gen_requirements_all.py
View File

@ -129,6 +129,7 @@ TEST_REQUIREMENTS = (
'python-forecastio',
'python-nest',
'python_awair',
'python-velbus',
'pytradfri[async]',
'pyunifi',
'pyupnp-async',

1
tests/components/velbus/__init__.py Normal file
View File

@ -0,0 +1 @@
"""Tests for the Velbus component."""

98
tests/components/velbus/test_config_flow.py Normal file
View File

@ -0,0 +1,98 @@
"""Tests for the Velbus config flow."""
from unittest.mock import patch, Mock
import pytest
from homeassistant import data_entry_flow
from homeassistant.components.velbus import config_flow
from homeassistant.const import CONF_PORT, CONF_NAME
from tests.common import MockConfigEntry
PORT_SERIAL = '/dev/ttyACME100'
PORT_TCP = '127.0.1.0.1:3788'
@pytest.fixture(name='controller_assert')
def mock_controller_assert():
"""Mock the velbus controller with an assert."""
with patch('velbus.Controller', side_effect=Exception()):
yield
@pytest.fixture(name='controller')
def mock_controller():
"""Mock a successfull velbus controller."""
controller = Mock()
with patch('velbus.Controller', return_value=controller):
yield controller
def init_config_flow(hass):
"""Init a configuration flow."""
flow = config_flow.VelbusConfigFlow()
flow.hass = hass
return flow
async def test_user(hass, controller):
"""Test user config."""
flow = init_config_flow(hass)
result = await flow.async_step_user()
assert result['type'] == data_entry_flow.RESULT_TYPE_FORM
assert result['step_id'] == 'user'
result = await flow.async_step_user({
CONF_NAME: 'Velbus Test Serial', CONF_PORT: PORT_SERIAL})
assert result['type'] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result['title'] == 'velbus_test_serial'
assert result['data'][CONF_PORT] == PORT_SERIAL
result = await flow.async_step_user({
CONF_NAME: 'Velbus Test TCP', CONF_PORT: PORT_TCP})
assert result['type'] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result['title'] == 'velbus_test_tcp'
assert result['data'][CONF_PORT] == PORT_TCP
async def test_user_fail(hass, controller_assert):
"""Test user config."""
flow = init_config_flow(hass)
result = await flow.async_step_user({
CONF_NAME: 'Velbus Test Serial', CONF_PORT: PORT_SERIAL})
assert result['type'] == data_entry_flow.RESULT_TYPE_FORM
assert result['errors'] == {CONF_PORT: 'connection_failed'}
result = await flow.async_step_user({
CONF_NAME: 'Velbus Test TCP', CONF_PORT: PORT_TCP})
assert result['type'] == data_entry_flow.RESULT_TYPE_FORM
assert result['errors'] == {CONF_PORT: 'connection_failed'}
async def test_import(hass, controller):
"""Test import step."""
flow = init_config_flow(hass)
result = await flow.async_step_import({CONF_PORT: PORT_TCP})
assert result['type'] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result['title'] == 'velbus_import'
async def test_abort_if_already_setup(hass):
"""Test we abort if Daikin is already setup."""
flow = init_config_flow(hass)
MockConfigEntry(domain='velbus',
data={CONF_PORT: PORT_TCP,
CONF_NAME: 'velbus home'}).add_to_hass(hass)
result = await flow.async_step_import(
{CONF_PORT: PORT_TCP, CONF_NAME: 'velbus import test'})
assert result['type'] == data_entry_flow.RESULT_TYPE_ABORT
assert result['reason'] == 'port_exists'
result = await flow.async_step_user(
{CONF_PORT: PORT_TCP, CONF_NAME: 'velbus import test'})
assert result['type'] == data_entry_flow.RESULT_TYPE_FORM
assert result['errors'] == {'port': 'port_exists'}