Use Typed dict for device_description (#1678)

* Use Typed dict for device_description

* Fixes

* Update changelog.md

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)
 

hahomematic/caches/persistent.py

@@ -21,7 +21,7 @@
     FILE_PARAMSETS,
     INIT_DATETIME,
     DataOperationResult,
-    Description,
+    DeviceDescription,
     ParameterData,
     ParamsetKey,
 )
@@ -150,34 +150,34 @@ 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,
         )
         # {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:
             self._raw_device_descriptions[interface_id] = []
 
         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)
 
         self._convert_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,15 +270,15 @@ 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:
             self._addresses[interface_id] = {}
         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]:

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(

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,21 +766,19 @@ 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. "
                 "Not updating paramsets for %s",
                 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: