Compare commits

...

5 Commits

Author SHA1 Message Date
epenet
bc1bac9bc8 Adjust 2026-03-13 15:02:45 +00:00
epenet
97979864ab Use DPCodeInvertedPercentageWrapper 2026-03-13 15:00:15 +00:00
epenet
1eb0c91041 Use DPCodeNonZeroPercentageWrapper 2026-03-13 15:00:15 +00:00
epenet
c7b147464d Use DPCodeInSetWrapper 2026-03-13 15:00:14 +00:00
epenet
f5259ca883 Use DPCodeRoundedIntegerWrapper 2026-03-13 15:00:14 +00:00
5 changed files with 28 additions and 163 deletions

View File

@@ -5,11 +5,11 @@ from __future__ import annotations
from dataclasses import dataclass
from tuya_device_handlers.device_wrapper.base import DeviceWrapper
from tuya_device_handlers.device_wrapper.binary_sensor import DPCodeBitmapBitWrapper
from tuya_device_handlers.device_wrapper.common import (
DPCodeBooleanWrapper,
DPCodeWrapper,
from tuya_device_handlers.device_wrapper.binary_sensor import (
DPCodeBitmapBitWrapper,
DPCodeInSetWrapper,
)
from tuya_device_handlers.device_wrapper.common import DPCodeBooleanWrapper
from tuya_sharing import CustomerDevice, Manager
from homeassistant.components.binary_sensor import (
@@ -376,29 +376,10 @@ BINARY_SENSORS: dict[DeviceCategory, tuple[TuyaBinarySensorEntityDescription, ..
}
class _CustomDPCodeWrapper(DPCodeWrapper[bool]):
"""Custom DPCode Wrapper to check for values in a set."""
_valid_values: set[bool | float | int | str]
def __init__(
self, dpcode: str, valid_values: set[bool | float | int | str]
) -> None:
"""Init CustomDPCodeBooleanWrapper."""
super().__init__(dpcode)
self._valid_values = valid_values
def read_device_status(self, device: CustomerDevice) -> bool | None:
"""Read the device value for the dpcode."""
if (raw_value := device.status.get(self.dpcode)) is None:
return None
return raw_value in self._valid_values
def _get_dpcode_wrapper(
device: CustomerDevice,
description: TuyaBinarySensorEntityDescription,
) -> DPCodeWrapper | None:
) -> DeviceWrapper[bool] | None:
"""Get DPCode wrapper for an entity description."""
dpcode = description.dpcode or description.key
if description.bitmap_key is not None:
@@ -412,7 +393,7 @@ def _get_dpcode_wrapper(
# Legacy / compatibility
if dpcode not in device.status:
return None
return _CustomDPCodeWrapper(
return DPCodeInSetWrapper(
dpcode,
description.on_value
if isinstance(description.on_value, set)

View File

@@ -12,6 +12,7 @@ from tuya_device_handlers.device_wrapper.common import (
DPCodeEnumWrapper,
DPCodeIntegerWrapper,
)
from tuya_device_handlers.device_wrapper.extended import DPCodeRoundedIntegerWrapper
from tuya_device_handlers.type_information import EnumTypeInformation
from tuya_sharing import CustomerDevice, Manager
@@ -54,16 +55,6 @@ TUYA_HVAC_TO_HA = {
}
class _RoundedIntegerWrapper(DPCodeIntegerWrapper[int]):
"""An integer that always rounds its value."""
def read_device_status(self, device: CustomerDevice) -> int | None:
"""Read and round the device status."""
if (value := self._read_dpcode_value(device)) is None:
return None
return round(value)
@dataclass(kw_only=True)
class _SwingModeWrapper(DeviceWrapper[str]):
"""Wrapper for managing climate swing mode operations across multiple DPCodes."""
@@ -358,7 +349,7 @@ async def async_setup_entry(
device,
manager,
CLIMATE_DESCRIPTIONS[device.category],
current_humidity_wrapper=_RoundedIntegerWrapper.find_dpcode(
current_humidity_wrapper=DPCodeRoundedIntegerWrapper.find_dpcode(
device, DPCode.HUMIDITY_CURRENT
),
current_temperature_wrapper=temperature_wrappers[0],
@@ -378,7 +369,7 @@ async def async_setup_entry(
switch_wrapper=DPCodeBooleanWrapper.find_dpcode(
device, DPCode.SWITCH, prefer_function=True
),
target_humidity_wrapper=_RoundedIntegerWrapper.find_dpcode(
target_humidity_wrapper=DPCodeRoundedIntegerWrapper.find_dpcode(
device, DPCode.HUMIDITY_SET, prefer_function=True
),
temperature_unit=temperature_wrappers[2],

View File

@@ -9,13 +9,12 @@ from tuya_device_handlers.device_wrapper.base import DeviceWrapper
from tuya_device_handlers.device_wrapper.common import (
DPCodeBooleanWrapper,
DPCodeEnumWrapper,
DPCodeIntegerWrapper,
)
from tuya_device_handlers.type_information import (
EnumTypeInformation,
IntegerTypeInformation,
from tuya_device_handlers.device_wrapper.extended import (
DPCodeInvertedPercentageWrapper,
DPCodePercentageWrapper,
)
from tuya_device_handlers.utils import RemapHelper
from tuya_device_handlers.type_information import EnumTypeInformation
from tuya_sharing import CustomerDevice, Manager
from homeassistant.components.cover import (
@@ -35,48 +34,10 @@ from .const import TUYA_DISCOVERY_NEW, DeviceCategory, DPCode
from .entity import TuyaEntity
class _DPCodePercentageMappingWrapper(DPCodeIntegerWrapper[int]):
"""Wrapper for DPCode position values mapping to 0-100 range."""
def __init__(self, dpcode: str, type_information: IntegerTypeInformation) -> None:
"""Init DPCodeIntegerWrapper."""
super().__init__(dpcode, type_information)
self._remap_helper = RemapHelper.from_type_information(type_information, 0, 100)
def _position_reversed(self, device: CustomerDevice) -> bool:
"""Check if the position and direction should be reversed."""
return False
def read_device_status(self, device: CustomerDevice) -> int | None:
if (value := device.status.get(self.dpcode)) is None:
return None
return round(
self._remap_helper.remap_value_to(
value, reverse=self._position_reversed(device)
)
)
def _convert_value_to_raw_value(self, device: CustomerDevice, value: Any) -> Any:
return round(
self._remap_helper.remap_value_from(
value, reverse=self._position_reversed(device)
)
)
class _InvertedPercentageMappingWrapper(_DPCodePercentageMappingWrapper):
"""Wrapper for DPCode position values mapping to 0-100 range."""
def _position_reversed(self, device: CustomerDevice) -> bool:
"""Check if the position and direction should be reversed."""
return True
class _ControlBackModePercentageMappingWrapper(_DPCodePercentageMappingWrapper):
class _ControlBackModePercentageMappingWrapper(DPCodePercentageWrapper):
"""Wrapper for DPCode position values with control_back_mode support."""
def _position_reversed(self, device: CustomerDevice) -> bool:
def _remap_inverted(self, device: CustomerDevice) -> bool:
"""Check if the position and direction should be reversed."""
return device.status.get(DPCode.CONTROL_BACK_MODE) != "back"
@@ -149,9 +110,7 @@ class TuyaCoverEntityDescription(CoverEntityDescription):
)
current_position: DPCode | tuple[DPCode, ...] | None = None
instruction_wrapper: type[_InstructionEnumWrapper] = _InstructionEnumWrapper
position_wrapper: type[_DPCodePercentageMappingWrapper] = (
_InvertedPercentageMappingWrapper
)
position_wrapper: type[DPCodePercentageWrapper] = DPCodeInvertedPercentageWrapper
set_position: DPCode | None = None

View File

@@ -8,25 +8,17 @@ from tuya_device_handlers.device_wrapper.base import DeviceWrapper
from tuya_device_handlers.device_wrapper.common import (
DPCodeBooleanWrapper,
DPCodeEnumWrapper,
DPCodeIntegerWrapper,
)
from tuya_device_handlers.type_information import IntegerTypeInformation
from tuya_device_handlers.utils import RemapHelper
from tuya_device_handlers.device_wrapper.fan import (
FanSpeedEnumWrapper,
FanSpeedIntegerWrapper,
)
from tuya_sharing import CustomerDevice, Manager
from homeassistant.components.fan import (
DIRECTION_FORWARD,
DIRECTION_REVERSE,
FanEntity,
FanEntityFeature,
)
from homeassistant.components.fan import FanEntity, FanEntityFeature
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.util.percentage import (
ordered_list_item_to_percentage,
percentage_to_ordered_list_item,
)
from . import TuyaConfigEntry
from .const import TUYA_DISCOVERY_NEW, DeviceCategory, DPCode
@@ -54,19 +46,6 @@ TUYA_SUPPORT_TYPE: set[DeviceCategory] = {
}
class _DirectionEnumWrapper(DPCodeEnumWrapper):
"""Wrapper for fan direction DP code."""
def read_device_status(self, device: CustomerDevice) -> str | None:
"""Read the device status and return the direction string."""
if (value := self._read_dpcode_value(device)) and value in {
DIRECTION_FORWARD,
DIRECTION_REVERSE,
}:
return value
return None
def _has_a_valid_dpcode(device: CustomerDevice) -> bool:
"""Check if the device has at least one valid DP code."""
properties_to_check: list[DPCode | tuple[DPCode, ...] | None] = [
@@ -80,50 +59,15 @@ def _has_a_valid_dpcode(device: CustomerDevice) -> bool:
return any(get_dpcode(device, code) for code in properties_to_check)
class _FanSpeedEnumWrapper(DPCodeEnumWrapper[int]):
"""Wrapper for fan speed DP code (from an enum)."""
def read_device_status(self, device: CustomerDevice) -> int | None:
"""Get the current speed as a percentage."""
if (value := self._read_dpcode_value(device)) is None:
return None
return ordered_list_item_to_percentage(self.options, value)
def _convert_value_to_raw_value(self, device: CustomerDevice, value: Any) -> Any:
"""Convert a Home Assistant value back to a raw device value."""
return percentage_to_ordered_list_item(self.options, value)
class _FanSpeedIntegerWrapper(DPCodeIntegerWrapper[int]):
"""Wrapper for fan speed DP code (from an integer)."""
def __init__(self, dpcode: str, type_information: IntegerTypeInformation) -> None:
"""Init DPCodeIntegerWrapper."""
super().__init__(dpcode, type_information)
self._remap_helper = RemapHelper.from_type_information(type_information, 1, 100)
def read_device_status(self, device: CustomerDevice) -> int | None:
"""Get the current speed as a percentage."""
if (value := self._read_dpcode_value(device)) is None:
return None
return round(self._remap_helper.remap_value_to(value))
def _convert_value_to_raw_value(self, device: CustomerDevice, value: Any) -> Any:
"""Convert a Home Assistant value back to a raw device value."""
return round(self._remap_helper.remap_value_from(value))
def _get_speed_wrapper(
device: CustomerDevice,
) -> _FanSpeedEnumWrapper | _FanSpeedIntegerWrapper | None:
) -> DeviceWrapper[int] | None:
"""Get the speed wrapper for the device."""
if int_wrapper := _FanSpeedIntegerWrapper.find_dpcode(
if int_wrapper := FanSpeedIntegerWrapper.find_dpcode(
device, _SPEED_DPCODES, prefer_function=True
):
return int_wrapper
return _FanSpeedEnumWrapper.find_dpcode(
device, _SPEED_DPCODES, prefer_function=True
)
return FanSpeedEnumWrapper.find_dpcode(device, _SPEED_DPCODES, prefer_function=True)
async def async_setup_entry(
@@ -145,7 +89,7 @@ async def async_setup_entry(
TuyaFanEntity(
device,
manager,
direction_wrapper=_DirectionEnumWrapper.find_dpcode(
direction_wrapper=DPCodeEnumWrapper.find_dpcode(
device, _DIRECTION_DPCODES, prefer_function=True
),
mode_wrapper=DPCodeEnumWrapper.find_dpcode(

View File

@@ -9,8 +9,8 @@ from tuya_device_handlers.device_wrapper.base import DeviceWrapper
from tuya_device_handlers.device_wrapper.common import (
DPCodeBooleanWrapper,
DPCodeEnumWrapper,
DPCodeIntegerWrapper,
)
from tuya_device_handlers.device_wrapper.extended import DPCodeRoundedIntegerWrapper
from tuya_sharing import CustomerDevice, Manager
from homeassistant.components.humidifier import (
@@ -29,16 +29,6 @@ from .entity import TuyaEntity
from .util import ActionDPCodeNotFoundError, get_dpcode
class _RoundedIntegerWrapper(DPCodeIntegerWrapper[int]):
"""An integer that always rounds its value."""
def read_device_status(self, device: CustomerDevice) -> int | None:
"""Read and round the device status."""
if (value := self._read_dpcode_value(device)) is None:
return None
return round(value)
@dataclass(frozen=True)
class TuyaHumidifierEntityDescription(HumidifierEntityDescription):
"""Describe an Tuya (de)humidifier entity."""
@@ -104,7 +94,7 @@ async def async_setup_entry(
device,
manager,
description,
current_humidity_wrapper=_RoundedIntegerWrapper.find_dpcode(
current_humidity_wrapper=DPCodeRoundedIntegerWrapper.find_dpcode(
device, description.current_humidity
),
mode_wrapper=DPCodeEnumWrapper.find_dpcode(
@@ -115,7 +105,7 @@ async def async_setup_entry(
description.dpcode or description.key,
prefer_function=True,
),
target_humidity_wrapper=_RoundedIntegerWrapper.find_dpcode(
target_humidity_wrapper=DPCodeRoundedIntegerWrapper.find_dpcode(
device, description.humidity, prefer_function=True
),
)