Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
89 changes: 49 additions & 40 deletions homeassistant/components/velbus/__init__.py
Original file line number Diff line number Diff line change
@@ -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,
Expand All @@ -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:
Expand Down Expand Up @@ -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/"),
Expand All @@ -77,15 +82,15 @@ 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
raise vol.Invalid(
"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(
Expand All @@ -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(
Expand All @@ -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
Expand Down Expand Up @@ -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]
6 changes: 4 additions & 2 deletions homeassistant/components/velbus/climate.py
Original file line number Diff line number Diff line change
@@ -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,
Expand Down Expand Up @@ -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()

Expand All @@ -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
Expand Down
15 changes: 10 additions & 5 deletions homeassistant/components/velbus/config_flow.py
Original file line number Diff line number Diff line change
@@ -1,20 +1,23 @@
"""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

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)
Expand All @@ -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)
Expand All @@ -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:
Expand All @@ -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]
Expand Down
37 changes: 25 additions & 12 deletions homeassistant/components/velbus/light.py
Original file line number Diff line number Diff line change
@@ -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,
Expand All @@ -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"):
Expand All @@ -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
Expand All @@ -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",
Expand All @@ -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:
Expand All @@ -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)
Loading