From 9e6c6faf67a96f5c932b9f6c12b97d70d5cf36df Mon Sep 17 00:00:00 2001 From: epenet <6771947+epenet@users.noreply.github.com> Date: Thu, 4 Dec 2025 13:47:01 +0000 Subject: [PATCH 1/3] Move Tuya type information classes to separate module --- homeassistant/components/tuya/light.py | 20 +- homeassistant/components/tuya/models.py | 234 +----------------- homeassistant/components/tuya/number.py | 5 +- homeassistant/components/tuya/sensor.py | 4 +- .../components/tuya/type_information.py | 225 +++++++++++++++++ 5 files changed, 253 insertions(+), 235 deletions(-) create mode 100644 homeassistant/components/tuya/type_information.py diff --git a/homeassistant/components/tuya/light.py b/homeassistant/components/tuya/light.py index 7678b831d90729..bcbc467c2e92e8 100644 --- a/homeassistant/components/tuya/light.py +++ b/homeassistant/components/tuya/light.py @@ -35,8 +35,8 @@ DPCodeEnumWrapper, DPCodeIntegerWrapper, DPCodeJsonWrapper, - IntegerTypeData, ) +from .type_information import IntegerTypeInformation from .util import remap_value @@ -138,24 +138,24 @@ def _convert_value_to_raw_value(self, device: CustomerDevice, value: Any) -> Any ) -DEFAULT_H_TYPE = IntegerTypeData( +DEFAULT_H_TYPE = IntegerTypeInformation( dpcode=DPCode.COLOUR_DATA_HSV, min=1, scale=0, max=360, step=1 ) -DEFAULT_S_TYPE = IntegerTypeData( +DEFAULT_S_TYPE = IntegerTypeInformation( dpcode=DPCode.COLOUR_DATA_HSV, min=1, scale=0, max=255, step=1 ) -DEFAULT_V_TYPE = IntegerTypeData( +DEFAULT_V_TYPE = IntegerTypeInformation( dpcode=DPCode.COLOUR_DATA_HSV, min=1, scale=0, max=255, step=1 ) -DEFAULT_H_TYPE_V2 = IntegerTypeData( +DEFAULT_H_TYPE_V2 = IntegerTypeInformation( dpcode=DPCode.COLOUR_DATA_HSV, min=1, scale=0, max=360, step=1 ) -DEFAULT_S_TYPE_V2 = IntegerTypeData( +DEFAULT_S_TYPE_V2 = IntegerTypeInformation( dpcode=DPCode.COLOUR_DATA_HSV, min=1, scale=0, max=1000, step=1 ) -DEFAULT_V_TYPE_V2 = IntegerTypeData( +DEFAULT_V_TYPE_V2 = IntegerTypeInformation( dpcode=DPCode.COLOUR_DATA_HSV, min=1, scale=0, max=1000, step=1 ) @@ -578,15 +578,15 @@ def _get_color_data_wrapper( if function_data := json_loads_object( cast(str, color_data_wrapper.type_information.type_data) ): - color_data_wrapper.h_type = IntegerTypeData( + color_data_wrapper.h_type = IntegerTypeInformation( dpcode=color_data_wrapper.dpcode, **cast(dict, function_data["h"]), ) - color_data_wrapper.s_type = IntegerTypeData( + color_data_wrapper.s_type = IntegerTypeInformation( dpcode=color_data_wrapper.dpcode, **cast(dict, function_data["s"]), ) - color_data_wrapper.v_type = IntegerTypeData( + color_data_wrapper.v_type = IntegerTypeInformation( dpcode=color_data_wrapper.dpcode, **cast(dict, function_data["v"]), ) diff --git a/homeassistant/components/tuya/models.py b/homeassistant/components/tuya/models.py index 975d86a9d57187..f157a52c1f4701 100644 --- a/homeassistant/components/tuya/models.py +++ b/homeassistant/components/tuya/models.py @@ -3,15 +3,19 @@ from __future__ import annotations import base64 -from dataclasses import dataclass -from typing import Any, Literal, Self, cast, overload +from typing import Any, Self from tuya_sharing import CustomerDevice -from homeassistant.util.json import json_loads, json_loads_object +from homeassistant.util.json import json_loads from .const import LOGGER, DPType -from .util import parse_dptype, remap_value +from .type_information import ( + EnumTypeInformation, + IntegerTypeInformation, + TypeInformation, + find_dpcode, +) # Dictionary to track logged warnings to avoid spamming logs # Keyed by device ID @@ -32,139 +36,6 @@ def _should_log_warning(device_id: str, warning_key: str) -> bool: return True -@dataclass(kw_only=True) -class TypeInformation: - """Type information. - - As provided by the SDK, from `device.function` / `device.status_range`. - """ - - dpcode: str - type_data: str | None = None - - @classmethod - def from_json(cls, dpcode: str, type_data: str) -> Self | None: - """Load JSON string and return a TypeInformation object.""" - return cls(dpcode=dpcode, type_data=type_data) - - -@dataclass(kw_only=True) -class IntegerTypeData(TypeInformation): - """Integer Type Data.""" - - min: int - max: int - scale: int - 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) - - def scale_value_back(self, value: float) -> int: - """Return raw value for scaled.""" - return round(value * (10**self.scale)) - - def remap_value_to( - self, - value: float, - to_min: float = 0, - to_max: float = 255, - reverse: bool = False, - ) -> float: - """Remap a value from this range to a new range.""" - return remap_value(value, self.min, self.max, to_min, to_max, reverse) - - def remap_value_from( - self, - value: float, - from_min: float = 0, - from_max: float = 255, - reverse: bool = False, - ) -> float: - """Remap a value from its current range to this range.""" - return remap_value(value, from_min, from_max, self.min, self.max, reverse) - - @classmethod - def from_json(cls, dpcode: str, type_data: str) -> Self | None: - """Load JSON string and return a IntegerTypeData object.""" - if not (parsed := cast(dict[str, Any] | None, json_loads_object(type_data))): - return None - - return cls( - dpcode=dpcode, - type_data=type_data, - min=int(parsed["min"]), - max=int(parsed["max"]), - scale=int(parsed["scale"]), - step=int(parsed["step"]), - unit=parsed.get("unit"), - ) - - -@dataclass(kw_only=True) -class BitmapTypeInformation(TypeInformation): - """Bitmap type information.""" - - label: list[str] - - @classmethod - def from_json(cls, dpcode: str, type_data: str) -> Self | None: - """Load JSON string and return a BitmapTypeInformation object.""" - if not (parsed := json_loads_object(type_data)): - return None - return cls( - dpcode=dpcode, - type_data=type_data, - **cast(dict[str, list[str]], parsed), - ) - - -@dataclass(kw_only=True) -class EnumTypeData(TypeInformation): - """Enum Type Data.""" - - range: list[str] - - @classmethod - def from_json(cls, dpcode: str, type_data: str) -> Self | None: - """Load JSON string and return a EnumTypeData object.""" - if not (parsed := json_loads_object(type_data)): - return None - return cls( - dpcode=dpcode, - type_data=type_data, - **cast(dict[str, list[str]], parsed), - ) - - -_TYPE_INFORMATION_MAPPINGS: dict[DPType, type[TypeInformation]] = { - DPType.BITMAP: BitmapTypeInformation, - DPType.BOOLEAN: TypeInformation, - DPType.ENUM: EnumTypeData, - DPType.INTEGER: IntegerTypeData, - DPType.JSON: TypeInformation, - DPType.RAW: TypeInformation, - DPType.STRING: TypeInformation, -} - - class DeviceWrapper: """Base device wrapper.""" @@ -303,8 +174,8 @@ def read_json(self, device: CustomerDevice) -> Any | None: return json_loads(raw_value) -class DPCodeEnumWrapper(DPCodeTypeInformationWrapper[EnumTypeData]): - """Simple wrapper for EnumTypeData values.""" +class DPCodeEnumWrapper(DPCodeTypeInformationWrapper[EnumTypeInformation]): + """Simple wrapper for EnumTypeInformation values.""" DPTYPE = DPType.ENUM @@ -342,12 +213,12 @@ def _convert_value_to_raw_value(self, device: CustomerDevice, value: Any) -> Any ) -class DPCodeIntegerWrapper(DPCodeTypeInformationWrapper[IntegerTypeData]): - """Simple wrapper for IntegerTypeData values.""" +class DPCodeIntegerWrapper(DPCodeTypeInformationWrapper[IntegerTypeInformation]): + """Simple wrapper for IntegerTypeInformation values.""" DPTYPE = DPType.INTEGER - def __init__(self, dpcode: str, type_information: IntegerTypeData) -> None: + def __init__(self, dpcode: str, type_information: IntegerTypeInformation) -> None: """Init DPCodeIntegerWrapper.""" super().__init__(dpcode, type_information) self.native_unit = type_information.unit @@ -414,82 +285,3 @@ def find_dpcode( type_information.dpcode, type_information.label.index(bitmap_key) ) return None - - -@overload -def find_dpcode( - device: CustomerDevice, - dpcodes: str | tuple[str, ...] | None, - *, - prefer_function: bool = False, - dptype: Literal[DPType.BITMAP], -) -> BitmapTypeInformation | None: ... - - -@overload -def find_dpcode( - device: CustomerDevice, - dpcodes: str | tuple[str, ...] | None, - *, - prefer_function: bool = False, - dptype: Literal[DPType.ENUM], -) -> EnumTypeData | None: ... - - -@overload -def find_dpcode( - device: CustomerDevice, - dpcodes: str | tuple[str, ...] | None, - *, - prefer_function: bool = False, - dptype: Literal[DPType.INTEGER], -) -> IntegerTypeData | None: ... - - -@overload -def find_dpcode( - device: CustomerDevice, - dpcodes: str | tuple[str, ...] | None, - *, - prefer_function: bool = False, - dptype: Literal[DPType.BOOLEAN, DPType.JSON, DPType.RAW], -) -> TypeInformation | None: ... - - -def find_dpcode( - device: CustomerDevice, - dpcodes: str | tuple[str, ...] | None, - *, - prefer_function: bool = False, - dptype: DPType, -) -> TypeInformation | None: - """Find type information for a matching DP code available for this device.""" - if not (type_information_cls := _TYPE_INFORMATION_MAPPINGS.get(dptype)): - raise NotImplementedError(f"find_dpcode not supported for {dptype}") - - if dpcodes is None: - return None - - if not isinstance(dpcodes, tuple): - dpcodes = (dpcodes,) - - lookup_tuple = ( - (device.function, device.status_range) - if prefer_function - else (device.status_range, device.function) - ) - - for dpcode in dpcodes: - for device_specs in lookup_tuple: - if ( - (current_definition := device_specs.get(dpcode)) - and parse_dptype(current_definition.type) is dptype - and ( - type_information := type_information_cls.from_json( - dpcode=dpcode, type_data=current_definition.values - ) - ) - ): - return type_information - - return None diff --git a/homeassistant/components/tuya/number.py b/homeassistant/components/tuya/number.py index 0be3770048c98c..2fb3b6cdfa1310 100644 --- a/homeassistant/components/tuya/number.py +++ b/homeassistant/components/tuya/number.py @@ -25,7 +25,8 @@ DPCode, ) from .entity import TuyaEntity -from .models import DPCodeIntegerWrapper, IntegerTypeData +from .models import DPCodeIntegerWrapper +from .type_information import IntegerTypeInformation NUMBERS: dict[DeviceCategory, tuple[NumberEntityDescription, ...]] = { DeviceCategory.BH: ( @@ -483,7 +484,7 @@ def async_discover_device(device_ids: list[str]) -> None: class TuyaNumberEntity(TuyaEntity, NumberEntity): """Tuya Number Entity.""" - _number: IntegerTypeData | None = None + _number: IntegerTypeInformation | None = None def __init__( self, diff --git a/homeassistant/components/tuya/sensor.py b/homeassistant/components/tuya/sensor.py index 0a50a0a5063494..5ec76ff5a8ef14 100644 --- a/homeassistant/components/tuya/sensor.py +++ b/homeassistant/components/tuya/sensor.py @@ -46,12 +46,12 @@ DPCodeJsonWrapper, DPCodeTypeInformationWrapper, DPCodeWrapper, - EnumTypeData, ) from .raw_data_models import ElectricityData +from .type_information import EnumTypeInformation -class _WindDirectionWrapper(DPCodeTypeInformationWrapper[EnumTypeData]): +class _WindDirectionWrapper(DPCodeTypeInformationWrapper[EnumTypeInformation]): """Custom DPCode Wrapper for converting enum to wind direction.""" DPTYPE = DPType.ENUM diff --git a/homeassistant/components/tuya/type_information.py b/homeassistant/components/tuya/type_information.py new file mode 100644 index 00000000000000..897b37935beebc --- /dev/null +++ b/homeassistant/components/tuya/type_information.py @@ -0,0 +1,225 @@ +"""Tuya Home Assistant type information classes.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any, Literal, Self, cast, overload + +from tuya_sharing import CustomerDevice + +from homeassistant.util.json import json_loads_object + +from .const import DPType +from .util import parse_dptype, remap_value + + +@dataclass(kw_only=True) +class TypeInformation: + """Type information. + + As provided by the SDK, from `device.function` / `device.status_range`. + """ + + dpcode: str + type_data: str | None = None + + @classmethod + def from_json(cls, dpcode: str, type_data: str) -> Self | None: + """Load JSON string and return a TypeInformation object.""" + return cls(dpcode=dpcode, type_data=type_data) + + +@dataclass(kw_only=True) +class BitmapTypeInformation(TypeInformation): + """Bitmap type information.""" + + label: list[str] + + @classmethod + def from_json(cls, dpcode: str, type_data: str) -> Self | None: + """Load JSON string and return a BitmapTypeInformation object.""" + if not (parsed := json_loads_object(type_data)): + return None + return cls( + dpcode=dpcode, + type_data=type_data, + **cast(dict[str, list[str]], parsed), + ) + + +@dataclass(kw_only=True) +class EnumTypeInformation(TypeInformation): + """Enum type information.""" + + range: list[str] + + @classmethod + def from_json(cls, dpcode: str, type_data: str) -> Self | None: + """Load JSON string and return a EnumTypeInformation object.""" + if not (parsed := json_loads_object(type_data)): + return None + return cls( + dpcode=dpcode, + type_data=type_data, + **cast(dict[str, list[str]], parsed), + ) + + +@dataclass(kw_only=True) +class IntegerTypeInformation(TypeInformation): + """Integer type information.""" + + min: int + max: int + scale: int + 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) + + def scale_value_back(self, value: float) -> int: + """Return raw value for scaled.""" + return round(value * (10**self.scale)) + + def remap_value_to( + self, + value: float, + to_min: float = 0, + to_max: float = 255, + reverse: bool = False, + ) -> float: + """Remap a value from this range to a new range.""" + return remap_value(value, self.min, self.max, to_min, to_max, reverse) + + def remap_value_from( + self, + value: float, + from_min: float = 0, + from_max: float = 255, + reverse: bool = False, + ) -> float: + """Remap a value from its current range to this range.""" + return remap_value(value, from_min, from_max, self.min, self.max, reverse) + + @classmethod + def from_json(cls, dpcode: str, type_data: str) -> Self | None: + """Load JSON string and return a IntegerTypeInformation object.""" + if not (parsed := cast(dict[str, Any] | None, json_loads_object(type_data))): + return None + + return cls( + dpcode=dpcode, + type_data=type_data, + min=int(parsed["min"]), + max=int(parsed["max"]), + scale=int(parsed["scale"]), + step=int(parsed["step"]), + unit=parsed.get("unit"), + ) + + +_TYPE_INFORMATION_MAPPINGS: dict[DPType, type[TypeInformation]] = { + DPType.BITMAP: BitmapTypeInformation, + DPType.BOOLEAN: TypeInformation, + DPType.ENUM: EnumTypeInformation, + DPType.INTEGER: IntegerTypeInformation, + DPType.JSON: TypeInformation, + DPType.RAW: TypeInformation, + DPType.STRING: TypeInformation, +} + + +@overload +def find_dpcode( + device: CustomerDevice, + dpcodes: str | tuple[str, ...] | None, + *, + prefer_function: bool = False, + dptype: Literal[DPType.BITMAP], +) -> BitmapTypeInformation | None: ... + + +@overload +def find_dpcode( + device: CustomerDevice, + dpcodes: str | tuple[str, ...] | None, + *, + prefer_function: bool = False, + dptype: Literal[DPType.ENUM], +) -> EnumTypeInformation | None: ... + + +@overload +def find_dpcode( + device: CustomerDevice, + dpcodes: str | tuple[str, ...] | None, + *, + prefer_function: bool = False, + dptype: Literal[DPType.INTEGER], +) -> IntegerTypeInformation | None: ... + + +@overload +def find_dpcode( + device: CustomerDevice, + dpcodes: str | tuple[str, ...] | None, + *, + prefer_function: bool = False, + dptype: Literal[DPType.BOOLEAN, DPType.JSON, DPType.RAW], +) -> TypeInformation | None: ... + + +def find_dpcode( + device: CustomerDevice, + dpcodes: str | tuple[str, ...] | None, + *, + prefer_function: bool = False, + dptype: DPType, +) -> TypeInformation | None: + """Find type information for a matching DP code available for this device.""" + if not (type_information_cls := _TYPE_INFORMATION_MAPPINGS.get(dptype)): + raise NotImplementedError(f"find_dpcode not supported for {dptype}") + + if dpcodes is None: + return None + + if not isinstance(dpcodes, tuple): + dpcodes = (dpcodes,) + + lookup_tuple = ( + (device.function, device.status_range) + if prefer_function + else (device.status_range, device.function) + ) + + for dpcode in dpcodes: + for device_specs in lookup_tuple: + if ( + (current_definition := device_specs.get(dpcode)) + and parse_dptype(current_definition.type) is dptype + and ( + type_information := type_information_cls.from_json( + dpcode=dpcode, type_data=current_definition.values + ) + ) + ): + return type_information + + return None From aa954fe36267a10b460e76097f14e4b5d6216ee3 Mon Sep 17 00:00:00 2001 From: epenet <6771947+epenet@users.noreply.github.com> Date: Thu, 4 Dec 2025 15:13:52 +0100 Subject: [PATCH 2/3] docstring Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> --- homeassistant/components/tuya/type_information.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/homeassistant/components/tuya/type_information.py b/homeassistant/components/tuya/type_information.py index 897b37935beebc..0e2af878c2de7d 100644 --- a/homeassistant/components/tuya/type_information.py +++ b/homeassistant/components/tuya/type_information.py @@ -1,4 +1,4 @@ -"""Tuya Home Assistant type information classes.""" +"""Type information classes for the Tuya integration.""" from __future__ import annotations @@ -55,7 +55,7 @@ class EnumTypeInformation(TypeInformation): @classmethod def from_json(cls, dpcode: str, type_data: str) -> Self | None: - """Load JSON string and return a EnumTypeInformation object.""" + """Load JSON string and return an EnumTypeInformation object.""" if not (parsed := json_loads_object(type_data)): return None return cls( @@ -120,7 +120,7 @@ def remap_value_from( @classmethod def from_json(cls, dpcode: str, type_data: str) -> Self | None: - """Load JSON string and return a IntegerTypeInformation object.""" + """Load JSON string and return an IntegerTypeInformation object.""" if not (parsed := cast(dict[str, Any] | None, json_loads_object(type_data))): return None From 0748f26fa6271db6cf30a2258d61432b63019b88 Mon Sep 17 00:00:00 2001 From: epenet <6771947+epenet@users.noreply.github.com> Date: Thu, 4 Dec 2025 14:15:27 +0000 Subject: [PATCH 3/3] Cleanup --- homeassistant/components/tuya/number.py | 3 --- 1 file changed, 3 deletions(-) diff --git a/homeassistant/components/tuya/number.py b/homeassistant/components/tuya/number.py index 2fb3b6cdfa1310..c5bdbd0f466e0d 100644 --- a/homeassistant/components/tuya/number.py +++ b/homeassistant/components/tuya/number.py @@ -26,7 +26,6 @@ ) from .entity import TuyaEntity from .models import DPCodeIntegerWrapper -from .type_information import IntegerTypeInformation NUMBERS: dict[DeviceCategory, tuple[NumberEntityDescription, ...]] = { DeviceCategory.BH: ( @@ -484,8 +483,6 @@ def async_discover_device(device_ids: list[str]) -> None: class TuyaNumberEntity(TuyaEntity, NumberEntity): """Tuya Number Entity.""" - _number: IntegerTypeInformation | None = None - def __init__( self, device: CustomerDevice,