From 7044e6a29cf1c04b2572278027d3db9443dbf0e2 Mon Sep 17 00:00:00 2001 From: Snuffy2 Date: Mon, 16 Dec 2024 21:09:41 -0500 Subject: [PATCH] Enable mypy --- custom_components/keymaster/__init__.py | 37 +- custom_components/keymaster/binary_sensor.py | 7 +- custom_components/keymaster/button.py | 12 +- custom_components/keymaster/config_flow.py | 51 +-- custom_components/keymaster/const.py | 3 +- custom_components/keymaster/coordinator.py | 344 ++++++++++--------- custom_components/keymaster/datetime.py | 27 +- custom_components/keymaster/entity.py | 24 +- custom_components/keymaster/helpers.py | 27 +- custom_components/keymaster/lock.py | 14 +- custom_components/keymaster/lovelace.py | 69 ++-- custom_components/keymaster/migrate.py | 59 ++-- custom_components/keymaster/number.py | 19 +- custom_components/keymaster/sensor.py | 13 +- custom_components/keymaster/services.py | 18 +- custom_components/keymaster/switch.py | 82 +++-- custom_components/keymaster/text.py | 29 +- custom_components/keymaster/time.py | 69 ++-- pyproject.toml | 19 +- requirements_test.txt | 3 +- 20 files changed, 520 insertions(+), 406 deletions(-) diff --git a/custom_components/keymaster/__init__.py b/custom_components/keymaster/__init__.py index d6863fe9..9612d2f3 100644 --- a/custom_components/keymaster/__init__.py +++ b/custom_components/keymaster/__init__.py @@ -3,10 +3,11 @@ from __future__ import annotations import asyncio -from collections.abc import Mapping +from collections.abc import MutableMapping from datetime import datetime, timedelta import functools import logging +from typing import Any from homeassistant.config_entries import ConfigEntry from homeassistant.core import HomeAssistant @@ -54,8 +55,8 @@ async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> b updated_config = config_entry.data.copy() - updated_config[CONF_SLOTS] = int(updated_config.get(CONF_SLOTS)) - updated_config[CONF_START] = int(updated_config.get(CONF_START)) + # updated_config[CONF_SLOTS] = int(updated_config[CONF_SLOTS]) + # updated_config[CONF_START] = int(updated_config[CONF_START]) if config_entry.data.get(CONF_PARENT) in {None, "(none)"}: updated_config[CONF_PARENT] = None @@ -77,10 +78,10 @@ async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> b ) elif isinstance( updated_config.get(CONF_NOTIFY_SCRIPT_NAME), str - ) and updated_config.get(CONF_NOTIFY_SCRIPT_NAME).startswith("script."): - updated_config[CONF_NOTIFY_SCRIPT_NAME] = updated_config.get( + ) and updated_config[CONF_NOTIFY_SCRIPT_NAME].startswith("script."): + updated_config[CONF_NOTIFY_SCRIPT_NAME] = updated_config[ CONF_NOTIFY_SCRIPT_NAME - ).split(".", maxsplit=1)[1] + ].split(".", maxsplit=1)[1] if updated_config.get(CONF_DOOR_SENSOR_ENTITY_ID) == DEFAULT_DOOR_SENSOR: updated_config[CONF_DOOR_SENSOR_ENTITY_ID] = None @@ -114,7 +115,7 @@ async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> b device_registry = dr.async_get(hass) - via_device: str | None = None + via_device: tuple[str, Any] | None = None if config_entry.data.get(CONF_PARENT_ENTRY_ID): via_device = (DOMAIN, config_entry.data.get(CONF_PARENT_ENTRY_ID)) @@ -135,12 +136,12 @@ async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> b # _LOGGER.debug(f"[init async_setup_entry] device: {device}") - code_slots: Mapping[int, KeymasterCodeSlot] = {} + code_slots: MutableMapping[int, KeymasterCodeSlot] = {} for x in range( config_entry.data[CONF_START], config_entry.data[CONF_START] + config_entry.data[CONF_SLOTS], ): - dow_slots: Mapping[int, KeymasterCodeSlotDayOfWeek] = {} + dow_slots: MutableMapping[int, KeymasterCodeSlotDayOfWeek] = {} for i, dow in enumerate( [ "Monday", @@ -158,8 +159,8 @@ async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> b code_slots[x] = KeymasterCodeSlot(number=x, accesslimit_day_of_week=dow_slots) kmlock = KeymasterLock( - lock_name=config_entry.data.get(CONF_LOCK_NAME), - lock_entity_id=config_entry.data.get(CONF_LOCK_ENTITY_ID), + lock_name=config_entry.data[CONF_LOCK_NAME], + lock_entity_id=config_entry.data[CONF_LOCK_ENTITY_ID], keymaster_config_entry_id=config_entry.entry_id, alarm_level_or_user_code_entity_id=config_entry.data.get( CONF_ALARM_LEVEL_OR_USER_CODE_ENTITY_ID @@ -168,8 +169,8 @@ async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> b CONF_ALARM_TYPE_OR_ACCESS_CONTROL_ENTITY_ID ), door_sensor_entity_id=config_entry.data.get(CONF_DOOR_SENSOR_ENTITY_ID), - number_of_code_slots=config_entry.data.get(CONF_SLOTS), - starting_code_slot=config_entry.data.get(CONF_START), + number_of_code_slots=config_entry.data[CONF_SLOTS], + starting_code_slot=config_entry.data[CONF_START], code_slots=code_slots, parent_name=config_entry.data.get(CONF_PARENT), parent_config_entry_id=config_entry.data.get(CONF_PARENT_ENTRY_ID), @@ -184,12 +185,12 @@ async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> b await hass.config_entries.async_forward_entry_setups(config_entry, PLATFORMS) await generate_lovelace( hass=hass, - kmlock_name=config_entry.data.get(CONF_LOCK_NAME), + kmlock_name=config_entry.data[CONF_LOCK_NAME], keymaster_config_entry_id=config_entry.entry_id, parent_config_entry_id=config_entry.data.get(CONF_PARENT_ENTRY_ID), - code_slot_start=config_entry.data.get(CONF_START), - code_slots=config_entry.data.get(CONF_SLOTS), - lock_entity=config_entry.data.get(CONF_LOCK_ENTITY_ID), + code_slot_start=config_entry.data[CONF_START], + code_slots=config_entry.data[CONF_SLOTS], + lock_entity=config_entry.data[CONF_LOCK_ENTITY_ID], door_sensor=config_entry.data.get(CONF_DOOR_SENSOR_ENTITY_ID), ) @@ -207,7 +208,7 @@ async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> b async def async_unload_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool: """Handle removal of an entry.""" - lockname: str = config_entry.data.get(CONF_LOCK_NAME) + lockname: str = config_entry.data[CONF_LOCK_NAME] _LOGGER.info("Unloading %s", lockname) unload_ok: bool = all( await asyncio.gather( diff --git a/custom_components/keymaster/binary_sensor.py b/custom_components/keymaster/binary_sensor.py index f0e7e8fc..08ebb479 100644 --- a/custom_components/keymaster/binary_sensor.py +++ b/custom_components/keymaster/binary_sensor.py @@ -17,7 +17,6 @@ from .coordinator import KeymasterCoordinator from .entity import KeymasterEntity, KeymasterEntityDescription from .helpers import async_using_zwave_js -from .lock import KeymasterLock _LOGGER: logging.Logger = logging.getLogger(__name__) @@ -29,7 +28,7 @@ async def async_setup_entry( ): """Create the keymaster Binary Sensors.""" coordinator: KeymasterCoordinator = hass.data[DOMAIN][COORDINATOR] - kmlock: KeymasterLock = await coordinator.get_lock_by_config_entry_id( + kmlock = await coordinator.get_lock_by_config_entry_id( config_entry.entry_id ) entities: list = [] @@ -74,7 +73,7 @@ async def async_setup_entry( return True -@dataclass(kw_only=True) +@dataclass(frozen=True, kw_only=True) class KeymasterBinarySensorEntityDescription( KeymasterEntityDescription, BinarySensorEntityDescription ): @@ -84,6 +83,8 @@ class KeymasterBinarySensorEntityDescription( class KeymasterBinarySensor(KeymasterEntity, BinarySensorEntity): """Keymaster Binary Sensor Class.""" + entity_description: KeymasterBinarySensorEntityDescription + def __init__( self, entity_description: KeymasterBinarySensorEntityDescription, diff --git a/custom_components/keymaster/button.py b/custom_components/keymaster/button.py index 629595c8..50f6bdb7 100644 --- a/custom_components/keymaster/button.py +++ b/custom_components/keymaster/button.py @@ -1,5 +1,6 @@ """Support for keymaster buttons.""" +from collections.abc import MutableMapping from dataclasses import dataclass import logging @@ -56,10 +57,9 @@ async def async_setup_entry( ] ) async_add_entities(entities, True) - return True -@dataclass(kw_only=True) +@dataclass(frozen=True, kw_only=True) class KeymasterButtonEntityDescription( KeymasterEntityDescription, ButtonEntityDescription ): @@ -69,6 +69,8 @@ class KeymasterButtonEntityDescription( class KeymasterButton(KeymasterEntity, ButtonEntity): """Representation of a keymaster button.""" + entity_description: KeymasterButtonEntityDescription + def __init__( self, entity_description: KeymasterButtonEntityDescription, @@ -81,13 +83,13 @@ def __init__( @callback def _handle_coordinator_update(self) -> None: - if not self._kmlock.connected: + if not self._kmlock or not self._kmlock.connected: self._attr_available = False self.async_write_ha_state() return if ( - ".code_slots" in self._property + ".code_slots" in self._property and isinstance(self._kmlock.code_slots, MutableMapping) and self._code_slot not in self._kmlock.code_slots ): self._attr_available = False @@ -103,7 +105,7 @@ async def async_press(self) -> None: await self.coordinator.reset_lock( config_entry_id=self._config_entry.entry_id, ) - elif self._property.endswith(".reset"): + elif self._property.endswith(".reset") and self._code_slot: await self.coordinator.reset_code_slot( config_entry_id=self._config_entry.entry_id, code_slot=self._code_slot, diff --git a/custom_components/keymaster/config_flow.py b/custom_components/keymaster/config_flow.py index e197a028..e2b080e7 100644 --- a/custom_components/keymaster/config_flow.py +++ b/custom_components/keymaster/config_flow.py @@ -2,7 +2,7 @@ from __future__ import annotations -from collections.abc import Mapping +from collections.abc import MutableMapping import contextlib import logging from typing import TYPE_CHECKING, Any @@ -14,6 +14,7 @@ from homeassistant.components.lock import DOMAIN as LOCK_DOMAIN from homeassistant.components.script import DOMAIN as SCRIPT_DOMAIN from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN +from homeassistant.config_entries import ConfigFlowResult from homeassistant.core import HomeAssistant, callback from homeassistant.util import slugify @@ -44,11 +45,13 @@ _LOGGER: logging.Logger = logging.getLogger(__name__) -class KeymasterFlowHandler(config_entries.ConfigFlow, domain=DOMAIN): +class KeymasterFlowHandler(config_entries.ConfigFlow): """Config flow for keymaster.""" - VERSION = 3 - DEFAULTS: Mapping[str, Any] = { + domain: str = DOMAIN + + VERSION: int = 3 + DEFAULTS: MutableMapping[str, Any] = { CONF_SLOTS: DEFAULT_CODE_SLOTS, CONF_START: DEFAULT_START, CONF_DOOR_SENSOR_ENTITY_ID: DEFAULT_DOOR_SENSOR, @@ -57,7 +60,7 @@ class KeymasterFlowHandler(config_entries.ConfigFlow, domain=DOMAIN): CONF_HIDE_PINS: DEFAULT_HIDE_PINS, } - async def get_unique_name_error(self, user_input) -> Mapping[str, str]: + async def get_unique_name_error(self, user_input) -> MutableMapping[str, str]: """Check if name is unique, returning dictionary error if so.""" # Validate that lock name is unique existing_entry = await self.async_set_unique_id( @@ -68,13 +71,13 @@ async def get_unique_name_error(self, user_input) -> Mapping[str, str]: return {} async def async_step_user( - self, user_input: Mapping[str, Any] | None = None - ) -> Mapping[str, Any]: + self, user_input: MutableMapping[str, Any] | None = None + ) -> ConfigFlowResult: """Handle a flow initialized by the user.""" return await _start_config_flow( cls=self, step_id="user", - title=user_input[CONF_LOCK_NAME] if user_input else None, + title=user_input[CONF_LOCK_NAME] if user_input else "", user_input=user_input, defaults=self.DEFAULTS, ) @@ -95,7 +98,7 @@ def __init__(self, config_entry: config_entries.ConfigEntry) -> None: """Initialize.""" self.config_entry = config_entry - async def get_unique_name_error(self, user_input) -> Mapping[str, str]: + async def get_unique_name_error(self, user_input) -> MutableMapping[str, str]: """Check if name is unique, returning dictionary error if so.""" # If lock name has changed, make sure new name isn't already being used # otherwise show an error @@ -106,15 +109,15 @@ async def get_unique_name_error(self, user_input) -> Mapping[str, str]: return {} async def async_step_init( - self, user_input: Mapping[str, Any] | None = None - ) -> Mapping[str, Any]: + self, user_input: MutableMapping[str, Any] | None = None + ) -> MutableMapping[str, Any]: """Handle a flow initialized by the user.""" return await _start_config_flow( cls=self, step_id="init", title="", user_input=user_input, - defaults=self.config_entry.data, + defaults=dict(self.config_entry.data), entry_id=self.config_entry.entry_id, ) @@ -182,8 +185,8 @@ def _get_locks_in_use(hass: HomeAssistant, exclude: str | None = None) -> list[s def _get_schema( hass: HomeAssistant, - user_input: Mapping[str, Any] | None, - default_dict: Mapping[str, Any], + user_input: MutableMapping[str, Any] | None, + default_dict: MutableMapping[str, Any], entry_id: str | None = None, ) -> vol.Schema: """Get a schema using the default_dict as a backup.""" @@ -191,7 +194,7 @@ def _get_schema( user_input = {} if CONF_PARENT in default_dict and default_dict[CONF_PARENT] is None: - check_dict: Mapping[str, Any] = default_dict.copy() + check_dict: MutableMapping[str, Any] = dict(default_dict).copy() check_dict.pop(CONF_PARENT, None) default_dict = check_dict @@ -310,13 +313,14 @@ async def _start_config_flow( cls: KeymasterFlowHandler | KeymasterOptionsFlow, step_id: str, title: str, - user_input: Mapping[str, Any], - defaults: Mapping[str, Any] | None = None, + user_input: MutableMapping[str, Any] | None, + defaults: MutableMapping[str, Any] | None = None, entry_id: str | None = None, ): """Start a config flow.""" - errors = {} - description_placeholders = {} + errors: dict[str, Any] = {} + description_placeholders: dict[str, Any] = {} + defaults = defaults or {} if user_input is not None: _LOGGER.debug( @@ -325,8 +329,8 @@ async def _start_config_flow( user_input, errors, ) - user_input[CONF_SLOTS] = int(user_input.get(CONF_SLOTS)) - user_input[CONF_START] = int(user_input.get(CONF_START)) + user_input[CONF_SLOTS] = int(user_input[CONF_SLOTS]) + user_input[CONF_START] = int(user_input[CONF_START]) # Convert (none) to None if user_input[CONF_PARENT] == "(none)": @@ -341,8 +345,9 @@ async def _start_config_flow( step_id, user_input, ) - if step_id == "user": + if step_id == "user" or not entry_id: return cls.async_create_entry(title=title, data=user_input) + assert isinstance(cls, KeymasterOptionsFlow) cls.hass.config_entries.async_update_entry( cls.config_entry, data=user_input ) @@ -351,7 +356,7 @@ async def _start_config_flow( return cls.async_show_form( step_id=step_id, - data_schema=_get_schema(cls.hass, user_input, defaults, entry_id), + data_schema=_get_schema(hass=cls.hass, user_input=user_input, default_dict=defaults, entry_id=entry_id), errors=errors, description_placeholders=description_placeholders, ) diff --git a/custom_components/keymaster/const.py b/custom_components/keymaster/const.py index 59bfa6d0..eaf0ce0d 100644 --- a/custom_components/keymaster/const.py +++ b/custom_components/keymaster/const.py @@ -20,6 +20,7 @@ ] THROTTLE_SECONDS: int = 5 SYNC_STATUS_THRESHOLD: int = 15 +QUICK_REFRESH_SECONDS: int = 15 # hass.data attributes CHILD_LOCKS = "child_locks" @@ -128,7 +129,7 @@ }, } -LOCK_STATE_MAP = { +LOCK_STATE_MAP: dict[str, dict[str, int]] = { ALARM_TYPE: { LockState.LOCKED: 24, LockState.UNLOCKED: 25, diff --git a/custom_components/keymaster/coordinator.py b/custom_components/keymaster/coordinator.py index eab93210..43fb7e1b 100644 --- a/custom_components/keymaster/coordinator.py +++ b/custom_components/keymaster/coordinator.py @@ -4,7 +4,7 @@ import asyncio import base64 -from collections.abc import Callable, Mapping +from collections.abc import Callable, MutableMapping import contextlib from dataclasses import fields, is_dataclass from datetime import datetime, time as dt_time, timedelta @@ -68,6 +68,7 @@ EVENT_KEYMASTER_LOCK_STATE_CHANGED, ISSUE_URL, LOCK_STATE_MAP, + QUICK_REFRESH_SECONDS, SYNC_STATUS_THRESHOLD, THROTTLE_SECONDS, VERSION, @@ -102,13 +103,13 @@ def __init__(self, hass: HomeAssistant) -> None: """Initialize keymaster Coordinator.""" self._device_registry: dr.DeviceRegistry = dr.async_get(hass) self._entity_registry: er.EntityRegistry = er.async_get(hass) - self.kmlocks: Mapping[str, KeymasterLock] = {} - self._prev_kmlocks_dict: Mapping[str, Any] = {} + self.kmlocks: MutableMapping[str, KeymasterLock] = {} + self._prev_kmlocks_dict: MutableMapping[str, Any] = {} self._initial_setup_done_event = asyncio.Event() self._throttle = Throttle() self._sync_status_counter: int = 0 - self._refresh_in_15: bool = False - self._cancel_refresh_in_15: Callable | None = None + self._quick_refresh: bool = False + self._cancel_quick_refresh: Callable | None = None super().__init__( hass, @@ -122,7 +123,7 @@ def __init__(self, hass: HomeAssistant) -> None: ) self._json_filename: str = f"{DOMAIN}_kmlocks.json" - async def initial_setup(self) -> bool: + async def initial_setup(self) -> None: """Trigger the initial async_setup.""" await self._async_setup() @@ -165,8 +166,8 @@ def _create_json_folder(self) -> None: # e, # ) - def _get_dict_from_json_file(self) -> Mapping: - config: Mapping = {} + def _get_dict_from_json_file(self) -> MutableMapping: + config: MutableMapping = {} try: file_path: Path = Path(self._json_folder) / self._json_filename with file_path.open(encoding="utf-8") as jsonfile: @@ -201,7 +202,7 @@ def _get_dict_from_json_file(self) -> Mapping: ) # _LOGGER.debug(f"[get_dict_from_json_file] Imported JSON: {config}") - kmlocks: Mapping = { + kmlocks: MutableMapping = { key: self._dict_to_kmlocks(value, KeymasterLock) for key, value in config.items() } @@ -225,10 +226,11 @@ def _decode_pin(encoded_pin: str, unique_id: str) -> str: def _dict_to_kmlocks(self, data: dict, cls: type) -> Any: """Recursively convert a dictionary to a dataclass instance.""" if hasattr(cls, "__dataclass_fields__"): - field_values: Mapping = {} + field_values: MutableMapping = {} for field in fields(cls): field_name: str = field.name + assert isinstance(field.type, type) field_type: type = keymasterlock_type_lookup.get(field_name, field.type) field_value: Any = data.get(field_name) @@ -270,23 +272,23 @@ def _dict_to_kmlocks(self, data: dict, cls: type) -> Any: # _LOGGER.debug(f"[dict_to_kmlocks] isinstance(origin_type, type): {isinstance(origin_type, type)}") # if isinstance(origin_type, type): - # _LOGGER.debug(f"[dict_to_kmlocks] issubclass(origin_type, Mapping): {issubclass(origin_type, Mapping)}, origin_type == dict: {origin_type == dict}") + # _LOGGER.debug(f"[dict_to_kmlocks] issubclass(origin_type, MutableMapping): {issubclass(origin_type, MutableMapping)}, origin_type == dict: {origin_type == dict}") - # Handle Mapping types: when origin_type is Mapping + # Handle MutableMapping types: when origin_type is MutableMapping if isinstance(origin_type, type) and ( - issubclass(origin_type, Mapping) or origin_type is dict + issubclass(origin_type, MutableMapping) or origin_type is dict ): # Define key_type and value_type from type_args if len(type_args) == 2: key_type, value_type = type_args # _LOGGER.debug( - # f"[dict_to_kmlocks] field_name: {field_name}: Is Mapping or dict. key_type: {key_type}, " + # f"[dict_to_kmlocks] field_name: {field_name}: Is MutableMapping or dict. key_type: {key_type}, " # f"value_type: {value_type}, isinstance(field_value, dict): {isinstance(field_value, dict)}, " # f"is_dataclass(value_type): {is_dataclass(value_type)}" # ) if isinstance(field_value, dict): # If the value_type is a dataclass, recursively process it - if is_dataclass(value_type): + if is_dataclass(value_type) and isinstance(value_type, type): # _LOGGER.debug(f"[dict_to_kmlocks] Recursively converting dict items for {field_name}") field_value = { ( @@ -312,14 +314,14 @@ def _dict_to_kmlocks(self, data: dict, cls: type) -> Any: } # Handle nested dataclasses - elif isinstance(field_value, dict) and is_dataclass(field_type): + elif isinstance(field_value, dict) and is_dataclass(field_type) and isinstance(field_type, type): # _LOGGER.debug(f"[dict_to_kmlocks] Recursively converting nested dataclass: {field_name}") field_value = self._dict_to_kmlocks(field_value, field_type) # Handle list of nested dataclasses elif isinstance(field_value, list) and type_args: list_type = type_args[0] - if is_dataclass(list_type): + if is_dataclass(list_type) and isinstance(list_type, type): # _LOGGER.debug(f"[dict_to_kmlocks] Recursively converting list of dataclasses: {field_name}") field_value = [ ( @@ -336,10 +338,10 @@ def _dict_to_kmlocks(self, data: dict, cls: type) -> Any: return data - def _kmlocks_to_dict(self, instance: object) -> Mapping: + def _kmlocks_to_dict(self, instance: object) -> object: """Recursively convert a dataclass instance to a dictionary for JSON export.""" - if hasattr(instance, "__dataclass_fields__"): - result: Mapping = {} + if is_dataclass(instance): + result: MutableMapping = {} for field in fields(instance): field_name: str = field.name field_value: Any = getattr(instance, field_name) @@ -401,7 +403,7 @@ def delete_json(self) -> None: _LOGGER.debug("JSON config file deleted: %s", self._json_filename) def _write_config_to_json(self) -> bool: - config: Mapping = { + config: MutableMapping = { key: self._kmlocks_to_dict(kmlock) for key, kmlock in self.kmlocks.items() } for lock in config.values(): @@ -484,8 +486,8 @@ async def _handle_zwave_js_lock_event( # Get lock state to provide as part of event data new_state = None - if self.hass.states.get(kmlock.lock_entity_id): - new_state = self.hass.states.get(kmlock.lock_entity_id).state + if temp_new_state := self.hass.states.get(kmlock.lock_entity_id): + new_state = temp_new_state.state params = event.data.get(ATTR_PARAMETERS) or {} code_slot = params.get("userId", 0) @@ -538,12 +540,12 @@ async def _handle_lock_state_change( if changed_entity != kmlock.lock_entity_id: return - old_state = None - if event.data.get("old_state"): - old_state: str = event.data.get("old_state").state - new_state = None - if event.data.get("new_state"): - new_state: str = event.data.get("new_state").state + old_state: str | None = None + if temp_old_state := event.data.get("old_state"): + old_state = temp_old_state.state + new_state: str | None = None + if temp_new_state := event.data.get("new_state"): + new_state = temp_new_state.state # Determine action type to set appropriate action text using ACTION_MAP action_type: str = "" @@ -560,19 +562,22 @@ async def _handle_lock_state_change( action_type = ACCESS_CONTROL # Get alarm_level/usercode and alarm_type/access_control states - alarm_level_state = self.hass.states.get( - kmlock.alarm_level_or_user_code_entity_id - ) + alarm_level_state = None + if kmlock.alarm_level_or_user_code_entity_id: + alarm_level_state = self.hass.states.get( + kmlock.alarm_level_or_user_code_entity_id + ) alarm_level_value: int | None = ( int(alarm_level_state.state) if alarm_level_state and alarm_level_state.state not in {STATE_UNKNOWN, STATE_UNAVAILABLE} else None ) - - alarm_type_state = self.hass.states.get( - kmlock.alarm_type_or_access_control_entity_id - ) + alarm_type_state = None + if kmlock.alarm_type_or_access_control_entity_id: + alarm_type_state = self.hass.states.get( + kmlock.alarm_type_or_access_control_entity_id + ) alarm_type_value: int | None = ( int(alarm_type_state.state) if alarm_type_state @@ -595,6 +600,8 @@ async def _handle_lock_state_change( # in a while set action_value to RF lock/unlock if ( alarm_level_state is not None + and alarm_type_state is not None + and new_state and int(alarm_level_state.state) == 0 and dt_util.utcnow() - dt_util.as_utc(alarm_type_state.last_changed) > timedelta(seconds=5) @@ -660,12 +667,12 @@ async def _handle_door_state_change( if changed_entity != kmlock.door_sensor_entity_id: return - old_state = None - if event.data.get("old_state"): - old_state: str = event.data.get("old_state").state - new_state = None - if event.data.get("new_state"): - new_state: str = event.data.get("new_state").state + old_state: str | None = None + if temp_old_state := event.data.get("old_state"): + old_state = temp_old_state.state + new_state: str | None = None + if temp_new_state := event.data.get("new_state"): + new_state = temp_new_state.state _LOGGER.debug( "[handle_door_state_change] %s: old_state: %s, new_state: %s", kmlock.lock_name, @@ -707,7 +714,7 @@ async def _create_listeners( ) ) - if kmlock.door_sensor_entity_id not in {None, DEFAULT_DOOR_SENSOR}: + if kmlock.door_sensor_entity_id is not None and kmlock.door_sensor_entity_id != DEFAULT_DOOR_SENSOR: _LOGGER.debug( "[create_listeners] %s: Creating handle_door_state_change listener", kmlock.lock_name, @@ -829,15 +836,14 @@ async def _lock_unlocked( ) if ( isinstance(parent_kmlock, KeymasterLock) + and parent_kmlock.code_slots + and code_slot and code_slot in parent_kmlock.code_slots and parent_kmlock.code_slots[code_slot].accesslimit_count_enabled - and isinstance( - parent_kmlock.code_slots[code_slot].accesslimit_count, int - ) - and parent_kmlock.code_slots[code_slot].accesslimit_count > 0 ): - parent_kmlock.code_slots[code_slot].accesslimit_count -= 1 - + accesslimit_count: int | None = parent_kmlock.code_slots[code_slot].accesslimit_count + if accesslimit_count is not None and accesslimit_count > 0: + parent_kmlock.code_slots[code_slot].accesslimit_count = int(accesslimit_count) - 1 elif ( kmlock.code_slots[code_slot].accesslimit_count_enabled and isinstance(kmlock.code_slots[code_slot].accesslimit_count, int) @@ -966,12 +972,12 @@ async def _door_closed(self, kmlock) -> None: async def _lock_lock(self, kmlock: KeymasterLock): _LOGGER.debug("[lock_lock] %s: Locking", kmlock.lock_name) kmlock.pending_retry_lock = False - target: Mapping[str, Any] = {ATTR_ENTITY_ID: kmlock.lock_entity_id} + target: MutableMapping[str, Any] = {ATTR_ENTITY_ID: kmlock.lock_entity_id} await call_hass_service( hass=self.hass, domain=LOCK_DOMAIN, service=SERVICE_LOCK, - target=target, + target=dict(target), ) async def _setup_timers(self) -> None: @@ -1013,8 +1019,8 @@ async def _update_door_and_lock_state( for kmlock in self.kmlocks.values(): if isinstance(kmlock.lock_entity_id, str) and kmlock.lock_entity_id: lock_state = None - if self.hass.states.get(kmlock.lock_entity_id): - lock_state = self.hass.states.get(kmlock.lock_entity_id).state + if temp_lock_state := self.hass.states.get(kmlock.lock_entity_id): + lock_state = temp_lock_state.state if lock_state in { LockState.LOCKED, LockState.UNLOCKED, @@ -1054,35 +1060,34 @@ async def _update_door_and_lock_state( and kmlock.door_sensor_entity_id and kmlock.door_sensor_entity_id != DEFAULT_DOOR_SENSOR ): - door_state: str = self.hass.states.get( - kmlock.door_sensor_entity_id - ).state - if door_state in {STATE_OPEN, STATE_CLOSED}: - if ( - kmlock.door_state - in { - STATE_OPEN, - STATE_CLOSED, - } - and kmlock.door_state != door_state - ): - _LOGGER.debug( - "[update_door_and_lock_state] Door Status out of sync: " - "kmlock.door_state: %s, door_state: %s", - kmlock.door_state, - door_state, - ) - if ( - trigger_actions_if_changed - and kmlock.door_state in {STATE_OPEN, STATE_CLOSED} - and kmlock.door_state != door_state - ): - if door_state == STATE_OPEN: - await self._door_opened(kmlock=kmlock) - elif door_state == STATE_CLOSED: - await self._door_closed(kmlock=kmlock) - else: - kmlock.door_state = door_state + if temp_door_state := self.hass.states.get(kmlock.door_sensor_entity_id): + door_state: str = temp_door_state.state + if door_state in {STATE_OPEN, STATE_CLOSED}: + if ( + kmlock.door_state + in { + STATE_OPEN, + STATE_CLOSED, + } + and kmlock.door_state != door_state + ): + _LOGGER.debug( + "[update_door_and_lock_state] Door Status out of sync: " + "kmlock.door_state: %s, door_state: %s", + kmlock.door_state, + door_state, + ) + if ( + trigger_actions_if_changed + and kmlock.door_state in {STATE_OPEN, STATE_CLOSED} + and kmlock.door_state != door_state + ): + if door_state == STATE_OPEN: + await self._door_opened(kmlock=kmlock) + elif door_state == STATE_CLOSED: + await self._door_closed(kmlock=kmlock) + else: + kmlock.door_state = door_state async def add_lock(self, kmlock: KeymasterLock, update: bool = False) -> None: """Add a new kmlock.""" @@ -1124,6 +1129,8 @@ async def _update_lock(self, new: KeymasterLock) -> bool: ) return False old: KeymasterLock = self.kmlocks[new.keymaster_config_entry_id] + if not old.starting_code_slot or not old.number_of_code_slots or not new.number_of_code_slots or not new.starting_code_slot or not new.code_slots or not old.code_slots: + return False await KeymasterCoordinator._unsubscribe_listeners(old) # _LOGGER.debug(f"[update_lock] {new.lock_name}: old: {old}") # _LOGGER.debug(f"[update_lock] {new.lock_name}: new: {new}") @@ -1144,35 +1151,39 @@ async def _update_lock(self, new: KeymasterLock) -> bool: new.autolock_min_night = old.autolock_min_night new.retry_lock = old.retry_lock for num, new_slot in new.code_slots.items(): - if num in old.code_slots: - old_slot: KeymasterCodeSlot = old.code_slots[num] - new_slot.enabled = old_slot.enabled - new_slot.name = old_slot.name - new_slot.override_parent = old_slot.override_parent - new_slot.notifications = old_slot.notifications - new_slot.accesslimit_count_enabled = old_slot.accesslimit_count_enabled - new_slot.accesslimit_count = old_slot.accesslimit_count - new_slot.accesslimit_date_range_enabled = ( - old_slot.accesslimit_date_range_enabled - ) - new_slot.accesslimit_date_range_start = ( - old_slot.accesslimit_date_range_start - ) - new_slot.accesslimit_date_range_end = ( - old_slot.accesslimit_date_range_end - ) - new_slot.accesslimit_day_of_week_enabled = ( - old_slot.accesslimit_day_of_week_enabled - ) - for dow_num, new_dow in new_slot.accesslimit_day_of_week.items(): + if num not in old.code_slots: + continue + old_slot: KeymasterCodeSlot = old.code_slots[num] + new_slot.enabled = old_slot.enabled + new_slot.name = old_slot.name + new_slot.override_parent = old_slot.override_parent + new_slot.notifications = old_slot.notifications + new_slot.accesslimit_count_enabled = old_slot.accesslimit_count_enabled + new_slot.accesslimit_count = old_slot.accesslimit_count + new_slot.accesslimit_date_range_enabled = ( + old_slot.accesslimit_date_range_enabled + ) + new_slot.accesslimit_date_range_start = ( + old_slot.accesslimit_date_range_start + ) + new_slot.accesslimit_date_range_end = ( + old_slot.accesslimit_date_range_end + ) + new_slot.accesslimit_day_of_week_enabled = ( + old_slot.accesslimit_day_of_week_enabled + ) + if not new_slot.accesslimit_day_of_week: + continue + for dow_num, new_dow in new_slot.accesslimit_day_of_week.items(): + if old_slot.accesslimit_day_of_week: old_dow: KeymasterCodeSlotDayOfWeek = ( old_slot.accesslimit_day_of_week[dow_num] ) - new_dow.dow_enabled = old_dow.dow_enabled - new_dow.limit_by_time = old_dow.limit_by_time - new_dow.include_exclude = old_dow.include_exclude - new_dow.time_start = old_dow.time_start - new_dow.time_end = old_dow.time_end + new_dow.dow_enabled = old_dow.dow_enabled + new_dow.limit_by_time = old_dow.limit_by_time + new_dow.include_exclude = old_dow.include_exclude + new_dow.time_start = old_dow.time_start + new_dow.time_end = old_dow.time_end self.kmlocks[new.keymaster_config_entry_id] = new _LOGGER.debug("[update_lock] Code slot entities to delete: %s", del_code_slots) for x in del_code_slots: @@ -1192,13 +1203,13 @@ async def _delete_lock(self, kmlock: KeymasterLock, _: datetime) -> None: await self._initial_setup_done_event.wait() _LOGGER.debug("[delete_lock] %s: Triggered", kmlock.lock_name) if kmlock.keymaster_config_entry_id not in self.kmlocks: - return True + return if not kmlock.pending_delete: _LOGGER.debug( "[delete_lock] %s: Appears to be a reload, delete cancelled", kmlock.lock_name, ) - return None + return _LOGGER.debug("[delete_lock] %s: Deleting", kmlock.lock_name) await self.hass.async_add_executor_job( delete_lovelace, self.hass, kmlock.lock_name @@ -1212,7 +1223,7 @@ async def _delete_lock(self, kmlock: KeymasterLock, _: datetime) -> None: await self._rebuild_lock_relationships() await self.hass.async_add_executor_job(self._write_config_to_json) await self.async_refresh() - return None + return async def delete_lock_by_config_entry_id(self, config_entry_id: str) -> None: """Delete a keymaster lock by entry_id.""" @@ -1231,7 +1242,7 @@ async def delete_lock_by_config_entry_id(self, config_entry_id: str) -> None: kmlock.listeners.append( async_call_later( hass=self.hass, - delay=15, + delay=QUICK_REFRESH_SECONDS, action=functools.partial(self._delete_lock, kmlock), ) ) @@ -1282,7 +1293,7 @@ async def set_pin_on_lock( ) return False - if code_slot not in kmlock.code_slots: + if not kmlock.code_slots or code_slot not in kmlock.code_slots: _LOGGER.debug( "[set_pin_on_lock] %s: Code Slot %s: Code slot doesn't exist", kmlock.lock_name, @@ -1332,8 +1343,8 @@ async def set_pin_on_lock( ) kmlock.code_slots[code_slot].synced = Synced.ADDING - self._refresh_in_15 = True - if async_using_zwave_js(hass=self.hass, entity_id=kmlock.lock_entity_id): + self._quick_refresh = True + if async_using_zwave_js(hass=self.hass, entity_id=kmlock.lock_entity_id) and kmlock.zwave_js_lock_node: try: await set_usercode(kmlock.zwave_js_lock_node, code_slot, pin) @@ -1374,7 +1385,7 @@ async def clear_pin_from_lock( ) return False - if code_slot not in kmlock.code_slots: + if not kmlock.code_slots or code_slot not in kmlock.code_slots: _LOGGER.debug( "[clear_pin_from_lock] %s: Code Slot %s: Code slot doesn't exist", kmlock.lock_name, @@ -1405,8 +1416,8 @@ async def clear_pin_from_lock( ) kmlock.code_slots[code_slot].synced = Synced.DELETING - self._refresh_in_15 = True - if async_using_zwave_js(hass=self.hass, entity_id=kmlock.lock_entity_id): + self._quick_refresh = True + if async_using_zwave_js(hass=self.hass, entity_id=kmlock.lock_entity_id) and kmlock.zwave_js_lock_node: try: await clear_usercode(kmlock.zwave_js_lock_node, code_slot) except BaseZwaveJSServerError as e: @@ -1442,10 +1453,11 @@ async def reset_lock(self, config_entry_id: str) -> None: kmlock.autolock_min_day = None kmlock.autolock_min_night = None kmlock.retry_lock = False - for x in kmlock.code_slots: - await self.reset_code_slot( - config_entry_id=kmlock.keymaster_config_entry_id, code_slot=x - ) + if kmlock.code_slots: + for x in kmlock.code_slots: + await self.reset_code_slot( + config_entry_id=kmlock.keymaster_config_entry_id, code_slot=x + ) await self.async_refresh() async def reset_code_slot(self, config_entry_id: str, code_slot: int) -> None: @@ -1458,7 +1470,7 @@ async def reset_code_slot(self, config_entry_id: str, code_slot: int) -> None: ) return - if code_slot not in kmlock.code_slots: + if not kmlock.code_slots or code_slot not in kmlock.code_slots: _LOGGER.error( "[Coordinator] %s: Code Slot %s: Code slot doesn't exist", kmlock.lock_name, @@ -1476,7 +1488,7 @@ async def reset_code_slot(self, config_entry_id: str, code_slot: int) -> None: override=True, ) - dow_slots: Mapping[int, KeymasterCodeSlotDayOfWeek] = {} + dow_slots: MutableMapping[int, KeymasterCodeSlotDayOfWeek] = {} for i, dow in enumerate( [ "Monday", @@ -1519,7 +1531,7 @@ async def _is_slot_active(slot: KeymasterCodeSlot) -> bool: ): return False - if slot.accesslimit_day_of_week_enabled: + if slot.accesslimit_day_of_week_enabled and slot.accesslimit_day_of_week: today_index: int = datetime.now().astimezone().weekday() today: KeymasterCodeSlotDayOfWeek = slot.accesslimit_day_of_week[ today_index @@ -1558,7 +1570,7 @@ async def _is_slot_active(slot: KeymasterCodeSlot) -> bool: return True - async def _trigger_refresh_in_15(self, _: datetime): + async def _trigger_quick_refresh(self, _: datetime): await self.async_request_refresh() async def update_slot_active_state( @@ -1576,7 +1588,7 @@ async def update_slot_active_state( ) return False - if code_slot not in kmlock.code_slots: + if not kmlock.code_slots or code_slot not in kmlock.code_slots: _LOGGER.debug( "[update_slot_active_state] %s: " "Keymaster code slot %s doesn't exist.", @@ -1606,12 +1618,20 @@ async def _connect_and_update_lock(self, kmlock: KeymasterLock) -> bool: return False kmlock.lock_config_entry_id = lock_ent_reg_entry.config_entry_id - + assert kmlock.lock_config_entry_id is not None try: zwave_entry = self.hass.config_entries.async_get_entry( kmlock.lock_config_entry_id ) - client = zwave_entry.runtime_data[ZWAVE_JS_DATA_CLIENT] + if zwave_entry: + client = zwave_entry.runtime_data[ZWAVE_JS_DATA_CLIENT] + else: + _LOGGER.error( + "[Coordinator] %s: Can't access the Z-Wave JS client.", + kmlock.lock_name, + ) + kmlock.connected = False + return False except (KeyError, TypeError) as e: _LOGGER.error( "[Coordinator] %s: Can't access the Z-Wave JS client. %s: %s", @@ -1655,9 +1675,11 @@ async def _connect_and_update_lock(self, kmlock: KeymasterLock) -> bool: kmlock.connected = False return False - lock_dev_reg_entry = self._device_registry.async_get( - lock_ent_reg_entry.device_id - ) + lock_dev_reg_entry = None + if lock_ent_reg_entry and lock_ent_reg_entry.device_id: + lock_dev_reg_entry = self._device_registry.async_get( + lock_ent_reg_entry.device_id + ) if not lock_dev_reg_entry: _LOGGER.error( "[Coordinator] %s: Can't find the lock in the Device Registry", @@ -1681,10 +1703,10 @@ async def _connect_and_update_lock(self, kmlock: KeymasterLock) -> bool: kmlock.zwave_js_lock_device = lock_dev_reg_entry return True - async def _async_update_data(self) -> Mapping[str, Any]: + async def _async_update_data(self) -> dict[str, Any]: """Update all keymaster locks.""" await self._initial_setup_done_event.wait() - self._refresh_in_15 = False + self._quick_refresh = False self._sync_status_counter += 1 # Clear any pending refresh callback @@ -1692,7 +1714,7 @@ async def _async_update_data(self) -> Mapping[str, Any]: # Update all keymaster locks for kmlock in self.kmlocks.values(): - await self._update_lock(kmlock) + await self._update_lock_data(kmlock) # Propagate parent kmlock settings to child kmlocks for kmlock in self.kmlocks.values(): @@ -1709,28 +1731,28 @@ async def _async_update_data(self) -> Mapping[str, Any]: # Schedule next refresh if needed await self._schedule_quick_refresh_if_needed() - return self.kmlocks + return dict(self.kmlocks) async def _clear_pending_quick_refresh(self): """Clear any pending refresh callback.""" - if self._cancel_refresh_in_15: - self._cancel_refresh_in_15() - self._cancel_refresh_in_15 = None + if self._cancel_quick_refresh: + self._cancel_quick_refresh() + self._cancel_quick_refresh = None - async def _update_lock(self, kmlock: KeymasterLock): + async def _update_lock_data(self, kmlock: KeymasterLock): """Update a single keymaster lock.""" await self._connect_and_update_lock(kmlock) if not kmlock.connected: _LOGGER.error("[Coordinator] %s: Not Connected", kmlock.lock_name) - self._set_code_slots_to_disconnected(kmlock) + # self._set_code_slots_to_disconnected(kmlock) return if not async_using_zwave_js(hass=self.hass, kmlock=kmlock): _LOGGER.error("[Coordinator] %s: Not using Z-Wave JS", kmlock.lock_name) return - node: ZwaveJSNode = kmlock.zwave_js_lock_node + node: ZwaveJSNode | None = kmlock.zwave_js_lock_node if node is None: _LOGGER.error("[Coordinator] %s: Z-Wave JS Node not defined", kmlock.lock_name) return @@ -1756,8 +1778,9 @@ async def _get_usercodes_from_node(node: ZwaveJSNode, kmlock: KeymasterLock) -> async def _update_code_slots(self, kmlock: KeymasterLock, usercodes: list) -> None: """Update the code slots for a keymaster lock.""" # Check active status of code slots and set/clear PINs on Z-Wave JS Lock - for num, slot in kmlock.code_slots.items(): - await self._update_slot(kmlock, slot, num) + if kmlock.code_slots: + for num, slot in kmlock.code_slots.items(): + await self._update_slot(kmlock, slot, num) # Get usercodes from Z-Wave JS Lock and update kmlock PINs for slot in usercodes: @@ -1786,11 +1809,11 @@ async def _update_slot(self, kmlock: KeymasterLock, slot, num: int) -> None: async def _sync_usercode(self, kmlock: KeymasterLock, slot) -> None: """Sync a usercode from Z-Wave JS.""" - code_slot = int(slot[ATTR_CODE_SLOT]) - usercode = slot[ATTR_USERCODE] - in_use = slot[ATTR_IN_USE] + code_slot: int = int(slot[ATTR_CODE_SLOT]) + usercode: str = slot[ATTR_USERCODE] + in_use: bool = slot[ATTR_IN_USE] - if code_slot not in kmlock.code_slots: + if not kmlock.code_slots or code_slot not in kmlock.code_slots: return if in_use is None and code_slot in kmlock.code_slots: @@ -1802,6 +1825,8 @@ async def _sync_usercode(self, kmlock: KeymasterLock, slot) -> None: async def _sync_pin(self, kmlock: KeymasterLock, code_slot: int, usercode: str): """Sync the pin with the lock based on conditions.""" + if not kmlock.code_slots: + return if not usercode: if ( not kmlock.code_slots[code_slot].enabled @@ -1809,11 +1834,12 @@ async def _sync_pin(self, kmlock: KeymasterLock, code_slot: int, usercode: str): or not kmlock.code_slots[code_slot].pin ): kmlock.code_slots[code_slot].synced = Synced.DISCONNECTED - else: + elif kmlock.code_slots[code_slot].pin is not None: + pin: str = str(kmlock.code_slots[code_slot].pin) await self.set_pin_on_lock( config_entry_id=kmlock.keymaster_config_entry_id, code_slot=code_slot, - pin=kmlock.code_slots[code_slot].pin, + pin=pin, override=True, ) elif ( @@ -1834,7 +1860,7 @@ async def _sync_pin(self, kmlock: KeymasterLock, code_slot: int, usercode: str): and kmlock.code_slots[code_slot].pin != usercode ): kmlock.code_slots[code_slot].synced = Synced.OUT_OF_SYNC - self._refresh_in_15 = True + self._quick_refresh = True async def _sync_child_locks(self, kmlock: KeymasterLock) -> None: """Propagate parent lock settings to child locks.""" @@ -1883,10 +1909,12 @@ async def _sync_child_lock(self, kmlock: KeymasterLock, child_entry_id: str) -> async def _update_child_code_slots(self, kmlock: KeymasterLock, child_kmlock: KeymasterLock) -> None: """Update code slots on a child lock based on parent settings.""" + if not kmlock.code_slots: + return for num, slot in kmlock.code_slots.items(): - if num not in child_kmlock.code_slots: + if not child_kmlock.code_slots or num not in child_kmlock.code_slots: continue - if child_kmlock.code_slots[num].override_parent: + if not child_kmlock.code_slots or child_kmlock.code_slots[num].override_parent: continue prev_enabled = child_kmlock.code_slots[num].enabled @@ -1912,7 +1940,7 @@ async def _update_child_code_slots(self, kmlock: KeymasterLock, child_kmlock: Ke or prev_enabled != child_kmlock.code_slots[num].enabled or prev_active != child_kmlock.code_slots[num].active ): - self._refresh_in_15 = True + self._quick_refresh = True if not slot.enabled or not slot.active or not slot.pin: await self.clear_pin_from_lock( config_entry_id=child_kmlock.keymaster_config_entry_id, @@ -1929,8 +1957,8 @@ async def _update_child_code_slots(self, kmlock: KeymasterLock, child_kmlock: Ke async def _schedule_quick_refresh_if_needed(self) -> None: """Schedule quick refresh if required.""" - if self._refresh_in_15: - _LOGGER.debug("[Coordinator] Scheduling refresh in 15 seconds") - self._cancel_refresh_in_15 = async_call_later( - self.hass, 15, self._async_update_data + if self._quick_refresh: + _LOGGER.debug("[Coordinator] Scheduling refresh in %s seconds", QUICK_REFRESH_SECONDS) + self._cancel_quick_refresh = async_call_later( + hass=self.hass, delay=QUICK_REFRESH_SECONDS, action=self._trigger_quick_refresh ) diff --git a/custom_components/keymaster/datetime.py b/custom_components/keymaster/datetime.py index 08d820dc..33fe3463 100644 --- a/custom_components/keymaster/datetime.py +++ b/custom_components/keymaster/datetime.py @@ -29,7 +29,7 @@ async def async_setup_entry( config_entry.data[CONF_START], config_entry.data[CONF_START] + config_entry.data[CONF_SLOTS], ): - entities.append( + entities.extend([ KeymasterDateTime( entity_description=KeymasterDateTimeEntityDescription( key=f"datetime.code_slots:{x}.accesslimit_date_range_start", @@ -40,9 +40,7 @@ async def async_setup_entry( config_entry=config_entry, coordinator=coordinator, ), - ) - ) - entities.append( + ), KeymasterDateTime( entity_description=KeymasterDateTimeEntityDescription( key=f"datetime.code_slots:{x}.accesslimit_date_range_end", @@ -54,13 +52,12 @@ async def async_setup_entry( coordinator=coordinator, ), ) - ) + ]) async_add_entities(entities, True) - return True -@dataclass(kw_only=True) +@dataclass(frozen=True, kw_only=True) class KeymasterDateTimeEntityDescription( KeymasterEntityDescription, DateTimeEntityDescription ): @@ -70,6 +67,8 @@ class KeymasterDateTimeEntityDescription( class KeymasterDateTime(KeymasterEntity, DateTimeEntity): """Keymaster DateTime Class.""" + entity_description: KeymasterDateTimeEntityDescription + def __init__( self, entity_description: KeymasterDateTimeEntityDescription, @@ -83,7 +82,7 @@ def __init__( @callback def _handle_coordinator_update(self) -> None: # _LOGGER.debug(f"[DateTime handle_coordinator_update] self.coordinator.data: {self.coordinator.data}") - if not self._kmlock.connected: + if not self._kmlock or not self._kmlock.connected: self._attr_available = False self.async_write_ha_state() return @@ -91,7 +90,7 @@ def _handle_coordinator_update(self) -> None: if ( ".code_slots" in self._property and self._kmlock.parent_name is not None - and not self._kmlock.code_slots[self._code_slot].override_parent + and (not self._kmlock.code_slots or not self._code_slot or not self._kmlock.code_slots[self._code_slot].override_parent) ): self._attr_available = False self.async_write_ha_state() @@ -99,7 +98,7 @@ def _handle_coordinator_update(self) -> None: if ( ".code_slots" in self._property - and self._code_slot not in self._kmlock.code_slots + and (not self._kmlock.code_slots or self._code_slot not in self._kmlock.code_slots) ): self._attr_available = False self.async_write_ha_state() @@ -108,9 +107,9 @@ def _handle_coordinator_update(self) -> None: if ( self._property.endswith(".accesslimit_date_range_start") or self._property.endswith(".accesslimit_date_range_end") - ) and not self._kmlock.code_slots[ + ) and (not self._kmlock.code_slots or not self._code_slot or not self._kmlock.code_slots[ self._code_slot - ].accesslimit_date_range_enabled: + ].accesslimit_date_range_enabled): self._attr_available = False self.async_write_ha_state() return @@ -130,8 +129,8 @@ async def async_set_value(self, value: datetime) -> None: if ( ".code_slots" in self._property - and self._kmlock.parent_name is not None - and not self._kmlock.code_slots[self._code_slot].override_parent + and self._kmlock and self._kmlock.parent_name + and (not self._kmlock.code_slots or not self._code_slot or not self._kmlock.code_slots[self._code_slot].override_parent) ): _LOGGER.debug( "[DateTime async_set_value] %s: " diff --git a/custom_components/keymaster/entity.py b/custom_components/keymaster/entity.py index 9f363b53..d98331c2 100644 --- a/custom_components/keymaster/entity.py +++ b/custom_components/keymaster/entity.py @@ -1,19 +1,19 @@ """Base entity for keymaster.""" +from __future__ import annotations -from collections.abc import Mapping from dataclasses import dataclass import logging from typing import Any from homeassistant.config_entries import ConfigEntry from homeassistant.core import HomeAssistant +from homeassistant.helpers.device_registry import DeviceInfo from homeassistant.helpers.entity import EntityDescription from homeassistant.helpers.update_coordinator import CoordinatorEntity from homeassistant.util import slugify from .const import DOMAIN from .coordinator import KeymasterCoordinator -from .lock import KeymasterLock _LOGGER: logging.Logger = logging.getLogger(__name__) @@ -26,22 +26,24 @@ class KeymasterEntity(CoordinatorEntity[KeymasterCoordinator]): """Base entity for Keymaster.""" - def __init__(self, entity_description: EntityDescription) -> None: + def __init__(self, entity_description: KeymasterEntityDescription) -> None: """Initialize base keymaster entity.""" self.hass: HomeAssistant = entity_description.hass self.coordinator: KeymasterCoordinator = entity_description.coordinator self._config_entry: ConfigEntry = entity_description.config_entry - self.entity_description: EntityDescription = entity_description + self.entity_description: KeymasterEntityDescription = entity_description self._attr_available = False self._property: str = ( entity_description.key ) # ..:.: *Only if needed - self._kmlock: KeymasterLock = self.coordinator.sync_get_lock_by_config_entry_id( + self._kmlock = self.coordinator.sync_get_lock_by_config_entry_id( self._config_entry.entry_id ) - self._attr_name: str = ( - f"{self._kmlock.lock_name} {self.entity_description.name}" - ) + self._attr_name: str | None = None + if self._attr_name: + self._attr_name = ( + f"{self._kmlock.lock_name} {self.entity_description.name}" + ) # _LOGGER.debug(f"[Entity init] entity_description.name: {self.entity_description.name}, name: {self.name}") self._attr_unique_id: str = ( f"{self._config_entry.entry_id}_{slugify(self._property)}" @@ -51,8 +53,8 @@ def __init__(self, entity_description: EntityDescription) -> None: self._code_slot: None | int = self._get_code_slots_num() if "accesslimit_day_of_week" in self._property: self._day_of_week_num: None | int = self._get_day_of_week_num() - self._attr_extra_state_attributes: Mapping[str, Any] = {} - self._attr_device_info: Mapping[str, Any] = { + self._attr_extra_state_attributes: dict[str, Any] = {} + self._attr_device_info: DeviceInfo = { "identifiers": {(DOMAIN, self._config_entry.entry_id)}, } # _LOGGER.debug(f"[Entity init] Entity created: {self.name}, device_info: {self.device_info}") @@ -135,7 +137,7 @@ def _get_day_of_week_num(self) -> None | int: return None -@dataclass(kw_only=True) +@dataclass(frozen=True, kw_only=True) class KeymasterEntityDescription(EntityDescription): """Base keymaster Entitity Description.""" diff --git a/custom_components/keymaster/helpers.py b/custom_components/keymaster/helpers.py index 60f726e6..a87706ca 100644 --- a/custom_components/keymaster/helpers.py +++ b/custom_components/keymaster/helpers.py @@ -2,7 +2,7 @@ from __future__ import annotations -from collections.abc import Callable, Mapping +from collections.abc import Callable, MutableMapping from datetime import datetime, timedelta import logging import time @@ -31,7 +31,7 @@ class Throttle: def __init__(self) -> None: """Initialize Throttle class.""" - self._cooldowns = ( + self._cooldowns: MutableMapping = ( {} ) # Nested dictionary: {function_name: {key: last_called_time}} @@ -101,7 +101,7 @@ async def start(self) -> bool: ) return True - async def cancel(self, timer_elapsed: datetime | None = None) -> bool: + async def cancel(self, timer_elapsed: datetime | None = None) -> None: """Cancel a timer.""" if timer_elapsed: _LOGGER.debug("[KeymasterTimer] Timer elapsed") @@ -142,7 +142,7 @@ def is_setup(self) -> bool: unsub() self._unsub_events = [] self._end_time = None - return self.hass and self._kmlock and self._call_action + return bool(self.hass and self._kmlock and self._call_action) @property def end_time(self) -> datetime | None: @@ -176,7 +176,7 @@ def remaining_seconds(self) -> int | None: self._unsub_events = [] self._end_time = None return None - return (datetime.now().astimezone() - self._end_time).total_seconds() + return round((datetime.now().astimezone() - self._end_time).total_seconds()) @callback @@ -190,12 +190,15 @@ def _async_using( if not (kmlock or entity_id): raise TypeError("Missing arguments") ent_reg = er.async_get(hass) - if kmlock: + if kmlock and kmlock.lock_entity_id: entity = ent_reg.async_get(kmlock.lock_entity_id) - else: + elif entity_id: entity = ent_reg.async_get(entity_id) - - return entity and entity.platform == domain + else: + return False + if not entity: + return False + return bool(entity) and bool(entity.platform == domain) @callback @@ -272,7 +275,7 @@ async def delete_code_slot_entities( f"time.code_slots:{code_slot}.accesslimit_day_of_week:{dow}.time_end", ] for prop in dow_prop: - entity_id: str | None = entity_registry.async_get_entity_id( + entity_id = entity_registry.async_get_entity_id( domain=prop.split(".", maxsplit=1)[0], platform=DOMAIN, unique_id=f"{keymaster_config_entry_id}_{slugify(prop)}", @@ -300,8 +303,8 @@ async def call_hass_service( hass: HomeAssistant, domain: str, service: str, - service_data: Mapping[str, Any] | None = None, - target: Mapping[str, Any] | None = None, + service_data: dict[str, Any] | None = None, + target: dict[str, Any] | None = None, ) -> None: """Call a hass service and log a failure on an error.""" _LOGGER.debug( diff --git a/custom_components/keymaster/lock.py b/custom_components/keymaster/lock.py index dd233c56..3702f4ba 100644 --- a/custom_components/keymaster/lock.py +++ b/custom_components/keymaster/lock.py @@ -2,10 +2,10 @@ from __future__ import annotations -from collections.abc import Callable, Mapping +from collections.abc import Callable, MutableMapping from dataclasses import dataclass, field from datetime import datetime, time as dt_time -from typing import TYPE_CHECKING, Any +from typing import TYPE_CHECKING from zwave_js_server.model.node import Node as ZwaveJSNode @@ -48,7 +48,7 @@ class KeymasterCodeSlot: accesslimit_date_range_start: datetime | None = None accesslimit_date_range_end: datetime | None = None accesslimit_day_of_week_enabled: bool = False - accesslimit_day_of_week: Mapping[int, KeymasterCodeSlotDayOfWeek] | None = None + accesslimit_day_of_week: MutableMapping[int, KeymasterCodeSlotDayOfWeek] | None = None @dataclass @@ -67,7 +67,7 @@ class KeymasterLock: zwave_js_lock_device: DeviceEntry | None = None number_of_code_slots: int | None = None starting_code_slot: int = 1 - code_slots: Mapping[int, KeymasterCodeSlot] | None = None + code_slots: MutableMapping[int, KeymasterCodeSlot] | None = None lock_notifications: bool = False door_notifications: bool = False notify_script_name: str | None = None @@ -86,7 +86,7 @@ class KeymasterLock: pending_delete: bool = False -keymasterlock_type_lookup: Mapping[str, Any] = { +keymasterlock_type_lookup: MutableMapping[str, type] = { "lock_name": str, "lock_entity_id": str, "keymaster_config_entry_id": str, @@ -99,7 +99,7 @@ class KeymasterLock: # "zwave_js_lock_device": DeviceEntry, "number_of_code_slots": int, "starting_code_slot": int, - "code_slots": Mapping[int, KeymasterCodeSlot], + "code_slots": MutableMapping[int, KeymasterCodeSlot], "lock_notifications": bool, "door_notifications": bool, "notify_script_name": str, @@ -137,5 +137,5 @@ class KeymasterLock: "accesslimit_date_range_start": datetime, "accesslimit_date_range_end": datetime, "accesslimit_day_of_week_enabled": bool, - "accesslimit_day_of_week": Mapping[int, KeymasterCodeSlotDayOfWeek], + "accesslimit_day_of_week": MutableMapping[int, KeymasterCodeSlotDayOfWeek], } diff --git a/custom_components/keymaster/lovelace.py b/custom_components/keymaster/lovelace.py index f702fae9..ece51242 100644 --- a/custom_components/keymaster/lovelace.py +++ b/custom_components/keymaster/lovelace.py @@ -2,7 +2,7 @@ from __future__ import annotations -from collections.abc import Callable, Mapping +from collections.abc import Callable, MutableMapping import functools import logging from pathlib import Path @@ -34,42 +34,43 @@ async def generate_lovelace( filename: str = f"{kmlock_name}.yaml" await hass.async_add_executor_job(_create_lovelace_folder, folder) - badges_list: list[Mapping[str, Any]] = await _generate_lock_badges( + badges_list: list[MutableMapping[str, Any]] = await _generate_lock_badges( child=bool(parent_config_entry_id), door=bool(door_sensor not in {None, DEFAULT_DOOR_SENSOR}), ) - mapped_badges_list: list[Mapping[str, Any]] = await _map_property_to_entity_id( + mapped_badges_list: MutableMapping[str, Any] | list[MutableMapping[str, Any]] = await _map_property_to_entity_id( hass=hass, lovelace_entities=badges_list, keymaster_config_entry_id=keymaster_config_entry_id, parent_config_entry_id=parent_config_entry_id, ) - await _add_lock_and_door_to_badges( - badges_list=mapped_badges_list, - lock_entity=lock_entity, - door_sensor=door_sensor, - ) - code_slot_list: list[Mapping[str, Any]] = [] + if isinstance(mapped_badges_list, list): + await _add_lock_and_door_to_badges( + badges_list=mapped_badges_list, + lock_entity=lock_entity, + door_sensor=door_sensor, + ) + code_slot_list: list[MutableMapping[str, Any]] = [] for x in range( code_slot_start, code_slot_start + code_slots, ): if parent_config_entry_id: - code_slot_dict: Mapping[str, Any] = await _generate_child_code_slot_dict( + code_slot_dict: MutableMapping[str, Any] = await _generate_child_code_slot_dict( code_slot=x ) else: - code_slot_dict: Mapping[str, Any] = await _generate_code_slot_dict( + code_slot_dict = await _generate_code_slot_dict( code_slot=x ) code_slot_list.append(code_slot_dict) - lovelace_list: list[Mapping[str, Any]] = await _map_property_to_entity_id( + lovelace_list: MutableMapping[str, Any] | list[MutableMapping[str, Any]] = await _map_property_to_entity_id( hass=hass, lovelace_entities=code_slot_list, keymaster_config_entry_id=keymaster_config_entry_id, parent_config_entry_id=parent_config_entry_id, ) - lovelace: list[Mapping[str, Any]] = [ + lovelace: list[MutableMapping[str, Any]] = [ { "type": "sections", "max_columns": 4, @@ -169,17 +170,17 @@ def _write_lovelace_yaml(folder: str, filename: str, lovelace: Any) -> None: async def _map_property_to_entity_id( hass: HomeAssistant, - lovelace_entities: list[Mapping[str, Any]] | Mapping[str, Any], + lovelace_entities: list[MutableMapping[str, Any]] | MutableMapping[str, Any], keymaster_config_entry_id: str, parent_config_entry_id: str | None = None, -) -> Mapping[str, Any] | list[Mapping[str, Any]]: +) -> MutableMapping[str, Any] | list[MutableMapping[str, Any]]: """Update all the entities with the entity_id for the keymaster lock.""" # _LOGGER.debug( # f"[map_property_to_entity_id] keymaster_config_entry_id: {keymaster_config_entry_id}, " # f"parent_config_entry_id: {parent_config_entry_id}" # ) entity_registry: er.EntityRegistry = er.async_get(hass) - lovelace_list: list[Mapping[str, Any]] | Mapping[str, Any] = ( + lovelace_list: list[MutableMapping[str, Any]] | MutableMapping[str, Any] = ( await _process_entities( lovelace_entities, "entity", @@ -220,13 +221,15 @@ async def _process_entities(data: Any, key_to_find: str, process_func: Callable) async def _get_entity_id( entity_registry: er.EntityRegistry, keymaster_config_entry_id: str, - parent_config_entry_id: str, + parent_config_entry_id: str | None, prop: str, -) -> str: +) -> str | None: """Lookup the entity_id from the property.""" if not prop: return None if prop.split(".", maxsplit=1)[0] == "parent": + if not parent_config_entry_id: + return None prop = prop.split(".", maxsplit=1)[1] # _LOGGER.debug( # f"[get_entity_id] Looking up parent ({parent_config_entry_id}) property: {prop}" @@ -238,7 +241,7 @@ async def _get_entity_id( ) else: # _LOGGER.debug(f"[get_entity_id] Looking up property: {prop}") - entity_id: str | None = entity_registry.async_get_entity_id( + entity_id = entity_registry.async_get_entity_id( domain=prop.split(".", maxsplit=1)[0], platform=DOMAIN, unique_id=f"{keymaster_config_entry_id}_{slugify(prop)}", @@ -246,9 +249,9 @@ async def _get_entity_id( return entity_id -async def _generate_code_slot_dict(code_slot, child=False) -> Mapping[str, Any]: +async def _generate_code_slot_dict(code_slot, child=False) -> MutableMapping[str, Any]: """Build the dict for the code slot.""" - code_slot_dict: Mapping[str, Any] = { + code_slot_dict: MutableMapping[str, Any] = { "type": "grid", "cards": [ { @@ -413,7 +416,7 @@ async def _generate_code_slot_dict(code_slot, child=False) -> Mapping[str, Any]: ] ) - dow_list: list[Mapping[str, Any]] = await _generate_dow_entities( + dow_list: list[MutableMapping[str, Any]] = await _generate_dow_entities( code_slot=code_slot ) code_slot_dict["cards"][1]["card"]["entities"].extend(dow_list) @@ -421,7 +424,7 @@ async def _generate_code_slot_dict(code_slot, child=False) -> Mapping[str, Any]: async def _add_lock_and_door_to_badges( - badges_list: list[Mapping[str, Any]], + badges_list: list[MutableMapping[str, Any]], lock_entity: str, door_sensor: str | None = None, ) -> None: @@ -434,8 +437,8 @@ async def _add_lock_and_door_to_badges( async def _generate_lock_badges( child: bool = False, door: bool = False -) -> list[Mapping[str, Any]]: - badges: list[Mapping[str, Any]] = [ +) -> list[MutableMapping[str, Any]]: + badges: list[MutableMapping[str, Any]] = [ { "type": "entity", "show_name": False, @@ -591,9 +594,9 @@ async def _generate_lock_badges( return badges -async def _generate_dow_entities(code_slot) -> list[Mapping[str, Any]]: +async def _generate_dow_entities(code_slot) -> list[MutableMapping[str, Any]]: """Build the day of week entities for the code slot.""" - dow_list: list[Mapping[str, Any]] = [] + dow_list: list[MutableMapping[str, Any]] = [] for dow_num, dow in enumerate( [ "Monday", @@ -725,15 +728,15 @@ async def _generate_dow_entities(code_slot) -> list[Mapping[str, Any]]: return dow_list -async def _generate_child_code_slot_dict(code_slot) -> Mapping[str, Any]: +async def _generate_child_code_slot_dict(code_slot) -> MutableMapping[str, Any]: """Build the dict for the code slot of a child keymaster lock.""" - normal_code_slot_dict: Mapping[str, Any] = await _generate_code_slot_dict( + normal_code_slot_dict: MutableMapping[str, Any] = await _generate_code_slot_dict( code_slot=code_slot, child=True ) override_code_slot_dict = normal_code_slot_dict["cards"][1] - code_slot_dict: Mapping[str, Any] = { + code_slot_dict: MutableMapping[str, Any] = { "type": "grid", "cards": [ { @@ -913,16 +916,16 @@ async def _generate_child_code_slot_dict(code_slot) -> Mapping[str, Any]: ], } - dow_list: list[Mapping[str, Any]] = await _generate_child_dow_entities( + dow_list: list[MutableMapping[str, Any]] = await _generate_child_dow_entities( code_slot=code_slot ) code_slot_dict["cards"][1]["card"]["entities"].extend(dow_list) return code_slot_dict -async def _generate_child_dow_entities(code_slot) -> list[Mapping[str, Any]]: +async def _generate_child_dow_entities(code_slot) -> list[MutableMapping[str, Any]]: """Build the day of week entities for a child code slot.""" - dow_list: list[Mapping[str, Any]] = [] + dow_list: list[MutableMapping[str, Any]] = [] for dow_num, dow in enumerate( [ "Monday", diff --git a/custom_components/keymaster/migrate.py b/custom_components/keymaster/migrate.py index 5c3b1b6c..e819eb83 100644 --- a/custom_components/keymaster/migrate.py +++ b/custom_components/keymaster/migrate.py @@ -3,8 +3,8 @@ from __future__ import annotations import asyncio -from collections.abc import Mapping -from datetime import datetime, time as dt_time, timezone +from collections.abc import MutableMapping +from datetime import datetime, time as dt_time, timedelta, timezone import logging from pathlib import Path from typing import Any @@ -54,18 +54,18 @@ async def migrate_2to3(hass: HomeAssistant, config_entry: ConfigEntry) -> bool: _LOGGER.info("[migrate_2to3] Starting Migration from Config 2 to 3") _LOGGER.debug("[migrate_2to3] config_entry: %s", config_entry) _LOGGER.debug("[migrate_2to3] config_entry.data: %s", config_entry.data) - entity_registry = er.async_get(hass) + entity_registry: er.EntityRegistry = er.async_get(hass) # Move states from helpers into kmlock _LOGGER.info("[migrate_2to3] Moving states from helpers into kmlock") kmlock: KeymasterLock = await _migrate_2to3_create_kmlock(config_entry=config_entry) - crosswalk_dict: Mapping[str, str] = await _migrate_2to3_build_crosswalk_dict( + crosswalk_dict: MutableMapping[str, str] = await _migrate_2to3_build_crosswalk_dict( lock_name=config_entry.data[CONF_LOCK_NAME], starting_slot=config_entry.data[CONF_START], num_slots=config_entry.data[CONF_SLOTS], ) - for entity_id, prop in crosswalk_dict.items(): - ent = hass.states.get(entity_id) + for ent_id, prop in crosswalk_dict.items(): + ent = hass.states.get(ent_id) if not ent: continue await _migrate_2to3_set_property_value( @@ -97,10 +97,10 @@ async def migrate_2to3(hass: HomeAssistant, config_entry: ConfigEntry) -> bool: # Delete existing integration entities _LOGGER.info("[migrate_2to3] Deleting existing integration entities") - for ent in er.async_entries_for_config_entry( - entity_registry, config_entry.entry_id + for del_ent in er.async_entries_for_config_entry( + registry=entity_registry, config_entry_id=config_entry.entry_id ): - entity_id: str = ent.entity_id + entity_id: str = del_ent.entity_id try: entity_registry.async_remove(entity_id) _LOGGER.info("[migrate_2to3] Removed entity_id: %s", entity_id) @@ -138,12 +138,12 @@ async def migrate_2to3(hass: HomeAssistant, config_entry: ConfigEntry) -> bool: async def _migrate_2to3_create_kmlock(config_entry: ConfigEntry) -> KeymasterLock: - code_slots: Mapping[int, KeymasterCodeSlot] = {} + code_slots: MutableMapping[int, KeymasterCodeSlot] = {} for x in range( config_entry.data[CONF_START], config_entry.data[CONF_START] + config_entry.data[CONF_SLOTS], ): - dow_slots: Mapping[int, KeymasterCodeSlotDayOfWeek] = {} + dow_slots: MutableMapping[int, KeymasterCodeSlotDayOfWeek] = {} for i, dow in enumerate( [ "Monday", @@ -161,8 +161,8 @@ async def _migrate_2to3_create_kmlock(config_entry: ConfigEntry) -> KeymasterLoc code_slots[x] = KeymasterCodeSlot(number=x, accesslimit_day_of_week=dow_slots) return KeymasterLock( - lock_name=config_entry.data.get(CONF_LOCK_NAME), - lock_entity_id=config_entry.data.get(CONF_LOCK_ENTITY_ID), + lock_name=config_entry.data[CONF_LOCK_NAME], + lock_entity_id=config_entry.data[CONF_LOCK_ENTITY_ID], keymaster_config_entry_id=config_entry.entry_id, alarm_level_or_user_code_entity_id=config_entry.data.get( CONF_ALARM_LEVEL_OR_USER_CODE_ENTITY_ID @@ -171,8 +171,8 @@ async def _migrate_2to3_create_kmlock(config_entry: ConfigEntry) -> KeymasterLoc CONF_ALARM_TYPE_OR_ACCESS_CONTROL_ENTITY_ID ), door_sensor_entity_id=config_entry.data.get(CONF_DOOR_SENSOR_ENTITY_ID), - number_of_code_slots=config_entry.data.get(CONF_SLOTS), - starting_code_slot=config_entry.data.get(CONF_START), + number_of_code_slots=config_entry.data[CONF_SLOTS], + starting_code_slot=config_entry.data[CONF_START], code_slots=code_slots, parent_name=config_entry.data.get(CONF_PARENT), parent_config_entry_id=config_entry.data.get(CONF_PARENT_ENTRY_ID), @@ -216,15 +216,12 @@ async def _migrate_2to3_set_property_value( return True -async def _migrate_2to3_validate_and_convert_property(prop, attr, value) -> Any: - if isinstance(value, keymasterlock_type_lookup.get(attr)): +async def _migrate_2to3_validate_and_convert_property(prop: str, attr: str, value) -> Any: + if keymasterlock_type_lookup.get(attr) is not None and isinstance(value, keymasterlock_type_lookup.get(attr, object)): # return value pass elif keymasterlock_type_lookup.get(attr) is bool and isinstance(value, str): - if value == "on": - value = True - else: - value = False + value = bool(value == "on") elif keymasterlock_type_lookup.get(attr) is int and isinstance(value, str): try: value = float(value) @@ -247,9 +244,9 @@ async def _migrate_2to3_validate_and_convert_property(prop, attr, value) -> Any: value = round(value) elif keymasterlock_type_lookup.get(attr) == datetime and isinstance(value, str): try: - value: datetime = datetime.fromisoformat(value) - value = value.replace( - tzinfo=timezone(datetime.now().astimezone().utcoffset()) + value_notz: datetime = datetime.fromisoformat(value) + value = value_notz.replace( + tzinfo=timezone(datetime.now().astimezone().utcoffset() or timedelta()) ) except ValueError: _LOGGER.debug( @@ -263,7 +260,7 @@ async def _migrate_2to3_validate_and_convert_property(prop, attr, value) -> Any: return None elif keymasterlock_type_lookup.get(attr) == dt_time and isinstance(value, str): try: - value: dt_time = dt_time.fromisoformat(value) + value = dt_time.fromisoformat(value) except ValueError: _LOGGER.debug( "[migrate_2to3_set_property_value] Value Type Mismatch, cannot convert str to time. Property: %s, final_prop: %s, value: %s. Type: %s, Expected Type: %s", @@ -305,7 +302,7 @@ def _migrate_2to3_delete_lock_and_base_folder( base_path.rmdir() -def _migrate_2to3_delete_folder(absolute_path: str, *relative_paths: str) -> None: +def _migrate_2to3_delete_folder(absolute_path: Path, *relative_paths: str) -> None: """Recursively delete folder and all children files and folders (depth first).""" path: Path = Path(absolute_path) / Path(*relative_paths) @@ -329,10 +326,10 @@ async def _migrate_2to3_reload_package_platforms(hass: HomeAssistant) -> bool: TEMPLATE_DOMAIN, TIMER_DOMAIN, ]: - if domain in hass.services: - await hass.services.async_call(domain, SERVICE_RELOAD, blocking=True) + if hass.services.has_service(domain=domain, service=SERVICE_RELOAD): + await hass.services.async_call(domain=domain, service=SERVICE_RELOAD, blocking=True) else: - _LOGGER.warning("Service not found for domain: %s", domain) + _LOGGER.warning("Reload service not found for domain: %s", domain) return False return True @@ -485,8 +482,8 @@ async def _migrate_2to3_build_delete_list( async def _migrate_2to3_build_crosswalk_dict( lock_name: str, starting_slot: int, num_slots: int -) -> Mapping[str, str]: - crosswalk_dict: Mapping[str, str] = { +) -> MutableMapping[str, str]: + crosswalk_dict: MutableMapping[str, str] = { f"input_boolean.keymaster_{lock_name}_autolock": "switch.autolock_enabled", f"input_boolean.keymaster_{lock_name}_retry": "switch.retry_lock", f"input_boolean.{lock_name}_dooraccess_notifications": "switch.door_notifications", diff --git a/custom_components/keymaster/number.py b/custom_components/keymaster/number.py index 8337f0b6..a79fafd2 100644 --- a/custom_components/keymaster/number.py +++ b/custom_components/keymaster/number.py @@ -92,10 +92,9 @@ async def async_setup_entry( ) async_add_entities(entities, True) - return True -@dataclass(kw_only=True) +@dataclass(frozen=True, kw_only=True) class KeymasterNumberEntityDescription( KeymasterEntityDescription, NumberEntityDescription ): @@ -105,6 +104,8 @@ class KeymasterNumberEntityDescription( class KeymasterNumber(KeymasterEntity, NumberEntity): """Class for keymaster Number.""" + entity_description: KeymasterNumberEntityDescription + def __init__( self, entity_description: KeymasterNumberEntityDescription, @@ -118,15 +119,15 @@ def __init__( @callback def _handle_coordinator_update(self) -> None: # _LOGGER.debug(f"[Number handle_coordinator_update] self.coordinator.data: {self.coordinator.data}") - if not self._kmlock.connected: + if not self._kmlock or not self._kmlock.connected: self._attr_available = False self.async_write_ha_state() return if ( ".code_slots" in self._property - and self._kmlock.parent_name is not None - and not self._kmlock.code_slots[self._code_slot].override_parent + and self._kmlock and self._kmlock.parent_name + and (not self._kmlock.code_slots or not self._code_slot or not self._kmlock.code_slots[self._code_slot].override_parent) ): self._attr_available = False self.async_write_ha_state() @@ -134,7 +135,7 @@ def _handle_coordinator_update(self) -> None: if ( ".code_slots" in self._property - and self._code_slot not in self._kmlock.code_slots + and (not self._kmlock.code_slots or self._code_slot not in self._kmlock.code_slots) ): self._attr_available = False self.async_write_ha_state() @@ -142,7 +143,7 @@ def _handle_coordinator_update(self) -> None: if ( self._property.endswith(".accesslimit_count") - and not self._kmlock.code_slots[self._code_slot].accesslimit_count_enabled + and (not self._kmlock.code_slots or not self._code_slot or not self._kmlock.code_slots[self._code_slot].accesslimit_count_enabled) ): self._attr_available = False self.async_write_ha_state() @@ -169,8 +170,8 @@ async def async_set_native_value(self, value: float) -> None: ) if ( self._property.endswith(".accesslimit_count") - and self._kmlock.parent_name is not None - and not self._kmlock.code_slots[self._code_slot].override_parent + and self._kmlock and self._kmlock.parent_name + and (not self._kmlock.code_slots or not self._code_slot or not self._kmlock.code_slots[self._code_slot].override_parent) ): _LOGGER.debug( "[Number async_set_value] %s: Child lock and code slot %s not set to override parent. Ignoring change", diff --git a/custom_components/keymaster/sensor.py b/custom_components/keymaster/sensor.py index 4da9db38..d51c91a8 100644 --- a/custom_components/keymaster/sensor.py +++ b/custom_components/keymaster/sensor.py @@ -11,7 +11,6 @@ from .const import CONF_SLOTS, CONF_START, COORDINATOR, DOMAIN from .coordinator import KeymasterCoordinator from .entity import KeymasterEntity, KeymasterEntityDescription -from .lock import KeymasterLock _LOGGER: logging.Logger = logging.getLogger(__name__) @@ -22,7 +21,7 @@ async def async_setup_entry( """Create keymaster Sensor entities.""" coordinator: KeymasterCoordinator = hass.data[DOMAIN][COORDINATOR] - kmlock: KeymasterLock = await coordinator.get_lock_by_config_entry_id( + kmlock = await coordinator.get_lock_by_config_entry_id( config_entry.entry_id ) entities: list = [] @@ -41,7 +40,7 @@ async def async_setup_entry( ) ) - if hasattr(kmlock, "parent_name") and kmlock.parent_name is not None: + if kmlock and hasattr(kmlock, "parent_name") and kmlock.parent_name is not None: entities.append( KeymasterSensor( entity_description=KeymasterSensorEntityDescription( @@ -80,7 +79,7 @@ async def async_setup_entry( return True -@dataclass(kw_only=True) +@dataclass(frozen=True, kw_only=True) class KeymasterSensorEntityDescription( KeymasterEntityDescription, SensorEntityDescription ): @@ -90,6 +89,8 @@ class KeymasterSensorEntityDescription( class KeymasterSensor(KeymasterEntity, SensorEntity): """Class for keymaster Sensors.""" + entity_description: KeymasterSensorEntityDescription + def __init__( self, entity_description: KeymasterSensorEntityDescription, @@ -103,14 +104,14 @@ def __init__( @callback def _handle_coordinator_update(self) -> None: # _LOGGER.debug(f"[Sensor handle_coordinator_update] self.coordinator.data: {self.coordinator.data}") - if not self._kmlock.connected: + if not self._kmlock or not self._kmlock.connected: self._attr_available = False self.async_write_ha_state() return if ( ".code_slots" in self._property - and self._code_slot not in self._kmlock.code_slots + and (not self._kmlock.code_slots or self._code_slot not in self._kmlock.code_slots) ): self._attr_available = False self.async_write_ha_state() diff --git a/custom_components/keymaster/services.py b/custom_components/keymaster/services.py index e97b80d3..ef505ef9 100644 --- a/custom_components/keymaster/services.py +++ b/custom_components/keymaster/services.py @@ -49,8 +49,8 @@ async def async_setup_services(hass: HomeAssistant) -> None: async def service_update_pin(service: ServiceCall) -> None: """Update a PIN in a Code Slot.""" _LOGGER.debug("[service_update_pin] service.data: %s", service.data) - code_slot: int = service.data.get(ATTR_CODE_SLOT) - pin: str = service.data.get(ATTR_PIN) + code_slot: int = service.data[ATTR_CODE_SLOT] + pin: str = service.data[ATTR_PIN] if not pin or not pin.isdigit() or len(pin) < 4: _LOGGER.error( "[service_update_pin] Code Slot %s: PIN not valid: %s. Must be 4 or more digits", @@ -61,7 +61,7 @@ async def service_update_pin(service: ServiceCall) -> None: f"Update PIN Error. PIN not valid: {pin}. Must be 4 or more digits" ) await coordinator.set_pin_on_lock( - config_entry_id=service.data.get(ATTR_CONFIG_ENTRY_ID), + config_entry_id=service.data[ATTR_CONFIG_ENTRY_ID], code_slot=code_slot, pin=pin, set_in_kmlock=True, @@ -70,9 +70,9 @@ async def service_update_pin(service: ServiceCall) -> None: async def service_clear_pin(service: ServiceCall) -> None: """Clear a PIN from a Code Slot.""" _LOGGER.debug("[service_clear_pin] service.data: %s", service.data) - code_slot: int = service.data.get(ATTR_CODE_SLOT) + code_slot: int = service.data[ATTR_CODE_SLOT] await coordinator.clear_pin_from_lock( - config_entry_id=service.data.get(ATTR_CONFIG_ENTRY_ID), + config_entry_id=service.data[ATTR_CONFIG_ENTRY_ID], code_slot=code_slot, clear_from_kmlock=True, ) @@ -82,12 +82,12 @@ async def service_regenerate_lovelace(_: ServiceCall) -> None: for config_entry in entries: await generate_lovelace( hass=hass, - kmlock_name=config_entry.data.get(CONF_LOCK_NAME), + kmlock_name=config_entry.data[CONF_LOCK_NAME], keymaster_config_entry_id=config_entry.entry_id, parent_config_entry_id=config_entry.data.get(CONF_PARENT_ENTRY_ID), - code_slot_start=config_entry.data.get(CONF_START), - code_slots=config_entry.data.get(CONF_SLOTS), - lock_entity=config_entry.data.get(CONF_LOCK_ENTITY_ID), + code_slot_start=config_entry.data[CONF_START], + code_slots=config_entry.data[CONF_SLOTS], + lock_entity=config_entry.data[CONF_LOCK_ENTITY_ID], door_sensor=config_entry.data.get(CONF_DOOR_SENSOR_ENTITY_ID), ) diff --git a/custom_components/keymaster/switch.py b/custom_components/keymaster/switch.py index bcc078ec..4bf2cde1 100644 --- a/custom_components/keymaster/switch.py +++ b/custom_components/keymaster/switch.py @@ -1,6 +1,6 @@ """Switch for keymaster.""" -from collections.abc import Mapping +from collections.abc import MutableMapping from dataclasses import dataclass import logging @@ -19,7 +19,7 @@ from .coordinator import KeymasterCoordinator from .entity import KeymasterEntity, KeymasterEntityDescription from .helpers import async_using_zwave_js -from .lock import KeymasterLock +from .lock import KeymasterCodeSlot, KeymasterCodeSlotDayOfWeek, KeymasterLock _LOGGER: logging.Logger = logging.getLogger(__name__) @@ -27,13 +27,13 @@ async def async_setup_entry(hass, config_entry, async_add_entities): """Create keymaster Switches.""" coordinator: KeymasterCoordinator = hass.data[DOMAIN][COORDINATOR] - kmlock: KeymasterLock = await coordinator.get_lock_by_config_entry_id( + kmlock: KeymasterLock | None = await coordinator.get_lock_by_config_entry_id( config_entry.entry_id ) entities: list = [] if async_using_zwave_js(hass=hass, kmlock=kmlock): - lock_switch_entities: list[Mapping[str, str]] = [ + lock_switch_entities: list[MutableMapping[str, str]] = [ { "prop": "switch.autolock_enabled", "name": "Auto Lock", @@ -84,7 +84,7 @@ async def async_setup_entry(hass, config_entry, async_add_entities): config_entry.data[CONF_START], config_entry.data[CONF_START] + config_entry.data[CONF_SLOTS], ): - if kmlock.parent_name: + if kmlock and kmlock.parent_name: entities.append( KeymasterSwitch( entity_description=KeymasterSwitchEntityDescription( @@ -98,7 +98,7 @@ async def async_setup_entry(hass, config_entry, async_add_entities): ) ) ) - code_slot_switch_entities: list[Mapping[str, str]] = [ + code_slot_switch_entities: list[MutableMapping[str, str]] = [ { "prop": f"switch.code_slots:{x}.enabled", "name": f"Code Slot {x}: Enabled", @@ -152,7 +152,7 @@ async def async_setup_entry(hass, config_entry, async_add_entities): "Sunday", ] ): - dow_switch_entities: list[Mapping[str, str]] = [ + dow_switch_entities: list[MutableMapping[str, str]] = [ { "prop": f"switch.code_slots:{x}.accesslimit_day_of_week:{i}.dow_enabled", "name": f"Code Slot {x}: {dow}", @@ -194,7 +194,7 @@ async def async_setup_entry(hass, config_entry, async_add_entities): return True -@dataclass(kw_only=True) +@dataclass(frozen=True, kw_only=True) class KeymasterSwitchEntityDescription( KeymasterEntityDescription, SwitchEntityDescription ): @@ -204,6 +204,8 @@ class KeymasterSwitchEntityDescription( class KeymasterSwitch(KeymasterEntity, SwitchEntity): """Class for keymaster Switches.""" + entity_description: KeymasterSwitchEntityDescription + def __init__( self, entity_description: KeymasterSwitchEntityDescription, @@ -212,12 +214,12 @@ def __init__( super().__init__( entity_description=entity_description, ) - self._attr_is_on = False + self._attr_is_on: bool = False @callback def _handle_coordinator_update(self) -> None: # _LOGGER.debug(f"[Switch handle_coordinator_update] self.coordinator.data: {self.coordinator.data}") - if not self._kmlock.connected: + if not self._kmlock or not self._kmlock.connected: self._attr_available = False self.async_write_ha_state() return @@ -229,7 +231,7 @@ def _handle_coordinator_update(self) -> None: or self._property.endswith(".notifications") ) and self._kmlock.parent_name is not None - and not self._kmlock.code_slots[self._code_slot].override_parent + and (not self._kmlock.code_slots or not self._code_slot or not self._kmlock.code_slots[self._code_slot].override_parent) ): self._attr_available = False self.async_write_ha_state() @@ -238,7 +240,7 @@ def _handle_coordinator_update(self) -> None: if ( not self._property.endswith(".enabled") and ".code_slots" in self._property - and self._code_slot not in self._kmlock.code_slots + and (not self._kmlock.code_slots or self._code_slot not in self._kmlock.code_slots) ): self._attr_available = False self.async_write_ha_state() @@ -247,38 +249,64 @@ def _handle_coordinator_update(self) -> None: if ( ".accesslimit_day_of_week" in self._property and not self._property.endswith(".accesslimit_day_of_week_enabled") - and not self._kmlock.code_slots[ + and (not self._kmlock.code_slots or not self._code_slot or not self._kmlock.code_slots[ self._code_slot - ].accesslimit_day_of_week_enabled + ].accesslimit_day_of_week_enabled) ): self._attr_available = False self.async_write_ha_state() return + code_slots: MutableMapping[int, KeymasterCodeSlot] | None = self._kmlock.code_slots + accesslimit_dow: MutableMapping[int, KeymasterCodeSlotDayOfWeek] | None = None + if self._code_slot is not None and code_slots and self._code_slot in code_slots: + accesslimit_dow = code_slots[self._code_slot].accesslimit_day_of_week + if ( self._property.endswith(".limit_by_time") - and not self._kmlock.code_slots[self._code_slot] - .accesslimit_day_of_week[self._day_of_week_num] - .dow_enabled + and ( + not code_slots + or self._code_slot is None + or not accesslimit_dow + or self._day_of_week_num is None + or self._day_of_week_num not in accesslimit_dow + or not accesslimit_dow[self._day_of_week_num].dow_enabled + ) ): self._attr_available = False self.async_write_ha_state() return - if self._property.endswith(".include_exclude") and ( - not self._kmlock.code_slots[self._code_slot] - .accesslimit_day_of_week[self._day_of_week_num] - .dow_enabled - or not self._kmlock.code_slots[self._code_slot] - .accesslimit_day_of_week[self._day_of_week_num] - .limit_by_time + # if self._property.endswith(".include_exclude") and ( + # not self._kmlock.code_slots[self._code_slot] + # .accesslimit_day_of_week[self._day_of_week_num] + # .dow_enabled + # or not self._kmlock.code_slots[self._code_slot] + # .accesslimit_day_of_week[self._day_of_week_num] + # .limit_by_time + # ): + # self._attr_available = False + # self.async_write_ha_state() + # return + + if ( + self._property.endswith(".include_exclude") + and ( + not code_slots + or self._code_slot is None + or not accesslimit_dow + or self._day_of_week_num is None + or self._day_of_week_num not in accesslimit_dow + or not accesslimit_dow[self._day_of_week_num].dow_enabled + or not accesslimit_dow[self._day_of_week_num].limit_by_time + ) ): self._attr_available = False self.async_write_ha_state() return self._attr_available = True - self._attr_is_on: bool = self._get_property_value() + self._attr_is_on = self._get_property_value() self.async_write_ha_state() async def async_turn_on(self, **_) -> None: @@ -294,7 +322,7 @@ async def async_turn_on(self, **_) -> None: if self._set_property_value(True): self._attr_is_on = True - if self._property.endswith(".enabled"): + if self._property.endswith(".enabled") and self._kmlock and self._code_slot and self._kmlock.code_slots: await self.coordinator.update_slot_active_state( config_entry_id=self._config_entry.entry_id, code_slot=self._code_slot, @@ -321,7 +349,7 @@ async def async_turn_off(self, **_) -> None: if self._set_property_value(False): self._attr_is_on = False - if self._property.endswith(".enabled"): + if self._property.endswith(".enabled") and self._code_slot: await self.coordinator.update_slot_active_state( config_entry_id=self._config_entry.entry_id, code_slot=self._code_slot, diff --git a/custom_components/keymaster/text.py b/custom_components/keymaster/text.py index 07eacbcd..b6fb6d65 100644 --- a/custom_components/keymaster/text.py +++ b/custom_components/keymaster/text.py @@ -29,7 +29,7 @@ async def async_setup_entry( config_entry.data[CONF_START], config_entry.data[CONF_START] + config_entry.data[CONF_SLOTS], ): - entities.append( + entities.extend([ KeymasterText( entity_description=KeymasterTextEntityDescription( key=f"text.code_slots:{x}.name", @@ -40,9 +40,7 @@ async def async_setup_entry( config_entry=config_entry, coordinator=coordinator, ), - ) - ) - entities.append( + ), KeymasterText( entity_description=KeymasterTextEntityDescription( key=f"text.code_slots:{x}.pin", @@ -59,13 +57,12 @@ async def async_setup_entry( coordinator=coordinator, ), ) - ) + ]) async_add_entities(entities, True) - return True -@dataclass(kw_only=True) +@dataclass(frozen=True, kw_only=True) class KeymasterTextEntityDescription(KeymasterEntityDescription, TextEntityDescription): """Entity Description for keymaster Text entities.""" @@ -73,6 +70,8 @@ class KeymasterTextEntityDescription(KeymasterEntityDescription, TextEntityDescr class KeymasterText(KeymasterEntity, TextEntity): """Class for keymaster Text entities.""" + entity_description: KeymasterTextEntityDescription + def __init__( self, entity_description: KeymasterTextEntityDescription, @@ -81,12 +80,12 @@ def __init__( super().__init__( entity_description=entity_description, ) - self._attr_native_value: str = None + self._attr_native_value: str | None = None @callback def _handle_coordinator_update(self) -> None: # _LOGGER.debug(f"[Text handle_coordinator_update] self.coordinator.data: {self.coordinator.data}") - if not self._kmlock.connected: + if not self._kmlock or not self._kmlock.connected: self._attr_available = False self.async_write_ha_state() return @@ -94,7 +93,7 @@ def _handle_coordinator_update(self) -> None: if ( ".code_slots" in self._property and self._kmlock.parent_name is not None - and not self._kmlock.code_slots[self._code_slot].override_parent + and (not self._kmlock.code_slots or not self._code_slot or not self._kmlock.code_slots[self._code_slot].override_parent) ): self._attr_available = False self.async_write_ha_state() @@ -102,7 +101,7 @@ def _handle_coordinator_update(self) -> None: if ( ".code_slots" in self._property - and self._code_slot not in self._kmlock.code_slots + and (not self._kmlock.code_slots or self._code_slot not in self._kmlock.code_slots) ): self._attr_available = False self.async_write_ha_state() @@ -120,13 +119,13 @@ async def async_set_value(self, value: str) -> None: value, ) if self._property.endswith(".pin"): - if value and value.isdigit() and len(value) >= 4: + if value and value.isdigit() and len(value) >= 4 and self._code_slot: await self.coordinator.set_pin_on_lock( config_entry_id=self._config_entry.entry_id, code_slot=self._code_slot, pin=value, ) - elif not value: + elif not value and self._code_slot: await self.coordinator.clear_pin_from_lock( config_entry_id=self._config_entry.entry_id, code_slot=self._code_slot, @@ -135,8 +134,8 @@ async def async_set_value(self, value: str) -> None: return elif ( self._property.endswith(".name") - and self._kmlock.parent_name is not None - and not self._kmlock.code_slots[self._code_slot].override_parent + and self._kmlock and self._kmlock.parent_name + and (not self._kmlock.code_slots or not self._code_slot or not self._kmlock.code_slots[self._code_slot].override_parent) ): _LOGGER.debug( "[Text async_set_value] %s: " diff --git a/custom_components/keymaster/time.py b/custom_components/keymaster/time.py index d22de41f..04dc7b3a 100644 --- a/custom_components/keymaster/time.py +++ b/custom_components/keymaster/time.py @@ -1,6 +1,6 @@ """Support for keymaster Time.""" -from collections.abc import Mapping +from collections.abc import MutableMapping from dataclasses import dataclass from datetime import time as dt_time import logging @@ -13,6 +13,7 @@ from .const import CONF_SLOTS, CONF_START, COORDINATOR, DOMAIN from .coordinator import KeymasterCoordinator from .entity import KeymasterEntity, KeymasterEntityDescription +from .lock import KeymasterCodeSlot, KeymasterCodeSlotDayOfWeek _LOGGER: logging.Logger = logging.getLogger(__name__) @@ -41,7 +42,7 @@ async def async_setup_entry( "Sunday", ] ): - dow_time_entities: list[Mapping[str, str]] = [ + dow_time_entities: list[MutableMapping[str, str]] = [ { "prop": f"time.code_slots:{x}.accesslimit_day_of_week:{i}.time_start", "name": f"Code Slot {x}: {dow} - Start Time", @@ -71,10 +72,9 @@ async def async_setup_entry( ) async_add_entities(entities, True) - return True -@dataclass(kw_only=True) +@dataclass(frozen=True, kw_only=True) class KeymasterTimeEntityDescription(KeymasterEntityDescription, TimeEntityDescription): """Entity Description for keymaster Time entities.""" @@ -82,6 +82,8 @@ class KeymasterTimeEntityDescription(KeymasterEntityDescription, TimeEntityDescr class KeymasterTime(KeymasterEntity, TimeEntity): """Class for keymaster Time entities.""" + entity_description: KeymasterTimeEntityDescription + def __init__( self, entity_description: KeymasterTimeEntityDescription, @@ -95,7 +97,7 @@ def __init__( @callback def _handle_coordinator_update(self) -> None: # _LOGGER.debug(f"[Time handle_coordinator_update] self.coordinator.data: {self.coordinator.data}") - if not self._kmlock.connected: + if not self._kmlock or not self._kmlock.connected: self._attr_available = False self.async_write_ha_state() return @@ -103,7 +105,7 @@ def _handle_coordinator_update(self) -> None: if ( ".code_slots" in self._property and self._kmlock.parent_name is not None - and not self._kmlock.code_slots[self._code_slot].override_parent + and (not self._kmlock.code_slots or not self._code_slot or not self._kmlock.code_slots[self._code_slot].override_parent) ): self._attr_available = False self.async_write_ha_state() @@ -111,7 +113,7 @@ def _handle_coordinator_update(self) -> None: if ( ".code_slots" in self._property - and self._code_slot not in self._kmlock.code_slots + and (not self._kmlock.code_slots or self._code_slot not in self._kmlock.code_slots) ): self._attr_available = False self.async_write_ha_state() @@ -119,10 +121,10 @@ def _handle_coordinator_update(self) -> None: if ( ".accesslimit_day_of_week" in self._property - and not self._kmlock.code_slots[ + and (not self._kmlock.code_slots or not self._code_slot or not self._kmlock.code_slots[ self._code_slot ].accesslimit_day_of_week_enabled - ): + )): self._attr_available = False self.async_write_ha_state() return @@ -130,17 +132,42 @@ def _handle_coordinator_update(self) -> None: if ( self._property.endswith(".time_start") or self._property.endswith(".time_end") - ) and ( - not self._kmlock.code_slots[self._code_slot] - .accesslimit_day_of_week[self._day_of_week_num] - .dow_enabled - or not self._kmlock.code_slots[self._code_slot] - .accesslimit_day_of_week[self._day_of_week_num] - .limit_by_time ): - self._attr_available = False - self.async_write_ha_state() - return + code_slots: MutableMapping[int, KeymasterCodeSlot] | None = self._kmlock.code_slots + if self._code_slot is None or code_slots is None or self._code_slot not in code_slots: + self._attr_available = False + self.async_write_ha_state() + return + + accesslimit_day_of_week: MutableMapping[int, KeymasterCodeSlotDayOfWeek] | None = code_slots[self._code_slot].accesslimit_day_of_week + if self._day_of_week_num is None or accesslimit_day_of_week is None or self._day_of_week_num not in accesslimit_day_of_week: + self._attr_available = False + self.async_write_ha_state() + return + + day_of_week: KeymasterCodeSlotDayOfWeek | None = accesslimit_day_of_week[self._day_of_week_num] + if ( + day_of_week is None or not day_of_week.dow_enabled + or not day_of_week.limit_by_time + ): + self._attr_available = False + self.async_write_ha_state() + return + + # if ( + # self._property.endswith(".time_start") + # or self._property.endswith(".time_end") + # ) and ( + # not self._kmlock.code_slots[self._code_slot] + # .accesslimit_day_of_week[self._day_of_week_num] + # .dow_enabled + # or not self._kmlock.code_slots[self._code_slot] + # .accesslimit_day_of_week[self._day_of_week_num] + # .limit_by_time + # ): + # self._attr_available = False + # self.async_write_ha_state() + # return self._attr_available = True self._attr_native_value = self._get_property_value() @@ -158,8 +185,8 @@ async def async_set_value(self, value: dt_time) -> None: self._property.endswith(".time_start") or self._property.endswith(".time_end") ) - and self._kmlock.parent_name is not None - and not self._kmlock.code_slots[self._code_slot].override_parent + and self._kmlock and self._kmlock.parent_name + and (not self._kmlock.code_slots or not self._code_slot or not self._kmlock.code_slots[self._code_slot].override_parent) ): _LOGGER.debug( "[Time async_set_value] %s: Child lock and code slot %s not set to override parent. Ignoring change", diff --git a/pyproject.toml b/pyproject.toml index e7832ebc..e9cb6547 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -3,14 +3,29 @@ show_missing = true fail_under = 80 [tool.mypy] +platform = "linux" python_version = "3.13" -show_error_codes = true -ignore_errors = true follow_imports = "silent" ignore_missing_imports = true +check_untyped_defs = true +disable_error_code = ["annotation-unchecked", "import-not-found", "import-untyped", "call-arg"] +disallow_incomplete_defs = false +disallow_subclassing_any = false +disallow_untyped_calls = false +disallow_untyped_decorators = false +disallow_untyped_defs = false +enable_error_code = ["deprecated", "ignore-without-code", "redundant-self", "truthy-iterable"] +extra_checks = false +local_partial_types = true +no_implicit_optional = true +show_error_codes = true +strict_equality = true warn_incomplete_stub = true warn_redundant_casts = true +warn_return_any = false +warn_unreachable = false warn_unused_configs = true +warn_unused_ignores = true [tool.pylint] ignore = ["tests"] diff --git a/requirements_test.txt b/requirements_test.txt index 3e43c29c..9a43dee4 100644 --- a/requirements_test.txt +++ b/requirements_test.txt @@ -8,4 +8,5 @@ zeroconf tox mypy pylint -ruff \ No newline at end of file +ruff +types-PyYAML \ No newline at end of file