Source code for nwp500.device_capabilities
"""Device capability checking for MQTT commands.
This module provides a generalized framework for checking device capabilities
before executing MQTT commands. It uses a mapping-based approach to validate
that a device supports specific controllable features without requiring
individual checker functions.
"""
from __future__ import annotations
from collections.abc import Callable
from typing import TYPE_CHECKING
from .exceptions import DeviceCapabilityError
if TYPE_CHECKING:
from .models import DeviceFeature
__author__ = "Emmanuel Levijarvi"
# Type for capability check functions
CapabilityCheckFn = Callable[["DeviceFeature"], bool]
[docs]
class MqttDeviceCapabilityChecker:
"""Generalized MQTT device capability checker using a capability map.
This class uses a mapping of controllable feature names to their check
functions, allowing capabilities to be validated in a centralized,
extensible way without requiring individual methods for each control.
"""
# Map of controllable features to their check functions
# Capability names MUST match DeviceFeature attribute names exactly
# for traceability: capability name -> DeviceFeature.{name}
_CAPABILITY_MAP: dict[str, CapabilityCheckFn] = {
"power_use": lambda f: bool(f.power_use),
"dhw_use": lambda f: bool(f.dhw_use),
"dhw_temperature_setting_use": lambda f: _check_dhw_temperature_control(
f
),
"holiday_use": lambda f: bool(f.holiday_use),
"program_reservation_use": lambda f: bool(f.program_reservation_use),
"recirculation_use": lambda f: bool(f.recirculation_use),
"recirc_reservation_use": lambda f: bool(f.recirc_reservation_use),
"anti_legionella_setting_use": lambda f: bool(
f.anti_legionella_setting_use
),
}
[docs]
@classmethod
def supports(cls, feature: str, device_features: DeviceFeature) -> bool:
"""Check if device supports control of a specific feature.
Args:
feature: Name of the controllable feature to check
device_features: Device feature information
Returns:
True if feature control is supported, False otherwise
Raises:
ValueError: If feature is not recognized
"""
if feature not in cls._CAPABILITY_MAP:
valid_features = ", ".join(sorted(cls._CAPABILITY_MAP.keys()))
raise ValueError(
f"Unknown controllable feature: {feature}. "
f"Valid features: {valid_features}"
)
return cls._CAPABILITY_MAP[feature](device_features)
[docs]
@classmethod
def assert_supported(
cls, feature: str, device_features: DeviceFeature
) -> None:
"""Assert that device supports control of a feature.
Args:
feature: Name of the controllable feature to check
device_features: Device feature information
Raises:
DeviceCapabilityError: If feature control is not supported
ValueError: If feature is not recognized
"""
if not cls.supports(feature, device_features):
raise DeviceCapabilityError(feature)
[docs]
@classmethod
def register_capability(
cls, name: str, check_fn: CapabilityCheckFn
) -> None:
"""Register a custom controllable feature check.
This allows extensions or applications to define custom capability
checks without modifying the core library.
Args:
name: Feature name
check_fn: Function that takes DeviceFeature and returns bool
"""
cls._CAPABILITY_MAP[name] = check_fn
[docs]
@classmethod
def get_available_controls(
cls, device_features: DeviceFeature
) -> dict[str, bool]:
"""Get all controllable features available on a device.
Args:
device_features: Device feature information
Returns:
Dictionary mapping feature names to whether they can be controlled
"""
return {
feature: cls.supports(feature, device_features)
for feature in cls._CAPABILITY_MAP
}
def _check_dhw_temperature_control(features: DeviceFeature) -> bool:
"""Check if device supports DHW temperature control.
Returns True if temperature control is enabled (not UNKNOWN or DISABLE).
"""
from .enums import DHWControlTypeFlag
return features.dhw_temperature_setting_use not in (
DHWControlTypeFlag.UNKNOWN,
DHWControlTypeFlag.DISABLE,
)