feat: allow command selection based on features and other attributes

alryaz · alryaz · commit 7640d77da9cf · 2024-03-14T02:40:39.000+03:00
refactor: upgraded command selection engine
diff --git a/custom_components/pandora_cas/button.py b/custom_components/pandora_cas/button.py
@@ -22,7 +22,8 @@
     PandoraCASEntity,
     PandoraCASEntityDescription,
     CommandOptions,
-    parse_description_command_id,
+    resolve_command_id,
+    has_device_type,
 )
 from pandora_cas.device import PandoraOnlineDevice
 from pandora_cas.enums import PandoraDeviceTypes, CommandID
@@ -44,10 +45,10 @@ class PandoraCASButtonEntityDescription(
     PandoraCASButtonEntityDescription(
         key="erase_errors",
         name="Erase Errors",
-        command={
-            None: CommandID.ERASE_DTC,
-            PandoraDeviceTypes.NAV12: CommandID.NAV12_RESET_ERRORS,
-        },
+        command=[
+            (has_device_type(PandoraDeviceTypes.NAV12), CommandID.NAV12_RESET_ERRORS),
+            (None, CommandID.ERASE_DTC),
+        ],
         entity_category=EntityCategory.DIAGNOSTIC,
         icon="mdi:eraser",
     ),
@@ -159,9 +160,14 @@ async def async_press(self) -> None:
             raise HomeAssistantError(
                 "Simultaneous commands not allowed, wait until command completes"
             )
-        command_id = parse_description_command_id(
-            self.entity_description.command, self.pandora_device.type
+        command_id = resolve_command_id(
+            self.pandora_device,
+            self.entity_description.command,
+            raise_exceptions=False,
         )
+        if command_id is None:
+            raise HomeAssistantError("could not determine command to run")
+
         self._is_pressing = True
         await self.run_device_command(command_id)
 
diff --git a/custom_components/pandora_cas/entity.py b/custom_components/pandora_cas/entity.py
@@ -1,3 +1,5 @@
+"""Base entity platform for Pandora Car Alarm system component."""
+
 import asyncio
 import dataclasses
 import logging
@@ -23,6 +25,7 @@
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.const import CONF_USERNAME, ATTR_DEVICE_ID
 from homeassistant.core import HomeAssistant, callback, Event
+from homeassistant.exceptions import HomeAssistantError
 from homeassistant.helpers.entity import EntityDescription, DeviceInfo, Entity
 from homeassistant.helpers.entity_platform import (
     AddEntitiesCallback,
@@ -38,6 +41,7 @@
     ATTR_COMMAND_ID,
     CONF_OFFLINE_AS_UNAVAILABLE,
     DEFAULT_WAITER_TIMEOUT,
+    EVENT_TYPE_COMMAND,
 )
 from pandora_cas.device import PandoraOnlineDevice
 from pandora_cas.enums import PandoraDeviceTypes, CommandID, Features
@@ -47,24 +51,96 @@
 
 _LOGGER: Final = logging.getLogger(__name__)
 
+DeviceValidator = Callable[[PandoraOnlineDevice], bool]
+CommandType = CommandID | int | Callable[[PandoraOnlineDevice], Awaitable]
+CommandOptions = CommandType | Sequence[tuple[DeviceValidator | None, CommandType]]
+
+
+# noinspection PyShadowingNames
+def check_for_device(
+    has_features: Features | None = None,
+    not_features: Features | None = None,
+    has_device_type: str | tuple[str, ...] | None = None,
+) -> DeviceValidator:
+    """
+    Creates a checker that checks something is suitable for device.
+    :param has_features: Has specified features.
+    :param not_features:
+    :param has_device_type:
+    :return:
+    """
+    if has_features is None and not_features is None and has_device_type is None:
+        raise ValueError(
+            "You must specify either has_features or not_features or has_device_type"
+        )
+
+    if isinstance(has_device_type, str):
+        has_device_type = (has_device_type,)
+
+    def _perform_check_on_device(device: PandoraOnlineDevice) -> bool:
+        if has_device_type is not None:
+            if device.type not in has_device_type:
+                return False
+        if not (has_features is None and not_features is None):
+            if (features := device.features) is None:
+                return False
+            if not (has_features is None or has_features & features):
+                return False
+            if not (not_features is None or not (not_features & features)):
+                return False
+        return True
+
+    return _perform_check_on_device
+
+
+def has_features(features: Features) -> DeviceValidator:
+    """Shortcut for has_features argument."""
+    return check_for_device(has_features=features)
+
+
+def not_features(features: Features) -> DeviceValidator:
+    """Shortcut for not_features argument."""
+    return check_for_device(not_features=features)
+
+
+def has_device_type(device_type: str | tuple[str, ...]) -> DeviceValidator:
+    """Shortcut for has_device_type argument."""
+    return check_for_device(has_device_type=device_type)
+
 
-def parse_description_command_id(value: Any, device_type: str | None = None) -> int:
+def resolve_command_id(
+    device: PandoraOnlineDevice, value: Any, raise_exceptions: bool = False
+) -> int | None:
     """Retrieve command from definition."""
     if value is None:
-        raise NotImplementedError("command not defined")
+        return None
 
-    if isinstance(value, Mapping):
-        try:
-            value = value[device_type]
-        except KeyError:
-            if device_type is None:
-                raise NotImplementedError("command not defined")
+    if isinstance(value, list):
+        for condition, command in value:
+            if condition is None:
+                return command
             try:
-                value = value[None]
-            except KeyError:
-                raise NotImplementedError("command not defined")
-
-    return value if callable(value) else int(value)
+                if condition(device):
+                    return command
+            except BaseException as exc:
+                if raise_exceptions:
+                    raise
+                _LOGGER.warning(
+                    f"Exception occurred while checking command for device {device}: {exc}",
+                    exc_info=exc,
+                )
+        return None
+
+    try:
+        return value if callable(value) else int(value)
+    except BaseException as exc:
+        if raise_exceptions:
+            raise
+        _LOGGER.warning(
+            f"Exception occurred while checking command for device {device}: {exc}",
+            exc_info=exc,
+        )
+        return None
 
 
 async def async_platform_setup_entry(
@@ -139,10 +215,6 @@ def __post_init__(self):
             self.translation_key = self.key
 
 
-CommandType = CommandID | int | Callable[[PandoraOnlineDevice], Awaitable]
-CommandOptions = CommandType | Mapping[str, CommandType]
-
-
 class BasePandoraCASEntity(Entity):
     ENTITY_ID_FORMAT: ClassVar[str] = NotImplemented
 
@@ -316,14 +388,21 @@ def _process_command_response(self, event: Union[Event, datetime]) -> None:
     def _add_command_listener(self, command: CommandOptions | None) -> None:
         if command is None:
             return None
-        command_id = parse_description_command_id(command, self.pandora_device.type)
+
+        command_id = resolve_command_id(
+            self.pandora_device,
+            command,
+            raise_exceptions=False,
+        )
+
         if not isinstance(command_id, int):
             return None
+
         if (listeners := self._command_listeners) is None:
             self._command_listeners = listeners = []
         listeners.append(
             self.hass.bus.async_listen(
-                event_type=f"{DOMAIN}_command",
+                event_type=EVENT_TYPE_COMMAND,
                 listener=self._process_command_response,
                 event_filter=callback(
                     lambda x: int(x.data[ATTR_COMMAND_ID]) == command_id
@@ -464,15 +543,19 @@ async def run_binary_command(self, enable: bool) -> None:
         :param enable: Whether to run 'on' or 'off'.
         """
         # Determine command to run
-        command_id = parse_description_command_id(
+        command_id = resolve_command_id(
+            self.pandora_device,
             (
                 self.entity_description.command_on
                 if enable or self.entity_description.command_off is None
                 else self.entity_description.command_off
             ),
-            self.pandora_device.type,
+            raise_exceptions=False,
         )
 
+        if command_id is None:
+            raise HomeAssistantError("could not determine command to run")
+
         self._is_turning_on = enable
         self._is_turning_off = not enable
 
diff --git a/custom_components/pandora_cas/switch.py b/custom_components/pandora_cas/switch.py
@@ -18,6 +18,7 @@
     async_platform_setup_entry,
     PandoraCASBooleanEntityDescription,
     PandoraCASBooleanEntity,
+    has_device_type,
 )
 from pandora_cas.data import CurrentState
 from pandora_cas.enums import PandoraDeviceTypes, CommandID, BitStatus, Features
@@ -66,14 +67,20 @@ class PandoraCASSwitchEntityDescription(
         icon_off="mdi:radiator-off",
         attribute=CurrentState.bit_state,
         flag=BitStatus.BLOCK_HEATER_ACTIVE,
-        command_on={
-            None: CommandID.TURN_ON_BLOCK_HEATER,
-            PandoraDeviceTypes.NAV12: CommandID.NAV12_TURN_ON_BLOCK_HEATER,
-        },
-        command_off={
-            None: CommandID.TURN_OFF_BLOCK_HEATER,
-            PandoraDeviceTypes.NAV12: CommandID.NAV12_TURN_OFF_BLOCK_HEATER,
-        },
+        command_on=[
+            (
+                has_device_type(PandoraDeviceTypes.NAV12),
+                CommandID.NAV12_TURN_ON_BLOCK_HEATER,
+            ),
+            (None, CommandID.TURN_ON_BLOCK_HEATER),
+        ],
+        command_off=[
+            (
+                has_device_type(PandoraDeviceTypes.NAV12),
+                CommandID.NAV12_TURN_OFF_BLOCK_HEATER,
+            ),
+            (None, CommandID.TURN_OFF_BLOCK_HEATER),
+        ],
         features=Features.HEATER,
     ),
     PandoraCASSwitchEntityDescription(
@@ -95,14 +102,17 @@ class PandoraCASSwitchEntityDescription(
         icon_off="mdi:wrench",
         attribute=CurrentState.bit_state,
         flag=BitStatus.SERVICE_MODE_ACTIVE,
-        command_on={
-            None: CommandID.ENABLE_SERVICE_MODE,
-            PandoraDeviceTypes.NAV12: CommandID.NAV12_ENABLE_SERVICE_MODE,
-        },
-        command_off={
-            None: CommandID.DISABLE_SERVICE_MODE,
-            PandoraDeviceTypes.NAV12: CommandID.DISABLE_SERVICE_MODE,
-        },
+        command_on=[
+            (
+                has_device_type(PandoraDeviceTypes.NAV12),
+                CommandID.NAV12_ENABLE_SERVICE_MODE,
+            ),
+            (None, CommandID.ENABLE_SERVICE_MODE),
+        ],
+        command_off=[
+            (has_device_type(PandoraDeviceTypes.NAV12), CommandID.DISABLE_SERVICE_MODE),
+            (None, CommandID.DISABLE_SERVICE_MODE),
+        ],
     ),
     PandoraCASSwitchEntityDescription(
         key="ext_channel",
@@ -120,14 +130,20 @@ class PandoraCASSwitchEntityDescription(
         icon="mdi:led-off",
         # icon_turning_on="",
         # icon_turning_off="",
-        command_on={
-            None: CommandID.ENABLE_STATUS_OUTPUT,
-            PandoraDeviceTypes.NAV12: CommandID.NAV12_ENABLE_STATUS_OUTPUT,
-        },
-        command_off={
-            None: CommandID.DISABLE_STATUS_OUTPUT,
-            PandoraDeviceTypes.NAV12: CommandID.NAV12_DISABLE_STATUS_OUTPUT,
-        },
+        command_on=[
+            (
+                has_device_type(PandoraDeviceTypes.NAV12),
+                CommandID.NAV12_ENABLE_STATUS_OUTPUT,
+            ),
+            (None, CommandID.ENABLE_STATUS_OUTPUT),
+        ],
+        command_off=[
+            (
+                has_device_type(PandoraDeviceTypes.NAV12),
+                CommandID.NAV12_DISABLE_STATUS_OUTPUT,
+            ),
+            (None, CommandID.DISABLE_STATUS_OUTPUT),
+        ],
         assumed_state=True,
     ),
     PandoraCASSwitchEntityDescription(
