This commit is contained in:
epenet
2025-12-11 11:59:58 +00:00
parent 14e5e416ed
commit f70bc20582
6 changed files with 162 additions and 200 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, 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
@@ -64,7 +93,7 @@ class DPCodeWrapper(DeviceWrapper):
return [
{
"code": self.dpcode,
"value": self._convert_value_to_raw_value(device, value),
"value": self._convert_value_to_raw_value(value),
}
]
@@ -82,13 +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:
def _convert_value_to_raw_value(self, value: Any) -> Any:
"""Convert a Home Assistant value back to a raw device value."""
return self.type_information.process_value_back(value)
return value
@classmethod
def find_dpcode(
@@ -113,12 +140,72 @@ class DPCodeBooleanWrapper(DPCodeTypeInformationWrapper[BooleanTypeInformation])
_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, value: bool) -> Any | None:
"""Convert a Home Assistant value back to a raw device value."""
if value in (True, False):
return value
# Currently only called with boolean values
# Safety net in case of future changes
raise ValueError(f"Invalid boolean value `{value}`")
class DPCodeEnumWrapper(DPCodeTypeInformationWrapper[EnumTypeInformation]):
"""Simple wrapper for EnumTypeInformation values."""
_DPTYPE = EnumTypeInformation
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, value: str) -> Any | None:
"""Convert a Home Assistant value back to a raw device value."""
if value in self.type_information.range:
return value
# Guarded by select option validation
# Safety net in case of future changes
raise ValueError(
f"Enum value `{value}` out of range: {self.type_information.range}"
)
class DPCodeIntegerWrapper(DPCodeTypeInformationWrapper[IntegerTypeInformation]):
"""Simple wrapper for IntegerTypeInformation values."""
@@ -129,6 +216,46 @@ 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 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, value: float) -> int:
"""Convert a Home Assistant value back to a raw device value."""
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
# Safety net in case of future changes
raise ValueError(
f"Value `{new_value}` (converted from `{value}`) out of range:"
f" ({self.type_information.min}-{self.type_information.max})"
)
class DPCodeJsonWrapper(DPCodeTypeInformationWrapper[JsonTypeInformation]):
@@ -136,12 +263,24 @@ class DPCodeJsonWrapper(DPCodeTypeInformationWrapper[JsonTypeInformation]):
_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]):
"""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]):
"""Simple wrapper for StringTypeInformation values."""

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,24 +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
def process_value_back(self, value: T) -> Any:
"""Convert display value back to a raw device value.
Base implementation does no validation, subclasses may override to provide
specific validation.
"""
return value
@classmethod
def _from_json(cls, dpcode: str, type_data: str) -> Self | None:
"""Load JSON string and return a TypeInformation object."""
@@ -129,37 +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
def process_value_back(self, value: bool) -> Any | None:
"""Convert a Home Assistant value back to a raw device value."""
if value in (True, False):
return value
# Currently only called with boolean values
# Safety net in case of future changes
raise ValueError(f"Invalid boolean value `{value}`")
@dataclass(kw_only=True)
class EnumTypeInformation(TypeInformation[str]):
@@ -169,37 +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
def process_value_back(self, value: str) -> Any | None:
"""Convert a Home Assistant value back to a raw device value."""
if value in self.range:
return value
# Guarded by select option validation
# Safety net in case of future changes
raise ValueError(f"Enum value `{value}` out of range: {self.range}")
@classmethod
def _from_json(cls, dpcode: str, type_data: str) -> Self | None:
"""Load JSON string and return an EnumTypeInformation object."""
@@ -224,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)
@@ -267,43 +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 self.scale_value(raw_value)
def process_value_back(self, value: float) -> int:
"""Convert a Home Assistant value back to a raw device value."""
new_value = self.scale_value_back(value)
if self.min <= new_value <= self.max:
return new_value
# Guarded by number validation
# Safety net in case of future changes
raise ValueError(
f"Value `{new_value}` (converted from `{value}`) out of range:"
f" ({self.min}-{self.max})"
)
@classmethod
def _from_json(cls, dpcode: str, type_data: str) -> Self | None:
"""Load JSON string and return an IntegerTypeInformation object."""
@@ -327,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]):
@@ -342,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]):