Compare commits

...

7 Commits

Author SHA1 Message Date
epenet
5200d66299 Adjust 2025-12-11 12:02:41 +00:00
epenet
f70bc20582 Adjust 2025-12-11 11:59:58 +00:00
epenet
14e5e416ed Simplify 2025-12-11 11:02:08 +00:00
epenet
3004b49a4f Simplify 2025-12-11 10:59:40 +00:00
epenet
b1e410a211 Drop usused scale_value_back 2025-12-11 10:28:33 +00:00
epenet
8d79b9aa8b Tweak 2025-12-11 10:26:40 +00:00
epenet
aa7e113aa5 Improve Tuya action validation 2025-12-11 10:09:56 +00:00
6 changed files with 148 additions and 177 deletions

View File

@@ -361,11 +361,9 @@ class TuyaClimateEntity(TuyaEntity, ClimateEntity):
# it to define min, max & step temperatures
if self._set_temperature:
self._attr_supported_features |= ClimateEntityFeature.TARGET_TEMPERATURE
self._attr_max_temp = self._set_temperature.type_information.max_scaled
self._attr_min_temp = self._set_temperature.type_information.min_scaled
self._attr_target_temperature_step = (
self._set_temperature.type_information.step_scaled
)
self._attr_max_temp = self._set_temperature.max
self._attr_min_temp = self._set_temperature.min
self._attr_target_temperature_step = self._set_temperature.step
# Determine HVAC modes
self._attr_hvac_modes: list[HVACMode] = []
@@ -394,12 +392,8 @@ class TuyaClimateEntity(TuyaEntity, ClimateEntity):
# Determine dpcode to use for setting the humidity
if target_humidity_wrapper:
self._attr_supported_features |= ClimateEntityFeature.TARGET_HUMIDITY
self._attr_min_humidity = round(
target_humidity_wrapper.type_information.min_scaled
)
self._attr_max_humidity = round(
target_humidity_wrapper.type_information.max_scaled
)
self._attr_min_humidity = round(target_humidity_wrapper.min)
self._attr_max_humidity = round(target_humidity_wrapper.max)
# Determine fan modes
if fan_mode_wrapper:

View File

@@ -14,7 +14,7 @@ from homeassistant.util import dt as dt_util
from . import TuyaConfigEntry
from .const import DOMAIN, DPCode
from .type_information import DEVICE_WARNINGS
from .models import DEVICE_WARNINGS
_REDACTED_DPCODES = {
DPCode.ALARM_MESSAGE,

View File

@@ -153,12 +153,8 @@ class TuyaHumidifierEntity(TuyaEntity, HumidifierEntity):
# Determine humidity parameters
if target_humidity_wrapper:
self._attr_min_humidity = round(
target_humidity_wrapper.type_information.min_scaled
)
self._attr_max_humidity = round(
target_humidity_wrapper.type_information.max_scaled
)
self._attr_min_humidity = round(target_humidity_wrapper.min)
self._attr_max_humidity = round(target_humidity_wrapper.max)
# Determine mode support and provided modes
if mode_wrapper:

View File

@@ -2,10 +2,13 @@
from __future__ import annotations
import base64
import json
from typing import Any, Self
from tuya_sharing import CustomerDevice
from .const import LOGGER
from .type_information import (
BitmapTypeInformation,
BooleanTypeInformation,
@@ -17,6 +20,24 @@ from .type_information import (
TypeInformation,
)
# Dictionary to track logged warnings to avoid spamming logs
# Keyed by device ID
DEVICE_WARNINGS: dict[str, set[str]] = {}
def _should_log_warning(device_id: str, warning_key: str) -> bool:
"""Check if a warning has already been logged for a device and add it if not.
Returns: True if the warning should be logged, False if it was already logged.
"""
if (device_warnings := DEVICE_WARNINGS.get(device_id)) is None:
device_warnings = set()
DEVICE_WARNINGS[device_id] = device_warnings
if warning_key in device_warnings:
return False
DEVICE_WARNINGS[device_id].add(warning_key)
return True
class DeviceWrapper:
"""Base device wrapper."""
@@ -46,13 +67,21 @@ class DPCodeWrapper(DeviceWrapper):
"""Init DPCodeWrapper."""
self.dpcode = dpcode
def _convert_value_to_raw_value(self, device: CustomerDevice, value: Any) -> Any:
"""Convert a Home Assistant value back to a raw device value.
def read_device_status(self, device: CustomerDevice) -> Any:
"""Read and process raw value against this type information.
This is called by `get_update_commands` to prepare the value for sending
back to the device, and should be implemented in concrete classes if needed.
Base implementation does no validation, subclasses may override to provide
specific validation.
"""
raise NotImplementedError
return device.status.get(self.dpcode)
def _convert_value_to_raw_value(self, device: CustomerDevice, value: Any) -> Any:
"""Convert display value back to a raw device value.
Base implementation does no validation, subclasses may override to provide
specific validation.
"""
return value
def get_update_commands(
self, device: CustomerDevice, value: Any
@@ -82,9 +111,11 @@ class DPCodeTypeInformationWrapper[T: TypeInformation](DPCodeWrapper):
def read_device_status(self, device: CustomerDevice) -> Any | None:
"""Read the device value for the dpcode."""
return self.type_information.process_raw_value(
device.status.get(self.dpcode), device
)
return device.status.get(self.dpcode)
def _convert_value_to_raw_value(self, device: CustomerDevice, value: Any) -> Any:
"""Convert a Home Assistant value back to a raw device value."""
return value
@classmethod
def find_dpcode(
@@ -105,15 +136,33 @@ class DPCodeTypeInformationWrapper[T: TypeInformation](DPCodeWrapper):
class DPCodeBooleanWrapper(DPCodeTypeInformationWrapper[BooleanTypeInformation]):
"""Simple wrapper for boolean values.
Supports True/False only.
"""
"""Simple wrapper for BooleanTypeInformation values."""
_DPTYPE = BooleanTypeInformation
def read_device_status(self, device: CustomerDevice) -> bool | None:
"""Read and process raw value against this type information."""
if (raw_value := device.status.get(self.dpcode)) is None:
return None
# Validate input against defined range
if raw_value not in (True, False):
if _should_log_warning(
device.id, f"boolean_out_range|{self.dpcode}|{raw_value}"
):
LOGGER.warning(
"Found invalid boolean value `%s` for datapoint `%s` in product "
"id `%s`, expected one of `%s`; please report this defect to "
"Tuya support",
raw_value,
self.dpcode,
device.product_id,
(True, False),
)
return None
return raw_value
def _convert_value_to_raw_value(
self, device: CustomerDevice, value: Any
self, device: CustomerDevice, value: bool
) -> Any | None:
"""Convert a Home Assistant value back to a raw device value."""
if value in (True, False):
@@ -123,18 +172,35 @@ class DPCodeBooleanWrapper(DPCodeTypeInformationWrapper[BooleanTypeInformation])
raise ValueError(f"Invalid boolean value `{value}`")
class DPCodeJsonWrapper(DPCodeTypeInformationWrapper[JsonTypeInformation]):
"""Wrapper to extract information from a JSON value."""
_DPTYPE = JsonTypeInformation
class DPCodeEnumWrapper(DPCodeTypeInformationWrapper[EnumTypeInformation]):
"""Simple wrapper for EnumTypeInformation values."""
_DPTYPE = EnumTypeInformation
def _convert_value_to_raw_value(self, device: CustomerDevice, value: Any) -> Any:
def read_device_status(self, device: CustomerDevice) -> str | None:
"""Read and process raw value against this type information."""
if (raw_value := device.status.get(self.dpcode)) is None:
return None
# Validate input against defined range
if raw_value not in self.type_information.range:
if _should_log_warning(
device.id, f"enum_out_range|{self.dpcode}|{raw_value}"
):
LOGGER.warning(
"Found invalid enum value `%s` for datapoint `%s` in product "
"id `%s`, expected one of `%s`; please report this defect to "
"Tuya support",
raw_value,
self.dpcode,
device.product_id,
self.type_information.range,
)
return None
return raw_value
def _convert_value_to_raw_value(
self, device: CustomerDevice, value: str
) -> Any | None:
"""Convert a Home Assistant value back to a raw device value."""
if value in self.type_information.range:
return value
@@ -154,10 +220,38 @@ class DPCodeIntegerWrapper(DPCodeTypeInformationWrapper[IntegerTypeInformation])
"""Init DPCodeIntegerWrapper."""
super().__init__(dpcode, type_information)
self.native_unit = type_information.unit
self.min = self.type_information.scale_value(type_information.min)
self.max = self.type_information.scale_value(type_information.max)
self.step = self.type_information.scale_value(type_information.step)
def _convert_value_to_raw_value(self, device: CustomerDevice, value: Any) -> Any:
def read_device_status(self, device: CustomerDevice) -> float | None:
"""Read and process raw value against this type information."""
if (raw_value := device.status.get(self.dpcode)) is None:
return None
# Validate input against defined range
if not isinstance(raw_value, int) or not (
self.type_information.min <= raw_value <= self.type_information.max
):
if _should_log_warning(
device.id, f"integer_out_range|{self.dpcode}|{raw_value}"
):
LOGGER.warning(
"Found invalid integer value `%s` for datapoint `%s` in product "
"id `%s`, expected integer value between %s and %s; please report "
"this defect to Tuya support",
raw_value,
self.dpcode,
device.product_id,
self.type_information.min,
self.type_information.max,
)
return None
return self.type_information.scale_value(raw_value)
def _convert_value_to_raw_value(self, device: CustomerDevice, value: float) -> int:
"""Convert a Home Assistant value back to a raw device value."""
new_value = round(value * (10**self.type_information.scale))
new_value = self.type_information.scale_value_back(value)
if self.type_information.min <= new_value <= self.type_information.max:
return new_value
# Guarded by number validation
@@ -168,14 +262,32 @@ class DPCodeIntegerWrapper(DPCodeTypeInformationWrapper[IntegerTypeInformation])
)
class DPCodeJsonWrapper(DPCodeTypeInformationWrapper[JsonTypeInformation]):
"""Simple wrapper for JsonTypeInformation values."""
_DPTYPE = JsonTypeInformation
def read_device_status(self, device: CustomerDevice) -> dict[str, Any] | None:
"""Read and process raw value against this type information."""
if (raw_value := device.status.get(self.dpcode)) is None:
return None
return json.loads(raw_value)
class DPCodeRawWrapper(DPCodeTypeInformationWrapper[RawTypeInformation]):
"""Wrapper to extract information from a RAW/binary value."""
"""Simple wrapper for RawTypeInformation values."""
_DPTYPE = RawTypeInformation
def read_device_status(self, device: CustomerDevice) -> bytes | None:
"""Read and process raw value against this type information."""
if (raw_value := device.status.get(self.dpcode)) is None:
return None
return base64.b64decode(raw_value)
class DPCodeStringWrapper(DPCodeTypeInformationWrapper[StringTypeInformation]):
"""Wrapper to extract information from a STRING value."""
"""Simple wrapper for StringTypeInformation values."""
_DPTYPE = StringTypeInformation

View File

@@ -496,9 +496,9 @@ class TuyaNumberEntity(TuyaEntity, NumberEntity):
self._attr_unique_id = f"{super().unique_id}{description.key}"
self._dpcode_wrapper = dpcode_wrapper
self._attr_native_max_value = dpcode_wrapper.type_information.max_scaled
self._attr_native_min_value = dpcode_wrapper.type_information.min_scaled
self._attr_native_step = dpcode_wrapper.type_information.step_scaled
self._attr_native_max_value = dpcode_wrapper.max
self._attr_native_min_value = dpcode_wrapper.min
self._attr_native_step = dpcode_wrapper.step
if description.native_unit_of_measurement is None:
self._attr_native_unit_of_measurement = dpcode_wrapper.native_unit

View File

@@ -2,7 +2,6 @@
from __future__ import annotations
import base64
from dataclasses import dataclass
from typing import Any, ClassVar, Self, cast
@@ -10,27 +9,9 @@ from tuya_sharing import CustomerDevice
from homeassistant.util.json import json_loads_object
from .const import LOGGER, DPType
from .const import DPType
from .util import parse_dptype, remap_value
# Dictionary to track logged warnings to avoid spamming logs
# Keyed by device ID
DEVICE_WARNINGS: dict[str, set[str]] = {}
def _should_log_warning(device_id: str, warning_key: str) -> bool:
"""Check if a warning has already been logged for a device and add it if not.
Returns: True if the warning should be logged, False if it was already logged.
"""
if (device_warnings := DEVICE_WARNINGS.get(device_id)) is None:
device_warnings = set()
DEVICE_WARNINGS[device_id] = device_warnings
if warning_key in device_warnings:
return False
DEVICE_WARNINGS[device_id].add(warning_key)
return True
@dataclass(kw_only=True)
class TypeInformation[T]:
@@ -43,16 +24,6 @@ class TypeInformation[T]:
dpcode: str
type_data: str | None = None
def process_raw_value(
self, raw_value: Any | None, device: CustomerDevice
) -> T | None:
"""Read and process raw value against this type information.
Base implementation does no validation, subclasses may override to provide
specific validation.
"""
return raw_value
@classmethod
def _from_json(cls, dpcode: str, type_data: str) -> Self | None:
"""Load JSON string and return a TypeInformation object."""
@@ -121,29 +92,6 @@ class BooleanTypeInformation(TypeInformation[bool]):
_DPTYPE = DPType.BOOLEAN
def process_raw_value(
self, raw_value: Any | None, device: CustomerDevice
) -> bool | None:
"""Read and process raw value against this type information."""
if raw_value is None:
return None
# Validate input against defined range
if raw_value not in (True, False):
if _should_log_warning(
device.id, f"boolean_out_range|{self.dpcode}|{raw_value}"
):
LOGGER.warning(
"Found invalid boolean value `%s` for datapoint `%s` in product "
"id `%s`, expected one of `%s`; please report this defect to "
"Tuya support",
raw_value,
self.dpcode,
device.product_id,
(True, False),
)
return None
return raw_value
@dataclass(kw_only=True)
class EnumTypeInformation(TypeInformation[str]):
@@ -153,29 +101,6 @@ class EnumTypeInformation(TypeInformation[str]):
range: list[str]
def process_raw_value(
self, raw_value: Any | None, device: CustomerDevice
) -> str | None:
"""Read and process raw value against this type information."""
if raw_value is None:
return None
# Validate input against defined range
if raw_value not in self.range:
if _should_log_warning(
device.id, f"enum_out_range|{self.dpcode}|{raw_value}"
):
LOGGER.warning(
"Found invalid enum value `%s` for datapoint `%s` in product "
"id `%s`, expected one of `%s`; please report this defect to "
"Tuya support",
raw_value,
self.dpcode,
device.product_id,
self.range,
)
return None
return raw_value
@classmethod
def _from_json(cls, dpcode: str, type_data: str) -> Self | None:
"""Load JSON string and return an EnumTypeInformation object."""
@@ -200,21 +125,6 @@ class IntegerTypeInformation(TypeInformation[float]):
step: int
unit: str | None = None
@property
def max_scaled(self) -> float:
"""Return the max scaled."""
return self.scale_value(self.max)
@property
def min_scaled(self) -> float:
"""Return the min scaled."""
return self.scale_value(self.min)
@property
def step_scaled(self) -> float:
"""Return the step scaled."""
return self.step / (10**self.scale)
def scale_value(self, value: int) -> float:
"""Scale a value."""
return value / (10**self.scale)
@@ -243,31 +153,6 @@ class IntegerTypeInformation(TypeInformation[float]):
"""Remap a value from its current range to this range."""
return remap_value(value, from_min, from_max, self.min, self.max, reverse)
def process_raw_value(
self, raw_value: Any | None, device: CustomerDevice
) -> float | None:
"""Read and process raw value against this type information."""
if raw_value is None:
return None
# Validate input against defined range
if not isinstance(raw_value, int) or not (self.min <= raw_value <= self.max):
if _should_log_warning(
device.id, f"integer_out_range|{self.dpcode}|{raw_value}"
):
LOGGER.warning(
"Found invalid integer value `%s` for datapoint `%s` in product "
"id `%s`, expected integer value between %s and %s; please report "
"this defect to Tuya support",
raw_value,
self.dpcode,
device.product_id,
self.min,
self.max,
)
return None
return raw_value / (10**self.scale)
@classmethod
def _from_json(cls, dpcode: str, type_data: str) -> Self | None:
"""Load JSON string and return an IntegerTypeInformation object."""
@@ -291,14 +176,6 @@ class JsonTypeInformation(TypeInformation[dict[str, Any]]):
_DPTYPE = DPType.JSON
def process_raw_value(
self, raw_value: Any | None, device: CustomerDevice
) -> dict[str, Any] | None:
"""Read and process raw value against this type information."""
if raw_value is None:
return None
return json_loads_object(raw_value)
@dataclass(kw_only=True)
class RawTypeInformation(TypeInformation[bytes]):
@@ -306,14 +183,6 @@ class RawTypeInformation(TypeInformation[bytes]):
_DPTYPE = DPType.RAW
def process_raw_value(
self, raw_value: Any | None, device: CustomerDevice
) -> bytes | None:
"""Read and process raw value against this type information."""
if raw_value is None:
return None
return base64.b64decode(raw_value)
@dataclass(kw_only=True)
class StringTypeInformation(TypeInformation[str]):