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

Add type hints to LCN (#52509)

Browse source 
* Add type hints to LCN

* Fix requested review changes
This commit is contained in:
Andre Lengwenus 2021-07-06 09:54:35 +02:00 • committed by GitHub
parent b496469a2f
commit e16ef10af5
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
13 changed files with 333 additions and 172 deletions
Download patch file Download diff file Expand all files Collapse all files

1
.strict-typing
View file

@ -48,6 +48,7 @@ homeassistant.components.image_processing.*
homeassistant.components.integration.*
homeassistant.components.knx.*
homeassistant.components.kraken.*
homeassistant.components.lcn.*
homeassistant.components.light.*
homeassistant.components.local_ip.*
homeassistant.components.lock.*

41
homeassistant/components/lcn/__init__.py
View file

@ -1,5 +1,8 @@
"""Support for LCN devices."""
from __future__ import annotations
import logging
from typing import Callable
import pypck
@ -14,16 +17,22 @@ from homeassistant.const import (
)
from homeassistant.helpers import entity_registry as er
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.typing import ConfigType, HomeAssistantType
from .const import CONF_DIM_MODE, CONF_SK_NUM_TRIES, CONNECTION, DOMAIN, PLATFORMS
from .helpers import generate_unique_id, import_lcn_config
from .helpers import (
DeviceConnectionType,
InputType,
generate_unique_id,
import_lcn_config,
)
from .schemas import CONFIG_SCHEMA # noqa: F401
from .services import SERVICES
_LOGGER = logging.getLogger(__name__)
async def async_setup(hass, config):
async def async_setup(hass: HomeAssistantType, config: ConfigType) -> bool:
"""Set up the LCN component."""
if DOMAIN not in config:
return True
@ -43,7 +52,9 @@ async def async_setup(hass, config):
return True
async def async_setup_entry(hass, config_entry):
async def async_setup_entry(
hass: HomeAssistantType, config_entry: config_entries.ConfigEntry
) -> bool:
"""Set up a connection to PCHK host from a config entry."""
hass.data.setdefault(DOMAIN, {})
if config_entry.entry_id in hass.data[DOMAIN]:
@ -104,7 +115,9 @@ async def async_setup_entry(hass, config_entry):
return True
async def async_unload_entry(hass, config_entry):
async def async_unload_entry(
hass: HomeAssistantType, config_entry: config_entries.ConfigEntry
) -> bool:
"""Close connection to PCHK host represented by config_entry."""
# forward unloading to platforms
unload_ok = await hass.config_entries.async_unload_platforms(
@ -126,16 +139,18 @@ async def async_unload_entry(hass, config_entry):
class LcnEntity(Entity):
"""Parent class for all entities associated with the LCN component."""
def __init__(self, config, entry_id, device_connection):
def __init__(
self, config: ConfigType, entry_id: str, device_connection: DeviceConnectionType
) -> None:
"""Initialize the LCN device."""
self.config = config
self.entry_id = entry_id
self.device_connection = device_connection
self._unregister_for_inputs = None
self._name = config[CONF_NAME]
self._unregister_for_inputs: Callable | None = None
self._name: str = config[CONF_NAME]
@property
def unique_id(self):
def unique_id(self) -> str:
"""Return a unique ID."""
unique_device_id = generate_unique_id(
(
@ -147,26 +162,26 @@ class LcnEntity(Entity):
return f"{self.entry_id}-{unique_device_id}-{self.config[CONF_RESOURCE]}"
@property
def should_poll(self):
def should_poll(self) -> bool:
"""Lcn device entity pushes its state to HA."""
return False
async def async_added_to_hass(self):
async def async_added_to_hass(self) -> None:
"""Run when entity about to be added to hass."""
if not self.device_connection.is_group:
self._unregister_for_inputs = self.device_connection.register_for_inputs(
self.input_received
)
async def async_will_remove_from_hass(self):
async def async_will_remove_from_hass(self) -> None:
"""Run when entity will be removed from hass."""
if self._unregister_for_inputs is not None:
self._unregister_for_inputs()
@property
def name(self):
def name(self) -> str:
"""Return the name of the device."""
return self._name
def input_received(self, input_obj):
def input_received(self, input_obj: InputType) -> None:
"""Set state/value when LCN input object (command) is received."""

55
homeassistant/components/lcn/binary_sensor.py
View file

@ -1,21 +1,28 @@
"""Support for LCN binary sensors."""
from __future__ import annotations
import pypck
from homeassistant.components.binary_sensor import (
DOMAIN as DOMAIN_BINARY_SENSOR,
BinarySensorEntity,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_ADDRESS, CONF_DOMAIN, CONF_ENTITIES, CONF_SOURCE
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, HomeAssistantType
from . import LcnEntity
from .const import BINSENSOR_PORTS, CONF_DOMAIN_DATA, SETPOINTS
from .helpers import get_device_connection
from .helpers import DeviceConnectionType, InputType, get_device_connection
def create_lcn_binary_sensor_entity(hass, entity_config, config_entry):
def create_lcn_binary_sensor_entity(
hass: HomeAssistantType, entity_config: ConfigType, config_entry: ConfigEntry
) -> LcnEntity:
"""Set up an entity for this domain."""
device_connection = get_device_connection(
hass, tuple(entity_config[CONF_ADDRESS]), config_entry
hass, entity_config[CONF_ADDRESS], config_entry
)
if entity_config[CONF_DOMAIN_DATA][CONF_SOURCE] in SETPOINTS:
@ -28,7 +35,11 @@ def create_lcn_binary_sensor_entity(hass, entity_config, config_entry):
return LcnLockKeysSensor(entity_config, config_entry.entry_id, device_connection)
async def async_setup_entry(hass, config_entry, async_add_entities):
async def async_setup_entry(
hass: HomeAssistantType,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up LCN switch entities from a config entry."""
entities = []
@ -44,7 +55,9 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
class LcnRegulatorLockSensor(LcnEntity, BinarySensorEntity):
"""Representation of a LCN binary sensor for regulator locks."""
def __init__(self, config, entry_id, device_connection):
def __init__(
self, config: ConfigType, entry_id: str, device_connection: DeviceConnectionType
) -> None:
"""Initialize the LCN binary sensor."""
super().__init__(config, entry_id, device_connection)
@ -54,7 +67,7 @@ class LcnRegulatorLockSensor(LcnEntity, BinarySensorEntity):
self._value = None
async def async_added_to_hass(self):
async def async_added_to_hass(self) -> None:
"""Run when entity about to be added to hass."""
await super().async_added_to_hass()
if not self.device_connection.is_group:
@ -62,7 +75,7 @@ class LcnRegulatorLockSensor(LcnEntity, BinarySensorEntity):
self.setpoint_variable
)
async def async_will_remove_from_hass(self):
async def async_will_remove_from_hass(self) -> None:
"""Run when entity will be removed from hass."""
await super().async_will_remove_from_hass()
if not self.device_connection.is_group:
@ -71,11 +84,11 @@ class LcnRegulatorLockSensor(LcnEntity, BinarySensorEntity):
)
@property
def is_on(self):
def is_on(self) -> bool | None:
"""Return true if the binary sensor is on."""
return self._value
def input_received(self, input_obj):
def input_received(self, input_obj: InputType) -> None:
"""Set sensor value when LCN input object (command) is received."""
if (
not isinstance(input_obj, pypck.inputs.ModStatusVar)
@ -90,7 +103,9 @@ class LcnRegulatorLockSensor(LcnEntity, BinarySensorEntity):
class LcnBinarySensor(LcnEntity, BinarySensorEntity):
"""Representation of a LCN binary sensor for binary sensor ports."""
def __init__(self, config, entry_id, device_connection):
def __init__(
self, config: ConfigType, entry_id: str, device_connection: DeviceConnectionType
) -> None:
"""Initialize the LCN binary sensor."""
super().__init__(config, entry_id, device_connection)
@ -100,7 +115,7 @@ class LcnBinarySensor(LcnEntity, BinarySensorEntity):
self._value = None
async def async_added_to_hass(self):
async def async_added_to_hass(self) -> None:
"""Run when entity about to be added to hass."""
await super().async_added_to_hass()
if not self.device_connection.is_group:
@ -108,7 +123,7 @@ class LcnBinarySensor(LcnEntity, BinarySensorEntity):
self.bin_sensor_port
)
async def async_will_remove_from_hass(self):
async def async_will_remove_from_hass(self) -> None:
"""Run when entity will be removed from hass."""
await super().async_will_remove_from_hass()
if not self.device_connection.is_group:
@ -117,11 +132,11 @@ class LcnBinarySensor(LcnEntity, BinarySensorEntity):
)
@property
def is_on(self):
def is_on(self) -> bool | None:
"""Return true if the binary sensor is on."""
return self._value
def input_received(self, input_obj):
def input_received(self, input_obj: InputType) -> None:
"""Set sensor value when LCN input object (command) is received."""
if not isinstance(input_obj, pypck.inputs.ModStatusBinSensors):
return
@ -133,31 +148,33 @@ class LcnBinarySensor(LcnEntity, BinarySensorEntity):
class LcnLockKeysSensor(LcnEntity, BinarySensorEntity):
"""Representation of a LCN sensor for key locks."""
def __init__(self, config, entry_id, device_connection):
def __init__(
self, config: ConfigType, entry_id: str, device_connection: DeviceConnectionType
) -> None:
"""Initialize the LCN sensor."""
super().__init__(config, entry_id, device_connection)
self.source = pypck.lcn_defs.Key[config[CONF_DOMAIN_DATA][CONF_SOURCE]]
self._value = None
async def async_added_to_hass(self):
async def async_added_to_hass(self) -> None:
"""Run when entity about to be added to hass."""
await super().async_added_to_hass()
if not self.device_connection.is_group:
await self.device_connection.activate_status_request_handler(self.source)
async def async_will_remove_from_hass(self):
async def async_will_remove_from_hass(self) -> None:
"""Run when entity will be removed from hass."""
await super().async_will_remove_from_hass()
if not self.device_connection.is_group:
await self.device_connection.cancel_status_request_handler(self.source)
@property
def is_on(self):
def is_on(self) -> bool | None:
"""Return true if the binary sensor is on."""
return self._value
def input_received(self, input_obj):
def input_received(self, input_obj: InputType) -> None:
"""Set sensor value when LCN input object (command) is received."""
if (
not isinstance(input_obj, pypck.inputs.ModStatusKeyLocks)

57
homeassistant/components/lcn/climate.py
View file

@ -1,4 +1,8 @@
"""Support for LCN climate control."""
from __future__ import annotations
from typing import Any, cast
import pypck
from homeassistant.components.climate import (
@ -6,6 +10,7 @@ from homeassistant.components.climate import (
ClimateEntity,
const,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
ATTR_TEMPERATURE,
CONF_ADDRESS,
@ -14,6 +19,8 @@ from homeassistant.const import (
CONF_SOURCE,
CONF_UNIT_OF_MEASUREMENT,
)
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, HomeAssistantType
from . import LcnEntity
from .const import (
@ -23,21 +30,27 @@ from .const import (
CONF_MIN_TEMP,
CONF_SETPOINT,
)
from .helpers import get_device_connection
from .helpers import DeviceConnectionType, InputType, get_device_connection
PARALLEL_UPDATES = 0
def create_lcn_climate_entity(hass, entity_config, config_entry):
def create_lcn_climate_entity(
hass: HomeAssistantType, entity_config: ConfigType, config_entry: ConfigEntry
) -> LcnEntity:
"""Set up an entity for this domain."""
device_connection = get_device_connection(
hass, tuple(entity_config[CONF_ADDRESS]), config_entry
hass, entity_config[CONF_ADDRESS], config_entry
)
return LcnClimate(entity_config, config_entry.entry_id, device_connection)
async def async_setup_entry(hass, config_entry, async_add_entities):
async def async_setup_entry(
hass: HomeAssistantType,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up LCN switch entities from a config entry."""
entities = []
@ -53,7 +66,9 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
class LcnClimate(LcnEntity, ClimateEntity):
"""Representation of a LCN climate device."""
def __init__(self, config, entry_id, device_connection):
def __init__(
self, config: ConfigType, entry_id: str, device_connection: DeviceConnectionType
) -> None:
"""Initialize of a LCN climate device."""
super().__init__(config, entry_id, device_connection)
@ -72,14 +87,14 @@ class LcnClimate(LcnEntity, ClimateEntity):
self._target_temperature = None
self._is_on = True
async def async_added_to_hass(self):
async def async_added_to_hass(self) -> None:
"""Run when entity about to be added to hass."""
await super().async_added_to_hass()
if not self.device_connection.is_group:
await self.device_connection.activate_status_request_handler(self.variable)
await self.device_connection.activate_status_request_handler(self.setpoint)
async def async_will_remove_from_hass(self):
async def async_will_remove_from_hass(self) -> None:
"""Run when entity will be removed from hass."""
await super().async_will_remove_from_hass()
if not self.device_connection.is_group:
@ -87,27 +102,27 @@ class LcnClimate(LcnEntity, ClimateEntity):
await self.device_connection.cancel_status_request_handler(self.setpoint)
@property
def supported_features(self):
def supported_features(self) -> int:
"""Return the list of supported features."""
return const.SUPPORT_TARGET_TEMPERATURE
@property
def temperature_unit(self):
def temperature_unit(self) -> str:
"""Return the unit of measurement."""
return self.unit.value
return cast(str, self.unit.value)
@property
def current_temperature(self):
def current_temperature(self) -> float | None:
"""Return the current temperature."""
return self._current_temperature
@property
def target_temperature(self):
def target_temperature(self) -> float | None:
"""Return the temperature we try to reach."""
return self._target_temperature
@property
def hvac_mode(self):
def hvac_mode(self) -> str:
"""Return hvac operation ie. heat, cool mode.
Need to be one of HVAC_MODE_*.
@ -117,7 +132,7 @@ class LcnClimate(LcnEntity, ClimateEntity):
return const.HVAC_MODE_OFF
@property
def hvac_modes(self):
def hvac_modes(self) -> list[str]:
"""Return the list of available hvac operation modes.
Need to be a subset of HVAC_MODES.
@ -128,16 +143,16 @@ class LcnClimate(LcnEntity, ClimateEntity):
return modes
@property
def max_temp(self):
def max_temp(self) -> float:
"""Return the maximum temperature."""
return self._max_temp
return cast(float, self._max_temp)
@property
def min_temp(self):
def min_temp(self) -> float:
"""Return the minimum temperature."""
return self._min_temp
return cast(float, self._min_temp)
async def async_set_hvac_mode(self, hvac_mode):
async def async_set_hvac_mode(self, hvac_mode: str) -> None:
"""Set new target hvac mode."""
if hvac_mode == const.HVAC_MODE_HEAT:
if not await self.device_connection.lock_regulator(
@ -153,7 +168,7 @@ class LcnClimate(LcnEntity, ClimateEntity):
self._target_temperature = None
self.async_write_ha_state()
async def async_set_temperature(self, **kwargs):
async def async_set_temperature(self, **kwargs: Any) -> None:
"""Set new target temperature."""
temperature = kwargs.get(ATTR_TEMPERATURE)
if temperature is None:
@ -166,7 +181,7 @@ class LcnClimate(LcnEntity, ClimateEntity):
self._target_temperature = temperature
self.async_write_ha_state()
def input_received(self, input_obj):
def input_received(self, input_obj: InputType) -> None:
"""Set temperature value when LCN input object is received."""
if not isinstance(input_obj, pypck.inputs.ModStatusVar):
return

12
homeassistant/components/lcn/config_flow.py
View file

@ -1,4 +1,6 @@
"""Config flow to configure the LCN integration."""
from __future__ import annotations
import logging
import pypck
@ -11,13 +13,17 @@ from homeassistant.const import (
CONF_PORT,
CONF_USERNAME,
)
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers.typing import ConfigType, HomeAssistantType
from .const import CONF_DIM_MODE, CONF_SK_NUM_TRIES, DOMAIN
_LOGGER = logging.getLogger(__name__)
def get_config_entry(hass, data):
def get_config_entry(
hass: HomeAssistantType, data: ConfigType
) -> config_entries.ConfigEntry | None:
"""Check config entries for already configured entries based on the ip address/port."""
return next(
(
@ -30,7 +36,7 @@ def get_config_entry(hass, data):
)
async def validate_connection(host_name, data):
async def validate_connection(host_name: str, data: ConfigType) -> ConfigType:
"""Validate if a connection to LCN can be established."""
host = data[CONF_IP_ADDRESS]
port = data[CONF_PORT]
@ -62,7 +68,7 @@ class LcnFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
VERSION = 1
async def async_step_import(self, data):
async def async_step_import(self, data: ConfigType) -> FlowResult:
"""Import existing configuration from LCN."""
host_name = data[CONF_HOST]
# validate the imported connection parameters

68
homeassistant/components/lcn/cover.py
View file

@ -1,21 +1,29 @@
"""Support for LCN covers."""
from __future__ import annotations
from typing import Any
import pypck
from homeassistant.components.cover import DOMAIN as DOMAIN_COVER, CoverEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_ADDRESS, CONF_DOMAIN, CONF_ENTITIES
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, HomeAssistantType
from . import LcnEntity
from .const import CONF_DOMAIN_DATA, CONF_MOTOR, CONF_REVERSE_TIME
from .helpers import get_device_connection
from .helpers import DeviceConnectionType, InputType, get_device_connection
PARALLEL_UPDATES = 0
def create_lcn_cover_entity(hass, entity_config, config_entry):
def create_lcn_cover_entity(
hass: HomeAssistantType, entity_config: ConfigType, config_entry: ConfigEntry
) -> LcnEntity:
"""Set up an entity for this domain."""
device_connection = get_device_connection(
hass, tuple(entity_config[CONF_ADDRESS]), config_entry
hass, entity_config[CONF_ADDRESS], config_entry
)
if entity_config[CONF_DOMAIN_DATA][CONF_MOTOR] in "OUTPUTS":
@ -24,7 +32,11 @@ def create_lcn_cover_entity(hass, entity_config, config_entry):
return LcnRelayCover(entity_config, config_entry.entry_id, device_connection)
async def async_setup_entry(hass, config_entry, async_add_entities):
async def async_setup_entry(
hass: HomeAssistantType,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up LCN cover entities from a config entry."""
entities = []
@ -38,7 +50,9 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
class LcnOutputsCover(LcnEntity, CoverEntity):
"""Representation of a LCN cover connected to output ports."""
def __init__(self, config, entry_id, device_connection):
def __init__(
self, config: ConfigType, entry_id: str, device_connection: DeviceConnectionType
) -> None:
"""Initialize the LCN cover."""
super().__init__(config, entry_id, device_connection)
@ -57,7 +71,7 @@ class LcnOutputsCover(LcnEntity, CoverEntity):
self._is_closing = False
self._is_opening = False
async def async_added_to_hass(self):
async def async_added_to_hass(self) -> None:
"""Run when entity about to be added to hass."""
await super().async_added_to_hass()
if not self.device_connection.is_group:
@ -68,7 +82,7 @@ class LcnOutputsCover(LcnEntity, CoverEntity):
pypck.lcn_defs.OutputPort["OUTPUTDOWN"]
)
async def async_will_remove_from_hass(self):
async def async_will_remove_from_hass(self) -> None:
"""Run when entity will be removed from hass."""
await super().async_will_remove_from_hass()
if not self.device_connection.is_group:
@ -80,26 +94,26 @@ class LcnOutputsCover(LcnEntity, CoverEntity):
)
@property
def is_closed(self):
def is_closed(self) -> bool:
"""Return if the cover is closed."""
return self._is_closed
@property
def is_opening(self):
def is_opening(self) -> bool:
"""Return if the cover is opening or not."""
return self._is_opening
@property
def is_closing(self):
def is_closing(self) -> bool:
"""Return if the cover is closing or not."""
return self._is_closing
@property
def assumed_state(self):
def assumed_state(self) -> bool:
"""Return True if unable to access real state of the entity."""
return True
async def async_close_cover(self, **kwargs):
async def async_close_cover(self, **kwargs: Any) -> None:
"""Close the cover."""
state = pypck.lcn_defs.MotorStateModifier.DOWN
if not await self.device_connection.control_motors_outputs(
@ -110,7 +124,7 @@ class LcnOutputsCover(LcnEntity, CoverEntity):
self._is_closing = True
self.async_write_ha_state()
async def async_open_cover(self, **kwargs):
async def async_open_cover(self, **kwargs: Any) -> None:
"""Open the cover."""
state = pypck.lcn_defs.MotorStateModifier.UP
if not await self.device_connection.control_motors_outputs(
@ -122,7 +136,7 @@ class LcnOutputsCover(LcnEntity, CoverEntity):
self._is_closing = False
self.async_write_ha_state()
async def async_stop_cover(self, **kwargs):
async def async_stop_cover(self, **kwargs: Any) -> None:
"""Stop the cover."""
state = pypck.lcn_defs.MotorStateModifier.STOP
if not await self.device_connection.control_motors_outputs(state):
@ -131,7 +145,7 @@ class LcnOutputsCover(LcnEntity, CoverEntity):
self._is_opening = False
self.async_write_ha_state()
def input_received(self, input_obj):
def input_received(self, input_obj: InputType) -> None:
"""Set cover states when LCN input object (command) is received."""
if (
not isinstance(input_obj, pypck.inputs.ModStatusOutput)
@ -159,7 +173,9 @@ class LcnOutputsCover(LcnEntity, CoverEntity):
class LcnRelayCover(LcnEntity, CoverEntity):
"""Representation of a LCN cover connected to relays."""
def __init__(self, config, entry_id, device_connection):
def __init__(
self, config: ConfigType, entry_id: str, device_connection: DeviceConnectionType
) -> None:
"""Initialize the LCN cover."""
super().__init__(config, entry_id, device_connection)
@ -171,39 +187,39 @@ class LcnRelayCover(LcnEntity, CoverEntity):
self._is_closing = False
self._is_opening = False
async def async_added_to_hass(self):
async def async_added_to_hass(self) -> None:
"""Run when entity about to be added to hass."""
await super().async_added_to_hass()
if not self.device_connection.is_group:
await self.device_connection.activate_status_request_handler(self.motor)
async def async_will_remove_from_hass(self):
async def async_will_remove_from_hass(self) -> None:
"""Run when entity will be removed from hass."""
await super().async_will_remove_from_hass()
if not self.device_connection.is_group:
await self.device_connection.cancel_status_request_handler(self.motor)
@property
def is_closed(self):
def is_closed(self) -> bool:
"""Return if the cover is closed."""
return self._is_closed
@property
def is_opening(self):
def is_opening(self) -> bool:
"""Return if the cover is opening or not."""
return self._is_opening
@property
def is_closing(self):
def is_closing(self) -> bool:
"""Return if the cover is closing or not."""
return self._is_closing
@property
def assumed_state(self):
def assumed_state(self) -> bool:
"""Return True if unable to access real state of the entity."""
return True
async def async_close_cover(self, **kwargs):
async def async_close_cover(self, **kwargs: Any) -> None:
"""Close the cover."""
states = [pypck.lcn_defs.MotorStateModifier.NOCHANGE] * 4
states[self.motor.value] = pypck.lcn_defs.MotorStateModifier.DOWN
@ -213,7 +229,7 @@ class LcnRelayCover(LcnEntity, CoverEntity):
self._is_closing = True
self.async_write_ha_state()
async def async_open_cover(self, **kwargs):
async def async_open_cover(self, **kwargs: Any) -> None:
"""Open the cover."""
states = [pypck.lcn_defs.MotorStateModifier.NOCHANGE] * 4
states[self.motor.value] = pypck.lcn_defs.MotorStateModifier.UP
@ -224,7 +240,7 @@ class LcnRelayCover(LcnEntity, CoverEntity):
self._is_closing = False
self.async_write_ha_state()
async def async_stop_cover(self, **kwargs):
async def async_stop_cover(self, **kwargs: Any) -> None:
"""Stop the cover."""
states = [pypck.lcn_defs.MotorStateModifier.NOCHANGE] * 4
states[self.motor.value] = pypck.lcn_defs.MotorStateModifier.STOP
@ -234,7 +250,7 @@ class LcnRelayCover(LcnEntity, CoverEntity):
self._is_opening = False
self.async_write_ha_state()
def input_received(self, input_obj):
def input_received(self, input_obj: InputType) -> None:
"""Set cover states when LCN input object (command) is received."""
if not isinstance(input_obj, pypck.inputs.ModStatusRelays):
return

34
homeassistant/components/lcn/helpers.py
View file

@ -1,9 +1,13 @@
"""Helpers for LCN component."""
from __future__ import annotations
import re
from typing import Tuple, Type, Union, cast
import pypck
import voluptuous as vol
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
CONF_ADDRESS,
CONF_BINARY_SENSORS,
@ -21,6 +25,7 @@ from homeassistant.const import (
CONF_SWITCHES,
CONF_USERNAME,
)
from homeassistant.helpers.typing import ConfigType, HomeAssistantType
from .const import (
CONF_CLIMATES,
@ -38,6 +43,13 @@ from .const import (
DOMAIN,
)
# typing
AddressType = Tuple[int, int, bool]
DeviceConnectionType = Union[
pypck.module.ModuleConnection, pypck.module.GroupConnection
]
InputType = Type[pypck.inputs.Input]
# Regex for address validation
PATTERN_ADDRESS = re.compile(
"^((?P<conn_id>\\w+)\\.)?s?(?P<seg_id>\\d+)\\.(?P<type>m|g)?(?P<id>\\d+)$"
@ -55,21 +67,23 @@ DOMAIN_LOOKUP = {
}
def get_device_connection(hass, address, config_entry):
def get_device_connection(
hass: HomeAssistantType, address: AddressType, config_entry: ConfigEntry
) -> DeviceConnectionType | None:
"""Return a lcn device_connection."""
host_connection = hass.data[DOMAIN][config_entry.entry_id][CONNECTION]
addr = pypck.lcn_addr.LcnAddr(*address)
return host_connection.get_address_conn(addr)
def get_resource(domain_name, domain_data):
def get_resource(domain_name: str, domain_data: ConfigType) -> str:
"""Return the resource for the specified domain_data."""
if domain_name in ["switch", "light"]:
return domain_data["output"]
return cast(str, domain_data["output"])
if domain_name in ["binary_sensor", "sensor"]:
return domain_data["source"]
return cast(str, domain_data["source"])
if domain_name == "cover":
return domain_data["motor"]
return cast(str, domain_data["motor"])
if domain_name == "climate":
return f'{domain_data["source"]}.{domain_data["setpoint"]}'
if domain_name == "scene":
@ -77,13 +91,13 @@ def get_resource(domain_name, domain_data):
raise ValueError("Unknown domain")
def generate_unique_id(address):
def generate_unique_id(address: AddressType) -> str:
"""Generate a unique_id from the given parameters."""
is_group = "g" if address[2] else "m"
return f"{is_group}{address[0]:03d}{address[1]:03d}"
def import_lcn_config(lcn_config):
def import_lcn_config(lcn_config: ConfigType) -> list[ConfigType]:
"""Convert lcn settings from configuration.yaml to config_entries data.
Create a list of config_entry data structures like:
@ -185,7 +199,7 @@ def import_lcn_config(lcn_config):
return list(data.values())
def has_unique_host_names(hosts):
def has_unique_host_names(hosts: list[ConfigType]) -> list[ConfigType]:
"""Validate that all connection names are unique.
Use 'pchk' as default connection_name (or add a numeric suffix if
@ -206,7 +220,7 @@ def has_unique_host_names(hosts):
return hosts
def is_address(value):
def is_address(value: str) -> tuple[AddressType, str]:
"""Validate the given address string.
Examples for S000M005 at myhome:
@ -227,7 +241,7 @@ def is_address(value):
raise ValueError(f"{value} is not a valid address string")
def is_states_string(states_string):
def is_states_string(states_string: str) -> list[str]:
"""Validate the given states string and return states list."""
if len(states_string) != 8:
raise ValueError("Invalid length of states string")

58
homeassistant/components/lcn/light.py
View file

@ -1,4 +1,7 @@
"""Support for LCN lights."""
from __future__ import annotations
from typing import Any
import pypck
@ -10,7 +13,10 @@ from homeassistant.components.light import (
SUPPORT_TRANSITION,
LightEntity,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_ADDRESS, CONF_DOMAIN, CONF_ENTITIES
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, HomeAssistantType
from . import LcnEntity
from .const import (
@ -20,15 +26,17 @@ from .const import (
CONF_TRANSITION,
OUTPUT_PORTS,
)
from .helpers import get_device_connection
from .helpers import DeviceConnectionType, InputType, get_device_connection
PARALLEL_UPDATES = 0
def create_lcn_light_entity(hass, entity_config, config_entry):
def create_lcn_light_entity(
hass: HomeAssistantType, entity_config: ConfigType, config_entry: ConfigEntry
) -> LcnEntity:
"""Set up an entity for this domain."""
device_connection = get_device_connection(
hass, tuple(entity_config[CONF_ADDRESS]), config_entry
hass, entity_config[CONF_ADDRESS], config_entry
)
if entity_config[CONF_DOMAIN_DATA][CONF_OUTPUT] in OUTPUT_PORTS:
@ -37,7 +45,11 @@ def create_lcn_light_entity(hass, entity_config, config_entry):
return LcnRelayLight(entity_config, config_entry.entry_id, device_connection)
async def async_setup_entry(hass, config_entry, async_add_entities):
async def async_setup_entry(
hass: HomeAssistantType,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up LCN light entities from a config entry."""
entities = []
@ -51,7 +63,9 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
class LcnOutputLight(LcnEntity, LightEntity):
"""Representation of a LCN light for output ports."""
def __init__(self, config, entry_id, device_connection):
def __init__(
self, config: ConfigType, entry_id: str, device_connection: DeviceConnectionType
) -> None:
"""Initialize the LCN light."""
super().__init__(config, entry_id, device_connection)
@ -66,36 +80,36 @@ class LcnOutputLight(LcnEntity, LightEntity):
self._is_on = False
self._is_dimming_to_zero = False
async def async_added_to_hass(self):
async def async_added_to_hass(self) -> None:
"""Run when entity about to be added to hass."""
await super().async_added_to_hass()
if not self.device_connection.is_group:
await self.device_connection.activate_status_request_handler(self.output)
async def async_will_remove_from_hass(self):
async def async_will_remove_from_hass(self) -> None:
"""Run when entity will be removed from hass."""
await super().async_will_remove_from_hass()
if not self.device_connection.is_group:
await self.device_connection.cancel_status_request_handler(self.output)
@property
def supported_features(self):
def supported_features(self) -> int:
"""Flag supported features."""
if self.dimmable:
return SUPPORT_TRANSITION | SUPPORT_BRIGHTNESS
return SUPPORT_TRANSITION
@property
def brightness(self):
def brightness(self) -> int | None:
"""Return the brightness of this light between 0..255."""
return self._brightness
@property
def is_on(self):
def is_on(self) -> bool:
"""Return True if entity is on."""
return self._is_on
async def async_turn_on(self, **kwargs):
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn the entity on."""
if ATTR_BRIGHTNESS in kwargs:
percent = int(kwargs[ATTR_BRIGHTNESS] / 255.0 * 100)
@ -116,7 +130,7 @@ class LcnOutputLight(LcnEntity, LightEntity):
self._is_dimming_to_zero = False
self.async_write_ha_state()
async def async_turn_off(self, **kwargs):
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn the entity off."""
if ATTR_TRANSITION in kwargs:
transition = pypck.lcn_defs.time_to_ramp_value(
@ -133,7 +147,7 @@ class LcnOutputLight(LcnEntity, LightEntity):
self._is_on = False
self.async_write_ha_state()
def input_received(self, input_obj):
def input_received(self, input_obj: InputType) -> None:
"""Set light state when LCN input object (command) is received."""
if (
not isinstance(input_obj, pypck.inputs.ModStatusOutput)
@ -144,7 +158,7 @@ class LcnOutputLight(LcnEntity, LightEntity):
self._brightness = int(input_obj.get_percent() / 100.0 * 255)
if self.brightness == 0:
self._is_dimming_to_zero = False
if not self._is_dimming_to_zero:
if not self._is_dimming_to_zero and self.brightness is not None:
self._is_on = self.brightness > 0
self.async_write_ha_state()
@ -152,7 +166,9 @@ class LcnOutputLight(LcnEntity, LightEntity):
class LcnRelayLight(LcnEntity, LightEntity):
"""Representation of a LCN light for relay ports."""
def __init__(self, config, entry_id, device_connection):
def __init__(
self, config: ConfigType, entry_id: str, device_connection: DeviceConnectionType
) -> None:
"""Initialize the LCN light."""
super().__init__(config, entry_id, device_connection)
@ -160,24 +176,24 @@ class LcnRelayLight(LcnEntity, LightEntity):
self._is_on = False
async def async_added_to_hass(self):
async def async_added_to_hass(self) -> None:
"""Run when entity about to be added to hass."""
await super().async_added_to_hass()
if not self.device_connection.is_group:
await self.device_connection.activate_status_request_handler(self.output)
async def async_will_remove_from_hass(self):
async def async_will_remove_from_hass(self) -> None:
"""Run when entity will be removed from hass."""
await super().async_will_remove_from_hass()
if not self.device_connection.is_group:
await self.device_connection.cancel_status_request_handler(self.output)
@property
def is_on(self):
def is_on(self) -> bool:
"""Return True if entity is on."""
return self._is_on
async def async_turn_on(self, **kwargs):
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn the entity on."""
states = [pypck.lcn_defs.RelayStateModifier.NOCHANGE] * 8
states[self.output.value] = pypck.lcn_defs.RelayStateModifier.ON
@ -186,7 +202,7 @@ class LcnRelayLight(LcnEntity, LightEntity):
self._is_on = True
self.async_write_ha_state()
async def async_turn_off(self, **kwargs):
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn the entity off."""
states = [pypck.lcn_defs.RelayStateModifier.NOCHANGE] * 8
states[self.output.value] = pypck.lcn_defs.RelayStateModifier.OFF
@ -195,7 +211,7 @@ class LcnRelayLight(LcnEntity, LightEntity):
self._is_on = False
self.async_write_ha_state()
def input_received(self, input_obj):
def input_received(self, input_obj: InputType) -> None:
"""Set light state when LCN input object (command) is received."""
if not isinstance(input_obj, pypck.inputs.ModStatusRelays):
return

26
homeassistant/components/lcn/scene.py
View file

@ -1,9 +1,15 @@
"""Support for LCN scenes."""
from __future__ import annotations
from typing import Any
import pypck
from homeassistant.components.scene import DOMAIN as DOMAIN_SCENE, Scene
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_ADDRESS, CONF_DOMAIN, CONF_ENTITIES, CONF_SCENE
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, HomeAssistantType
from . import LcnEntity
from .const import (
@ -13,21 +19,27 @@ from .const import (
CONF_TRANSITION,
OUTPUT_PORTS,
)
from .helpers import get_device_connection
from .helpers import DeviceConnectionType, get_device_connection
PARALLEL_UPDATES = 0
def create_lcn_scene_entity(hass, entity_config, config_entry):
def create_lcn_scene_entity(
hass: HomeAssistantType, entity_config: ConfigType, config_entry: ConfigEntry
) -> LcnEntity:
"""Set up an entity for this domain."""
device_connection = get_device_connection(
hass, tuple(entity_config[CONF_ADDRESS]), config_entry
hass, entity_config[CONF_ADDRESS], config_entry
)
return LcnScene(entity_config, config_entry.entry_id, device_connection)
async def async_setup_entry(hass, config_entry, async_add_entities):
async def async_setup_entry(
hass: HomeAssistantType,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up LCN switch entities from a config entry."""
entities = []
@ -41,7 +53,9 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
class LcnScene(LcnEntity, Scene):
"""Representation of a LCN scene."""
def __init__(self, config, entry_id, device_connection):
def __init__(
self, config: ConfigType, entry_id: str, device_connection: DeviceConnectionType
) -> None:
"""Initialize the LCN scene."""
super().__init__(config, entry_id, device_connection)
@ -63,7 +77,7 @@ class LcnScene(LcnEntity, Scene):
config[CONF_DOMAIN_DATA][CONF_TRANSITION]
)
async def async_activate(self, **kwargs):
async def async_activate(self, **kwargs: Any) -> None:
"""Activate scene."""
await self.device_connection.activate_scene(
self.register_id,

46
homeassistant/components/lcn/sensor.py
View file

@ -1,8 +1,10 @@
"""Support for LCN sensors."""
from __future__ import annotations
import pypck
from homeassistant.components.sensor import DOMAIN as DOMAIN_SENSOR, SensorEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
CONF_ADDRESS,
CONF_DOMAIN,
@ -10,6 +12,8 @@ from homeassistant.const import (
CONF_SOURCE,
CONF_UNIT_OF_MEASUREMENT,
)
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, HomeAssistantType
from . import LcnEntity
from .const import (
@ -20,13 +24,15 @@ from .const import (
THRESHOLDS,
VARIABLES,
)
from .helpers import get_device_connection
from .helpers import DeviceConnectionType, InputType, get_device_connection
def create_lcn_sensor_entity(hass, entity_config, config_entry):
def create_lcn_sensor_entity(
hass: HomeAssistantType, entity_config: ConfigType, config_entry: ConfigEntry
) -> LcnEntity:
"""Set up an entity for this domain."""
device_connection = get_device_connection(
hass, tuple(entity_config[CONF_ADDRESS]), config_entry
hass, entity_config[CONF_ADDRESS], config_entry
)
if (
@ -40,7 +46,11 @@ def create_lcn_sensor_entity(hass, entity_config, config_entry):
return LcnLedLogicSensor(entity_config, config_entry.entry_id, device_connection)
async def async_setup_entry(hass, config_entry, async_add_entities):
async def async_setup_entry(
hass: HomeAssistantType,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up LCN switch entities from a config entry."""
entities = []
@ -54,7 +64,9 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
class LcnVariableSensor(LcnEntity, SensorEntity):
"""Representation of a LCN sensor for variables."""
def __init__(self, config, entry_id, device_connection):
def __init__(
self, config: ConfigType, entry_id: str, device_connection: DeviceConnectionType
) -> None:
"""Initialize the LCN sensor."""
super().__init__(config, entry_id, device_connection)
@ -65,29 +77,29 @@ class LcnVariableSensor(LcnEntity, SensorEntity):
self._value = None
async def async_added_to_hass(self):
async def async_added_to_hass(self) -> None:
"""Run when entity about to be added to hass."""
await super().async_added_to_hass()
if not self.device_connection.is_group:
await self.device_connection.activate_status_request_handler(self.variable)
async def async_will_remove_from_hass(self):
async def async_will_remove_from_hass(self) -> None:
"""Run when entity will be removed from hass."""
await super().async_will_remove_from_hass()
if not self.device_connection.is_group:
await self.device_connection.cancel_status_request_handler(self.variable)
@property
def state(self):
def state(self) -> str | None:
"""Return the state of the entity."""
return self._value
@property
def unit_of_measurement(self):
def unit_of_measurement(self) -> str:
"""Return the unit of measurement of this entity, if any."""
return self.unit.value
return str(self.unit.value)
def input_received(self, input_obj):
def input_received(self, input_obj: InputType) -> None:
"""Set sensor value when LCN input object (command) is received."""
if (
not isinstance(input_obj, pypck.inputs.ModStatusVar)
@ -102,7 +114,9 @@ class LcnVariableSensor(LcnEntity, SensorEntity):
class LcnLedLogicSensor(LcnEntity, SensorEntity):
"""Representation of a LCN sensor for leds and logicops."""
def __init__(self, config, entry_id, device_connection):
def __init__(
self, config: ConfigType, entry_id: str, device_connection: DeviceConnectionType
) -> None:
"""Initialize the LCN sensor."""
super().__init__(config, entry_id, device_connection)
@ -115,24 +129,24 @@ class LcnLedLogicSensor(LcnEntity, SensorEntity):
self._value = None
async def async_added_to_hass(self):
async def async_added_to_hass(self) -> None:
"""Run when entity about to be added to hass."""
await super().async_added_to_hass()
if not self.device_connection.is_group:
await self.device_connection.activate_status_request_handler(self.source)
async def async_will_remove_from_hass(self):
async def async_will_remove_from_hass(self) -> None:
"""Run when entity will be removed from hass."""
await super().async_will_remove_from_hass()
if not self.device_connection.is_group:
await self.device_connection.cancel_status_request_handler(self.source)
@property
def state(self):
def state(self) -> str | None:
"""Return the state of the entity."""
return self._value
def input_received(self, input_obj):
def input_received(self, input_obj: InputType) -> None:
"""Set sensor value when LCN input object (command) is received."""
if not isinstance(input_obj, pypck.inputs.ModStatusLedsAndLogicOps):
return

40
homeassistant/components/lcn/services.py
View file

@ -12,6 +12,7 @@ from homeassistant.const import (
TIME_SECONDS,
)
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.typing import HomeAssistantType, ServiceCallType
from .const import (
CONF_KEYS,
@ -40,7 +41,12 @@ from .const import (
VAR_UNITS,
VARIABLES,
)
from .helpers import get_device_connection, is_address, is_states_string
from .helpers import (
DeviceConnectionType,
get_device_connection,
is_address,
is_states_string,
)
class LcnServiceCall:
@ -48,11 +54,11 @@ class LcnServiceCall:
schema = vol.Schema({vol.Required(CONF_ADDRESS): is_address})
def __init__(self, hass):
def __init__(self, hass: HomeAssistantType) -> None:
"""Initialize service call."""
self.hass = hass
def get_device_connection(self, service):
def get_device_connection(self, service: ServiceCallType) -> DeviceConnectionType:
"""Get address connection object."""
address, host_name = service.data[CONF_ADDRESS]
@ -66,7 +72,7 @@ class LcnServiceCall:
return device_connection
raise ValueError("Invalid host name.")
async def async_call_service(self, service):
async def async_call_service(self, service: ServiceCallType) -> None:
"""Execute service call."""
raise NotImplementedError
@ -86,7 +92,7 @@ class OutputAbs(LcnServiceCall):
}
)
async def async_call_service(self, service):
async def async_call_service(self, service: ServiceCallType) -> None:
"""Execute service call."""
output = pypck.lcn_defs.OutputPort[service.data[CONF_OUTPUT]]
brightness = service.data[CONF_BRIGHTNESS]
@ -110,7 +116,7 @@ class OutputRel(LcnServiceCall):
}
)
async def async_call_service(self, service):
async def async_call_service(self, service: ServiceCallType) -> None:
"""Execute service call."""
output = pypck.lcn_defs.OutputPort[service.data[CONF_OUTPUT]]
brightness = service.data[CONF_BRIGHTNESS]
@ -131,7 +137,7 @@ class OutputToggle(LcnServiceCall):
}
)
async def async_call_service(self, service):
async def async_call_service(self, service: ServiceCallType) -> None:
"""Execute service call."""
output = pypck.lcn_defs.OutputPort[service.data[CONF_OUTPUT]]
transition = pypck.lcn_defs.time_to_ramp_value(
@ -147,7 +153,7 @@ class Relays(LcnServiceCall):
schema = LcnServiceCall.schema.extend({vol.Required(CONF_STATE): is_states_string})
async def async_call_service(self, service):
async def async_call_service(self, service: ServiceCallType) -> None:
"""Execute service call."""
states = [
pypck.lcn_defs.RelayStateModifier[state]
@ -168,7 +174,7 @@ class Led(LcnServiceCall):
}
)
async def async_call_service(self, service):
async def async_call_service(self, service: ServiceCallType) -> None:
"""Execute service call."""
led = pypck.lcn_defs.LedPort[service.data[CONF_LED]]
led_state = pypck.lcn_defs.LedStatus[service.data[CONF_STATE]]
@ -196,7 +202,7 @@ class VarAbs(LcnServiceCall):
}
)
async def async_call_service(self, service):
async def async_call_service(self, service: ServiceCallType) -> None:
"""Execute service call."""
var = pypck.lcn_defs.Var[service.data[CONF_VARIABLE]]
value = service.data[CONF_VALUE]
@ -213,7 +219,7 @@ class VarReset(LcnServiceCall):
{vol.Required(CONF_VARIABLE): vol.All(vol.Upper, vol.In(VARIABLES + SETPOINTS))}
)
async def async_call_service(self, service):
async def async_call_service(self, service: ServiceCallType) -> None:
"""Execute service call."""
var = pypck.lcn_defs.Var[service.data[CONF_VARIABLE]]
@ -239,7 +245,7 @@ class VarRel(LcnServiceCall):
}
)
async def async_call_service(self, service):
async def async_call_service(self, service: ServiceCallType) -> None:
"""Execute service call."""
var = pypck.lcn_defs.Var[service.data[CONF_VARIABLE]]
value = service.data[CONF_VALUE]
@ -260,7 +266,7 @@ class LockRegulator(LcnServiceCall):
}
)
async def async_call_service(self, service):
async def async_call_service(self, service: ServiceCallType) -> None:
"""Execute service call."""
setpoint = pypck.lcn_defs.Var[service.data[CONF_SETPOINT]]
state = service.data[CONF_STATE]
@ -288,7 +294,7 @@ class SendKeys(LcnServiceCall):
}
)
async def async_call_service(self, service):
async def async_call_service(self, service: ServiceCallType) -> None:
"""Execute service call."""
device_connection = self.get_device_connection(service)
@ -331,7 +337,7 @@ class LockKeys(LcnServiceCall):
}
)
async def async_call_service(self, service):
async def async_call_service(self, service: ServiceCallType) -> None:
"""Execute service call."""
device_connection = self.get_device_connection(service)
@ -368,7 +374,7 @@ class DynText(LcnServiceCall):
}
)
async def async_call_service(self, service):
async def async_call_service(self, service: ServiceCallType) -> None:
"""Execute service call."""
row_id = service.data[CONF_ROW] - 1
text = service.data[CONF_TEXT]
@ -382,7 +388,7 @@ class Pck(LcnServiceCall):
schema = LcnServiceCall.schema.extend({vol.Required(CONF_PCK): str})
async def async_call_service(self, service):
async def async_call_service(self, service: ServiceCallType) -> None:
"""Execute service call."""
pck = service.data[CONF_PCK]
device_connection = self.get_device_connection(service)

56
homeassistant/components/lcn/switch.py
View file

@ -1,21 +1,29 @@
"""Support for LCN switches."""
from __future__ import annotations
from typing import Any
import pypck
from homeassistant.components.switch import DOMAIN as DOMAIN_SWITCH, SwitchEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_ADDRESS, CONF_DOMAIN, CONF_ENTITIES
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, HomeAssistantType
from . import LcnEntity
from .const import CONF_DOMAIN_DATA, CONF_OUTPUT, OUTPUT_PORTS
from .helpers import get_device_connection
from .helpers import DeviceConnectionType, InputType, get_device_connection
PARALLEL_UPDATES = 0
def create_lcn_switch_entity(hass, entity_config, config_entry):
def create_lcn_switch_entity(
hass: HomeAssistantType, entity_config: ConfigType, config_entry: ConfigEntry
) -> LcnEntity:
"""Set up an entity for this domain."""
device_connection = get_device_connection(
hass, tuple(entity_config[CONF_ADDRESS]), config_entry
hass, entity_config[CONF_ADDRESS], config_entry
)
if entity_config[CONF_DOMAIN_DATA][CONF_OUTPUT] in OUTPUT_PORTS:
@ -24,7 +32,11 @@ def create_lcn_switch_entity(hass, entity_config, config_entry):
return LcnRelaySwitch(entity_config, config_entry.entry_id, device_connection)
async def async_setup_entry(hass, config_entry, async_add_entities):
async def async_setup_entry(
hass: HomeAssistantType,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up LCN switch entities from a config entry."""
entities = []
@ -39,46 +51,48 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
class LcnOutputSwitch(LcnEntity, SwitchEntity):
"""Representation of a LCN switch for output ports."""
def __init__(self, config, entry_id, device_connection):
def __init__(
self, config: ConfigType, entry_id: str, device_connection: DeviceConnectionType
) -> None:
"""Initialize the LCN switch."""
super().__init__(config, entry_id, device_connection)
self.output = pypck.lcn_defs.OutputPort[config[CONF_DOMAIN_DATA][CONF_OUTPUT]]
self._is_on = None
self._is_on = False
async def async_added_to_hass(self):
async def async_added_to_hass(self) -> None:
"""Run when entity about to be added to hass."""
await super().async_added_to_hass()
if not self.device_connection.is_group:
await self.device_connection.activate_status_request_handler(self.output)
async def async_will_remove_from_hass(self):
async def async_will_remove_from_hass(self) -> None:
"""Run when entity will be removed from hass."""
await super().async_will_remove_from_hass()
if not self.device_connection.is_group:
await self.device_connection.cancel_status_request_handler(self.output)
@property
def is_on(self):
def is_on(self) -> bool:
"""Return True if entity is on."""
return self._is_on
async def async_turn_on(self, **kwargs):
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn the entity on."""
if not await self.device_connection.dim_output(self.output.value, 100, 0):
return
self._is_on = True
self.async_write_ha_state()
async def async_turn_off(self, **kwargs):
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn the entity off."""
if not await self.device_connection.dim_output(self.output.value, 0, 0):
return
self._is_on = False
self.async_write_ha_state()
def input_received(self, input_obj):
def input_received(self, input_obj: InputType) -> None:
"""Set switch state when LCN input object (command) is received."""
if (
not isinstance(input_obj, pypck.inputs.ModStatusOutput)
@ -93,32 +107,34 @@ class LcnOutputSwitch(LcnEntity, SwitchEntity):
class LcnRelaySwitch(LcnEntity, SwitchEntity):
"""Representation of a LCN switch for relay ports."""
def __init__(self, config, entry_id, device_connection):
def __init__(
self, config: ConfigType, entry_id: str, device_connection: DeviceConnectionType
) -> None:
"""Initialize the LCN switch."""
super().__init__(config, entry_id, device_connection)
self.output = pypck.lcn_defs.RelayPort[config[CONF_DOMAIN_DATA][CONF_OUTPUT]]
self._is_on = None
self._is_on = False
async def async_added_to_hass(self):
async def async_added_to_hass(self) -> None:
"""Run when entity about to be added to hass."""
await super().async_added_to_hass()
if not self.device_connection.is_group:
await self.device_connection.activate_status_request_handler(self.output)
async def async_will_remove_from_hass(self):
async def async_will_remove_from_hass(self) -> None:
"""Run when entity will be removed from hass."""
await super().async_will_remove_from_hass()
if not self.device_connection.is_group:
await self.device_connection.cancel_status_request_handler(self.output)
@property
def is_on(self):
def is_on(self) -> bool:
"""Return True if entity is on."""
return self._is_on
async def async_turn_on(self, **kwargs):
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn the entity on."""
states = [pypck.lcn_defs.RelayStateModifier.NOCHANGE] * 8
states[self.output.value] = pypck.lcn_defs.RelayStateModifier.ON
@ -127,7 +143,7 @@ class LcnRelaySwitch(LcnEntity, SwitchEntity):
self._is_on = True
self.async_write_ha_state()
async def async_turn_off(self, **kwargs):
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn the entity off."""
states = [pypck.lcn_defs.RelayStateModifier.NOCHANGE] * 8
states[self.output.value] = pypck.lcn_defs.RelayStateModifier.OFF
@ -136,7 +152,7 @@ class LcnRelaySwitch(LcnEntity, SwitchEntity):
self._is_on = False
self.async_write_ha_state()
def input_received(self, input_obj):
def input_received(self, input_obj: InputType) -> None:
"""Set switch state when LCN input object (command) is received."""
if not isinstance(input_obj, pypck.inputs.ModStatusRelays):
return

11
mypy.ini
View file

@ -539,6 +539,17 @@ no_implicit_optional = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.lcn.*]
check_untyped_defs = true
disallow_incomplete_defs = true
disallow_subclassing_any = true
disallow_untyped_calls = true
disallow_untyped_decorators = true
disallow_untyped_defs = true
no_implicit_optional = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.light.*]
check_untyped_defs = true
disallow_incomplete_defs = true