core/homeassistant/components/tuya/util.py

"""Utility methods for the Tuya integration."""

from __future__ import annotations

from dataclasses import dataclass
from typing import TYPE_CHECKING, Any

from tuya_sharing import CustomerDevice

from homeassistant.exceptions import ServiceValidationError

from .const import DOMAIN, DPCode, DPType

if TYPE_CHECKING:
    from .type_information import IntegerTypeInformation

_DPTYPE_MAPPING: dict[str, DPType] = {
    "bitmap": DPType.BITMAP,
    "bool": DPType.BOOLEAN,
    "enum": DPType.ENUM,
    "json": DPType.JSON,
    "raw": DPType.RAW,
    "string": DPType.STRING,
    "value": DPType.INTEGER,
}


def get_dpcode(
    device: CustomerDevice, dpcodes: str | tuple[str, ...] | None
) -> str | None:
    """Get the first matching DPCode from the device or return None."""
    if dpcodes is None:
        return None

    if not isinstance(dpcodes, tuple):
        dpcodes = (dpcodes,)

    for dpcode in dpcodes:
        if (
            dpcode in device.function
            or dpcode in device.status
            or dpcode in device.status_range
        ):
            return dpcode

    return None


def parse_dptype(dptype: str) -> DPType | None:
    """Parse DPType from device DPCode information."""
    try:
        return DPType(dptype)
    except ValueError:
        # Sometimes, we get ill-formed DPTypes from the cloud,
        # this fixes them and maps them to the correct DPType.
        return _DPTYPE_MAPPING.get(dptype)


@dataclass(kw_only=True)
class RemapHelper:
    """Helper class for remapping values."""

    source_min: int
    source_max: int
    target_min: int
    target_max: int

    @classmethod
    def from_type_information(
        cls,
        type_information: IntegerTypeInformation,
        target_min: int,
        target_max: int,
    ) -> RemapHelper:
        """Create RemapHelper from IntegerTypeInformation."""
        return cls(
            source_min=type_information.min,
            source_max=type_information.max,
            target_min=target_min,
            target_max=target_max,
        )

    @classmethod
    def from_function_data(
        cls, function_data: dict[str, Any], target_min: int, target_max: int
    ) -> RemapHelper:
        """Create RemapHelper from function_data."""
        return cls(
            source_min=function_data["min"],
            source_max=function_data["max"],
            target_min=target_min,
            target_max=target_max,
        )

    def remap_value_to(self, value: float, *, reverse: bool = False) -> float:
        """Remap a value from this range to a new range."""
        return self.remap_value(
            value,
            self.source_min,
            self.source_max,
            self.target_min,
            self.target_max,
            reverse=reverse,
        )

    def remap_value_from(self, value: float, *, reverse: bool = False) -> float:
        """Remap a value from its current range to this range."""
        return self.remap_value(
            value,
            self.target_min,
            self.target_max,
            self.source_min,
            self.source_max,
            reverse=reverse,
        )

    @staticmethod
    def remap_value(
        value: float,
        from_min: float,
        from_max: float,
        to_min: float,
        to_max: float,
        *,
        reverse: bool = False,
    ) -> float:
        """Remap a value from its current range, to a new range."""
        if reverse:
            value = from_max - value + from_min
        return ((value - from_min) / (from_max - from_min)) * (to_max - to_min) + to_min


class ActionDPCodeNotFoundError(ServiceValidationError):
    """Custom exception for action DP code not found errors."""

    def __init__(
        self, device: CustomerDevice, expected: str | tuple[str, ...] | None
    ) -> None:
        """Initialize the error with device and expected DP codes."""
        if expected is None:
            expected = ()  # empty tuple for no expected codes
        elif not isinstance(expected, tuple):
            expected = (expected,)

        super().__init__(
            translation_domain=DOMAIN,
            translation_key="action_dpcode_not_found",
            translation_placeholders={
                "expected": str(
                    sorted(
                        [dp.value if isinstance(dp, DPCode) else dp for dp in expected]
                    )
                ),
                "available": str(sorted(device.function.keys())),
            },
        )