diff --git a/homeassistant/components/velbus/__init__.py b/homeassistant/components/velbus/__init__.py index acc90116269f5e..a8395abdcb8fab 100644 --- a/homeassistant/components/velbus/__init__.py +++ b/homeassistant/components/velbus/__init__.py @@ -1,16 +1,20 @@ """Support for Velbus devices.""" from __future__ import annotations +from collections.abc import Callable +from dataclasses import dataclass import logging +from velbusaio.channels import Channel as VelbusChannel 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.core import HomeAssistant, ServiceCall import homeassistant.helpers.config_validation as cv -from homeassistant.helpers.entity import Entity +from homeassistant.helpers.entity import DeviceInfo, Entity, EntityDescription +from homeassistant.helpers.typing import ConfigType from .const import ( CONF_INTERFACE, @@ -30,7 +34,7 @@ PLATFORMS = ["switch", "sensor", "binary_sensor", "cover", "climate", "light"] -async def async_setup(hass, config): +async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool: """Set up the Velbus platform.""" # Import from the configuration file if needed if DOMAIN not in config: @@ -62,6 +66,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: """Establish connection with velbus.""" hass.data.setdefault(DOMAIN, {}) + # Init the velbus controller controller = Velbus( entry.data[CONF_PORT], cache_dir=hass.config.path(".storage/velbuscache/"), @@ -77,7 +82,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: if hass.services.has_service(DOMAIN, SERVICE_SCAN): return True - def check_entry_id(interface: str): + def check_entry_id(interface: str) -> str: for entry in hass.config_entries.async_entries(DOMAIN): if "port" in entry.data and entry.data["port"] == interface: return entry.entry_id @@ -85,7 +90,7 @@ def check_entry_id(interface: str): "The interface provided is not defined as a port in a Velbus integration" ) - async def scan(call): + async def scan(call: ServiceCall) -> None: await hass.data[DOMAIN][call.data[CONF_INTERFACE]]["cntrl"].scan() hass.services.async_register( @@ -95,7 +100,7 @@ async def scan(call): vol.Schema({vol.Required(CONF_INTERFACE): vol.All(cv.string, check_entry_id)}), ) - async def syn_clock(call): + async def syn_clock(call: ServiceCall) -> None: await hass.data[DOMAIN][call.data[CONF_INTERFACE]]["cntrl"].sync_clock() hass.services.async_register( @@ -105,7 +110,7 @@ async def syn_clock(call): vol.Schema({vol.Required(CONF_INTERFACE): vol.All(cv.string, check_entry_id)}), ) - async def set_memo_text(call): + async def set_memo_text(call: ServiceCall) -> None: """Handle Memo Text service call.""" memo_text = call.data[CONF_MEMO_TEXT] memo_text.hass = hass @@ -147,47 +152,51 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: class VelbusEntity(Entity): """Representation of a Velbus entity.""" - def __init__(self, channel): + _attr_should_poll = False + + def __init__( + self, + channel: VelbusChannel, + entity_description: EntityDescription | None = None, + ) -> None: """Initialize a Velbus entity.""" self._channel = channel - @property - def unique_id(self): - """Get unique ID.""" 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._channel.get_name() - - @property - def should_poll(self): - """Disable polling.""" - return False + self._attr_unique_id = f"{serial}-{self._channel.get_channel_number()}" + self._attr_name = self._channel.get_name() + + if entity_description is not None: + self.entity_description = entity_description + if entity_description.key is not None and entity_description.key != "": + self._attr_unique_id += f"-{entity_description.key}" + if entity_description.name is not None and entity_description.name != "": + self._attr_name = f"{self._attr_name}-{entity_description.name}" + + self._attr_device_info = DeviceInfo( + identifiers={ + ( + DOMAIN, + str(self._channel.get_module_address()), + ) + }, + name=self._channel.get_full_name(), + manufacturer="Velleman", + model=self._channel.get_module_type_name(), + sw_version=self._channel.get_module_sw_version(), + ) - async def async_added_to_hass(self): + async def async_added_to_hass(self) -> None: """Add listener for state changes.""" self._channel.on_status_update(self._on_update) - async def _on_update(self): + async def _on_update(self) -> None: self.async_write_ha_state() - @property - def device_info(self): - """Return the device info.""" - return { - "identifiers": { - ( - DOMAIN, - self._channel.get_module_address(), - self._channel.get_module_serial(), - ) - }, - "name": self._channel.get_full_name(), - "manufacturer": "Velleman", - "model": self._channel.get_module_type_name(), - "sw_version": self._channel.get_module_sw_version(), - } + +@dataclass +class VelbusEntityDescriptionMixin: + """Bases description for Velbus entities.""" + + suitable: Callable[[VelbusChannel], bool] diff --git a/homeassistant/components/velbus/climate.py b/homeassistant/components/velbus/climate.py index a065679c4e5f14..f4b5b842b9330f 100644 --- a/homeassistant/components/velbus/climate.py +++ b/homeassistant/components/velbus/climate.py @@ -1,6 +1,8 @@ """Support for Velbus thermostat.""" from __future__ import annotations +from typing import Any + from homeassistant.components.climate import ClimateEntity from homeassistant.components.climate.const import ( HVAC_MODE_HEAT, @@ -40,7 +42,7 @@ class VelbusClimate(VelbusEntity, ClimateEntity): _attr_preset_modes = list(PRESET_MODES) @property - def target_temperature(self) -> int | None: + def target_temperature(self) -> Any: """Return the temperature we try to reach.""" return self._channel.get_climate_target() @@ -56,7 +58,7 @@ def preset_mode(self) -> str | None: None, ) - async def async_set_temperature(self, **kwargs) -> None: + async def async_set_temperature(self, **kwargs: Any) -> None: """Set new target temperatures.""" if (temp := kwargs.get(ATTR_TEMPERATURE)) is None: return diff --git a/homeassistant/components/velbus/config_flow.py b/homeassistant/components/velbus/config_flow.py index 3ec5af14397439..2ab8a4995bb834 100644 --- a/homeassistant/components/velbus/config_flow.py +++ b/homeassistant/components/velbus/config_flow.py @@ -1,6 +1,8 @@ """Config flow for the Velbus platform.""" from __future__ import annotations +from typing import Any + import velbusaio from velbusaio.exceptions import VelbusConnectionFailed import voluptuous as vol @@ -8,13 +10,14 @@ from homeassistant import config_entries from homeassistant.const import CONF_NAME, CONF_PORT from homeassistant.core import HomeAssistant, callback +from homeassistant.data_entry_flow import FlowResult from homeassistant.util import slugify from .const import DOMAIN @callback -def velbus_entries(hass: HomeAssistant): +def velbus_entries(hass: HomeAssistant) -> set[Any]: """Return connections for Velbus domain.""" return { (entry.data[CONF_PORT]) for entry in hass.config_entries.async_entries(DOMAIN) @@ -30,11 +33,11 @@ def __init__(self) -> None: """Initialize the velbus config flow.""" self._errors: dict[str, str] = {} - def _create_device(self, name: str, prt: str): + def _create_device(self, name: str, prt: str) -> FlowResult: """Create an entry async.""" return self.async_create_entry(title=name, data={CONF_PORT: prt}) - async def _test_connection(self, prt): + async def _test_connection(self, prt: str) -> bool: """Try to connect to the velbus with the port specified.""" try: controller = velbusaio.controller.Velbus(prt) @@ -51,7 +54,9 @@ def _prt_in_configuration_exists(self, prt: str) -> bool: return True return False - async def async_step_user(self, user_input=None): + async def async_step_user( + self, user_input: dict[str, Any] | None = None + ) -> FlowResult: """Step when user initializes a integration.""" self._errors = {} if user_input is not None: @@ -78,7 +83,7 @@ async def async_step_user(self, user_input=None): errors=self._errors, ) - async def async_step_import(self, user_input=None): + async def async_step_import(self, user_input: dict[str, Any]) -> FlowResult: """Import a config entry.""" user_input[CONF_NAME] = "Velbus Import" prt = user_input[CONF_PORT] diff --git a/homeassistant/components/velbus/light.py b/homeassistant/components/velbus/light.py index 2c6cd8d87764bc..fa3bee04539182 100644 --- a/homeassistant/components/velbus/light.py +++ b/homeassistant/components/velbus/light.py @@ -1,4 +1,10 @@ """Support for Velbus light.""" +from __future__ import annotations + +from typing import Any + +from velbusaio.channels import Channel as VelbusChannel + from homeassistant.components.light import ( ATTR_BRIGHTNESS, ATTR_FLASH, @@ -10,16 +16,23 @@ SUPPORT_TRANSITION, LightEntity, ) +from homeassistant.config_entries import ConfigEntry +from homeassistant.core import HomeAssistant +from homeassistant.helpers.entity_platform import AddEntitiesCallback from . import VelbusEntity from .const import DOMAIN -async def async_setup_entry(hass, entry, async_add_entities): +async def async_setup_entry( + hass: HomeAssistant, + entry: ConfigEntry, + async_add_entities: AddEntitiesCallback, +) -> None: """Set up Velbus switch based on config_entry.""" await hass.data[DOMAIN][entry.entry_id]["tsk"] cntrl = hass.data[DOMAIN][entry.entry_id]["cntrl"] - entities = [] + entities: list[Any] = [] for channel in cntrl.get_all("light"): entities.append(VelbusLight(channel)) for channel in cntrl.get_all("led"): @@ -32,22 +45,22 @@ class VelbusLight(VelbusEntity, LightEntity): _attr_supported_feature = SUPPORT_BRIGHTNESS | SUPPORT_TRANSITION - def __init__(self, channel): + def __init__(self, channel: VelbusChannel) -> None: """Initialize the dimmer.""" super().__init__(channel) self._attr_name = self._channel.get_name() @property - def is_on(self): + def is_on(self) -> Any: """Return true if the light is on.""" return self._channel.is_on() @property - def brightness(self): + def brightness(self) -> int: """Return the brightness of the light.""" return int((self._channel.get_dimmer_state() * 255) / 100) - async def async_turn_on(self, **kwargs): + async def async_turn_on(self, **kwargs: Any) -> None: """Instruct the Velbus light to turn on.""" if ATTR_BRIGHTNESS in kwargs: # Make sure a low but non-zero value is not rounded down to zero @@ -67,7 +80,7 @@ async def async_turn_on(self, **kwargs): ) await getattr(self._channel, attr)(*args) - async def async_turn_off(self, **kwargs): + async def async_turn_off(self, **kwargs: Any) -> None: """Instruct the velbus light to turn off.""" attr, *args = ( "set_dimmer_state", @@ -83,22 +96,22 @@ class VelbusButtonLight(VelbusEntity, LightEntity): _attr_entity_registry_enabled_default = False _attr_supported_feature = SUPPORT_FLASH - def __init__(self, channel): + def __init__(self, channel: VelbusChannel) -> None: """Initialize the button light (led).""" super().__init__(channel) self._attr_name = f"LED {self._channel.get_name()}" @property - def is_on(self): + def is_on(self) -> Any: """Return true if the light is on.""" return self._channel.is_on() @property - def brightness(self): + def brightness(self) -> int: """Return the brightness of the light.""" return int((self._channel.get_dimmer_state() * 255) / 100) - async def async_turn_on(self, **kwargs): + async def async_turn_on(self, **kwargs: Any) -> None: """Instruct the Velbus light to turn on.""" if ATTR_FLASH in kwargs: if kwargs[ATTR_FLASH] == FLASH_LONG: @@ -111,7 +124,7 @@ async def async_turn_on(self, **kwargs): attr, *args = "set_led_state", "on" await getattr(self._channel, attr)(*args) - async def async_turn_off(self, **kwargs): + async def async_turn_off(self, **kwargs: Any) -> None: """Instruct the velbus light to turn off.""" attr, *args = "set_led_state", "off" await getattr(self._channel, attr)(*args) diff --git a/homeassistant/components/velbus/sensor.py b/homeassistant/components/velbus/sensor.py index 32f016b8ce333d..695f4b220a8125 100644 --- a/homeassistant/components/velbus/sensor.py +++ b/homeassistant/components/velbus/sensor.py @@ -1,92 +1,114 @@ """Support for Velbus sensors.""" from __future__ import annotations +from collections.abc import Callable +from dataclasses import dataclass +from typing import Final + +from velbusaio.channels import Channel as VelbusChannel + from homeassistant.components.sensor import ( STATE_CLASS_MEASUREMENT, STATE_CLASS_TOTAL_INCREASING, SensorEntity, + SensorEntityDescription, ) -from homeassistant.const import ( - DEVICE_CLASS_ENERGY, - DEVICE_CLASS_POWER, - DEVICE_CLASS_TEMPERATURE, -) +from homeassistant.config_entries import ConfigEntry +from homeassistant.const import DEVICE_CLASS_POWER, DEVICE_CLASS_TEMPERATURE +from homeassistant.core import HomeAssistant +from homeassistant.helpers.entity_platform import AddEntitiesCallback -from . import VelbusEntity +from . import VelbusEntity, VelbusEntityDescriptionMixin from .const import DOMAIN -async def async_setup_entry(hass, entry, async_add_entities): +@dataclass +class VelbusSensorEntityDescriptionMixin(VelbusEntityDescriptionMixin): + """Bases description for Velbus Sensor entities.""" + + native_value: Callable[[VelbusChannel], float | int | None] + + +@dataclass +class VelbusSensorEntityDescription( + SensorEntityDescription, VelbusSensorEntityDescriptionMixin +): + """Base velbus sensor entitydescription.""" + + +SENSOR_TYPES: Final[tuple[VelbusSensorEntityDescription, ...]] = ( + VelbusSensorEntityDescription( + key="", + state_class=STATE_CLASS_MEASUREMENT, + suitable=lambda channel: not channel.is_temperature(), + native_value=lambda channel: channel.get_state(), + ), + VelbusSensorEntityDescription( + key="", + device_class=DEVICE_CLASS_TEMPERATURE, + state_class=STATE_CLASS_MEASUREMENT, + suitable=lambda channel: channel.is_temperature(), + native_value=lambda channel: channel.get_state(), + ), + VelbusSensorEntityDescription( + name="max", + key="max", + device_class=DEVICE_CLASS_TEMPERATURE, + state_class=STATE_CLASS_MEASUREMENT, + suitable=lambda channel: channel.is_temperature(), + native_value=lambda channel: channel.get_max(), + entity_registry_enabled_default=False, + ), + VelbusSensorEntityDescription( + name="min", + key="min", + device_class=DEVICE_CLASS_TEMPERATURE, + state_class=STATE_CLASS_MEASUREMENT, + suitable=lambda channel: channel.is_temperature(), + native_value=lambda channel: channel.get_min(), + entity_registry_enabled_default=False, + ), + VelbusSensorEntityDescription( + name="counter", + key="counter", + device_class=DEVICE_CLASS_POWER, + state_class=STATE_CLASS_TOTAL_INCREASING, + suitable=lambda channel: channel.is_counter_channel(), + native_value=lambda channel: channel.get_counter_state(), + icon="mdi:counter", + ), +) + + +async def async_setup_entry( + hass: HomeAssistant, + entry: ConfigEntry, + async_add_entities: AddEntitiesCallback, +) -> None: """Set up Velbus switch based on config_entry.""" await hass.data[DOMAIN][entry.entry_id]["tsk"] cntrl = hass.data[DOMAIN][entry.entry_id]["cntrl"] entities = [] for channel in cntrl.get_all("sensor"): - entities.append(VelbusSensor(channel)) - if channel.is_counter_channel(): - entities.append(VelbusSensor(channel, True)) + for description in SENSOR_TYPES: + if description.suitable(channel): + entities.append(VelbusSensor(channel, description)) async_add_entities(entities) class VelbusSensor(VelbusEntity, SensorEntity): - """Representation of a sensor.""" + """The entity class for FRITZ!SmartHome sensors.""" - def __init__(self, channel, counter=False): - """Initialize a sensor Velbus entity.""" - super().__init__(channel) - self._is_counter = counter + entity_description: VelbusSensorEntityDescription - @property - def unique_id(self): - """Return unique ID for counter sensors.""" - unique_id = super().unique_id - if self._is_counter: - 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 + def __init__( + self, channel: VelbusChannel, description: VelbusSensorEntityDescription + ) -> None: + """Initialize the dimmer.""" + super().__init__(channel, description) + self._attr_native_unit_of_measurement = self._channel.get_unit() @property - def device_class(self): - """Return the device class of the sensor.""" - 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 - - @property - def native_value(self): + def native_value(self) -> float | int | None: """Return the state of the sensor.""" - if self._is_counter: - 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._channel.get_counter_unit() - return self._channel.get_unit() - - @property - def icon(self): - """Icon to use in the frontend.""" - 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 + return self.entity_description.native_value(self._channel)