mirror of
https://github.com/home-assistant/core.git
synced 2026-03-15 07:22:12 +01:00
Compare commits
5 Commits
dev
...
epenet/202
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
bc1bac9bc8 | ||
|
|
97979864ab | ||
|
|
1eb0c91041 | ||
|
|
c7b147464d | ||
|
|
f5259ca883 |
@@ -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)
|
||||
|
||||
@@ -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],
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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
|
||||
),
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user