Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Use TypedDict for parameter_data #1680

Merged
merged 3 commits into from
Aug 29, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 3 additions & 3 deletions changelog.md
Original file line number Diff line number Diff line change
@@ -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)

Expand Down
26 changes: 4 additions & 22 deletions hahomematic/caches/persistent.py
Original file line number Diff line number Diff line change
Expand Up @@ -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__)
Expand Down Expand Up @@ -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()
Expand All @@ -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,
)
)
Expand All @@ -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

Expand Down Expand Up @@ -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)
2 changes: 1 addition & 1 deletion hahomematic/central/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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 (
(
(
Expand Down
50 changes: 26 additions & 24 deletions hahomematic/client/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,6 @@
Backend,
CallSource,
CommandRxMode,
Description,
DeviceDescription,
ForcedDeviceAvailability,
InterfaceEventType,
Expand Down Expand Up @@ -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):
Expand Down Expand Up @@ -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}"
)
Expand Down Expand Up @@ -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",
Expand Down Expand Up @@ -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")
Expand Down Expand Up @@ -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(
Expand Down
49 changes: 13 additions & 36 deletions hahomematic/const.py
Original file line number Diff line number Diff line change
Expand Up @@ -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."""

Expand Down Expand Up @@ -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):
Expand Down
12 changes: 6 additions & 6 deletions hahomematic/platforms/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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
):
Expand Down
5 changes: 1 addition & 4 deletions hahomematic/platforms/device.py
Original file line number Diff line number Diff line change
Expand Up @@ -55,7 +55,6 @@
CacheEntry,
Channel,
check_or_create_directory,
device_paramset_description_export_converter,
get_entity_key,
get_rx_modes,
reduce_args,
Expand Down Expand Up @@ -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:
Expand Down
20 changes: 11 additions & 9 deletions hahomematic/platforms/entity.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"])
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)

Expand Down
2 changes: 1 addition & 1 deletion hahomematic/platforms/event.py
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Expand Down
6 changes: 3 additions & 3 deletions hahomematic/platforms/generic/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down Expand Up @@ -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
Expand Down
4 changes: 2 additions & 2 deletions hahomematic/platforms/support.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down
Loading