Use TypedDict for parameter_data

changelog.md

@@ -1,11 +1,11 @@
-# Version 2024.8.15 (2024-08-26)
+# Version 2024.8.15 (2024-08-29)
 
 - Avoid permanent cache save on remove device
 - Check rx_mode
 - 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
+- Use TypedDict for device_description
+- Use TypedDict for parameter_data
 
 # Version 2024.8.14 (2024-08-26)
 

hahomematic/caches/persistent.py

@@ -33,8 +33,6 @@
     get_device_address,
     get_split_channel_address,
     hash_sha256,
-    paramset_description_export_converter,
-    paramset_description_import_converter,
 )
 
 _LOGGER: Final = logging.getLogger(__name__)
@@ -69,14 +67,6 @@ def data_changed(self) -> bool:
         """Return if the data has changed."""
         return self.cache_hash != self.last_hash_saved
 
-    def _convert_date_after_load(self, data: Any) -> Any:
-        """Convert data load."""
-        return data
-
-    def _convert_date_before_save(self, data: Any) -> Any:
-        """Convert data save."""
-        return data
-
     async def save(self) -> DataOperationResult:
         """Save current name data in NAMES to disk."""
         self.last_save_triggered = datetime.now()
@@ -94,7 +84,7 @@ def _save() -> DataOperationResult:
             ) as fptr:
                 fptr.write(
                     orjson.dumps(
-                        self._convert_date_before_save(data=self._persistant_cache),
+                        self._persistant_cache,
                         option=orjson.OPT_NON_STR_KEYS,
                     )
                 )
@@ -119,11 +109,11 @@ def _load() -> DataOperationResult:
                 file=os.path.join(self._cache_dir, self._filename),
                 encoding=DEFAULT_ENCODING,
             ) as fptr:
-                converted_data = self._convert_date_after_load(data=orjson.loads(fptr.read()))
-                if (converted_hash := hash_sha256(value=converted_data)) == self.last_hash_saved:
+                data = orjson.loads(fptr.read())
+                if (converted_hash := hash_sha256(value=data)) == self.last_hash_saved:
                     return DataOperationResult.NO_LOAD
                 self._persistant_cache.clear()
-                self._persistant_cache.update(converted_data)
+                self._persistant_cache.update(data)
                 self.last_hash_saved = converted_hash
             return DataOperationResult.LOAD_SUCCESS
 
@@ -454,14 +444,6 @@ async def load(self) -> DataOperationResult:
         self._init_address_parameter_list()
         return result
 
-    def _convert_date_after_load(self, data: Any) -> Any:
-        """Convert data load."""
-        return paramset_description_import_converter(data=data)
-
     async def save(self) -> DataOperationResult:
         """Save current paramset descriptions to disk."""
         return await super().save()
-
-    def _convert_date_before_save(self, data: Any) -> Any:
-        """Convert data save."""
-        return paramset_description_export_converter(data=data)

hahomematic/central/__init__.py

@@ -1066,7 +1066,7 @@ def get_parameters(
                 for parameter, parameter_data in (
                     channels[channel_address].get(paramset_key, {}).items()
                 ):
-                    if all(parameter_data.operations & operation for operation in operations):
+                    if all(parameter_data["OPERATIONS"] & operation for operation in operations):
                         if un_ignore_candidates_only and (
                             (
                                 (

hahomematic/client/__init__.py

@@ -25,7 +25,6 @@
     Backend,
     CallSource,
     CommandRxMode,
-    Description,
     DeviceDescription,
     ForcedDeviceAvailability,
     InterfaceEventType,
@@ -53,11 +52,12 @@
 
 _LOGGER: Final = logging.getLogger(__name__)
 
-_ADDRESS: Final = "address"
-_CHANNELS: Final = "channels"
-_ID: Final = "id"
-_INTERFACE: Final = "interface"
-_NAME: Final = "name"
+_JSON_ADDRESS: Final = "address"
+_JSON_CHANNELS: Final = "channels"
+_JSON_ID: Final = "id"
+_JSON_INTERFACE: Final = "interface"
+_JSON_NAME: Final = "name"
+_NAME: Final = "NAME"
 
 
 class Client(ABC):
@@ -646,9 +646,11 @@ def _convert_value(
             paramset_key=paramset_key,
             parameter=parameter,
         ):
-            pd_type = parameter_data.hm_type
-            pd_value_list = tuple(parameter_data.value_list) if parameter_data.value_list else None
-            if not bool(int(parameter_data.operations) & operation):
+            pd_type = parameter_data["TYPE"]
+            pd_value_list = (
+                tuple(parameter_data["VALUE_LIST"]) if parameter_data.get("VALUE_LIST") else None
+            )
+            if not bool(int(parameter_data["OPERATIONS"]) & operation):
                 raise HaHomematicException(
                     f"Parameter {parameter} does not support the requested operation {operation.value}"
                 )
@@ -709,12 +711,10 @@ async def _get_paramset_description(
     ) -> dict[str, ParameterData] | None:
         """Get paramset description from CCU."""
         try:
-            if raw_paramset_description := await self._proxy_read.getParamsetDescription(
-                address, paramset_key
-            ):
-                return {
-                    key: ParameterData(value) for key, value in raw_paramset_description.items()
-                }
+            return cast(
+                dict[str, ParameterData],
+                await self._proxy_read.getParamsetDescription(address, paramset_key),
+            )
         except BaseHomematicException as ex:
             _LOGGER.debug(
                 "GET_PARAMSET_DESCRIPTIONS failed with %s [%s] for %s address %s",
@@ -819,21 +819,23 @@ async def fetch_device_details(self) -> None:
         """Get all names via JSON-RPS and store in data.NAMES."""
         if json_result := await self._json_rpc_client.get_device_details():
             for device in json_result:
-                device_address = device[_ADDRESS]
-                self.central.device_details.add_name(address=device_address, name=device[_NAME])
+                device_address = device[_JSON_ADDRESS]
+                self.central.device_details.add_name(
+                    address=device_address, name=device[_JSON_NAME]
+                )
                 self.central.device_details.add_device_channel_id(
-                    address=device_address, channel_id=device[_ID]
+                    address=device_address, channel_id=device[_JSON_ID]
                 )
-                for channel in device.get(_CHANNELS, []):
-                    channel_address = channel[_ADDRESS]
+                for channel in device.get(_JSON_CHANNELS, []):
+                    channel_address = channel[_JSON_ADDRESS]
                     self.central.device_details.add_name(
-                        address=channel_address, name=channel[_NAME]
+                        address=channel_address, name=channel[_JSON_NAME]
                     )
                     self.central.device_details.add_device_channel_id(
-                        address=channel_address, channel_id=channel[_ID]
+                        address=channel_address, channel_id=channel[_JSON_ID]
                     )
                 self.central.device_details.add_interface(
-                    address=device_address, interface=device[_INTERFACE]
+                    address=device_address, interface=device[_JSON_INTERFACE]
                 )
         else:
             _LOGGER.debug("FETCH_DEVICE_DETAILS: Unable to fetch device details via JSON-RPC")
@@ -968,7 +970,7 @@ async def fetch_device_details(self) -> None:
             try:
                 self.central.device_details.add_name(
                     address,
-                    await self._proxy_read.getMetadata(address, Description.NAME),
+                    await self._proxy_read.getMetadata(address, _NAME),
                 )
             except BaseHomematicException as ex:
                 _LOGGER.warning(

hahomematic/const.py

@@ -121,22 +121,6 @@ class DataOperationResult(Enum):
     NO_SAVE = 21
 
 
-class Description(StrEnum):
-    """Enum with homematic device/paramset description attributes."""
-
-    DEFAULT = "DEFAULT"
-    FLAGS = "FLAGS"
-    MAX = "MAX"
-    MIN = "MIN"
-    NAME = "NAME"
-    ID = "ID"
-    OPERATIONS = "OPERATIONS"
-    SPECIAL = "SPECIAL"
-    TYPE = "TYPE"
-    UNIT = "UNIT"
-    VALUE_LIST = "VALUE_LIST"
-
-
 class DeviceFirmwareState(StrEnum):
     """Enum with homematic device firmware states."""
 
@@ -563,26 +547,19 @@ class SystemInformation:
     serial: str | None = None
 
 
-@dataclass
-class ParameterData:
-    """Dataclass for parameter data."""
-
-    def __init__(self, data: Mapping[str, Any]) -> None:
-        """Init the dataclass from mapping."""
-        self.default: Final[Any] = data[Description.DEFAULT]
-        self.flags: Final[int] = data[Description.FLAGS]
-        self.id: Final[str | None] = data.get(Description.ID)
-        self.max: Final[Any] = data[Description.MAX]
-        self.min: Final[Any] = data[Description.MIN]
-        self.operations: int = data[Description.OPERATIONS]
-        self.special: Mapping[str, Any] | None = data.get(Description.SPECIAL)
-        self.hm_type = ParameterType(data[Description.TYPE])
-        self.unit: str | None = data.get(Description.UNIT)
-        self.value_list: Iterable[Any] | None = data.get(Description.VALUE_LIST)
-
-    def as_dict(self) -> dict[str, Any]:
-        """Return dataclass as dict."""
-        return {key.upper(): str(value) for key, value in self.__dict__.items() if value}
+class ParameterData(TypedDict, total=False):
+    """Typed dict for parameter data."""
+
+    DEFAULT: Any
+    FLAGS: int
+    ID: str
+    MAX: Any
+    MIN: Any
+    OPERATIONS: int
+    SPECIAL: Mapping[str, Any]
+    TYPE: ParameterType
+    UNIT: str
+    VALUE_LIST: Iterable[Any]
 
 
 class DeviceDescription(TypedDict, total=False):

hahomematic/platforms/__init__.py

@@ -68,10 +68,10 @@ def create_entities_and_append_to_device(device: hmd.HmDevice) -> None:
                         continue
 
                     # required to fix hm master paramset operation values
-                    if parameter_is_un_ignored and parameter_data.operations == 0:
-                        parameter_data.operations = 3
+                    if parameter_is_un_ignored and parameter_data["OPERATIONS"] == 0:
+                        parameter_data["OPERATIONS"] = 3
 
-                if parameter_data.operations & Operations.EVENT and (
+                if parameter_data["OPERATIONS"] & Operations.EVENT and (
                     parameter in CLICK_EVENTS
                     or parameter.startswith(DEVICE_ERROR_EVENTS)
                     or parameter in IMPULSE_EVENTS
@@ -83,10 +83,10 @@ def create_entities_and_append_to_device(device: hmd.HmDevice) -> None:
                         parameter_data=parameter_data,
                     )
                 if (
-                    not parameter_data.operations & Operations.EVENT
-                    and not parameter_data.operations & Operations.WRITE
+                    not parameter_data["OPERATIONS"] & Operations.EVENT
+                    and not parameter_data["OPERATIONS"] & Operations.WRITE
                 ) or (
-                    parameter_data.flags & Flag.INTERNAL
+                    parameter_data["FLAGS"] & Flag.INTERNAL
                     and parameter not in ALLOWED_INTERNAL_PARAMETERS
                     and not parameter_is_un_ignored
                 ):

hahomematic/platforms/device.py

@@ -55,7 +55,6 @@
     CacheEntry,
     Channel,
     check_or_create_directory,
-    device_paramset_description_export_converter,
     get_entity_key,
     get_rx_modes,
     reduce_args,
@@ -841,9 +840,7 @@ async def export_data(self) -> None:
         await self._save(
             file_dir=f"{self._storage_folder}/{DEFAULT_PARAMSET_DESCRIPTIONS_DIR}",
             filename=filename,
-            data=device_paramset_description_export_converter(
-                data=anonymize_paramset_descriptions
-            ),
+            data=anonymize_paramset_descriptions,
         )
 
     def _anonymize_address(self, address: str) -> str:

hahomematic/platforms/entity.py

@@ -436,17 +436,19 @@ def __init__(
 
     def _assign_parameter_data(self, parameter_data: ParameterData) -> None:
         """Assign parameter data to instance variables."""
-        self._type: ParameterType = ParameterType(parameter_data.hm_type)
-        self._values = tuple(parameter_data.value_list) if parameter_data.value_list else None
-        self._max: ParameterT = self._convert_value(parameter_data.max)
-        self._min: ParameterT = self._convert_value(parameter_data.min)
-        self._default: ParameterT = self._convert_value(parameter_data.default or self._min)
-        flags: int = parameter_data.flags
+        self._type: ParameterType = ParameterType(parameter_data["TYPE"])
+        self._values = (
+            tuple(parameter_data["VALUE_LIST"]) if parameter_data.get("VALUE_LIST") else None
+        )
+        self._max: ParameterT = self._convert_value(parameter_data["MAX"])
+        self._min: ParameterT = self._convert_value(parameter_data["MIN"])
+        self._default: ParameterT = self._convert_value(parameter_data["DEFAULT"] or self._min)
+        flags: int = parameter_data["FLAGS"]
         self._visible: bool = flags & Flag.VISIBLE == Flag.VISIBLE
         self._service: bool = flags & Flag.SERVICE == Flag.SERVICE
-        self._operations: int = parameter_data.operations
-        self._special: Mapping[str, Any] | None = parameter_data.special
-        self._raw_unit: str | None = parameter_data.unit
+        self._operations: int = parameter_data["OPERATIONS"]
+        self._special: Mapping[str, Any] | None = parameter_data.get("SPECIAL")
+        self._raw_unit: str | None = parameter_data.get("UNIT")
         self._unit: str | None = self._cleanup_unit(raw_unit=self._raw_unit)
         self._multiplier: int = self._get_multiplier(raw_unit=self._raw_unit)
 

hahomematic/platforms/event.py

@@ -159,7 +159,7 @@ def create_event_and_append_to_device(
         device.interface_id,
     )
     event_t: type[GenericEvent] | None = None
-    if parameter_data.operations & Operations.EVENT:
+    if parameter_data["OPERATIONS"] & Operations.EVENT:
         if parameter in CLICK_EVENTS:
             event_t = ClickEvent
         if parameter.startswith(DEVICE_ERROR_EVENTS):

hahomematic/platforms/generic/__init__.py

@@ -68,8 +68,8 @@ def create_entity_and_append_to_device(
         parameter,
         device.interface_id,
     )
-    p_type = parameter_data.hm_type
-    p_operations = parameter_data.operations
+    p_type = parameter_data["TYPE"]
+    p_operations = parameter_data["OPERATIONS"]
     entity_t: type[hmge.GenericEntity] | None = None
     if p_operations & Operations.WRITE:
         if p_type == ParameterType.ACTION:
@@ -97,7 +97,7 @@ def create_entity_and_append_to_device(
     elif parameter not in CLICK_EVENTS:
         # Also check, if sensor could be a binary_sensor due to.
         if is_binary_sensor(parameter_data):
-            parameter_data.hm_type = ParameterType.BOOL
+            parameter_data["TYPE"] = ParameterType.BOOL
             entity_t = HmBinarySensor
         else:
             entity_t = HmSensor

hahomematic/platforms/support.py

@@ -371,9 +371,9 @@ def convert_value(
 
 def is_binary_sensor(parameter_data: ParameterData) -> bool:
     """Check, if the sensor is a binary_sensor."""
-    if parameter_data.hm_type == ParameterType.BOOL:
+    if parameter_data["TYPE"] == ParameterType.BOOL:
         return True
-    if value_list := parameter_data.value_list:
+    if value_list := parameter_data.get("VALUE_LIST"):
         return tuple(value_list) in _BINARY_SENSOR_TRUE_VALUE_DICT_FOR_VALUE_LIST
     return False