mirror of
https://github.com/home-assistant/core.git
synced 2026-04-06 23:47:33 +00:00
Compare commits
7 Commits
hassio-spl
...
epenet-202
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
5200d66299 | ||
|
|
f70bc20582 | ||
|
|
14e5e416ed | ||
|
|
3004b49a4f | ||
|
|
b1e410a211 | ||
|
|
8d79b9aa8b | ||
|
|
aa7e113aa5 |
@@ -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:
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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]):
|
||||
|
||||
Reference in New Issue
Block a user