evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Switch velbus from python-velbus to velbusaio (#54032)

Browse source 
* initial commit

* use new release

* Update for sensors

* big update

* pylint fixes, bump dependancy to 2021.8.2

* New version to try to fix the tests

* Fix a lot of errors, bump version

* more work

* Bump version

* Adde dimmer support

* Make sure the counters are useable in the energy dashboard

* bump version

* Fix testcases

* Update after review

* Bump version to be able to have some decent exception catches, add the temperature device class

* Readd the import of the platform from config file, but add a deprecation warning

* More comments updated

* Fix lefover index

* Fix unique id to be backwards compatible

* Fix small bug in covers

* Fix testcases

* Changes for theenery dashboard

* Fixed services

* Fix memo text

* Make the interface for a service the port string instead of the device selector

* Fix set_memo_text

* added an async scan task, more comments

* Accidently disabled some paltforms

* More comments, bump version

* Bump version, add extra attributes, enable mypy

* Removed new features

* More comments

* Bump version

* Update homeassistant/components/velbus/__init__.py

Co-authored-by: brefra <frank_van_breugel@hotmail.com>

* Readd the import step

Co-authored-by: brefra <frank_van_breugel@hotmail.com>
This commit is contained in:
Maikel Punie 2021-09-13 08:22:46 +02:00 • committed by GitHub
parent 1f997fcd58
commit 7472fb2049
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
14 changed files with 252 additions and 220 deletions
Download patch file Download diff file Expand all files Collapse all files

145
homeassistant/components/velbus/__init__.py
View file

@ -1,22 +1,28 @@
"""Support for Velbus devices."""
from __future__ import annotations
import logging
import velbus
from velbusaio.controller import Velbus
import voluptuous as vol
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
from homeassistant.const import CONF_ADDRESS, CONF_NAME, CONF_PORT
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity import Entity
from .const import CONF_MEMO_TEXT, DOMAIN, SERVICE_SET_MEMO_TEXT
from .const import (
CONF_INTERFACE,
CONF_MEMO_TEXT,
DOMAIN,
SERVICE_SCAN,
SERVICE_SET_MEMO_TEXT,
SERVICE_SYNC,
)
_LOGGER = logging.getLogger(__name__)
VELBUS_MESSAGE = "velbus.message"
CONFIG_SCHEMA = vol.Schema(
{DOMAIN: vol.Schema({vol.Required(CONF_PORT): cv.string})}, extra=vol.ALLOW_EXTRA
)
@ -29,6 +35,9 @@ async def async_setup(hass, config):
# Import from the configuration file if needed
if DOMAIN not in config:
return True
_LOGGER.warning("Loading VELBUS via configuration.yaml is deprecated")
port = config[DOMAIN].get(CONF_PORT)
data = {}
@ -39,57 +48,67 @@ async def async_setup(hass, config):
DOMAIN, context={"source": SOURCE_IMPORT}, data=data
)
)
return True
async def velbus_connect_task(
controller: Velbus, hass: HomeAssistant, entry_id: str
) -> None:
"""Task to offload the long running connect."""
await controller.connect()
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Establish connection with velbus."""
hass.data.setdefault(DOMAIN, {})
def callback():
modules = controller.get_modules()
discovery_info = {"cntrl": controller}
for platform in PLATFORMS:
discovery_info[platform] = []
for module in modules:
for channel in range(1, module.number_of_channels() + 1):
for platform in PLATFORMS:
if platform in module.get_categories(channel):
discovery_info[platform].append(
(module.get_module_address(), channel)
controller = Velbus(entry.data[CONF_PORT])
hass.data[DOMAIN][entry.entry_id] = {}
hass.data[DOMAIN][entry.entry_id]["cntrl"] = controller
hass.data[DOMAIN][entry.entry_id]["tsk"] = hass.async_create_task(
velbus_connect_task(controller, hass, entry.entry_id)
)
hass.data[DOMAIN][entry.entry_id] = discovery_info
for platform in PLATFORMS:
hass.add_job(hass.config_entries.async_forward_entry_setup(entry, platform))
hass.config_entries.async_setup_platforms(entry, PLATFORMS)
try:
controller = velbus.Controller(entry.data[CONF_PORT])
controller.scan(callback)
except velbus.util.VelbusException as err:
_LOGGER.error("An error occurred: %s", err)
raise ConfigEntryNotReady from err
if hass.services.has_service(DOMAIN, SERVICE_SCAN):
return True
def syn_clock(self, service=None):
try:
controller.sync_clock()
except velbus.util.VelbusException as err:
_LOGGER.error("An error occurred: %s", err)
def check_entry_id(interface: str):
for entry in hass.config_entries.async_entries(DOMAIN):
if "port" in entry.data and entry.data["port"] == interface:
return entry.entry_id
raise vol.Invalid(
"The interface provided is not defined as a port in a Velbus integration"
)
hass.services.async_register(DOMAIN, "sync_clock", syn_clock, schema=vol.Schema({}))
async def scan(call):
await hass.data[DOMAIN][call.data[CONF_INTERFACE]]["cntrl"].scan()
def set_memo_text(service):
hass.services.async_register(
DOMAIN,
SERVICE_SCAN,
scan,
vol.Schema({vol.Required(CONF_INTERFACE): vol.All(cv.string, check_entry_id)}),
)
async def syn_clock(call):
await hass.data[DOMAIN][call.data[CONF_INTERFACE]]["cntrl"].sync_clock()
hass.services.async_register(
DOMAIN,
SERVICE_SYNC,
syn_clock,
vol.Schema({vol.Required(CONF_INTERFACE): vol.All(cv.string, check_entry_id)}),
)
async def set_memo_text(call):
"""Handle Memo Text service call."""
module_address = service.data[CONF_ADDRESS]
memo_text = service.data[CONF_MEMO_TEXT]
memo_text = call.data[CONF_MEMO_TEXT]
memo_text.hass = hass
try:
controller.get_module(module_address).set_memo_text(
memo_text.async_render()
)
except velbus.util.VelbusException as err:
_LOGGER.error("An error occurred while setting memo text: %s", err)
await hass.data[DOMAIN][call.data[CONF_INTERFACE]]["cntrl"].get_module(
call.data[CONF_ADDRESS]
).set_memo_text(memo_text.async_render())
hass.services.async_register(
DOMAIN,
@ -97,6 +116,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
set_memo_text,
vol.Schema(
{
vol.Required(CONF_INTERFACE): vol.All(cv.string, check_entry_id),
vol.Required(CONF_ADDRESS): vol.All(
vol.Coerce(int), vol.Range(min=0, max=255)
),
@ -111,35 +131,34 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry):
"""Remove the velbus connection."""
unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
hass.data[DOMAIN][entry.entry_id]["cntrl"].stop()
await hass.data[DOMAIN][entry.entry_id]["cntrl"].stop()
hass.data[DOMAIN].pop(entry.entry_id)
if not hass.data[DOMAIN]:
hass.data.pop(DOMAIN)
hass.services.async_remove(DOMAIN, SERVICE_SCAN)
hass.services.async_remove(DOMAIN, SERVICE_SYNC)
hass.services.async_remove(DOMAIN, SERVICE_SET_MEMO_TEXT)
return unload_ok
class VelbusEntity(Entity):
"""Representation of a Velbus entity."""
def __init__(self, module, channel):
def __init__(self, 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 f"{serial}-{self._channel}"
if (serial := self._channel.get_module_serial()) == 0:
serial = self._channel.get_module_address()
return f"{serial}-{self._channel.get_channel_number()}"
@property
def name(self):
"""Return the display name of this entity."""
return self._module.get_name(self._channel)
return self._channel.get_name()
@property
def should_poll(self):
@ -148,26 +167,24 @@ class VelbusEntity(Entity):
async def async_added_to_hass(self):
"""Add listener for state changes."""
self._module.on_status_update(self._channel, self._on_update)
self._channel.on_status_update(self._on_update)
def _on_update(self, state):
self.schedule_update_ha_state()
async def _on_update(self):
self.async_write_ha_state()
@property
def device_info(self):
"""Return the device info."""
return {
"identifiers": {
(DOMAIN, self._module.get_module_address(), self._module.serial)
(
DOMAIN,
self._channel.get_module_address(),
self._channel.get_module_serial(),
)
},
"name": "{} ({})".format(
self._module.get_module_name(), self._module.get_module_address()
),
"name": self._channel.get_full_name(),
"manufacturer": "Velleman",
"model": self._module.get_module_type_name(),
"sw_version": "{}.{}-{}".format(
self._module.memory_map_version,
self._module.build_year,
self._module.build_week,
),
"model": self._channel.get_module_type_name(),
"sw_version": self._channel.get_module_sw_version(),
}

13
homeassistant/components/velbus/binary_sensor.py
View file

@ -6,13 +6,12 @@ from .const import DOMAIN
async def async_setup_entry(hass, entry, async_add_entities):
"""Set up Velbus binary sensor based on config_entry."""
"""Set up Velbus switch based on config_entry."""
await hass.data[DOMAIN][entry.entry_id]["tsk"]
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))
for channel in cntrl.get_all("binary_sensor"):
entities.append(VelbusBinarySensor(channel))
async_add_entities(entities)
@ -20,6 +19,6 @@ class VelbusBinarySensor(VelbusEntity, BinarySensorEntity):
"""Representation of a Velbus Binary Sensor."""
@property
def is_on(self):
def is_on(self) -> bool:
"""Return true if the sensor is on."""
return self._module.is_closed(self._channel)
return self._channel.is_closed()

27
homeassistant/components/velbus/climate.py
View file

@ -1,14 +1,12 @@
"""Support for Velbus thermostat."""
import logging
from velbus.util import VelbusException
from homeassistant.components.climate import ClimateEntity
from homeassistant.components.climate.const import (
HVAC_MODE_HEAT,
SUPPORT_TARGET_TEMPERATURE,
)
from homeassistant.const import ATTR_TEMPERATURE, TEMP_CELSIUS, TEMP_FAHRENHEIT
from homeassistant.const import ATTR_TEMPERATURE, TEMP_CELSIUS
from . import VelbusEntity
from .const import DOMAIN
@ -17,13 +15,12 @@ _LOGGER = logging.getLogger(__name__)
async def async_setup_entry(hass, entry, async_add_entities):
"""Set up Velbus binary sensor based on config_entry."""
"""Set up Velbus switch based on config_entry."""
await hass.data[DOMAIN][entry.entry_id]["tsk"]
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))
for channel in cntrl.get_all("climate"):
entities.append(VelbusClimate(channel))
async_add_entities(entities)
@ -37,15 +34,13 @@ class VelbusClimate(VelbusEntity, ClimateEntity):
@property
def temperature_unit(self):
"""Return the unit this state is expressed in."""
if self._module.get_unit(self._channel) == TEMP_CELSIUS:
"""Return the unit."""
return TEMP_CELSIUS
return TEMP_FAHRENHEIT
@property
def current_temperature(self):
"""Return the current temperature."""
return self._module.get_state(self._channel)
return self._channel.get_state()
@property
def hvac_mode(self):
@ -66,18 +61,14 @@ class VelbusClimate(VelbusEntity, ClimateEntity):
@property
def target_temperature(self):
"""Return the temperature we try to reach."""
return self._module.get_climate_target()
return self._channel.get_climate_target()
def set_temperature(self, **kwargs):
"""Set new target temperatures."""
temp = kwargs.get(ATTR_TEMPERATURE)
if temp is None:
return
try:
self._module.set_temp(temp)
except VelbusException as err:
_LOGGER.error("A Velbus error occurred: %s", err)
return
self._channel.set_temp(temp)
self.schedule_update_ha_state()
def set_hvac_mode(self, hvac_mode):

14
homeassistant/components/velbus/config_flow.py
View file

@ -1,7 +1,8 @@
"""Config flow for the Velbus platform."""
from __future__ import annotations
import velbus
import velbusaio
from velbusaio.exceptions import VelbusConnectionFailed
import voluptuous as vol
from homeassistant import config_entries
@ -33,14 +34,15 @@ class VelbusConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Create an entry async."""
return self.async_create_entry(title=name, data={CONF_PORT: prt})
def _test_connection(self, prt):
async 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
controller = velbusaio.controller.Velbus(prt)
await controller.connect(True)
await controller.stop()
except VelbusConnectionFailed:
self._errors[CONF_PORT] = "cannot_connect"
return False
controller.stop()
return True
def _prt_in_configuration_exists(self, prt: str) -> bool:
@ -56,7 +58,7 @@ class VelbusConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
name = slugify(user_input[CONF_NAME])
prt = user_input[CONF_PORT]
if not self._prt_in_configuration_exists(prt):
if self._test_connection(prt):
if await self._test_connection(prt):
return self._create_device(name, prt)
else:
self._errors[CONF_PORT] = "already_configured"

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

@ -2,6 +2,9 @@
DOMAIN = "velbus"
CONF_INTERFACE = "interface"
CONF_MEMO_TEXT = "memo_text"
SERVICE_SCAN = "scan"
SERVICE_SYNC = "sync_clock"
SERVICE_SET_MEMO_TEXT = "set_memo_text"

47
homeassistant/components/velbus/cover.py
View file

@ -1,8 +1,6 @@
"""Support for Velbus covers."""
import logging
from velbus.util import VelbusException
from homeassistant.components.cover import (
ATTR_POSITION,
SUPPORT_CLOSE,
@ -19,13 +17,12 @@ _LOGGER = logging.getLogger(__name__)
async def async_setup_entry(hass, entry, async_add_entities):
"""Set up Velbus cover based on config_entry."""
"""Set up Velbus switch based on config_entry."""
await hass.data[DOMAIN][entry.entry_id]["tsk"]
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))
for channel in cntrl.get_all("cover"):
entities.append(VelbusCover(channel))
async_add_entities(entities)
@ -35,16 +32,14 @@ class VelbusCover(VelbusEntity, CoverEntity):
@property
def supported_features(self):
"""Flag supported features."""
if self._module.support_position():
if self._channel.support_position():
return SUPPORT_OPEN | SUPPORT_CLOSE | SUPPORT_STOP | SUPPORT_SET_POSITION
return SUPPORT_OPEN | SUPPORT_CLOSE | SUPPORT_STOP
@property
def is_closed(self):
"""Return if the cover is closed."""
if self._module.get_position(self._channel) == 100:
return True
return False
return self._channel.is_closed()
@property
def current_cover_position(self):
@ -53,33 +48,21 @@ class VelbusCover(VelbusEntity, CoverEntity):
None is unknown, 0 is closed, 100 is fully open
Velbus: 100 = closed, 0 = open
"""
pos = self._module.get_position(self._channel)
pos = self._channel.get_position()
return 100 - pos
def open_cover(self, **kwargs):
async def async_open_cover(self, **kwargs):
"""Open the cover."""
try:
self._module.open(self._channel)
except VelbusException as err:
_LOGGER.error("A Velbus error occurred: %s", err)
await self._channel.open()
def close_cover(self, **kwargs):
async def async_close_cover(self, **kwargs):
"""Close the cover."""
try:
self._module.close(self._channel)
except VelbusException as err:
_LOGGER.error("A Velbus error occurred: %s", err)
await self._channel.close()
def stop_cover(self, **kwargs):
async def async_stop_cover(self, **kwargs):
"""Stop the cover."""
try:
self._module.stop(self._channel)
except VelbusException as err:
_LOGGER.error("A Velbus error occurred: %s", err)
await self._channel.stop()
def set_cover_position(self, **kwargs):
async def async_set_cover_position(self, **kwargs):
"""Move the cover to a specific position."""
try:
self._module.set(self._channel, (100 - kwargs[ATTR_POSITION]))
except VelbusException as err:
_LOGGER.error("A Velbus error occurred: %s", err)
self._channel.set_position(100 - kwargs[ATTR_POSITION])

68
homeassistant/components/velbus/light.py
View file

@ -1,8 +1,6 @@
"""Support for Velbus light."""
import logging
from velbus.util import VelbusException
from homeassistant.components.light import (
ATTR_BRIGHTNESS,
ATTR_FLASH,
@ -22,62 +20,61 @@ _LOGGER = logging.getLogger(__name__)
async def async_setup_entry(hass, entry, async_add_entities):
"""Set up Velbus light based on config_entry."""
"""Set up Velbus switch based on config_entry."""
await hass.data[DOMAIN][entry.entry_id]["tsk"]
cntrl = hass.data[DOMAIN][entry.entry_id]["cntrl"]
modules_data = hass.data[DOMAIN][entry.entry_id]["light"]
entities = []
for address, channel in modules_data:
module = cntrl.get_module(address)
entities.append(VelbusLight(module, channel))
for channel in cntrl.get_all("light"):
entities.append(VelbusLight(channel, False))
for channel in cntrl.get_all("led"):
entities.append(VelbusLight(channel, True))
async_add_entities(entities)
class VelbusLight(VelbusEntity, LightEntity):
"""Representation of a Velbus light."""
def __init__(self, channel, led):
"""Initialize a light Velbus entity."""
super().__init__(channel)
self._is_led = led
@property
def name(self):
"""Return the display name of this entity."""
if self._module.light_is_buttonled(self._channel):
return f"LED {self._module.get_name(self._channel)}"
return self._module.get_name(self._channel)
if self._is_led:
return f"LED {self._channel.get_name()}"
return self._channel.get_name()
@property
def supported_features(self):
"""Flag supported features."""
if self._module.light_is_buttonled(self._channel):
if self._is_led:
return SUPPORT_FLASH
return SUPPORT_BRIGHTNESS | SUPPORT_TRANSITION
@property
def entity_registry_enabled_default(self):
"""Disable Button LEDs by default."""
if self._module.light_is_buttonled(self._channel):
return False
return True
@property
def is_on(self):
"""Return true if the light is on."""
return self._module.is_on(self._channel)
return self._channel.is_on()
@property
def brightness(self):
"""Return the brightness of the light."""
return int((self._module.get_dimmer_state(self._channel) * 255) / 100)
return int((self._channel.get_dimmer_state() * 255) / 100)
def turn_on(self, **kwargs):
async def async_turn_on(self, **kwargs):
"""Instruct the Velbus light to turn on."""
if self._module.light_is_buttonled(self._channel):
if self._is_led:
if ATTR_FLASH in kwargs:
if kwargs[ATTR_FLASH] == FLASH_LONG:
attr, *args = "set_led_state", self._channel, "slow"
attr, *args = "set_led_state", "slow"
elif kwargs[ATTR_FLASH] == FLASH_SHORT:
attr, *args = "set_led_state", self._channel, "fast"
attr, *args = "set_led_state", "fast"
else:
attr, *args = "set_led_state", self._channel, "on"
attr, *args = "set_led_state", "on"
else:
attr, *args = "set_led_state", self._channel, "on"
attr, *args = "set_led_state", "on"
else:
if ATTR_BRIGHTNESS in kwargs:
# Make sure a low but non-zero value is not rounded down to zero
@ -87,33 +84,24 @@ class VelbusLight(VelbusEntity, LightEntity):
brightness = max(int((kwargs[ATTR_BRIGHTNESS] * 100) / 255), 1)
attr, *args = (
"set_dimmer_state",
self._channel,
brightness,
kwargs.get(ATTR_TRANSITION, 0),
)
else:
attr, *args = (
"restore_dimmer_state",
self._channel,
kwargs.get(ATTR_TRANSITION, 0),
)
try:
getattr(self._module, attr)(*args)
except VelbusException as err:
_LOGGER.error("A Velbus error occurred: %s", err)
await getattr(self._channel, attr)(*args)
def turn_off(self, **kwargs):
async def async_turn_off(self, **kwargs):
"""Instruct the velbus light to turn off."""
if self._module.light_is_buttonled(self._channel):
attr, *args = "set_led_state", self._channel, "off"
if self._is_led:
attr, *args = "set_led_state", "off"
else:
attr, *args = (
"set_dimmer_state",
self._channel,
0,
kwargs.get(ATTR_TRANSITION, 0),
)
try:
getattr(self._module, attr)(*args)
except VelbusException as err:
_LOGGER.error("A Velbus error occurred: %s", err)
await getattr(self._channel, attr)(*args)

2
homeassistant/components/velbus/manifest.json
View file

@ -2,7 +2,7 @@
"domain": "velbus",
"name": "Velbus",
"documentation": "https://www.home-assistant.io/integrations/velbus",
"requirements": ["python-velbus==2.1.2"],
"requirements": ["velbus-aio==2021.9.1"],
"config_flow": true,
"codeowners": ["@Cereal2nd", "@brefra"],
"iot_class": "local_push"

62
homeassistant/components/velbus/sensor.py
View file

@ -1,30 +1,39 @@
"""Support for Velbus sensors."""
from homeassistant.components.sensor import SensorEntity
from homeassistant.const import DEVICE_CLASS_POWER, ENERGY_KILO_WATT_HOUR
from __future__ import annotations
from homeassistant.components.sensor import (
STATE_CLASS_MEASUREMENT,
STATE_CLASS_TOTAL_INCREASING,
SensorEntity,
)
from homeassistant.const import (
DEVICE_CLASS_ENERGY,
DEVICE_CLASS_POWER,
DEVICE_CLASS_TEMPERATURE,
)
from . import VelbusEntity
from .const import DOMAIN
async def async_setup_entry(hass, entry, async_add_entities):
"""Set up Velbus sensor based on config_entry."""
"""Set up Velbus switch based on config_entry."""
await hass.data[DOMAIN][entry.entry_id]["tsk"]
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))
if module.get_class(channel) == "counter":
entities.append(VelbusSensor(module, channel, True))
for channel in cntrl.get_all("sensor"):
entities.append(VelbusSensor(channel))
if channel.is_counter_channel():
entities.append(VelbusSensor(channel, True))
async_add_entities(entities)
class VelbusSensor(VelbusEntity, SensorEntity):
"""Representation of a sensor."""
def __init__(self, module, channel, counter=False):
def __init__(self, channel, counter=False):
"""Initialize a sensor Velbus entity."""
super().__init__(module, channel)
super().__init__(channel)
self._is_counter = counter
@property
@ -35,28 +44,38 @@ class VelbusSensor(VelbusEntity, SensorEntity):
unique_id = f"{unique_id}-counter"
return unique_id
@property
def name(self):
"""Return the name for the sensor."""
name = super().name
if self._is_counter:
name = f"{name}-counter"
return name
@property
def device_class(self):
"""Return the device class of the sensor."""
if self._module.get_class(self._channel) == "counter" and not self._is_counter:
if self._module.get_counter_unit(self._channel) == ENERGY_KILO_WATT_HOUR:
if self._is_counter:
return DEVICE_CLASS_ENERGY
if self._channel.is_counter_channel():
return DEVICE_CLASS_POWER
if self._channel.is_temperature():
return DEVICE_CLASS_TEMPERATURE
return None
return self._module.get_class(self._channel)
@property
def native_value(self):
"""Return the state of the sensor."""
if self._is_counter:
return self._module.get_counter_state(self._channel)
return self._module.get_state(self._channel)
return self._channel.get_counter_state()
return self._channel.get_state()
@property
def native_unit_of_measurement(self):
"""Return the unit this state is expressed in."""
if self._is_counter:
return self._module.get_counter_unit(self._channel)
return self._module.get_unit(self._channel)
return self._channel.get_counter_unit()
return self._channel.get_unit()
@property
def icon(self):
@ -64,3 +83,10 @@ class VelbusSensor(VelbusEntity, SensorEntity):
if self._is_counter:
return "mdi:counter"
return None
@property
def state_class(self):
"""Return the state class of this device."""
if self._is_counter:
return STATE_CLASS_TOTAL_INCREASING
return STATE_CLASS_MEASUREMENT

34
homeassistant/components/velbus/services.yaml
View file

@ -1,6 +1,28 @@
sync_clock:
name: Sync clock
description: Sync the velbus modules clock to the Home Assistant clock, this is the same as the 'sync clock' from VelbusLink
fields:
interface:
name: Interface
description: The velbus interface to send the command to, this will be the same value as used during configuration
required: true
example: "192.168.1.5:27015"
default: ''
selector:
text:
scan:
name: Scan
description: Scan the velbus modules, this will be need if you see unknown module warnings in the logs, or when you added new modules
fields:
interface:
name: Interface
description: The velbus interface to send the command to, this will be the same value as used during configuration
required: true
example: "192.168.1.5:27015"
default: ''
selector:
text:
set_memo_text:
name: Set memo text
@ -8,6 +30,14 @@ set_memo_text:
Set the memo text to the display of modules like VMBGPO, VMBGPOD
Be sure the page(s) of the module is configured to display the memo text.
fields:
interface:
name: Interface
description: The velbus interface to send the command to, this will be the same value as used during configuration
required: true
example: "192.168.1.5:27015"
default: ''
selector:
text:
address:
name: Address
description: >
@ -16,8 +46,8 @@ set_memo_text:
required: true
selector:
number:
min: 0
max: 255
min: 1
max: 254
memo_text:
name: Memo text
description: >

28
homeassistant/components/velbus/switch.py
View file

@ -1,7 +1,6 @@
"""Support for Velbus switches."""
import logging
from velbus.util import VelbusException
from typing import Any
from homeassistant.components.switch import SwitchEntity
@ -13,12 +12,11 @@ _LOGGER = logging.getLogger(__name__)
async def async_setup_entry(hass, entry, async_add_entities):
"""Set up Velbus switch based on config_entry."""
await hass.data[DOMAIN][entry.entry_id]["tsk"]
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))
for channel in cntrl.get_all("switch"):
entities.append(VelbusSwitch(channel))
async_add_entities(entities)
@ -26,20 +24,14 @@ class VelbusSwitch(VelbusEntity, SwitchEntity):
"""Representation of a switch."""
@property
def is_on(self):
def is_on(self) -> bool:
"""Return true if the switch is on."""
return self._module.is_on(self._channel)
return self._channel.is_on()
def turn_on(self, **kwargs):
async def async_turn_on(self, **kwargs: Any) -> None:
"""Instruct the switch to turn on."""
try:
self._module.turn_on(self._channel)
except VelbusException as err:
_LOGGER.error("A Velbus error occurred: %s", err)
await self._channel.turn_on()
def turn_off(self, **kwargs):
async def async_turn_off(self, **kwargs: Any) -> None:
"""Instruct the switch to turn off."""
try:
self._module.turn_off(self._channel)
except VelbusException as err:
_LOGGER.error("A Velbus error occurred: %s", err)
await self._channel.turn_off()

6
requirements_all.txt
View file

@ -1924,9 +1924,6 @@ python-telnet-vlc==2.0.1
# homeassistant.components.twitch
python-twitch-client==0.6.0
# homeassistant.components.velbus
python-velbus==2.1.2
# homeassistant.components.vlc
python-vlc==1.1.2
@ -2350,6 +2347,9 @@ uvcclient==0.11.0
# homeassistant.components.vallox
vallox-websocket-api==2.8.1
# homeassistant.components.velbus
velbus-aio==2021.9.1
# homeassistant.components.venstar
venstarcolortouch==0.14

6
requirements_test_all.txt
View file

@ -1090,9 +1090,6 @@ python-tado==0.10.0
# homeassistant.components.twitch
python-twitch-client==0.6.0
# homeassistant.components.velbus
python-velbus==2.1.2
# homeassistant.components.awair
python_awair==0.2.1
@ -1312,6 +1309,9 @@ url-normalize==1.4.1
# homeassistant.components.uvc
uvcclient==0.11.0
# homeassistant.components.velbus
velbus-aio==2021.9.1
# homeassistant.components.venstar
venstarcolortouch==0.14

9
tests/components/velbus/test_config_flow.py
View file

@ -1,7 +1,8 @@
"""Tests for the Velbus config flow."""
from unittest.mock import Mock, patch
from unittest.mock import AsyncMock, patch
import pytest
from velbusaio.exceptions import VelbusConnectionFailed
from homeassistant import data_entry_flow
from homeassistant.components.velbus import config_flow
@ -16,15 +17,15 @@ 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()):
with patch("velbusaio.controller.Velbus", side_effect=VelbusConnectionFailed()):
yield
@pytest.fixture(name="controller")
def mock_controller():
"""Mock a successful velbus controller."""
controller = Mock()
with patch("velbus.Controller", return_value=controller):
controller = AsyncMock()
with patch("velbusaio.controller.Velbus", return_value=controller):
yield controller