Switch typing of paramset_key from str to ParamsetKey (#1666)

* Add paramset_key to entity_key

* Switch typing of paramset_key from str to ParamsetKey

* Avoid unnecessary cast

changelog.md

@@ -1,3 +1,8 @@
+# Version 2024.8.14 (2024-08-26)
+
+- Add paramset_key to entity_key
+- Switch typing of paramset_key from str to ParamsetKey
+
 # Version 2024.8.13 (2024-08-25)
 
 - Check/convert values of manual executed put_paramset/set_value

hahomematic/caches/dynamic.py

@@ -59,19 +59,21 @@ def add_set_value(
 
         entity_key = get_entity_key(
             channel_address=channel_address,
+            paramset_key=ParamsetKey.VALUES,
             parameter=parameter,
         )
         self._last_send_command[entity_key] = (value, datetime.now())
         return {entity_key}
 
     def add_put_paramset(
-        self, channel_address: str, paramset_key: str, values: dict[str, Any]
+        self, channel_address: str, paramset_key: ParamsetKey, values: dict[str, Any]
     ) -> set[ENTITY_KEY]:
         """Add data from put paramset command."""
         entity_keys: set[ENTITY_KEY] = set()
         for parameter, value in values.items():
             entity_key = get_entity_key(
                 channel_address=channel_address,
+                paramset_key=paramset_key,
                 parameter=parameter,
             )
             self._last_send_command[entity_key] = (value, datetime.now())
@@ -254,7 +256,7 @@ async def load(self, direct_call: bool = False) -> None:
         for client in self._central.clients:
             await client.fetch_all_device_data()
 
-    async def refresh_entity_data(self, paramset_key: str | None = None) -> None:
+    async def refresh_entity_data(self, paramset_key: ParamsetKey | None = None) -> None:
         """Refresh entity data."""
         for entity in self._central.get_readable_generic_entities(paramset_key=paramset_key):
             await entity.load_entity_value(call_source=CallSource.HM_INIT)

hahomematic/caches/persistent.py

@@ -21,6 +21,7 @@
     INIT_DATETIME,
     DataOperationResult,
     Description,
+    ParamsetKey,
 )
 from hahomematic.platforms.device import HmDevice
 from hahomematic.support import (
@@ -278,7 +279,7 @@ def __init__(self, central: hmcu.CentralUnit) -> None:
         """Init the paramset description cache."""
         # {interface_id, {channel_address, paramsets}}
         self._raw_paramset_descriptions: Final[
-            dict[str, dict[str, dict[str, dict[str, Any]]]]
+            dict[str, dict[str, dict[ParamsetKey, dict[str, Any]]]]
         ] = {}
         super().__init__(
             central=central,
@@ -289,15 +290,15 @@ def __init__(self, central: hmcu.CentralUnit) -> None:
         self._address_parameter_cache: Final[dict[tuple[str, str], set[int | None]]] = {}
 
     @property
-    def raw_paramset_descriptions(self) -> dict[str, dict[str, dict[str, dict[str, Any]]]]:
+    def raw_paramset_descriptions(self) -> dict[str, dict[str, dict[ParamsetKey, dict[str, Any]]]]:
         """Return the paramset descriptions."""
         return self._raw_paramset_descriptions
 
     def add(
         self,
         interface_id: str,
         channel_address: str,
-        paramset_key: str,
+        paramset_key: ParamsetKey,
         paramset_description: dict[str, Any],
     ) -> None:
         """Add paramset description to cache."""
@@ -329,14 +330,16 @@ def has_interface_id(self, interface_id: str) -> bool:
         """Return if interface is in paramset_descriptions cache."""
         return interface_id in self._raw_paramset_descriptions
 
-    def get_paramset_keys(self, interface_id: str, channel_address: str) -> tuple[str, ...]:
+    def get_paramset_keys(
+        self, interface_id: str, channel_address: str
+    ) -> tuple[ParamsetKey, ...]:
         """Get paramset_keys from paramset descriptions cache."""
         return tuple(
             self._raw_paramset_descriptions.get(interface_id, {}).get(channel_address, [])
         )
 
     def get_paramset_descriptions(
-        self, interface_id: str, channel_address: str, paramset_key: str
+        self, interface_id: str, channel_address: str, paramset_key: ParamsetKey
     ) -> dict[str, Any]:
         """Get paramset descriptions from cache."""
         return (
@@ -346,7 +349,7 @@ def get_paramset_descriptions(
         )
 
     def get_parameter_data(
-        self, interface_id: str, channel_address: str, paramset_key: str, parameter: str
+        self, interface_id: str, channel_address: str, paramset_key: ParamsetKey, parameter: str
     ) -> Any:
         """Get parameter_data  from cache."""
         return (
@@ -368,17 +371,17 @@ def is_in_multiple_channels(self, channel_address: str, parameter: str) -> bool:
 
     def get_channel_addresses_by_paramset_key(
         self, interface_id: str, device_address: str
-    ) -> dict[str, list[str]]:
+    ) -> dict[ParamsetKey, list[str]]:
         """Get device channel addresses."""
-        channel_addresses: dict[str, list[str]] = {}
+        channel_addresses: dict[ParamsetKey, list[str]] = {}
         interface_paramset_descriptions = self._raw_paramset_descriptions[interface_id]
         for (
             channel_address,
             paramset_descriptions,
         ) in interface_paramset_descriptions.items():
             if channel_address.startswith(device_address):
-                for paramset_key in paramset_descriptions:
-                    if paramset_key not in channel_addresses:
+                for p_key in paramset_descriptions:
+                    if (paramset_key := ParamsetKey(p_key)) not in channel_addresses:
                         channel_addresses[paramset_key] = []
                     channel_addresses[paramset_key].append(channel_address)
 

hahomematic/caches/visibility.py

@@ -322,7 +322,7 @@ def __init__(
 
         # device_type, channel_no, paramset_key, set[parameter]
         self._un_ignore_parameters_by_device_paramset_key: Final[
-            dict[str, dict[int | None, dict[str, set[str]]]]
+            dict[str, dict[int | None, dict[ParamsetKey, set[str]]]]
         ] = {}
 
         # device_type, channel_no
@@ -373,7 +373,7 @@ def parameter_is_ignored(
         self,
         device_type: str,
         channel_no: int | None,
-        paramset_key: str,
+        paramset_key: ParamsetKey,
         parameter: str,
     ) -> bool:
         """Check if parameter can be ignored."""
@@ -437,7 +437,7 @@ def _parameter_is_un_ignored(
         self,
         device_type: str,
         channel_no: int | str | None,
-        paramset_key: str,
+        paramset_key: ParamsetKey,
         parameter: str,
         custom_only: bool = False,
     ) -> bool:
@@ -495,7 +495,7 @@ def parameter_is_un_ignored(
         self,
         device_type: str,
         channel_no: int | None,
-        paramset_key: str,
+        paramset_key: ParamsetKey,
         parameter: str,
         custom_only: bool = False,
     ) -> bool:
@@ -557,7 +557,7 @@ def _add_line_to_cache(self, line: str) -> None:
 
     def _get_un_ignore_line_details(
         self, line: str
-    ) -> tuple[str, int | str | None, str, str] | str | None:
+    ) -> tuple[str, int | str | None, str, ParamsetKey] | str | None:
         """
         Check the format of the line for un_ignore file.
 
@@ -566,7 +566,7 @@ def _get_un_ignore_line_details(
 
         device_type: str | None = None
         channel_no: int | str | None = None
-        paramset_key: str | None = None
+        paramset_key: ParamsetKey | None = None
         parameter: str | None = None
 
         if "@" in line:
@@ -576,7 +576,7 @@ def _get_un_ignore_line_details(
                     param_data = data[0].split(":")
                     if len(param_data) == 2:
                         parameter = param_data[0]
-                        paramset_key = param_data[1]
+                        paramset_key = ParamsetKey(param_data[1])
                     else:
                         _LOGGER.warning(
                             "GET_UN_IGNORE_LINE_DETAILS failed: Could not add line '%s' to un ignore cache. "
@@ -641,7 +641,11 @@ def _get_un_ignore_line_details(
         return line
 
     def _add_complex_un_ignore_entry(
-        self, device_type: str, channel_no: int | str | None, paramset_key: str, parameter: str
+        self,
+        device_type: str,
+        channel_no: int | str | None,
+        paramset_key: ParamsetKey,
+        parameter: str,
     ) -> None:
         """Add line to un ignore cache."""
         if paramset_key == ParamsetKey.MASTER:
@@ -676,7 +680,7 @@ def parameter_is_hidden(
         self,
         device_type: str,
         channel_no: int | None,
-        paramset_key: str,
+        paramset_key: ParamsetKey,
         parameter: str,
     ) -> bool:
         """
@@ -695,7 +699,7 @@ def parameter_is_hidden(
     def is_relevant_paramset(
         self,
         device_type: str,
-        paramset_key: str,
+        paramset_key: ParamsetKey,
         channel_no: int | None,
     ) -> bool:
         """

hahomematic/central/__init__.py

@@ -621,7 +621,7 @@ def get_entities(
         return tuple(all_entities)
 
     def get_readable_generic_entities(
-        self, paramset_key: str | None = None
+        self, paramset_key: ParamsetKey | None = None
     ) -> tuple[GenericEntity, ...]:
         """Return the readable generic entities."""
         return tuple(
@@ -909,6 +909,7 @@ async def event(
 
         entity_key = get_entity_key(
             channel_address=channel_address,
+            paramset_key=ParamsetKey.VALUES,
             parameter=parameter,
         )
 
@@ -989,7 +990,7 @@ async def fetch_program_data(self, include_internal: bool = False) -> None:
         await self._hub.fetch_program_data(include_internal=include_internal)
 
     @measure_execution_time
-    async def load_and_refresh_entity_data(self, paramset_key: str | None = None) -> None:
+    async def load_and_refresh_entity_data(self, paramset_key: ParamsetKey | None = None) -> None:
         """Refresh entity data."""
         if paramset_key != ParamsetKey.MASTER and self.data_cache.is_empty:
             await self.data_cache.load()
@@ -1038,16 +1039,14 @@ def get_parameters(
     ) -> list[str]:
         """Return all parameters from VALUES paramset."""
         parameters: set[str] = set()
-        for channels in self.paramset_descriptions.raw_paramset_descriptions.values():  # pylint: disable=too-many-nested-blocks
+        for channels in self.paramset_descriptions.raw_paramset_descriptions.values():
             for channel_address in channels:
                 device_type: str | None = None
                 if full_format:
                     device_type = self.device_descriptions.get_device_type(
                         device_address=get_device_address(channel_address)
                     )
-                for parameter, paramset in (
-                    channels[channel_address].get(paramset_key.value, {}).items()
-                ):
+                for parameter, paramset in channels[channel_address].get(paramset_key, {}).items():
                     p_operations = paramset[Description.OPERATIONS]
 
                     if all(p_operations & operation for operation in operations):
@@ -1093,11 +1092,13 @@ def _get_virtual_remote(self, device_address: str) -> HmDevice | None:
         return None
 
     def get_generic_entity(
-        self, channel_address: str, parameter: str, paramset_key: ParamsetKey = ParamsetKey.VALUES
+        self, channel_address: str, parameter: str, paramset_key: ParamsetKey | None = None
     ) -> GenericEntity | None:
         """Get entity by channel_address and parameter."""
         if device := self.get_device(address=channel_address):
-            return device.get_generic_entity(channel_address=channel_address, parameter=parameter)
+            return device.get_generic_entity(
+                channel_address=channel_address, parameter=parameter, paramset_key=paramset_key
+            )
         return None
 
     def get_event(self, channel_address: str, parameter: str) -> GenericEvent | None:

hahomematic/client/__init__.py

@@ -394,7 +394,7 @@ async def get_install_mode(self) -> int:
     async def get_value(
         self,
         channel_address: str,
-        paramset_key: str,
+        paramset_key: ParamsetKey,
         parameter: str,
         call_source: CallSource = CallSource.MANUAL_OR_SCHEDULED,
     ) -> Any:
@@ -479,7 +479,7 @@ async def _set_value(
             return set()
 
     def _check_set_value(
-        self, channel_address: str, paramset_key: str, parameter: str, value: Any
+        self, channel_address: str, paramset_key: ParamsetKey, parameter: str, value: Any
     ) -> Any:
         """Check set_value."""
         return self._convert_value(
@@ -493,7 +493,7 @@ def _check_set_value(
     async def set_value(
         self,
         channel_address: str,
-        paramset_key: str,
+        paramset_key: ParamsetKey,
         parameter: str,
         value: Any,
         wait_for_callback: int | None = WAIT_FOR_CALLBACK,
@@ -519,7 +519,7 @@ async def set_value(
             check_against_pd=check_against_pd,
         )
 
-    async def get_paramset(self, address: str, paramset_key: str) -> dict[str, Any]:
+    async def get_paramset(self, address: str, paramset_key: ParamsetKey) -> dict[str, Any]:
         """
         Return a paramset from CCU.
 
@@ -547,7 +547,7 @@ async def get_paramset(self, address: str, paramset_key: str) -> dict[str, Any]:
     async def put_paramset(
         self,
         channel_address: str,
-        paramset_key: str,
+        paramset_key: ParamsetKey,
         values: dict[str, Any],
         wait_for_callback: int | None = WAIT_FOR_CALLBACK,
         rx_mode: str | None = None,
@@ -606,7 +606,7 @@ async def put_paramset(
             return set()
 
     def _check_put_paramset(
-        self, channel_address: str, paramset_key: str, values: dict[str, Any]
+        self, channel_address: str, paramset_key: ParamsetKey, values: dict[str, Any]
     ) -> dict[str, Any]:
         """Check put_paramset."""
         checked_values: dict[str, Any] = {}
@@ -623,7 +623,7 @@ def _check_put_paramset(
     def _convert_value(
         self,
         channel_address: str,
-        paramset_key: str,
+        paramset_key: ParamsetKey,
         parameter: str,
         value: Any,
         operation: Operations,
@@ -652,7 +652,7 @@ def _convert_value(
         )
 
     async def fetch_paramset_description(
-        self, channel_address: str, paramset_key: str, save_to_file: bool = True
+        self, channel_address: str, paramset_key: ParamsetKey, save_to_file: bool = True
     ) -> None:
         """Fetch a specific paramset and add it to the known ones."""
         _LOGGER.debug("FETCH_PARAMSET_DESCRIPTION: %s for %s", paramset_key, channel_address)
@@ -679,7 +679,7 @@ async def fetch_paramset_descriptions(self, device_description: dict[str, Any])
                 self.central.paramset_descriptions.add(
                     interface_id=self.interface_id,
                     channel_address=address,
-                    paramset_key=paramset_key,
+                    paramset_key=ParamsetKey(paramset_key),
                     paramset_description=paramset_description,
                 )
 
@@ -714,7 +714,7 @@ async def get_paramset_descriptions(
         return paramsets
 
     async def _get_paramset_description(
-        self, address: str, paramset_key: str
+        self, address: str, paramset_key: ParamsetKey
     ) -> dict[str, Any] | None:
         """Get paramset description from CCU."""
         try:
@@ -1257,8 +1257,10 @@ def _async_event_changed(*args: Any, **kwargs: Any) -> None:
                 )
                 ev.set()
 
-    channel_address, parameter = entity_key
-    if entity := device.get_generic_entity(channel_address=channel_address, parameter=parameter):
+    channel_address, paramset_key, parameter = entity_key
+    if entity := device.get_generic_entity(
+        channel_address=channel_address, parameter=parameter, paramset_key=paramset_key
+    ):
         if not entity.supports_events:
             _LOGGER.debug(
                 "TRACK_SINGLE_ENTITY_STATE_CHANGE_OR_TIMEOUT: Entity supports no events %s",