From 72d8afcf7c2fc0fded935825e064dfd4e78215c2 Mon Sep 17 00:00:00 2001 From: SukramJ Date: Sun, 5 Dec 2021 13:28:04 +0100 Subject: [PATCH] fixes, typing, enums and more (#65) - Remove variables that are covered by other sensors (CCU only) - Remove dummy from service message (HmIP-RF always sends 0001D3C98DD4B6:3 unreach) - Rename Bidcos thermostats to SimpleRfThermostat and RfThermostat - Use more Enums (like HA does): HmPlatform, HmEventType - Use assignment expressions - Add more type hints (fix most mypy errors) --- changelog.txt | 8 ++ hahomematic/central_unit.py | 64 +++++----- hahomematic/client.py | 4 +- hahomematic/const.py | 87 +++++++------ hahomematic/data.py | 7 +- hahomematic/device.py | 74 +++++++----- hahomematic/devices/climate.py | 67 ++++++---- hahomematic/devices/cover.py | 30 ++--- hahomematic/devices/device_description.py | 48 ++++---- hahomematic/devices/light.py | 26 ++-- hahomematic/devices/lock.py | 18 +-- hahomematic/devices/switch.py | 16 ++- hahomematic/entity.py | 141 ++++++++++++++-------- hahomematic/hub.py | 51 +++++--- hahomematic/internal/action.py | 4 +- hahomematic/internal/text.py | 4 +- hahomematic/json_rpc_client.py | 6 +- hahomematic/platforms/binary_sensor.py | 4 +- hahomematic/platforms/button.py | 4 +- hahomematic/platforms/number.py | 4 +- hahomematic/platforms/select.py | 4 +- hahomematic/platforms/sensor.py | 4 +- hahomematic/platforms/switch.py | 4 +- hahomematic/xml_rpc_server.py | 20 ++- setup.py | 2 +- 25 files changed, 419 insertions(+), 282 deletions(-) diff --git a/changelog.txt b/changelog.txt index 735bdff5..6709f2d6 100644 --- a/changelog.txt +++ b/changelog.txt @@ -1,3 +1,11 @@ +Version 0.0.17 (2021-12-05) +- Remove variables that are covered by other sensors (CCU only) +- Remove dummy from service message (HmIP-RF always sends 0001D3C98DD4B6:3 unreach) +- Rename Bidcos thermostats to SimpleRfThermostat and RfThermostat +- Use more Enums (like HA does): HmPlatform, HmEventType +- Use assignment expressions +- Add more type hints (fix most mypy errors) + Version 0.0.16 (2021-12-02) - Don't use default entities for climate groups (already included in device) diff --git a/hahomematic/central_unit.py b/hahomematic/central_unit.py index fee5387a..be43a7f5 100644 --- a/hahomematic/central_unit.py +++ b/hahomematic/central_unit.py @@ -32,6 +32,7 @@ HM_VIRTUAL_REMOTE_HMIP, LOCALHOST, PRIMARY_PORTS, + HmPlatform, ) from hahomematic.data import INSTANCES from hahomematic.device import HmDevice, create_devices @@ -112,14 +113,14 @@ def __init__(self, central_config): self._load_caches() self.init_address_parameter_list() self._connection_checker = ConnectionChecker(self) - self.hub = None + self.hub: HmHub | None = None async def init_hub(self): """Init the hub.""" if self.model is not BACKEND_PYDEVCCU: self.hub = HmHub( self, - use_entities=self.central_config.enable_sensors_for_own_system_variables, + use_entities=self.central_config.enable_sensors_for_system_variables, ) await self.hub.fetch_data() else: @@ -147,8 +148,7 @@ def has_multiple_channels(self, address, parameter) -> bool: if ":" not in address: return False d_address = address.split(":")[0] - channels = self.address_parameter_cache.get((d_address, parameter)) - if channels: + if channels := self.address_parameter_cache.get((d_address, parameter)): return len(set(channels)) > 1 return False @@ -303,20 +303,26 @@ async def set_system_variable(self, name, value): async def get_service_messages(self): """Get service messages from CCU / Homegear.""" - await self.get_primary_client().get_service_messages() + service_messages = [] + for client in self.clients.values(): + if client.port in PRIMARY_PORTS: + if client_messages := await client.get_service_messages(): + service_messages.append(client_messages) + return _remove_dummy_service_message(service_messages) # pylint: disable=invalid-name async def set_install_mode( self, interface_id, on=True, t=60, mode=1, address=None ) -> None: """Activate or deactivate install-mode on CCU / Homegear.""" - await self.get_primary_client(interface_id).set_install_mode( - on=on, t=t, mode=mode, address=address - ) + if client := self.get_primary_client(interface_id): + await client.set_install_mode(on=on, t=t, mode=mode, address=address) async def get_install_mode(self, interface_id) -> int: """Get remaining time in seconds install mode is active from CCU / Homegear.""" - return await self.get_primary_client(interface_id).get_install_mode() + if client := self.get_primary_client(interface_id): + return await client.get_install_mode() + return 0 async def put_paramset(self, interface_id, address, paramset, value, rx_mode=None): """Set paramsets manually.""" @@ -349,14 +355,13 @@ async def press_virtual_remote_key(self, address, parameter): ) device_address = address.split(":")[0] - virtual_remote: HmDevice = self._get_virtual_remote(device_address) - if virtual_remote: + if virtual_remote := self._get_virtual_remote(device_address): virtual_remote_channel = virtual_remote.action_events.get( (address, parameter) ) await virtual_remote_channel.send_value(True) - def get_hm_entities_by_platform(self, platform): + def get_hm_entities_by_hmplatform(self, platform: HmPlatform): """ Return all hm-entities by platform """ @@ -367,7 +372,7 @@ def get_hm_entities_by_platform(self, platform): return hm_entities - def get_primary_client(self, interface_id=None) -> hm_client.Client: + def get_primary_client(self, interface_id=None) -> hm_client.Client | None: """Return the client by interface_id or the first with a primary port.""" try: if interface_id: @@ -382,15 +387,14 @@ def get_primary_client(self, interface_id=None) -> hm_client.Client: ) _LOGGER.warning(message) raise hm_client.ClientException(message) from err + return None def get_hm_entity_by_parameter(self, address, parameter) -> GenericEntity | None: """Get entity by address and parameter.""" if ":" in address: device_address = address.split(":")[0] - device = self.hm_devices.get(device_address) - if device: - entity = device.entities.get((address, parameter)) - if entity: + if device := self.hm_devices.get(device_address): + if entity := device.entities.get((address, parameter)): return entity return None @@ -430,8 +434,7 @@ def get_all_used_parameters(self): parameters = set() for entity in self.hm_entities.values(): if isinstance(entity, GenericEntity): - parameter = getattr(entity, "parameter", None) - if parameter: + if getattr(entity, "parameter", None): parameters.add(entity.parameter) return sorted(parameters) @@ -439,11 +442,9 @@ def get_all_used_parameters(self): def get_used_parameters(self, address): """Return used parameters""" parameters = set() - device = self.hm_devices.get(address) - if device: + if device := self.hm_devices.get(address): for entity in device.entities.values(): - parameter = getattr(entity, "parameter", None) - if parameter: + if getattr(entity, "parameter", None): parameters.add(entity.parameter) return sorted(parameters) @@ -713,7 +714,7 @@ def __init__( json_port=None, json_tls=DEFAULT_TLS, enable_virtual_channels=False, - enable_sensors_for_own_system_variables=False, + enable_sensors_for_system_variables=False, ): self.entry_id = entry_id self.loop = loop @@ -730,10 +731,19 @@ def __init__( self.json_port = json_port self.json_tls = json_tls self.enable_virtual_channels = enable_virtual_channels - self.enable_sensors_for_own_system_variables = ( - enable_sensors_for_own_system_variables - ) + self.enable_sensors_for_system_variables = enable_sensors_for_system_variables def get_central(self) -> CentralUnit: """Identify the used client.""" return CentralUnit(self) + + +def _remove_dummy_service_message(service_messages): + """Remove dummy SM, that hmip server always sends.""" + new_service_messages = [] + for client_messages in service_messages: + if "0001D3C98DD4B6:3" not in [ + client_message[0] for client_message in client_messages + ]: + new_service_messages.append(client_messages) + return new_service_messages diff --git a/hahomematic/client.py b/hahomematic/client.py index d9766dce..096bd446 100644 --- a/hahomematic/client.py +++ b/hahomematic/client.py @@ -177,6 +177,7 @@ async def proxy_re_init(self) -> int: de_init_status = await self.proxy_de_init() if de_init_status is not PROXY_DE_INIT_FAILED: return await self.proxy_init() + return PROXY_DE_INIT_FAILED def stop(self): """Stop depending services.""" @@ -268,12 +269,13 @@ async def set_install_mode(self, on=True, t=60, mode=1, address=None) -> None: except ProxyException: _LOGGER.exception("set_install_mode: ProxyException") - async def get_install_mode(self): + async def get_install_mode(self) -> int: """Get remaining time in seconds install mode is active from CCU / Homegear.""" try: return await self.proxy.getInstallMode() except ProxyException: _LOGGER.exception("get_install_mode: ProxyException") + return 0 async def get_all_metadata(self, address): """Get all metadata of device.""" diff --git a/hahomematic/const.py b/hahomematic/const.py index 9f26df7c..b3985f52 100644 --- a/hahomematic/const.py +++ b/hahomematic/const.py @@ -3,8 +3,12 @@ """ from __future__ import annotations -DEFAULT_ENCODING = "UTF-8" +from datetime import datetime +from enum import Enum +DEFAULT_ENCODING = "UTF-8" +HA_DOMAIN = "hahm" +INIT_DATETIME = datetime.strptime("01.01.1970 00:00:00", "%d.%m.%Y %H:%M:%S") LOCALHOST = "localhost" IP_LOCALHOST_V4 = "127.0.0.1" IP_LOCALHOST_V6 = "::1" @@ -57,8 +61,6 @@ # PARAMSET_MASTER, ] -HA_DOMAIN = "hahm" - HH_EVENT_DELETE_DEVICES = "deleteDevices" HH_EVENT_DEVICES_CREATED = "devicesCreated" HH_EVENT_ERROR = "error" @@ -85,10 +87,6 @@ EVENT_SEQUENCE_OK = "SEQUENCE_OK" EVENT_UN_REACH = "UNREACH" -EVENT_ALARM = "homematic.alarm" -EVENT_KEYPRESS = "homematic.keypress" -EVENT_IMPULSE = "homematic.impulse" - CLICK_EVENTS = [ EVENT_PRESS, EVENT_PRESS_SHORT, @@ -274,35 +272,56 @@ DEFAULT_TLS = False DEFAULT_VERIFY_TLS = False -HA_PLATFORM_ACTION = "action" -HA_PLATFORM_BINARY_SENSOR = "binary_sensor" -HA_PLATFORM_BUTTON = "button" -HA_PLATFORM_CLIMATE = "climate" -HA_PLATFORM_COVER = "cover" -HA_PLATFORM_EVENT = "event" -HA_PLATFORM_LIGHT = "light" -HA_PLATFORM_LOCK = "lock" -HA_PLATFORM_NUMBER = "number" -HA_PLATFORM_SELECT = "select" -HA_PLATFORM_SENSOR = "sensor" -HA_PLATFORM_SWITCH = "switch" -HA_PLATFORM_TEXT = "text" - -HA_PLATFORMS = [ - HA_PLATFORM_BINARY_SENSOR, - HA_PLATFORM_BUTTON, - HA_PLATFORM_CLIMATE, - HA_PLATFORM_COVER, - HA_PLATFORM_LIGHT, - HA_PLATFORM_LOCK, - HA_PLATFORM_NUMBER, - HA_PLATFORM_SELECT, - HA_PLATFORM_SENSOR, - HA_PLATFORM_SWITCH, -] - HM_ENTITY_UNIT_REPLACE = {'"': "", "100%": "%", "% rF": "%"} HM_VIRTUAL_REMOTE_HM = "BidCoS-RF" HM_VIRTUAL_REMOTE_HMIP = "HmIP-RCV-1" HM_VIRTUAL_REMOTES = [HM_VIRTUAL_REMOTE_HM, HM_VIRTUAL_REMOTE_HMIP] + + +class HmPlatform(Enum): + """Enum with platforms relevant for Home Assistant.""" + + ACTION = "action" + BINARY_SENSOR = "binary_sensor" + BUTTON = "button" + CLIMATE = "climate" + COVER = "cover" + EVENT = "event" + LIGHT = "light" + LOCK = "lock" + NUMBER = "number" + SELECT = "select" + SENSOR = "sensor" + SWITCH = "switch" + TEXT = "text" + + def __str__(self) -> str: + """Return self.value.""" + return str(self.value) + + +class HmEventType(Enum): + """Enum with hahm event types.""" + + ALARM = "homematic.alarm" + KEYPRESS = "homematic.keypress" + IMPULSE = "homematic.impulse" + + def __str__(self) -> str: + """Return self.value.""" + return str(self.value) + + +AVAILABLE_HM_PLATFORMS = [ + HmPlatform.BINARY_SENSOR, + HmPlatform.BUTTON, + HmPlatform.CLIMATE, + HmPlatform.COVER, + HmPlatform.LIGHT, + HmPlatform.LOCK, + HmPlatform.NUMBER, + HmPlatform.SELECT, + HmPlatform.SENSOR, + HmPlatform.SWITCH, +] diff --git a/hahomematic/data.py b/hahomematic/data.py index 4de5f33b..9582d811 100644 --- a/hahomematic/data.py +++ b/hahomematic/data.py @@ -3,13 +3,14 @@ """ from __future__ import annotations +import hahomematic.central_unit as hm_central + # {instance_name, central_unit} -INSTANCES = {} +INSTANCES: dict[str, hm_central.CentralUnit] = {} def get_client_by_interface_id(interface_id): """Return client by interface_id""" for central in INSTANCES.values(): - client = central.clients.get(interface_id) - if client: + if client := central.clients.get(interface_id): return client diff --git a/hahomematic/device.py b/hahomematic/device.py index 23ce9199..81a656dc 100644 --- a/hahomematic/device.py +++ b/hahomematic/device.py @@ -3,7 +3,8 @@ """ from __future__ import annotations -import datetime +from collections.abc import Callable +from datetime import datetime import logging import hahomematic.central_unit as hm_central @@ -23,6 +24,7 @@ IGNORED_PARAMETERS_WILDCARDS_END, IGNORED_PARAMETERS_WILDCARDS_START, IMPULSE_EVENTS, + INIT_DATETIME, OPERATION_EVENT, OPERATION_WRITE, PARAM_UN_REACH, @@ -41,6 +43,7 @@ AlarmEvent, BaseEntity, BaseEvent, + BaseParameterEntity, CallbackEntity, ClickEvent, CustomEntity, @@ -65,7 +68,9 @@ class HmDevice: Object to hold information about a device and associated entities. """ - def __init__(self, central: hm_central.CentralUnit, interface_id, address): + def __init__( + self, central: hm_central.CentralUnit, interface_id: str, address: str + ): """ Initialize the device object. """ @@ -83,21 +88,24 @@ def __init__(self, central: hm_central.CentralUnit, interface_id, address): self.entities: dict[tuple[str, str], GenericEntity] = {} self.custom_entities: dict[str, CustomEntity] = {} self.action_events: dict[tuple[str, str], BaseEvent] = {} - self.last_update = None + self.last_update: datetime = INIT_DATETIME self._available = True - self._update_callbacks = [] - self._remove_callbacks = [] - self.device_type = self.central.devices_raw_dict[self.interface_id][ + self._update_callbacks: list[Callable] = [] + self._remove_callbacks: list[Callable] = [] + self.device_type: str = self.central.devices_raw_dict[self.interface_id][ self.address ][ATTR_HM_TYPE] - self.sub_type = self.central.devices_raw_dict[self.interface_id][ + self.sub_type: str = self.central.devices_raw_dict[self.interface_id][ self.address ].get(ATTR_HM_SUBTYPE) # marker if device will be created as custom device - self.is_custom_device = device_desc_exists(self.device_type, self.sub_type) - self.firmware = self.central.devices_raw_dict[self.interface_id][self.address][ - ATTR_HM_FIRMWARE - ] + self.is_custom_device: bool = device_desc_exists( + self.device_type, self.sub_type + ) + self.firmware: str = self.central.devices_raw_dict[self.interface_id][ + self.address + ][ATTR_HM_FIRMWARE] + self.name: str = "" if self.address in self.central.names_cache.get(self.interface_id, {}): self.name = self.central.names_cache[self.interface_id][self.address] else: @@ -182,7 +190,7 @@ def remove_device(self, *args) -> None: _callback(*args) def _set_last_update(self) -> None: - self.last_update = datetime.datetime.now() + self.last_update = datetime.now() def get_hm_entity(self, address, parameter) -> GenericEntity | None: """Return a hm_entity from device.""" @@ -237,11 +245,11 @@ async def reload_paramsets(self) -> None: self.update_device() # pylint: disable=too-many-nested-blocks - def create_entities(self) -> set[GenericEntity] | None: + def create_entities(self) -> set[BaseEntity]: """ Create the entities associated to this device. """ - new_entities: set[GenericEntity] = set() + new_entities: list[BaseEntity] = [] for channel in self.channels: if channel not in self.central.paramsets_cache[self.interface_id]: _LOGGER.debug( @@ -255,6 +263,7 @@ def create_entities(self) -> set[GenericEntity] | None: for parameter, parameter_data in self.central.paramsets_cache[ self.interface_id ][channel][paramset].items(): + entity: BaseParameterEntity | None if ( not parameter_data[ATTR_HM_OPERATIONS] & OPERATION_EVENT and not parameter_data[ATTR_HM_OPERATIONS] & OPERATION_WRITE @@ -275,13 +284,13 @@ def create_entities(self) -> set[GenericEntity] | None: parameter_data=parameter_data, ) if self.address.startswith(tuple(HM_VIRTUAL_REMOTES)): - entity = self.create_buttons( + entity = self.create_button( address=channel, parameter=parameter, parameter_data=parameter_data, ) if entity is not None: - new_entities.add(entity) + new_entities.append(entity) if not (parameter in CLICK_EVENTS or parameter in IMPULSE_EVENTS): entity = self.create_entity( address=channel, @@ -289,7 +298,7 @@ def create_entities(self) -> set[GenericEntity] | None: parameter_data=parameter_data, ) if entity is not None: - new_entities.add(entity) + new_entities.append(entity) # create custom entities if self.is_custom_device: _LOGGER.debug( @@ -303,11 +312,13 @@ def create_entities(self) -> set[GenericEntity] | None: for (device_func, group_base_channels) in get_device_funcs( self.device_type, self.sub_type ): - custom_entities = device_func(self, self.address, group_base_channels) - new_entities.update(custom_entities) - return new_entities + custom_entities: list[CustomEntity] = device_func( + self, self.address, group_base_channels + ) + new_entities.extend(custom_entities) + return set(new_entities) - def create_buttons(self, address, parameter, parameter_data) -> HmButton | None: + def create_button(self, address, parameter, parameter_data) -> HmButton | None: """Create the buttons associated to this device""" unique_id = generate_unique_id( address, parameter, f"button_{self.central.instance_name}" @@ -319,16 +330,16 @@ def create_buttons(self, address, parameter, parameter_data) -> HmButton | None: self.interface_id, ) - button = HmButton( + if button := HmButton( device=self, unique_id=unique_id, address=address, parameter=parameter, parameter_data=parameter_data, - ) - if button: + ): button.add_to_collections() - return button + return button + return None def create_event(self, address, parameter, parameter_data) -> BaseEvent | None: """Create action event entity.""" @@ -345,7 +356,7 @@ def create_event(self, address, parameter, parameter_data) -> BaseEvent | None: parameter, self.interface_id, ) - action_event = None + action_event: BaseEvent | None = None if parameter_data[ATTR_HM_OPERATIONS] & OPERATION_EVENT: if parameter in CLICK_EVENTS: action_event = ClickEvent( @@ -402,7 +413,7 @@ def create_entity(self, address, parameter, parameter_data) -> GenericEntity | N parameter, self.interface_id, ) - entity = None + entity: GenericEntity | None = None if parameter_data[ATTR_HM_OPERATIONS] & OPERATION_WRITE: if parameter_data[ATTR_HM_TYPE] == TYPE_ACTION: if parameter_data[ATTR_HM_OPERATIONS] == OPERATION_WRITE: @@ -513,7 +524,7 @@ def create_devices(central: hm_central.CentralUnit) -> None: Trigger creation of the objects that expose the functionality. """ new_devices = set[str]() - new_entities = set[GenericEntity]() + new_entities: list[BaseEntity] = [] for interface_id, client in central.clients.items(): if not client: _LOGGER.warning( @@ -528,7 +539,7 @@ def create_devices(central: hm_central.CentralUnit) -> None: continue for device_address in central.devices[interface_id]: # Do we check for duplicates here? For now we do. - device = None + device: HmDevice | None = None if device_address in central.hm_devices: _LOGGER.debug( "create_devices: Skipping device %s on %s, already exists.", @@ -547,7 +558,8 @@ def create_devices(central: hm_central.CentralUnit) -> None: device_address, ) try: - new_entities.update(device.create_entities()) + if device: + new_entities.extend(device.create_entities()) except Exception: _LOGGER.exception( "create_devices: Failed to create entities: %s, %s", @@ -556,7 +568,7 @@ def create_devices(central: hm_central.CentralUnit) -> None: ) if callable(central.callback_system_event): central.callback_system_event( - HH_EVENT_DEVICES_CREATED, new_devices, new_entities + HH_EVENT_DEVICES_CREATED, new_devices, set(new_entities) ) diff --git a/hahomematic/devices/climate.py b/hahomematic/devices/climate.py index e31fc636..f30002ed 100644 --- a/hahomematic/devices/climate.py +++ b/hahomematic/devices/climate.py @@ -2,8 +2,9 @@ from __future__ import annotations import logging +from typing import Any -from hahomematic.const import ATTR_HM_MAX, ATTR_HM_MIN, HA_PLATFORM_CLIMATE +from hahomematic.const import ATTR_HM_MAX, ATTR_HM_MIN, HmPlatform from hahomematic.devices.device_description import ( FIELD_AUTO_MODE, FIELD_BOOST_MODE, @@ -16,7 +17,7 @@ FIELD_SET_POINT_MODE, FIELD_SETPOINT, FIELD_TEMPERATURE, - Devices, + DeviceDescription, make_custom_entity, ) from hahomematic.entity import CustomEntity @@ -45,7 +46,7 @@ SUPPORT_PRESET_MODE = 16 -class SimpleThermostat(CustomEntity): +class SimpleRfThermostat(CustomEntity): """Simple classic HomeMatic thermostat HM-CC-TC.""" def __init__( @@ -65,11 +66,11 @@ def __init__( device_enum=device_enum, device_desc=device_desc, entity_desc=entity_desc, - platform=HA_PLATFORM_CLIMATE, + platform=HmPlatform.CLIMATE, channel_no=channel_no, ) _LOGGER.debug( - "SimpleThermostat.__init__(%s, %s, %s)", + "SimpleRfThermostat.__init__(%s, %s, %s)", self._device.interface_id, address, unique_id, @@ -148,7 +149,7 @@ async def set_temperature(self, **kwargs): await self._send_value(FIELD_SETPOINT, float(temperature)) -class Thermostat(CustomEntity): +class RfThermostat(CustomEntity): """Classic HomeMatic thermostat like HM-CC-RT-DN.""" def __init__( @@ -168,11 +169,11 @@ def __init__( device_enum=device_enum, device_desc=device_desc, entity_desc=entity_desc, - platform=HA_PLATFORM_CLIMATE, + platform=HmPlatform.CLIMATE, channel_no=channel_no, ) _LOGGER.debug( - "Thermostat.__init__(%s, %s, %s)", + "RfThermostat.__init__(%s, %s, %s)", self._device.interface_id, address, unique_id, @@ -331,7 +332,7 @@ def __init__( device_enum=device_enum, device_desc=device_desc, entity_desc=entity_desc, - platform=HA_PLATFORM_CLIMATE, + platform=HmPlatform.CLIMATE, channel_no=channel_no, ) _LOGGER.debug( @@ -469,48 +470,64 @@ async def set_preset_mode(self, preset_mode): await self._send_value(FIELD_BOOST_MODE, False) -def make_simple_thermostat(device, address, group_base_channels: [int]): - """Creates SimpleThermostat entities.""" +def make_simple_thermostat(device, address, group_base_channels: list[int]): + """Creates SimpleRfThermostat entities.""" return make_custom_entity( device, address, - SimpleThermostat, - Devices.SIMPLE_RF_THERMOSTAT, + SimpleRfThermostat, + DeviceDescription.SIMPLE_RF_THERMOSTAT, group_base_channels, ) -def make_thermostat(device, address, group_base_channels: [int]): - """Creates Thermostat entities.""" +def make_thermostat(device, address, group_base_channels: list[int]): + """Creates RfThermostat entities.""" return make_custom_entity( - device, address, Thermostat, Devices.RF_THERMOSTAT, group_base_channels + device, + address, + RfThermostat, + DeviceDescription.RF_THERMOSTAT, + group_base_channels, ) -def make_thermostat_group(device, address, group_base_channels: [int]): - """Creates Thermostat group entities.""" +def make_thermostat_group(device, address, group_base_channels: list[int]): + """Creates RfThermostat group entities.""" return make_custom_entity( - device, address, Thermostat, Devices.RF_THERMOSTAT_GROUP, group_base_channels + device, + address, + RfThermostat, + DeviceDescription.RF_THERMOSTAT_GROUP, + group_base_channels, ) -def make_ip_thermostat(device, address, group_base_channels: [int]): +def make_ip_thermostat(device, address, group_base_channels: list[int]): """Creates IPThermostat entities.""" return make_custom_entity( - device, address, IPThermostat, Devices.IP_THERMOSTAT, group_base_channels + device, + address, + IPThermostat, + DeviceDescription.IP_THERMOSTAT, + group_base_channels, ) -def make_ip_thermostat_group(device, address, group_base_channels: [int]): +def make_ip_thermostat_group(device, address, group_base_channels: list[int]): """Creates IPThermostat group entities.""" return make_custom_entity( - device, address, IPThermostat, Devices.IP_THERMOSTAT_GROUP, group_base_channels + device, + address, + IPThermostat, + DeviceDescription.IP_THERMOSTAT_GROUP, + group_base_channels, ) # Case for device model is not relevant # device_type and sub_type(IP-only) can be used here -DEVICES = { +DEVICES: dict[str, tuple[Any, list[int]]] = { "BC-RT-TRX-CyG*": (make_thermostat, []), "BC-RT-TRX-CyN*": (make_thermostat, []), "BC-TC-C-WM*": (make_thermostat, []), @@ -525,6 +542,6 @@ def make_ip_thermostat_group(device, address, group_base_channels: [int]): "HmIP-WTH*": (make_ip_thermostat, []), "HmIPW-STH*": (make_ip_thermostat, []), "HmIPW-WTH*": (make_ip_thermostat, []), - "Thermostat AA*": (make_ip_thermostat, []), + "RfThermostat AA*": (make_ip_thermostat, []), "ZEL STG RM FWT": (make_simple_thermostat, []), } diff --git a/hahomematic/devices/cover.py b/hahomematic/devices/cover.py index 760b1707..f90ade42 100644 --- a/hahomematic/devices/cover.py +++ b/hahomematic/devices/cover.py @@ -5,7 +5,7 @@ import logging from typing import Any -from hahomematic.const import HA_PLATFORM_COVER +from hahomematic.const import HmPlatform from hahomematic.devices.device_description import ( FIELD_CHANNEL_LEVEL, FIELD_CHANNEL_LEVEL_2, @@ -14,7 +14,7 @@ FIELD_LEVEL, FIELD_LEVEL_2, FIELD_STOP, - Devices, + DeviceDescription, make_custom_entity, ) from hahomematic.entity import CustomEntity @@ -60,7 +60,7 @@ def __init__( device_enum=device_enum, device_desc=device_desc, entity_desc=entity_desc, - platform=HA_PLATFORM_COVER, + platform=HmPlatform.COVER, channel_no=channel_no, ) _LOGGER.debug( @@ -197,7 +197,7 @@ def __init__( device_enum=device_enum, device_desc=device_desc, entity_desc=entity_desc, - platform=HA_PLATFORM_COVER, + platform=HmPlatform.COVER, channel_no=channel_no, ) _LOGGER.debug( @@ -256,44 +256,44 @@ async def vent_cover(self) -> None: await self._send_value(FIELD_DOOR_COMMAND, GARAGE_DOOR_COMMAND_PARTIAL_OPEN) -def make_ip_cover(device, address, group_base_channels: [int]): +def make_ip_cover(device, address, group_base_channels: list[int]): """Creates homematic ip cover entities.""" return make_custom_entity( - device, address, HmCover, Devices.IP_COVER, group_base_channels + device, address, HmCover, DeviceDescription.IP_COVER, group_base_channels ) -def make_rf_cover(device, address, group_base_channels: [int]): +def make_rf_cover(device, address, group_base_channels: list[int]): """Creates homematic classic cover entities.""" return make_custom_entity( - device, address, HmCover, Devices.RF_COVER, group_base_channels + device, address, HmCover, DeviceDescription.RF_COVER, group_base_channels ) -def make_ip_blind(device, address, group_base_channels: [int]): +def make_ip_blind(device, address, group_base_channels: list[int]): """Creates homematic ip cover entities.""" return make_custom_entity( - device, address, HmBlind, Devices.IP_COVER, group_base_channels + device, address, HmBlind, DeviceDescription.IP_COVER, group_base_channels ) -def make_ip_garage(device, address, group_base_channels: [int]): +def make_ip_garage(device, address, group_base_channels: list[int]): """Creates homematic ip garage entities.""" return make_custom_entity( - device, address, HmGarage, Devices.IP_GARAGE, group_base_channels + device, address, HmGarage, DeviceDescription.IP_GARAGE, group_base_channels ) -def make_rf_blind(device, address, group_base_channels: [int]): +def make_rf_blind(device, address, group_base_channels: list[int]): """Creates homematic classic cover entities.""" return make_custom_entity( - device, address, HmBlind, Devices.RF_COVER, group_base_channels + device, address, HmBlind, DeviceDescription.RF_COVER, group_base_channels ) # Case for device model is not relevant # device_type and sub_type(IP-only) can be used here -DEVICES = { +DEVICES: dict[str, tuple[Any, list[int]]] = { "HmIP-BROLL": (make_ip_cover, [3]), "HmIP-FROLL": (make_ip_cover, [3]), "HmIP-BBL": (make_ip_blind, [3]), diff --git a/hahomematic/devices/device_description.py b/hahomematic/devices/device_description.py index 3ec9e4fe..30df169c 100644 --- a/hahomematic/devices/device_description.py +++ b/hahomematic/devices/device_description.py @@ -67,7 +67,7 @@ _LOGGER = logging.getLogger(__name__) -class Devices(Enum): +class DeviceDescription(Enum): """Enum for device_descriptions.""" IP_COVER = "IPCover" @@ -85,6 +85,10 @@ class Devices(Enum): RF_THERMOSTAT_GROUP = "RfThermostatGroup" SIMPLE_RF_THERMOSTAT = "SimpleRfThermostat" + def __str__(self) -> str: + """Return self.value.""" + return str(self.value) + SCHEMA_DD_FIELD_DETAILS = Schema({Optional(str): str, Optional(str): str}) @@ -112,7 +116,7 @@ class Devices(Enum): Required(DD_DEFAULT_ENTITIES): SCHEMA_DD_FIELD, Required(DD_DEVICES): Schema( { - Required(Devices): SCHEMA_DD_DEVICE_GROUPS, + Required(DeviceDescription): SCHEMA_DD_DEVICE_GROUPS, } ), } @@ -134,7 +138,7 @@ class Devices(Enum): } }, DD_DEVICES: { - Devices.IP_COVER: { + DeviceDescription.IP_COVER: { DD_DEVICE_GROUP: { DD_PHY_CHANNEL: [1], DD_VIRT_CHANNEL: [], @@ -151,7 +155,7 @@ class Devices(Enum): }, }, }, - Devices.IP_DIMMER: { + DeviceDescription.IP_DIMMER: { DD_DEVICE_GROUP: { DD_PHY_CHANNEL: [1], DD_VIRT_CHANNEL: [2, 3], @@ -165,7 +169,7 @@ class Devices(Enum): }, }, }, - Devices.IP_GARAGE: { + DeviceDescription.IP_GARAGE: { DD_DEVICE_GROUP: { DD_PHY_CHANNEL: [1], DD_VIRT_CHANNEL: [], @@ -176,7 +180,7 @@ class Devices(Enum): DD_FIELDS: {}, }, }, - Devices.IP_LIGHT_BSL: { + DeviceDescription.IP_LIGHT_BSL: { DD_DEVICE_GROUP: { DD_PHY_CHANNEL: [1], DD_VIRT_CHANNEL: [2, 3], @@ -192,7 +196,7 @@ class Devices(Enum): }, }, }, - Devices.IP_LIGHT_SWITCH: { + DeviceDescription.IP_LIGHT_SWITCH: { DD_DEVICE_GROUP: { DD_PHY_CHANNEL: [1], DD_VIRT_CHANNEL: [2, 3], @@ -216,7 +220,7 @@ class Devices(Enum): }, }, }, - Devices.IP_LOCK: { + DeviceDescription.IP_LOCK: { DD_DEVICE_GROUP: { DD_PHY_CHANNEL: [], DD_VIRT_CHANNEL: [], @@ -229,7 +233,7 @@ class Devices(Enum): }, }, }, - Devices.IP_THERMOSTAT: { + DeviceDescription.IP_THERMOSTAT: { DD_DEVICE_GROUP: { DD_PHY_CHANNEL: [1], DD_VIRT_CHANNEL: [], @@ -254,7 +258,7 @@ class Devices(Enum): }, }, }, - Devices.IP_THERMOSTAT_GROUP: { + DeviceDescription.IP_THERMOSTAT_GROUP: { DD_DEVICE_GROUP: { DD_PHY_CHANNEL: [1], DD_VIRT_CHANNEL: [], @@ -271,7 +275,7 @@ class Devices(Enum): }, DD_INCLUDE_DEFAULT_ENTITIES: False, }, - Devices.RF_COVER: { + DeviceDescription.RF_COVER: { DD_DEVICE_GROUP: { DD_PHY_CHANNEL: [1, 2, 3, 4], DD_VIRT_CHANNEL: [], @@ -282,7 +286,7 @@ class Devices(Enum): }, }, }, - Devices.RF_DIMMER: { + DeviceDescription.RF_DIMMER: { DD_DEVICE_GROUP: { DD_PHY_CHANNEL: [1, 2, 3, 4], DD_VIRT_CHANNEL: [], @@ -292,7 +296,7 @@ class Devices(Enum): DD_FIELDS: {}, }, }, - Devices.RF_LOCK: { + DeviceDescription.RF_LOCK: { DD_DEVICE_GROUP: { DD_PHY_CHANNEL: [], DD_VIRT_CHANNEL: [], @@ -305,7 +309,7 @@ class Devices(Enum): }, }, }, - Devices.RF_THERMOSTAT: { + DeviceDescription.RF_THERMOSTAT: { DD_DEVICE_GROUP: { DD_PHY_CHANNEL: [1, 2, 3, 4], DD_VIRT_CHANNEL: [], @@ -322,7 +326,7 @@ class Devices(Enum): }, }, }, - Devices.RF_THERMOSTAT_GROUP: { + DeviceDescription.RF_THERMOSTAT_GROUP: { DD_DEVICE_GROUP: { DD_PHY_CHANNEL: [1, 2, 3, 4], DD_VIRT_CHANNEL: [], @@ -340,7 +344,7 @@ class Devices(Enum): }, DD_INCLUDE_DEFAULT_ENTITIES: False, }, - Devices.SIMPLE_RF_THERMOSTAT: { + DeviceDescription.SIMPLE_RF_THERMOSTAT: { DD_DEVICE_GROUP: { DD_PHY_CHANNEL: [], DD_VIRT_CHANNEL: [], @@ -375,8 +379,8 @@ def make_custom_entity( device, address, custom_entity_class, - device_enum: Devices, - group_base_channels: [int], + device_enum: DeviceDescription, + group_base_channels: list[int], ): """ Creates custom_entities. @@ -458,7 +462,7 @@ def get_default_entities(): return copy(device_description[DD_DEFAULT_ENTITIES]) -def get_include_default_entities(device_enum: Devices) -> True: +def get_include_default_entities(device_enum: DeviceDescription) -> bool: """Return if default entities should be included.""" device = _get_device(device_enum) if device: @@ -466,7 +470,7 @@ def get_include_default_entities(device_enum: Devices) -> True: return DEFAULT_INCLUDE_DEFAULT_ENTITIES -def _get_device(device_enum: Devices): +def _get_device(device_enum: DeviceDescription): """Return device from device_descriptions.""" device = device_description[DD_DEVICES].get(device_enum) if device: @@ -474,7 +478,7 @@ def _get_device(device_enum: Devices): return None -def _get_device_group(device_enum: Devices, base_channel_no: int): +def _get_device_group(device_enum: DeviceDescription, base_channel_no: int): """Return the device group.""" device = _get_device(device_enum) group = {} @@ -501,7 +505,7 @@ def _get_device_group(device_enum: Devices, base_channel_no: int): return group -def _get_device_entities(device_enum: Devices, base_channel_no: int): +def _get_device_entities(device_enum: DeviceDescription, base_channel_no: int): """Return the device entities.""" additional_entities = ( device_description[DD_DEVICES] diff --git a/hahomematic/devices/light.py b/hahomematic/devices/light.py index 7acc64e5..597e763d 100644 --- a/hahomematic/devices/light.py +++ b/hahomematic/devices/light.py @@ -5,7 +5,7 @@ import logging from typing import Any -from hahomematic.const import HA_PLATFORM_LIGHT +from hahomematic.const import HmPlatform from hahomematic.devices.device_description import ( FIELD_CHANNEL_COLOR, FIELD_CHANNEL_LEVEL, @@ -13,7 +13,7 @@ FIELD_COLOR, FIELD_LEVEL, FIELD_STATE, - Devices, + DeviceDescription, make_custom_entity, ) from hahomematic.entity import CustomEntity @@ -52,7 +52,7 @@ def __init__( device_enum=device_enum, device_desc=device_desc, entity_desc=entity_desc, - platform=HA_PLATFORM_LIGHT, + platform=HmPlatform.LIGHT, channel_no=channel_no, ) _LOGGER.debug( @@ -69,7 +69,7 @@ def is_on(self) -> bool: ... @property - def brightness(self) -> int: + def brightness(self) -> int | None: """Return the brightness of this light between 0..255.""" return None @@ -299,37 +299,37 @@ def _convert_color(color: tuple) -> str: return bsl_color -def make_ip_dimmer(device, address, group_base_channels: [int]): +def make_ip_dimmer(device, address, group_base_channels: list[int]): """Creates homematic ip dimmer entities.""" return make_custom_entity( - device, address, HmDimmer, Devices.IP_DIMMER, group_base_channels + device, address, HmDimmer, DeviceDescription.IP_DIMMER, group_base_channels ) -def make_rf_dimmer(device, address, group_base_channels: [int]): +def make_rf_dimmer(device, address, group_base_channels: list[int]): """Creates homematic classic dimmer entities.""" return make_custom_entity( - device, address, HmDimmer, Devices.RF_DIMMER, group_base_channels + device, address, HmDimmer, DeviceDescription.RF_DIMMER, group_base_channels ) -def make_ip_light(device, address, group_base_channels: [int]): +def make_ip_light(device, address, group_base_channels: list[int]): """Creates homematic classic light entities.""" return make_custom_entity( - device, address, HmLight, Devices.IP_LIGHT_SWITCH, group_base_channels + device, address, HmLight, DeviceDescription.IP_LIGHT_SWITCH, group_base_channels ) -def make_ip_light_bsl(device, address, group_base_channels: [int]): +def make_ip_light_bsl(device, address, group_base_channels: list[int]): """Creates HmIP-BSL entities.""" return make_custom_entity( - device, address, IPLightBSL, Devices.IP_LIGHT_BSL, group_base_channels + device, address, IPLightBSL, DeviceDescription.IP_LIGHT_BSL, group_base_channels ) # Case for device model is not relevant # device_type and sub_type(IP-only) can be used here -DEVICES = { +DEVICES: dict[str, tuple[Any, list[int]]] = { "HmIP-BSL": (make_ip_light_bsl, [7, 11]), "HmIP-BSM": (make_ip_light, [3]), "HmIP-BDT": (make_ip_dimmer, [3]), diff --git a/hahomematic/devices/lock.py b/hahomematic/devices/lock.py index 47dd7a3f..9d272d87 100644 --- a/hahomematic/devices/lock.py +++ b/hahomematic/devices/lock.py @@ -5,13 +5,13 @@ import logging from typing import Any -from hahomematic.const import HA_PLATFORM_LOCK +from hahomematic.const import HmPlatform from hahomematic.devices.device_description import ( FIELD_LOCK_STATE, FIELD_LOCK_TARGET_LEVEL, FIELD_OPEN, FIELD_STATE, - Devices, + DeviceDescription, make_custom_entity, ) from hahomematic.entity import CustomEntity @@ -43,7 +43,7 @@ def __init__( device_enum=device_enum, device_desc=device_desc, entity_desc=entity_desc, - platform=HA_PLATFORM_LOCK, + platform=HmPlatform.LOCK, channel_no=channel_no, ) _LOGGER.debug( @@ -96,7 +96,7 @@ def __init__( device_enum=device_enum, device_desc=device_desc, entity_desc=entity_desc, - platform=HA_PLATFORM_LOCK, + platform=HmPlatform.LOCK, channel_no=channel_no, ) _LOGGER.debug( @@ -129,23 +129,23 @@ async def open(self, **kwargs: Any) -> None: await self._send_value(FIELD_OPEN, True) -def make_ip_lock(device, address, group_base_channels: [int]): +def make_ip_lock(device, address, group_base_channels: list[int]): """Creates homematic ip lock entities.""" return make_custom_entity( - device, address, IpLock, Devices.IP_LOCK, group_base_channels + device, address, IpLock, DeviceDescription.IP_LOCK, group_base_channels ) -def make_rf_lock(device, address, group_base_channels: [int]): +def make_rf_lock(device, address, group_base_channels: list[int]): """Creates homematic rf lock entities.""" return make_custom_entity( - device, address, RfLock, Devices.RF_LOCK, group_base_channels + device, address, RfLock, DeviceDescription.RF_LOCK, group_base_channels ) # Case for device model is not relevant # device_type and sub_type(IP-only) can be used here -DEVICES = { +DEVICES: dict[str, tuple[Any, list[int]]] = { "HmIP-DLD": (make_ip_lock, []), "HM-Sec-Key*": (make_rf_lock, []), } diff --git a/hahomematic/devices/switch.py b/hahomematic/devices/switch.py index da6421c7..61ba5751 100644 --- a/hahomematic/devices/switch.py +++ b/hahomematic/devices/switch.py @@ -4,11 +4,11 @@ import logging from typing import Any -from hahomematic.const import HA_PLATFORM_SWITCH +from hahomematic.const import HmPlatform from hahomematic.devices.device_description import ( FIELD_CHANNEL_STATE, FIELD_STATE, - Devices, + DeviceDescription, make_custom_entity, ) from hahomematic.entity import CustomEntity @@ -38,7 +38,7 @@ def __init__( device_enum=device_enum, device_desc=device_desc, entity_desc=entity_desc, - platform=HA_PLATFORM_SWITCH, + platform=HmPlatform.SWITCH, channel_no=channel_no, ) _LOGGER.debug( @@ -84,16 +84,20 @@ def extra_state_attributes(self) -> dict[str, Any]: return state_attr -def make_ip_switch(device, address, group_base_channels: [int]): +def make_ip_switch(device, address, group_base_channels: list[int]): """Creates homematic ip switch entities.""" return make_custom_entity( - device, address, HmSwitch, Devices.IP_LIGHT_SWITCH, group_base_channels + device, + address, + HmSwitch, + DeviceDescription.IP_LIGHT_SWITCH, + group_base_channels, ) # Case for device model is not relevant # device_type and sub_type(IP-only) can be used here -DEVICES = { +DEVICES: dict[str, tuple[Any, list[int]]] = { "HmIP-FSM*": (make_ip_switch, [1]), "HmIP-FSI*": (make_ip_switch, [2]), "HmIP-PS*": (make_ip_switch, [2]), diff --git a/hahomematic/entity.py b/hahomematic/entity.py index 0791d475..d9af7386 100644 --- a/hahomematic/entity.py +++ b/hahomematic/entity.py @@ -4,7 +4,8 @@ from __future__ import annotations from abc import ABC, abstractmethod -import datetime +from collections.abc import Callable +from datetime import datetime import logging from typing import Any @@ -26,26 +27,27 @@ DATA_LOAD_FAIL, DATA_LOAD_SUCCESS, DATA_NO_LOAD, - EVENT_ALARM, EVENT_CONFIG_PENDING, - EVENT_IMPULSE, - EVENT_KEYPRESS, EVENT_UN_REACH, FLAG_SERVICE, FLAG_VISIBLE, - HA_PLATFORM_EVENT, HIDDEN_PARAMETERS, HM_ENTITY_UNIT_REPLACE, + INIT_DATETIME, OPERATION_READ, TYPE_ACTION, TYPE_BOOL, TYPE_FLOAT, TYPE_INTEGER, TYPE_STRING, + HmEventType, + HmPlatform, ) +import hahomematic.device as hm_device from hahomematic.devices.device_description import ( DD_FIELDS, DD_FIELDS_REP, + DeviceDescription, get_default_entities, get_include_default_entities, ) @@ -58,9 +60,9 @@ class CallbackEntity(ABC): """Base class for callback entities.""" def __init__(self): - self.last_update = None - self._update_callbacks = [] - self._remove_callbacks = [] + self.last_update: datetime = INIT_DATETIME + self._update_callbacks: list[Callable] = [] + self._remove_callbacks: list[Callable] = [] def register_update_callback(self, update_callback) -> None: """register update callback""" @@ -99,12 +101,10 @@ def remove_entity(self, *args) -> None: _callback(*args) def _set_last_update(self) -> None: - self.last_update = datetime.datetime.now() + self.last_update = datetime.now() def _updated_within_minutes(self, minutes=10) -> bool: - if self.last_update is None: - return False - delta = datetime.datetime.now() - self.last_update + delta = datetime.now() - self.last_update if delta.seconds < minutes * 60: return True return False @@ -113,7 +113,13 @@ def _updated_within_minutes(self, minutes=10) -> bool: class BaseEntity(ABC): """Base class for regular entities.""" - def __init__(self, device, unique_id, address, platform): + def __init__( + self, + device: hm_device.HmDevice, + unique_id: str, + address: str, + platform: HmPlatform, + ): """ Initialize the entity. """ @@ -122,15 +128,15 @@ def __init__(self, device, unique_id, address, platform): self.address = address self.platform = platform self._central = self._device.central - self._interface_id = self._device.interface_id - self.device_type = self._device.device_type - self.sub_type = self._device.sub_type - self.create_in_ha = not self._device.is_custom_device + self._interface_id: str = self._device.interface_id + self.device_type: str = self._device.device_type + self.sub_type: str = self._device.sub_type + self.create_in_ha: bool = not self._device.is_custom_device self.client = self._central.clients[self._interface_id] self.proxy = self.client.proxy - self.name = self.client.central.names_cache.get(self._interface_id, {}).get( - self.address, self.unique_id - ) + self.name: str = self.client.central.names_cache.get( + self._interface_id, {} + ).get(self.address, self.unique_id) @property def available(self) -> bool: @@ -164,7 +170,15 @@ class BaseParameterEntity(BaseEntity): Base class for stateless entities. """ - def __init__(self, device, unique_id, address, parameter, parameter_data, platform): + def __init__( + self, + device: hm_device.HmDevice, + unique_id: str, + address: str, + parameter: str, + parameter_data: dict[str, Any], + platform: HmPlatform, + ): """ Initialize the entity. """ @@ -276,7 +290,15 @@ class GenericEntity(BaseParameterEntity, CallbackEntity): Base class for generic entities. """ - def __init__(self, device, unique_id, address, parameter, parameter_data, platform): + def __init__( + self, + device: hm_device.HmDevice, + unique_id: str, + address: str, + parameter: str, + parameter_data: dict[str, Any], + platform: HmPlatform, + ): """ Initialize the entity. """ @@ -388,14 +410,14 @@ class CustomEntity(BaseEntity, CallbackEntity): def __init__( self, - device, - unique_id, - address, - device_enum, - device_desc, - entity_desc, - platform, - channel_no=None, + device: hm_device.HmDevice, + unique_id: str, + address: str, + device_enum: DeviceDescription, + device_desc: dict[str, Any], + entity_desc: dict[str, Any], + platform: HmPlatform, + channel_no: int | None = None, ): """ Initialize the entity. @@ -460,7 +482,7 @@ def _mark_entity(self, field_desc): if entity: entity.create_in_ha = True - def _add_entity(self, f_name, entity: GenericEntity): + def _add_entity(self, f_name, entity: GenericEntity | None): """Add entity to collection and register callback""" if not entity: return @@ -468,7 +490,7 @@ def _add_entity(self, f_name, entity: GenericEntity): entity.register_update_callback(self.update_entity) self.data_entities[f_name] = entity - def _remove_entity(self, f_name, entity: GenericEntity): + def _remove_entity(self, f_name, entity: GenericEntity | None): """Remove entity from collection and un-register callback""" if not entity: return @@ -513,13 +535,12 @@ class BaseEvent(BaseParameterEntity): def __init__( self, - device, - unique_id, - address, - parameter, - parameter_data, - event_type, - platform, + device: hm_device.HmDevice, + unique_id: str, + address: str, + parameter: str, + parameter_data: dict[str, Any], + event_type: HmEventType, ): """ Initialize the event handler. @@ -530,7 +551,7 @@ def __init__( address=address, parameter=parameter, parameter_data=parameter_data, - platform=platform, + platform=HmPlatform.EVENT, ) self.name = get_entity_name( @@ -541,7 +562,7 @@ def __init__( unique_id=self.unique_id, ) self.event_type = event_type - self.last_update = None + self.last_update: datetime = INIT_DATETIME self._value = None # Subscribe for all action events of this device @@ -610,7 +631,7 @@ def add_to_collections(self) -> None: self._device.add_hm_action_event(self) def _set_last_update(self) -> None: - self.last_update = datetime.datetime.now() + self.last_update = datetime.now() @abstractmethod def get_event_data(self, value=None): @@ -632,7 +653,14 @@ class AlarmEvent(BaseEvent): class for handling alarm events. """ - def __init__(self, device, unique_id, address, parameter, parameter_data): + def __init__( + self, + device: hm_device.HmDevice, + unique_id: str, + address: str, + parameter: str, + parameter_data: dict[str, Any], + ): """ Initialize the event handler. """ @@ -642,8 +670,7 @@ def __init__(self, device, unique_id, address, parameter, parameter_data): address=address, parameter=parameter, parameter_data=parameter_data, - event_type=EVENT_ALARM, - platform=HA_PLATFORM_EVENT, + event_type=HmEventType.ALARM, ) def get_event_data(self, value=None): @@ -678,7 +705,14 @@ class ClickEvent(BaseEvent): class for handling click events. """ - def __init__(self, device, unique_id, address, parameter, parameter_data): + def __init__( + self, + device: hm_device.HmDevice, + unique_id: str, + address: str, + parameter: str, + parameter_data: dict[str, Any], + ): """ Initialize the event handler. """ @@ -688,8 +722,7 @@ def __init__(self, device, unique_id, address, parameter, parameter_data): address=address, parameter=parameter, parameter_data=parameter_data, - event_type=EVENT_KEYPRESS, - platform=HA_PLATFORM_EVENT, + event_type=HmEventType.KEYPRESS, ) def get_event_data(self, value=None): @@ -718,7 +751,14 @@ class ImpulseEvent(BaseEvent): class for handling impulse events. """ - def __init__(self, device, unique_id, address, parameter, parameter_data): + def __init__( + self, + device: hm_device.HmDevice, + unique_id: str, + address: str, + parameter: str, + parameter_data: dict[str, Any], + ): """ Initialize the event handler. """ @@ -728,8 +768,7 @@ def __init__(self, device, unique_id, address, parameter, parameter_data): address=address, parameter=parameter, parameter_data=parameter_data, - event_type=EVENT_IMPULSE, - platform=HA_PLATFORM_EVENT, + event_type=HmEventType.IMPULSE, ) def get_event_data(self, value=None): diff --git a/hahomematic/hub.py b/hahomematic/hub.py index 575396f2..ba8504e0 100644 --- a/hahomematic/hub.py +++ b/hahomematic/hub.py @@ -2,23 +2,29 @@ from __future__ import annotations from abc import ABC -import datetime +from datetime import datetime import logging from typing import Any -from hahomematic.const import HA_DOMAIN +from hahomematic.const import BACKEND_CCU, HA_DOMAIN, INIT_DATETIME from hahomematic.helpers import generate_unique_id _LOGGER = logging.getLogger(__name__) EXCLUDED_FROM_SENSOR = [ - "DutyCycle", - "OldVal", + "Connection", "pcCCUID", "RF-Gateway-Alarm", "WatchDog", ] +EXCLUDED = [ + "CarrierSense", + "DutyCycle", + "OldVal", + "Servicemeldungen", +] + class BaseHubEntity(ABC): """ @@ -33,7 +39,7 @@ def __init__(self, central, unique_id, name, state=None): self.unique_id = unique_id self.name = name self._state = state - self.last_update = None + self.last_update: datetime = INIT_DATETIME self._update_callbacks = [] self._remove_callbacks = [] self.create_in_ha = True @@ -109,7 +115,7 @@ def remove_entity(self) -> None: _callback(self.unique_id) def _set_last_update(self) -> None: - self.last_update = datetime.datetime.now() + self.last_update = datetime.now() class HmSystemVariable(BaseHubEntity): @@ -160,7 +166,7 @@ def __init__(self, central, use_entities=False): self._use_entities = use_entities @property - def device_info(self) -> dict[str, str]: + def device_info(self) -> dict[str, Any]: """Return device specific attributes.""" return { "config_entry_id": self._central.entry_id, @@ -189,8 +195,8 @@ async def fetch_data(self): async def _update_hub_state(self): """Retrieve latest state.""" - service_message = await self._central.get_service_messages() - state = 0 if service_message is None else len(service_message) + service_messages = await self._central.get_service_messages() + state = 0 if service_messages is None else len(service_messages) if self._state != state: self._state = state @@ -202,8 +208,16 @@ async def _update_entities(self): variables = await self._central.get_all_system_variables() if not variables: return + + # remove some variables in case of CCU Backend + # - DutyCycle/CarrierSense are covered by real sensors + # - OldValue(s) are for internal calculations + # - Servicemeldungen are wrong in case of hmip (contains a dummy) + if self._central.model is BACKEND_CCU: + variables = _clean_variables(variables) + for name, value in variables.items(): - if not self._use_entities or _is_excluded(name): + if not self._use_entities or _is_excluded(name, EXCLUDED_FROM_SENSOR): self._variables[name] = value continue @@ -250,7 +264,7 @@ def __init__(self, central, use_entities=False): self._use_entities = use_entities @property - def device_info(self) -> dict[str, str]: + def device_info(self) -> dict[str, Any]: """Return device specific attributes.""" return { "config_entry_id": self._central.entry_id, @@ -282,9 +296,18 @@ async def set_system_variable(self, name, value): return -def _is_excluded(variable): - """Check if variable is excluded by EXCLUDED_FROM_SENSOR.""" - for marker in EXCLUDED_FROM_SENSOR: +def _is_excluded(variable, exclude_list): + """Check if variable is excluded by exclude_list.""" + for marker in exclude_list: if marker in variable: return True return False + + +def _clean_variables(variables): + cleaned_variables = {} + for name, value in variables.items(): + if _is_excluded(name, EXCLUDED): + continue + cleaned_variables[name] = value + return cleaned_variables diff --git a/hahomematic/internal/action.py b/hahomematic/internal/action.py index b2ad1fde..99726f42 100644 --- a/hahomematic/internal/action.py +++ b/hahomematic/internal/action.py @@ -3,7 +3,7 @@ import logging -from hahomematic.const import DATA_LOAD_SUCCESS, HA_PLATFORM_ACTION +from hahomematic.const import DATA_LOAD_SUCCESS, HmPlatform from hahomematic.entity import GenericEntity _LOGGER = logging.getLogger(__name__) @@ -22,7 +22,7 @@ def __init__(self, device, unique_id, address, parameter, parameter_data): address=address, parameter=parameter, parameter_data=parameter_data, - platform=HA_PLATFORM_ACTION, + platform=HmPlatform.ACTION, ) @property diff --git a/hahomematic/internal/text.py b/hahomematic/internal/text.py index 46c60bca..bcd39eb8 100644 --- a/hahomematic/internal/text.py +++ b/hahomematic/internal/text.py @@ -3,7 +3,7 @@ import logging -from hahomematic.const import ATTR_HM_VALUE, HA_PLATFORM_TEXT +from hahomematic.const import ATTR_HM_VALUE, HmPlatform from hahomematic.entity import GenericEntity _LOGGER = logging.getLogger(__name__) @@ -22,7 +22,7 @@ def __init__(self, device, unique_id, address, parameter, parameter_data): address=address, parameter=parameter, parameter_data=parameter_data, - platform=HA_PLATFORM_TEXT, + platform=HmPlatform.TEXT, ) @property diff --git a/hahomematic/json_rpc_client.py b/hahomematic/json_rpc_client.py index 978db747..a1ba7e55 100644 --- a/hahomematic/json_rpc_client.py +++ b/hahomematic/json_rpc_client.py @@ -61,7 +61,7 @@ async def login_or_renew(self): self._session_id = await self._renew_login(self._session_id) return self._session_id is not None - async def _renew_login(self, session_id) -> str: + async def _renew_login(self, session_id) -> str | None: """Renew JSON-RPC session or perform login.""" try: response = await self._post( @@ -78,9 +78,9 @@ async def _renew_login(self, session_id) -> str: ) return None - async def _login(self) -> str: + async def _login(self) -> str | None: """Login to CCU and return session.""" - session_id = False + session_id = None try: params = { ATTR_USERNAME: self._username, diff --git a/hahomematic/platforms/binary_sensor.py b/hahomematic/platforms/binary_sensor.py index 519fe6e9..d9bf7f4e 100644 --- a/hahomematic/platforms/binary_sensor.py +++ b/hahomematic/platforms/binary_sensor.py @@ -6,7 +6,7 @@ import logging -from hahomematic.const import HA_PLATFORM_BINARY_SENSOR +from hahomematic.const import HmPlatform from hahomematic.entity import GenericEntity _LOGGER = logging.getLogger(__name__) @@ -25,7 +25,7 @@ def __init__(self, device, unique_id, address, parameter, parameter_data): address=address, parameter=parameter, parameter_data=parameter_data, - platform=HA_PLATFORM_BINARY_SENSOR, + platform=HmPlatform.BINARY_SENSOR, ) @property diff --git a/hahomematic/platforms/button.py b/hahomematic/platforms/button.py index 17591e93..643fcb09 100644 --- a/hahomematic/platforms/button.py +++ b/hahomematic/platforms/button.py @@ -6,7 +6,7 @@ import logging -from hahomematic.const import HA_PLATFORM_BUTTON +from hahomematic.const import HmPlatform from hahomematic.entity import BaseParameterEntity _LOGGER = logging.getLogger(__name__) @@ -25,7 +25,7 @@ def __init__(self, device, unique_id, address, parameter, parameter_data): address=address, parameter=parameter, parameter_data=parameter_data, - platform=HA_PLATFORM_BUTTON, + platform=HmPlatform.BUTTON, ) async def press(self) -> None: diff --git a/hahomematic/platforms/number.py b/hahomematic/platforms/number.py index 835fdb2d..08bb6008 100644 --- a/hahomematic/platforms/number.py +++ b/hahomematic/platforms/number.py @@ -6,7 +6,7 @@ import logging -from hahomematic.const import ATTR_HM_VALUE, HA_PLATFORM_NUMBER +from hahomematic.const import ATTR_HM_VALUE, HmPlatform from hahomematic.entity import GenericEntity _LOGGER = logging.getLogger(__name__) @@ -25,7 +25,7 @@ def __init__(self, device, unique_id, address, parameter, parameter_data): address=address, parameter=parameter, parameter_data=parameter_data, - platform=HA_PLATFORM_NUMBER, + platform=HmPlatform.NUMBER, ) @property diff --git a/hahomematic/platforms/select.py b/hahomematic/platforms/select.py index 7f103565..6eae0964 100644 --- a/hahomematic/platforms/select.py +++ b/hahomematic/platforms/select.py @@ -6,7 +6,7 @@ import logging -from hahomematic.const import HA_PLATFORM_SELECT +from hahomematic.const import HmPlatform from hahomematic.entity import GenericEntity _LOGGER = logging.getLogger(__name__) @@ -25,7 +25,7 @@ def __init__(self, device, unique_id, address, parameter, parameter_data): address=address, parameter=parameter, parameter_data=parameter_data, - platform=HA_PLATFORM_SELECT, + platform=HmPlatform.SELECT, ) @property diff --git a/hahomematic/platforms/sensor.py b/hahomematic/platforms/sensor.py index 46672805..f8329dbd 100644 --- a/hahomematic/platforms/sensor.py +++ b/hahomematic/platforms/sensor.py @@ -6,7 +6,7 @@ import logging -from hahomematic.const import HA_PLATFORM_SENSOR +from hahomematic.const import HmPlatform from hahomematic.entity import GenericEntity _LOGGER = logging.getLogger(__name__) @@ -25,7 +25,7 @@ def __init__(self, device, unique_id, address, parameter, parameter_data): address=address, parameter=parameter, parameter_data=parameter_data, - platform=HA_PLATFORM_SENSOR, + platform=HmPlatform.SENSOR, ) @property diff --git a/hahomematic/platforms/switch.py b/hahomematic/platforms/switch.py index 1b166016..62bf035a 100644 --- a/hahomematic/platforms/switch.py +++ b/hahomematic/platforms/switch.py @@ -6,7 +6,7 @@ import logging -from hahomematic.const import HA_PLATFORM_SWITCH, TYPE_ACTION +from hahomematic.const import TYPE_ACTION, HmPlatform from hahomematic.entity import GenericEntity _LOGGER = logging.getLogger(__name__) @@ -25,7 +25,7 @@ def __init__(self, device, unique_id, address, parameter, parameter_data): address=address, parameter=parameter, parameter_data=parameter_data, - platform=HA_PLATFORM_SWITCH, + platform=HmPlatform.SWITCH, ) @property diff --git a/hahomematic/xml_rpc_server.py b/hahomematic/xml_rpc_server.py index 112f338b..eec79b36 100644 --- a/hahomematic/xml_rpc_server.py +++ b/hahomematic/xml_rpc_server.py @@ -28,7 +28,7 @@ _LOGGER = logging.getLogger(__name__) -_XML_RPC_SERVER: XMLRPCServer = None +_XML_RPC_SERVER: XMLRPCServer | None = None # pylint: disable=invalid-name @@ -181,8 +181,7 @@ async def _async_delete_devices(): del central.devices_raw_dict[interface_id][address] del central.paramsets_cache[interface_id][address] del central.names_cache[interface_id][address] - ha_device = central.hm_devices.get(address) - if ha_device: + if ha_device := central.hm_devices.get(address): ha_device.remove_event_subscriptions() del central.hm_devices[address] except KeyError: @@ -314,12 +313,12 @@ def un_register_central(self, central: hm_central.CentralUnit): if self._centrals.get(central.instance_name): del self._centrals[central.instance_name] - def get_central(self, interface_id) -> hm_central.CentralUnit: + def get_central(self, interface_id) -> hm_central.CentralUnit | None: """Return a central by interface_id""" for central in self._centrals.values(): - client = central.clients.get(interface_id) - if client: + if central.clients.get(interface_id): return central + return None @property def no_central_registered(self) -> bool: @@ -327,12 +326,12 @@ def no_central_registered(self) -> bool: return len(self._centrals) == 0 -def get_xml_rpc_server() -> XMLRPCServer: +def get_xml_rpc_server() -> XMLRPCServer | None: """Return the XMLRPCServer.""" return _XML_RPC_SERVER -def _set_xml_rpc_server(xml_rpc_server: XMLRPCServer) -> None: +def _set_xml_rpc_server(xml_rpc_server: XMLRPCServer | None) -> None: """Add a XMLRPCServer.""" # pylint: disable=global-statement global _XML_RPC_SERVER @@ -341,8 +340,7 @@ def _set_xml_rpc_server(xml_rpc_server: XMLRPCServer) -> None: def register_xml_rpc_server(local_ip=IP_ANY_V4, local_port=PORT_ANY) -> XMLRPCServer: """Register the xml rpc server.""" - xml_rpc = get_xml_rpc_server() - if not xml_rpc: + if (xml_rpc := get_xml_rpc_server()) is None: xml_rpc = XMLRPCServer(local_ip, local_port) xml_rpc.start() _set_xml_rpc_server(xml_rpc) @@ -353,7 +351,7 @@ async def un_register_xml_rpc_server() -> bool: """Unregister the xml rpc server.""" xml_rpc = get_xml_rpc_server() _LOGGER.info("XMLRPCServer.stop: Shutting down server") - if xml_rpc.no_central_registered: + if xml_rpc and xml_rpc.no_central_registered: await xml_rpc.stop() _set_xml_rpc_server(None) _LOGGER.info("XMLRPCServer.stop: Server stopped") diff --git a/setup.py b/setup.py index 018bf79d..c1b06377 100644 --- a/setup.py +++ b/setup.py @@ -12,7 +12,7 @@ def readme(): PACKAGE_NAME = "hahomematic" HERE = os.path.abspath(os.path.dirname(__file__)) -VERSION = "0.0.16" +VERSION = "0.0.17" PACKAGES = find_packages(exclude=["tests", "tests.*", "dist", "build"])