From bee0b6cb0f787da27b46748326e257cb58537546 Mon Sep 17 00:00:00 2001 From: SukramJ Date: Thu, 29 Aug 2024 15:01:14 +0200 Subject: [PATCH] Use Typed dict for device_description (#1678) * Use Typed dict for device_description * Fixes * Update changelog.md --- changelog.md | 5 +- hahomematic/caches/persistent.py | 77 ++++++++++++------------ hahomematic/central/__init__.py | 12 ++-- hahomematic/client/__init__.py | 27 ++++----- hahomematic/const.py | 66 ++++++++++---------- hahomematic/platforms/device.py | 100 ++++++++++--------------------- 6 files changed, 127 insertions(+), 160 deletions(-) diff --git a/changelog.md b/changelog.md index b0045558..5ab74154 100644 --- a/changelog.md +++ b/changelog.md @@ -1,9 +1,10 @@ # Version 2024.8.15 (2024-08-26) -- Small definition fix for DALI -- Use ParameterData for paramset descriptions - Avoid permanent cache save on remove device - Ensure only one load/save of cache file at time +- Small definition fix for DALI +- Use ParameterData for paramset descriptions +- Use Typed dict for device_description # Version 2024.8.14 (2024-08-26) diff --git a/hahomematic/caches/persistent.py b/hahomematic/caches/persistent.py index d43358d4..2678afd1 100644 --- a/hahomematic/caches/persistent.py +++ b/hahomematic/caches/persistent.py @@ -21,7 +21,7 @@ FILE_PARAMSETS, INIT_DATETIME, DataOperationResult, - Description, + DeviceDescription, ParameterData, ParamsetKey, ) @@ -150,7 +150,7 @@ class DeviceDescriptionCache(BasePersistentCache): def __init__(self, central: hmcu.CentralUnit) -> None: """Init the device description cache.""" # {interface_id, [device_descriptions]} - self._raw_device_descriptions: Final[dict[str, list[dict[str, Any]]]] = {} + self._raw_device_descriptions: Final[dict[str, list[DeviceDescription]]] = {} super().__init__( central=central, persistant_cache=self._raw_device_descriptions, @@ -158,10 +158,10 @@ def __init__(self, central: hmcu.CentralUnit) -> None: # {interface_id, {device_address, [channel_address]}} self._addresses: Final[dict[str, dict[str, set[str]]]] = {} # {interface_id, {address, device_descriptions}} - self._device_descriptions: Final[dict[str, dict[str, dict[str, Any]]]] = {} + self._device_descriptions: Final[dict[str, dict[str, DeviceDescription]]] = {} def add_device_description( - self, interface_id: str, device_description: dict[str, Any] + self, interface_id: str, device_description: DeviceDescription ) -> None: """Add device_description to cache.""" if interface_id not in self._raw_device_descriptions: @@ -169,7 +169,7 @@ def add_device_description( self._remove_device( interface_id=interface_id, - deleted_addresses=[device_description[Description.ADDRESS]], + deleted_addresses=[device_description["ADDRESS"]], ) self._raw_device_descriptions[interface_id].append(device_description) @@ -177,7 +177,7 @@ def add_device_description( interface_id=interface_id, device_description=device_description ) - def get_raw_device_descriptions(self, interface_id: str) -> list[dict[str, Any]]: + def get_raw_device_descriptions(self, interface_id: str) -> list[DeviceDescription]: """Find raw device in cache.""" return self._raw_device_descriptions.get(interface_id, []) @@ -190,9 +190,9 @@ def remove_device(self, device: HmDevice) -> None: def _remove_device(self, interface_id: str, deleted_addresses: list[str]) -> None: """Remove device from cache.""" self._raw_device_descriptions[interface_id] = [ - raw_device - for raw_device in self.get_raw_device_descriptions(interface_id) - if raw_device[Description.ADDRESS] not in deleted_addresses + device_descriptions + for device_descriptions in self.get_raw_device_descriptions(interface_id) + if device_descriptions["ADDRESS"] not in deleted_addresses ] for address in deleted_addresses: @@ -212,57 +212,56 @@ def get_channels(self, interface_id: str, device_address: str) -> Mapping[str, C """Return the device channels by interface and device_address.""" channels: dict[str, Channel] = {} for channel_address in self._addresses.get(interface_id, {}).get(device_address, set()): - channel_name = str( - self.get_device_parameter( - interface_id=interface_id, - device_address=channel_address, - parameter=Description.TYPE, - ) + device_description = self.get_device_description( + interface_id=interface_id, + address=channel_address, + ) + channels[channel_address] = Channel( + type=device_description["TYPE"], address=channel_address ) - channels[channel_address] = Channel(type=channel_name, address=channel_address) return channels - def get_device_descriptions(self, interface_id: str) -> dict[str, dict[str, Any]]: + def get_device_descriptions(self, interface_id: str) -> dict[str, DeviceDescription]: """Return the devices by interface.""" return self._device_descriptions.get(interface_id, {}) - def get_device(self, interface_id: str, device_address: str) -> dict[str, Any]: - """Return the device dict by interface and device_address.""" - return self._device_descriptions.get(interface_id, {}).get(device_address, {}) + def find_device_description( + self, interface_id: str, device_address: str + ) -> DeviceDescription | None: + """Return the device description by interface and device_address.""" + return self._device_descriptions.get(interface_id, {}).get(device_address) + + def get_device_description(self, interface_id: str, address: str) -> DeviceDescription: + """Return the device description by interface and device_address.""" + return self._device_descriptions[interface_id][address] def get_device_with_channels( self, interface_id: str, device_address: str - ) -> Mapping[str, Any]: + ) -> Mapping[str, DeviceDescription]: """Return the device dict by interface and device_address.""" - data: dict[str, Any] = { - device_address: self._device_descriptions.get(interface_id, {}).get(device_address, {}) + device_descriptions: dict[str, DeviceDescription] = { + device_address: self.get_device_description( + interface_id=interface_id, address=device_address + ) } - children = data[device_address][Description.CHILDREN] + children = device_descriptions[device_address]["CHILDREN"] for channel_address in children: - data[channel_address] = self._device_descriptions.get(interface_id, {}).get( - channel_address, {} + device_descriptions[channel_address] = self.get_device_description( + interface_id=interface_id, address=channel_address ) - return data + return device_descriptions @lru_cache def get_device_type(self, device_address: str) -> str | None: """Return the device type.""" for data in self._device_descriptions.values(): if items := data.get(device_address): - return items[Description.TYPE] # type: ignore[no-any-return] + return items["TYPE"] return None - def get_device_parameter( - self, interface_id: str, device_address: str, parameter: str - ) -> Any | None: - """Return the device parameter by interface and device_address.""" - return ( - self._device_descriptions.get(interface_id, {}).get(device_address, {}).get(parameter) - ) - def _convert_device_descriptions( - self, interface_id: str, device_descriptions: list[dict[str, Any]] + self, interface_id: str, device_descriptions: list[DeviceDescription] ) -> None: """Convert provided list of device descriptions.""" for device_description in device_descriptions: @@ -271,7 +270,7 @@ def _convert_device_descriptions( ) def _convert_device_description( - self, interface_id: str, device_description: dict[str, Any] + self, interface_id: str, device_description: DeviceDescription ) -> None: """Convert provided dict of device descriptions.""" if interface_id not in self._addresses: @@ -279,7 +278,7 @@ def _convert_device_description( if interface_id not in self._device_descriptions: self._device_descriptions[interface_id] = {} - address = device_description[Description.ADDRESS] + address = device_description["ADDRESS"] self._device_descriptions[interface_id][address] = device_description if ":" not in address and address not in self._addresses[interface_id]: diff --git a/hahomematic/central/__init__.py b/hahomematic/central/__init__.py index 5eec2835..f80803c0 100644 --- a/hahomematic/central/__init__.py +++ b/hahomematic/central/__init__.py @@ -46,7 +46,7 @@ PORT_ANY, UN_IGNORE_WILDCARD, BackendSystemEvent, - Description, + DeviceDescription, DeviceFirmwareState, HmPlatform, HomematicEventType, @@ -795,7 +795,7 @@ async def delete_devices(self, interface_id: str, addresses: tuple[str, ...]) -> @callback_backend_system(system_event=BackendSystemEvent.NEW_DEVICES) async def add_new_devices( - self, interface_id: str, device_descriptions: tuple[dict[str, Any], ...] + self, interface_id: str, device_descriptions: tuple[DeviceDescription, ...] ) -> None: """Add new devices to central unit.""" await self._add_new_devices( @@ -804,7 +804,7 @@ async def add_new_devices( @measure_execution_time async def _add_new_devices( - self, interface_id: str, device_descriptions: tuple[dict[str, Any], ...] + self, interface_id: str, device_descriptions: tuple[DeviceDescription, ...] ) -> None: """Add new devices to central unit.""" _LOGGER.debug( @@ -823,7 +823,7 @@ async def _add_new_devices( async with self._sema_add_devices: # We need this to avoid adding duplicates. known_addresses = tuple( - dev_desc[Description.ADDRESS] + dev_desc["ADDRESS"] for dev_desc in self.device_descriptions.get_raw_device_descriptions( interface_id=interface_id ) @@ -837,7 +837,7 @@ async def _add_new_devices( interface_id=interface_id, device_description=dev_desc ) save_device_descriptions = True - if dev_desc[Description.ADDRESS] not in known_addresses: + if dev_desc["ADDRESS"] not in known_addresses: await client.fetch_paramset_descriptions(device_description=dev_desc) save_paramset_descriptions = True except Exception as err: # pragma: no cover @@ -953,7 +953,7 @@ async def event( ) @callback_backend_system(system_event=BackendSystemEvent.LIST_DEVICES) - def list_devices(self, interface_id: str) -> list[dict[str, Any]]: + def list_devices(self, interface_id: str) -> list[DeviceDescription]: """Return already existing devices to CCU / Homegear.""" result = self.device_descriptions.get_raw_device_descriptions(interface_id=interface_id) _LOGGER.debug( diff --git a/hahomematic/client/__init__.py b/hahomematic/client/__init__.py index bd8b196a..b9f24e15 100644 --- a/hahomematic/client/__init__.py +++ b/hahomematic/client/__init__.py @@ -25,6 +25,7 @@ Backend, CallSource, Description, + DeviceDescription, ForcedDeviceAvailability, InterfaceEventType, InterfaceName, @@ -334,7 +335,7 @@ def get_virtual_remote(self) -> HmDevice | None: return None @measure_execution_time - async def get_all_device_descriptions(self) -> tuple[dict[str, Any]] | None: + async def get_all_device_descriptions(self) -> tuple[DeviceDescription] | None: """Get device descriptions from CCU / Homegear.""" try: return tuple(await self._proxy.listDevices()) @@ -344,7 +345,7 @@ async def get_all_device_descriptions(self) -> tuple[dict[str, Any]] | None: ) return None - async def get_device_description(self, device_address: str) -> tuple[dict[str, Any]] | None: + async def get_device_description(self, device_address: str) -> tuple[DeviceDescription] | None: """Get device descriptions from CCU / Homegear.""" try: if device_description := await self._proxy_read.getDeviceDescription(device_address): @@ -658,7 +659,7 @@ async def fetch_paramset_description( paramset_description=paramset_description, ) - async def fetch_paramset_descriptions(self, device_description: dict[str, Any]) -> None: + async def fetch_paramset_descriptions(self, device_description: DeviceDescription) -> None: """Fetch paramsets for provided device description.""" data = await self.get_paramset_descriptions(device_description=device_description) for address, paramsets in data.items(): @@ -672,16 +673,14 @@ async def fetch_paramset_descriptions(self, device_description: dict[str, Any]) ) async def get_paramset_descriptions( - self, device_description: dict[str, Any] + self, device_description: DeviceDescription ) -> dict[str, dict[ParamsetKey, dict[str, ParameterData]]]: """Get paramsets for provided device description.""" - if not device_description: - return {} paramsets: dict[str, dict[ParamsetKey, dict[str, ParameterData]]] = {} - address = device_description[Description.ADDRESS] + address = device_description["ADDRESS"] paramsets[address] = {} _LOGGER.debug("GET_PARAMSET_DESCRIPTIONS for %s", address) - for p_key in device_description.get(Description.PARAMSETS, []): + for p_key in device_description["PARAMSETS"]: if (paramset_key := ParamsetKey(p_key)) not in DEFAULT_PARAMSETS: continue if paramset_description := await self._get_paramset_description( @@ -712,7 +711,7 @@ async def _get_paramset_description( return None async def get_all_paramset_descriptions( - self, device_descriptions: tuple[dict[str, Any], ...] + self, device_descriptions: tuple[DeviceDescription, ...] ) -> dict[str, dict[ParamsetKey, dict[str, ParameterData]]]: """Get all paramset descriptions for provided device descriptions.""" all_paramsets: dict[str, dict[ParamsetKey, dict[str, ParameterData]]] = {} @@ -767,9 +766,12 @@ async def update_paramset_descriptions(self, device_address: str) -> None: device_address, ) return - if not self.central.device_descriptions.get_device( + + if device_description := self.central.device_descriptions.find_device_description( interface_id=self.interface_id, device_address=device_address ): + await self.fetch_paramset_descriptions(device_description=device_description) + else: _LOGGER.warning( "UPDATE_PARAMSET_DESCRIPTIONS failed: " "Channel missing in central.cache. " @@ -777,11 +779,6 @@ async def update_paramset_descriptions(self, device_address: str) -> None: device_address, ) return - await self.fetch_paramset_descriptions( - device_description=self.central.device_descriptions.get_device( - interface_id=self.interface_id, device_address=device_address - ) - ) await self.central.save_caches(save_paramset_descriptions=True) def __str__(self) -> str: diff --git a/hahomematic/const.py b/hahomematic/const.py index ca5e76a0..62993dc4 100644 --- a/hahomematic/const.py +++ b/hahomematic/const.py @@ -6,7 +6,7 @@ from dataclasses import dataclass, field from datetime import datetime from enum import Enum, IntEnum, StrEnum -from typing import Any, Final +from typing import Any, Final, Required, TypedDict DEFAULT_CONNECTION_CHECKER_INTERVAL: Final = 15 # check if connection is available via rpc ping DEFAULT_CUSTOM_ID: Final = "custom_id" @@ -159,6 +159,7 @@ class DeviceFirmwareState(StrEnum): READY_FOR_UPDATE = "READY_FOR_UPDATE" DO_UPDATE_PENDING = "DO_UPDATE_PENDING" PERFORMING_UPDATE = "PERFORMING_UPDATE" + BACKGROUND_UPDATE_NOT_SUPPORTED = "BACKGROUND_UPDATE_NOT_SUPPORTED" class EntityUsage(StrEnum): @@ -558,19 +559,6 @@ class SystemInformation: class ParameterData: """Dataclass for parameter data.""" - __slots__ = ( - "default", - "flags", - "id", - "max", - "min", - "operations", - "special", - "hm_type", - "unit", - "value_list", - ) - def __init__(self, data: Mapping[str, Any]) -> None: """Init the dataclass from mapping.""" self.default: Final[Any] = data[Description.DEFAULT] @@ -586,20 +574,36 @@ def __init__(self, data: Mapping[str, Any]) -> None: def as_dict(self) -> dict[str, Any]: """Return dataclass as dict.""" - data_dict = { - Description.DEFAULT.value: self.default, - Description.FLAGS.value: self.flags, - Description.MAX.value: self.max, - Description.MIN.value: self.min, - Description.OPERATIONS.value: self.operations, - Description.TYPE.value: self.hm_type, - } - if self.id: - data_dict[Description.ID.value] = self.id - if self.special: - data_dict[Description.SPECIAL.value] = self.special - if self.unit: - data_dict[Description.UNIT.value] = self.unit - if self.value_list: - data_dict[Description.VALUE_LIST.value] = self.value_list - return dict(sorted(data_dict.items())) + return {key.upper(): str(value) for key, value in self.__dict__.items() if value} + + +class DeviceDescription(TypedDict, total=False): + """Typed dict for device descriptions.""" + + TYPE: Required[str] + SUBTYPE: str | None + ADDRESS: Required[str] + # RF_ADDRESS: int | None + CHILDREN: Required[list[str]] + PARENT: Required[str] + # PARENT_TYPE: str | None + # INDEX: int | None + # AES_ACTIVE: int | None + PARAMSETS: Required[list[str]] + FIRMWARE: str + AVAILABLE_FIRMWARE: str | None + UPDATABLE: bool | None + FIRMWARE_UPDATE_STATE: str | None + FIRMWARE_UPDATABLE: bool | None + # VERSION: Required[int] + # FLAGS: Required[int] + # LINK_SOURCE_ROLES: str | None + # LINK_TARGET_ROLES: str | None + # DIRECTION: int | None + # GROUP: str | None + # TEAM: str | None + # TEAM_TAG: str | None + # TEAM_CHANNELS: list + INTERFACE: str | None + # ROAMING: int | None + RX_MODE: int | None diff --git a/hahomematic/platforms/device.py b/hahomematic/platforms/device.py index b8768006..4eb21a10 100644 --- a/hahomematic/platforms/device.py +++ b/hahomematic/platforms/device.py @@ -30,7 +30,7 @@ VIRTUAL_REMOTE_TYPES, CallSource, DataOperationResult, - Description, + DeviceDescription, DeviceFirmwareState, EntityUsage, ForcedDeviceAvailability, @@ -38,6 +38,7 @@ HomematicEventType, Manufacturer, Parameter, + ParameterData, ParamsetKey, ProductGroup, ) @@ -91,20 +92,13 @@ def __init__(self, central: hmcu.CentralUnit, interface_id: str, device_address: self._forced_availability: ForcedDeviceAvailability = ForcedDeviceAvailability.NOT_SET self._device_updated_callbacks: Final[list[Callable]] = [] self._firmware_update_callbacks: Final[list[Callable]] = [] - self._device_type: Final = str( - self.central.device_descriptions.get_device_parameter( - interface_id=interface_id, - device_address=device_address, - parameter=Description.TYPE, - ) - ) - self._sub_type: Final = str( - central.device_descriptions.get_device_parameter( - interface_id=interface_id, - device_address=device_address, - parameter=Description.SUBTYPE, - ) + + device_description = self.central.device_descriptions.get_device_description( + interface_id=interface_id, address=device_address ) + self._device_type: Final = device_description["TYPE"] + self._sub_type: Final = device_description.get("SUBTYPE") + self._ignore_for_custom_entity: Final[bool] = ( central.parameter_visibility.device_type_is_ignored(device_type=self._device_type) ) @@ -136,42 +130,17 @@ def __init__(self, central: hmcu.CentralUnit, interface_id: str, device_address: def _update_firmware_data(self) -> None: """Update firmware related data from device descriptions.""" - self._available_firmware: str | None = ( - self.central.device_descriptions.get_device_parameter( - interface_id=self._interface_id, - device_address=self._device_address, - parameter=Description.AVAILABLE_FIRMWARE, - ) - or None + device_description = self.central.device_descriptions.get_device_description( + interface_id=self._interface_id, + address=self._device_address, ) - self._firmware = str( - self.central.device_descriptions.get_device_parameter( - interface_id=self._interface_id, - device_address=self._device_address, - parameter=Description.FIRMWARE, - ) + self._available_firmware = str(device_description.get("AVAILABLE_FIRMWARE", "")) + self._firmware = device_description["FIRMWARE"] + self._firmware_update_state = DeviceFirmwareState( + device_description.get("FIRMWARE_UPDATE_STATE") or DeviceFirmwareState.UP_TO_DATE ) - try: - self._firmware_update_state = DeviceFirmwareState( - str( - self.central.device_descriptions.get_device_parameter( - interface_id=self._interface_id, - device_address=self._device_address, - parameter=Description.FIRMWARE_UPDATE_STATE, - ) - ) - ) - except ValueError: - self._firmware_update_state = DeviceFirmwareState.UP_TO_DATE - - self._firmware_updatable = bool( - self.central.device_descriptions.get_device_parameter( - interface_id=self._interface_id, - device_address=self._device_address, - parameter=Description.FIRMWARE_UPDATABLE, - ) - ) + self._firmware_updatable = device_description.get("FIRMWARE_UPDATABLE") or False def _identify_manufacturer(self) -> Manufacturer: """Identify the manufacturer of a device.""" @@ -322,7 +291,7 @@ def rooms(self) -> set[str]: return self._rooms @config_property - def sub_type(self) -> str: + def sub_type(self) -> str | None: """Return the sub_type of the device.""" return self._sub_type @@ -816,41 +785,38 @@ def __init__(self, device: HmDevice) -> None: async def export_data(self) -> None: """Export data.""" - device_descriptions: Mapping[str, Any] = ( + device_descriptions: Mapping[str, DeviceDescription] = ( self._central.device_descriptions.get_device_with_channels( interface_id=self._interface_id, device_address=self._device_address ) ) - paramset_descriptions: Mapping[ - str, Any + paramset_descriptions: dict[ + str, dict[ParamsetKey, dict[str, ParameterData]] ] = await self._client.get_all_paramset_descriptions( device_descriptions=tuple(device_descriptions.values()) ) - device_type = device_descriptions[self._device_address][Description.TYPE] + device_type = device_descriptions[self._device_address]["TYPE"] filename = f"{device_type}.json" # anonymize device_descriptions - anonymize_device_descriptions: list[Any] = [] + anonymize_device_descriptions: list[DeviceDescription] = [] for device_description in device_descriptions.values(): - if device_description == {}: - continue # pragma: no cover - new_device_description = copy(device_description) - new_device_description[Description.ADDRESS] = self._anonymize_address( - address=new_device_description[Description.ADDRESS] + new_device_description: DeviceDescription = copy(device_description) + new_device_description["ADDRESS"] = self._anonymize_address( + address=new_device_description["ADDRESS"] ) - if new_device_description.get(Description.PARENT): - new_device_description[Description.PARENT] = new_device_description[ - Description.ADDRESS - ].split(":")[0] - elif new_device_description.get(Description.CHILDREN): - new_device_description[Description.CHILDREN] = [ - self._anonymize_address(a) - for a in new_device_description[Description.CHILDREN] + if new_device_description.get("PARENT"): + new_device_description["PARENT"] = new_device_description["ADDRESS"].split(":")[0] + elif new_device_description.get("CHILDREN"): + new_device_description["CHILDREN"] = [ + self._anonymize_address(a) for a in new_device_description["CHILDREN"] ] anonymize_device_descriptions.append(new_device_description) # anonymize paramset_descriptions - anonymize_paramset_descriptions: dict[str, Any] = {} + anonymize_paramset_descriptions: dict[ + str, dict[ParamsetKey, dict[str, ParameterData]] + ] = {} for address, paramset_description in paramset_descriptions.items(): anonymize_paramset_descriptions[self._anonymize_address(address=address)] = ( paramset_description