Refactor Tuya climate platform (#57609)
This commit is contained in:
parent
16b7375e60
commit
6a72af63c2
5 changed files with 402 additions and 298 deletions
|
@ -70,6 +70,7 @@ FAN_DIFFUSE = "diffuse"
|
||||||
|
|
||||||
|
|
||||||
# Possible swing state
|
# Possible swing state
|
||||||
|
SWING_ON = "on"
|
||||||
SWING_OFF = "off"
|
SWING_OFF = "off"
|
||||||
SWING_BOTH = "both"
|
SWING_BOTH = "both"
|
||||||
SWING_VERTICAL = "vertical"
|
SWING_VERTICAL = "vertical"
|
||||||
|
|
|
@ -170,9 +170,9 @@ class DeviceListener(TuyaDeviceListener):
|
||||||
"""Update device status."""
|
"""Update device status."""
|
||||||
if device.id in self.device_ids:
|
if device.id in self.device_ids:
|
||||||
_LOGGER.debug(
|
_LOGGER.debug(
|
||||||
"_update-->%s;->>%s",
|
"Received update for device %s: %s",
|
||||||
self,
|
|
||||||
device.id,
|
device.id,
|
||||||
|
self.device_manager.device_map[device.id].status,
|
||||||
)
|
)
|
||||||
dispatcher_send(self.hass, f"{TUYA_HA_SIGNAL_UPDATE_ENTITY}_{device.id}")
|
dispatcher_send(self.hass, f"{TUYA_HA_SIGNAL_UPDATE_ENTITY}_{device.id}")
|
||||||
|
|
||||||
|
|
|
@ -1,6 +1,9 @@
|
||||||
"""Tuya Home Assistant Base Device Model."""
|
"""Tuya Home Assistant Base Device Model."""
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from dataclasses import dataclass
|
||||||
|
import json
|
||||||
|
import logging
|
||||||
from typing import Any
|
from typing import Any
|
||||||
|
|
||||||
from tuya_iot import TuyaDevice, TuyaDeviceManager
|
from tuya_iot import TuyaDevice, TuyaDeviceManager
|
||||||
|
@ -10,6 +13,36 @@ from homeassistant.helpers.entity import DeviceInfo, Entity
|
||||||
|
|
||||||
from .const import DOMAIN, TUYA_HA_SIGNAL_UPDATE_ENTITY
|
from .const import DOMAIN, TUYA_HA_SIGNAL_UPDATE_ENTITY
|
||||||
|
|
||||||
|
_LOGGER = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class IntegerTypeData:
|
||||||
|
"""Integer Type Data."""
|
||||||
|
|
||||||
|
min: int
|
||||||
|
max: int
|
||||||
|
unit: str
|
||||||
|
scale: float
|
||||||
|
step: float
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def from_json(data: str) -> IntegerTypeData:
|
||||||
|
"""Load JSON string and return a IntegerTypeData object."""
|
||||||
|
return IntegerTypeData(**json.loads(data))
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class EnumTypeData:
|
||||||
|
"""Enum Type Data."""
|
||||||
|
|
||||||
|
range: list[str]
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def from_json(data: str) -> EnumTypeData:
|
||||||
|
"""Load JSON string and return a EnumTypeData object."""
|
||||||
|
return EnumTypeData(**json.loads(data))
|
||||||
|
|
||||||
|
|
||||||
class TuyaHaEntity(Entity):
|
class TuyaHaEntity(Entity):
|
||||||
"""Tuya base device."""
|
"""Tuya base device."""
|
||||||
|
@ -54,4 +87,12 @@ class TuyaHaEntity(Entity):
|
||||||
|
|
||||||
def _send_command(self, commands: list[dict[str, Any]]) -> None:
|
def _send_command(self, commands: list[dict[str, Any]]) -> None:
|
||||||
"""Send command to the device."""
|
"""Send command to the device."""
|
||||||
|
_LOGGER.debug(
|
||||||
|
"Sending commands for device %s: %s", self.tuya_device.id, commands
|
||||||
|
)
|
||||||
self.tuya_device_manager.send_commands(self.tuya_device.id, commands)
|
self.tuya_device_manager.send_commands(self.tuya_device.id, commands)
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def scale(value: float | int, scale: float | int) -> float:
|
||||||
|
"""Scale a value."""
|
||||||
|
return value * 1.0 / (10 ** scale)
|
||||||
|
|
|
@ -1,25 +1,28 @@
|
||||||
"""Support for Tuya Climate."""
|
"""Support for Tuya Climate."""
|
||||||
|
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
import json
|
from dataclasses import dataclass
|
||||||
import logging
|
|
||||||
from typing import Any
|
from typing import Any
|
||||||
|
|
||||||
from tuya_iot import TuyaDevice, TuyaDeviceManager
|
from tuya_iot import TuyaDevice, TuyaDeviceManager
|
||||||
|
|
||||||
from homeassistant.components.climate import ClimateEntity
|
from homeassistant.components.climate import ClimateEntity, ClimateEntityDescription
|
||||||
from homeassistant.components.climate.const import (
|
from homeassistant.components.climate.const import (
|
||||||
HVAC_MODE_AUTO,
|
|
||||||
HVAC_MODE_COOL,
|
HVAC_MODE_COOL,
|
||||||
HVAC_MODE_DRY,
|
HVAC_MODE_DRY,
|
||||||
HVAC_MODE_FAN_ONLY,
|
HVAC_MODE_FAN_ONLY,
|
||||||
HVAC_MODE_HEAT,
|
HVAC_MODE_HEAT,
|
||||||
|
HVAC_MODE_HEAT_COOL,
|
||||||
HVAC_MODE_OFF,
|
HVAC_MODE_OFF,
|
||||||
SUPPORT_FAN_MODE,
|
SUPPORT_FAN_MODE,
|
||||||
SUPPORT_SWING_MODE,
|
SUPPORT_SWING_MODE,
|
||||||
SUPPORT_TARGET_HUMIDITY,
|
SUPPORT_TARGET_HUMIDITY,
|
||||||
SUPPORT_TARGET_TEMPERATURE,
|
SUPPORT_TARGET_TEMPERATURE,
|
||||||
|
SWING_BOTH,
|
||||||
|
SWING_HORIZONTAL,
|
||||||
|
SWING_OFF,
|
||||||
|
SWING_ON,
|
||||||
|
SWING_VERTICAL,
|
||||||
)
|
)
|
||||||
from homeassistant.config_entries import ConfigEntry
|
from homeassistant.config_entries import ConfigEntry
|
||||||
from homeassistant.const import TEMP_CELSIUS, TEMP_FAHRENHEIT
|
from homeassistant.const import TEMP_CELSIUS, TEMP_FAHRENHEIT
|
||||||
|
@ -28,32 +31,53 @@ from homeassistant.helpers.dispatcher import async_dispatcher_connect
|
||||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||||
|
|
||||||
from . import HomeAssistantTuyaData
|
from . import HomeAssistantTuyaData
|
||||||
from .base import TuyaHaEntity
|
from .base import EnumTypeData, IntegerTypeData, TuyaHaEntity
|
||||||
from .const import DOMAIN, TUYA_DISCOVERY_NEW, DPCode
|
from .const import DOMAIN, TUYA_DISCOVERY_NEW, DPCode
|
||||||
|
|
||||||
_LOGGER = logging.getLogger(__name__)
|
|
||||||
|
|
||||||
SWING_OFF = "swing_off"
|
|
||||||
SWING_VERTICAL = "swing_vertical"
|
|
||||||
SWING_HORIZONTAL = "swing_horizontal"
|
|
||||||
SWING_BOTH = "swing_both"
|
|
||||||
|
|
||||||
DEFAULT_MIN_TEMP = 7
|
|
||||||
DEFAULT_MAX_TEMP = 35
|
|
||||||
|
|
||||||
TUYA_HVAC_TO_HA = {
|
TUYA_HVAC_TO_HA = {
|
||||||
"hot": HVAC_MODE_HEAT,
|
"auto": HVAC_MODE_HEAT_COOL,
|
||||||
"cold": HVAC_MODE_COOL,
|
"cold": HVAC_MODE_COOL,
|
||||||
|
"heat": HVAC_MODE_HEAT,
|
||||||
|
"hot": HVAC_MODE_HEAT,
|
||||||
|
"manual": HVAC_MODE_HEAT_COOL,
|
||||||
"wet": HVAC_MODE_DRY,
|
"wet": HVAC_MODE_DRY,
|
||||||
"wind": HVAC_MODE_FAN_ONLY,
|
"wind": HVAC_MODE_FAN_ONLY,
|
||||||
"auto": HVAC_MODE_AUTO,
|
|
||||||
}
|
}
|
||||||
|
|
||||||
# https://developer.tuya.com/en/docs/iot/standarddescription?id=K9i5ql6waswzq
|
|
||||||
TUYA_SUPPORT_TYPE = {
|
@dataclass
|
||||||
"kt", # Air conditioner
|
class TuyaClimateSensorDescriptionMixin:
|
||||||
"qn", # Heater
|
"""Define an entity description mixin for climate entities."""
|
||||||
"wk", # Thermostat
|
|
||||||
|
switch_only_hvac_mode: str
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class TuyaClimateEntityDescription(
|
||||||
|
ClimateEntityDescription, TuyaClimateSensorDescriptionMixin
|
||||||
|
):
|
||||||
|
"""Describe an Tuya climate entity."""
|
||||||
|
|
||||||
|
|
||||||
|
CLIMATE_DESCRIPTIONS: dict[str, TuyaClimateEntityDescription] = {
|
||||||
|
# Air conditioner
|
||||||
|
# https://developer.tuya.com/en/docs/iot/categorykt?id=Kaiuz0z71ov2n
|
||||||
|
"kt": TuyaClimateEntityDescription(
|
||||||
|
key="kt",
|
||||||
|
switch_only_hvac_mode=HVAC_MODE_COOL,
|
||||||
|
),
|
||||||
|
# Heater
|
||||||
|
# https://developer.tuya.com/en/docs/iot/f?id=K9gf46epy4j82
|
||||||
|
"qn": TuyaClimateEntityDescription(
|
||||||
|
key="qn",
|
||||||
|
switch_only_hvac_mode=HVAC_MODE_HEAT,
|
||||||
|
),
|
||||||
|
# Thermostat
|
||||||
|
# https://developer.tuya.com/en/docs/iot/f?id=K9gf45ld5l0t9
|
||||||
|
"wk": TuyaClimateEntityDescription(
|
||||||
|
key="wk",
|
||||||
|
switch_only_hvac_mode=HVAC_MODE_HEAT_COOL,
|
||||||
|
),
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
@ -66,11 +90,17 @@ async def async_setup_entry(
|
||||||
@callback
|
@callback
|
||||||
def async_discover_device(device_ids: list[str]) -> None:
|
def async_discover_device(device_ids: list[str]) -> None:
|
||||||
"""Discover and add a discovered Tuya climate."""
|
"""Discover and add a discovered Tuya climate."""
|
||||||
entities: list[TuyaHaClimate] = []
|
entities: list[TuyaClimateEntity] = []
|
||||||
for device_id in device_ids:
|
for device_id in device_ids:
|
||||||
device = hass_data.device_manager.device_map[device_id]
|
device = hass_data.device_manager.device_map[device_id]
|
||||||
if device and device.category in TUYA_SUPPORT_TYPE:
|
if device and device.category in CLIMATE_DESCRIPTIONS:
|
||||||
entities.append(TuyaHaClimate(device, hass_data.device_manager))
|
entities.append(
|
||||||
|
TuyaClimateEntity(
|
||||||
|
device,
|
||||||
|
hass_data.device_manager,
|
||||||
|
CLIMATE_DESCRIPTIONS[device.category],
|
||||||
|
)
|
||||||
|
)
|
||||||
async_add_entities(entities)
|
async_add_entities(entities)
|
||||||
|
|
||||||
async_discover_device([*hass_data.device_manager.device_map])
|
async_discover_device([*hass_data.device_manager.device_map])
|
||||||
|
@ -80,56 +110,203 @@ async def async_setup_entry(
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
class TuyaHaClimate(TuyaHaEntity, ClimateEntity):
|
class TuyaClimateEntity(TuyaHaEntity, ClimateEntity):
|
||||||
"""Tuya Switch Device."""
|
"""Tuya Climate Device."""
|
||||||
|
|
||||||
|
_current_humidity_dpcode: DPCode | None = None
|
||||||
|
_current_humidity_type: IntegerTypeData | None = None
|
||||||
|
_current_temperature_dpcode: DPCode | None = None
|
||||||
|
_current_temperature_type: IntegerTypeData | None = None
|
||||||
|
_hvac_to_tuya: dict[str, str]
|
||||||
|
_set_humidity_dpcode: DPCode | None = None
|
||||||
|
_set_humidity_type: IntegerTypeData | None = None
|
||||||
|
_set_temperature_dpcode: DPCode | None = None
|
||||||
|
_set_temperature_type: IntegerTypeData | None = None
|
||||||
|
entity_description: TuyaClimateEntityDescription
|
||||||
|
|
||||||
|
def __init__( # noqa: C901
|
||||||
|
self,
|
||||||
|
device: TuyaDevice,
|
||||||
|
device_manager: TuyaDeviceManager,
|
||||||
|
description: TuyaClimateEntityDescription,
|
||||||
|
) -> None:
|
||||||
|
"""Determine which values to use."""
|
||||||
|
self._attr_target_temperature_step = 1.0
|
||||||
|
self._attr_supported_features = 0
|
||||||
|
self.entity_description = description
|
||||||
|
|
||||||
def __init__(self, device: TuyaDevice, device_manager: TuyaDeviceManager) -> None:
|
|
||||||
"""Init Tuya Ha Climate."""
|
|
||||||
super().__init__(device, device_manager)
|
super().__init__(device, device_manager)
|
||||||
if DPCode.C_F in self.tuya_device.status:
|
|
||||||
self.dp_temp_unit = DPCode.C_F
|
|
||||||
else:
|
|
||||||
self.dp_temp_unit = DPCode.TEMP_UNIT_CONVERT
|
|
||||||
|
|
||||||
def get_temp_set_scale(self) -> int | None:
|
# If both temperature values for celsius and fahrenheit are present,
|
||||||
"""Get temperature set scale."""
|
# use whatever the device is set to, with a fallback to celsius.
|
||||||
dp_temp_set = DPCode.TEMP_SET if self.is_celsius() else DPCode.TEMP_SET_F
|
if all(
|
||||||
temp_set_value_range_item = self.tuya_device.status_range.get(dp_temp_set)
|
dpcode in device.status
|
||||||
if not temp_set_value_range_item:
|
for dpcode in (DPCode.TEMP_CURRENT, DPCode.TEMP_CURRENT_F)
|
||||||
return None
|
) or all(
|
||||||
|
dpcode in device.status for dpcode in (DPCode.TEMP_SET, DPCode.TEMP_SET_F)
|
||||||
|
):
|
||||||
|
self._attr_temperature_unit = TEMP_CELSIUS
|
||||||
|
if any(
|
||||||
|
"f" in device.status.get(dpcode, "").lower()
|
||||||
|
for dpcode in (DPCode.C_F, DPCode.TEMP_UNIT_CONVERT)
|
||||||
|
):
|
||||||
|
self._attr_temperature_unit = TEMP_FAHRENHEIT
|
||||||
|
|
||||||
temp_set_value_range = json.loads(temp_set_value_range_item.values)
|
# If any DPCode handling celsius is present, use celsius.
|
||||||
return temp_set_value_range.get("scale")
|
elif any(
|
||||||
|
dpcode in device.status for dpcode in (DPCode.TEMP_CURRENT, DPCode.TEMP_SET)
|
||||||
|
):
|
||||||
|
self._attr_temperature_unit = TEMP_CELSIUS
|
||||||
|
|
||||||
def get_temp_current_scale(self) -> int | None:
|
# If any DPCode handling fahrenheit is present, use celsius.
|
||||||
"""Get temperature current scale."""
|
elif any(
|
||||||
dp_temp_current = (
|
dpcode in device.status
|
||||||
DPCode.TEMP_CURRENT if self.is_celsius() else DPCode.TEMP_CURRENT_F
|
for dpcode in (DPCode.TEMP_CURRENT_F, DPCode.TEMP_SET_F)
|
||||||
)
|
):
|
||||||
temp_current_value_range_item = self.tuya_device.status_range.get(
|
self._attr_temperature_unit = TEMP_FAHRENHEIT
|
||||||
dp_temp_current
|
|
||||||
)
|
|
||||||
if not temp_current_value_range_item:
|
|
||||||
return None
|
|
||||||
|
|
||||||
temp_current_value_range = json.loads(temp_current_value_range_item.values)
|
# Determine dpcode to use for setting temperature
|
||||||
return temp_current_value_range.get("scale")
|
if all(
|
||||||
|
dpcode in device.status for dpcode in (DPCode.TEMP_SET, DPCode.TEMP_SET_F)
|
||||||
|
):
|
||||||
|
self._set_temperature_dpcode = DPCode.TEMP_SET
|
||||||
|
if self._attr_temperature_unit == TEMP_FAHRENHEIT:
|
||||||
|
self._set_temperature_dpcode = DPCode.TEMP_SET_F
|
||||||
|
elif DPCode.TEMP_SET in device.status:
|
||||||
|
self._set_temperature_dpcode = DPCode.TEMP_SET
|
||||||
|
elif DPCode.TEMP_SET_F in device.status:
|
||||||
|
self._set_temperature_dpcode = DPCode.TEMP_SET_F
|
||||||
|
|
||||||
# Functions
|
# Get integer type data for the dpcode to set temperature, use
|
||||||
|
# it to define min, max & step temperatures
|
||||||
|
if (
|
||||||
|
self._set_temperature_dpcode
|
||||||
|
and self._set_temperature_dpcode in device.status_range
|
||||||
|
):
|
||||||
|
type_data = IntegerTypeData.from_json(
|
||||||
|
device.status_range[self._set_temperature_dpcode].values
|
||||||
|
)
|
||||||
|
self._attr_supported_features |= SUPPORT_TARGET_TEMPERATURE
|
||||||
|
self._set_temperature_type = type_data
|
||||||
|
self._attr_max_temp = self.scale(type_data.max, type_data.scale)
|
||||||
|
self._attr_min_temp = self.scale(type_data.min, type_data.scale)
|
||||||
|
self._attr_target_temperature_step = self.scale(
|
||||||
|
type_data.step, type_data.scale
|
||||||
|
)
|
||||||
|
|
||||||
|
# Determine dpcode to use for getting the current temperature
|
||||||
|
if all(
|
||||||
|
dpcode in device.status
|
||||||
|
for dpcode in (DPCode.TEMP_CURRENT, DPCode.TEMP_CURRENT_F)
|
||||||
|
):
|
||||||
|
self._current_temperature_dpcode = DPCode.TEMP_CURRENT
|
||||||
|
if self._attr_temperature_unit == TEMP_FAHRENHEIT:
|
||||||
|
self._current_temperature_dpcode = DPCode.TEMP_CURRENT_F
|
||||||
|
elif DPCode.TEMP_CURRENT in device.status:
|
||||||
|
self._current_temperature_dpcode = DPCode.TEMP_CURRENT
|
||||||
|
elif DPCode.TEMP_CURRENT_F in device.status:
|
||||||
|
self._current_temperature_dpcode = DPCode.TEMP_CURRENT_F
|
||||||
|
|
||||||
|
# If we have a current temperature dpcode, get the integer type data
|
||||||
|
if (
|
||||||
|
self._current_temperature_dpcode
|
||||||
|
and self._current_temperature_dpcode in device.status_range
|
||||||
|
):
|
||||||
|
self._current_temperature_type = IntegerTypeData.from_json(
|
||||||
|
device.status_range[self._current_temperature_dpcode].values
|
||||||
|
)
|
||||||
|
|
||||||
|
# Determine HVAC modes
|
||||||
|
self._attr_hvac_modes = []
|
||||||
|
self._hvac_to_tuya = {}
|
||||||
|
if DPCode.MODE in device.function:
|
||||||
|
data_type = EnumTypeData.from_json(device.function[DPCode.MODE].values)
|
||||||
|
self._attr_hvac_modes = [HVAC_MODE_OFF]
|
||||||
|
for tuya_mode, ha_mode in TUYA_HVAC_TO_HA.items():
|
||||||
|
if tuya_mode in data_type.range:
|
||||||
|
self._hvac_to_tuya[ha_mode] = tuya_mode
|
||||||
|
self._attr_hvac_modes.append(ha_mode)
|
||||||
|
elif DPCode.SWITCH in device.function:
|
||||||
|
self._attr_hvac_modes = [
|
||||||
|
HVAC_MODE_OFF,
|
||||||
|
description.switch_only_hvac_mode,
|
||||||
|
]
|
||||||
|
|
||||||
|
# Determine dpcode to use for setting the humidity
|
||||||
|
if (
|
||||||
|
DPCode.HUMIDITY_SET in device.status
|
||||||
|
and DPCode.HUMIDITY_SET in device.status_range
|
||||||
|
):
|
||||||
|
self._attr_supported_features |= SUPPORT_TARGET_HUMIDITY
|
||||||
|
self._set_humidity_dpcode = DPCode.HUMIDITY_SET
|
||||||
|
type_data = IntegerTypeData.from_json(
|
||||||
|
device.status_range[DPCode.HUMIDITY_SET].values
|
||||||
|
)
|
||||||
|
self._set_humidity_type = type_data
|
||||||
|
self._attr_min_humidity = int(self.scale(type_data.max, type_data.scale))
|
||||||
|
self._attr_max_humidity = int(self.scale(type_data.min, type_data.scale))
|
||||||
|
|
||||||
|
# Determine dpcode to use for getting the current humidity
|
||||||
|
if (
|
||||||
|
DPCode.HUMIDITY_CURRENT in device.status
|
||||||
|
and DPCode.HUMIDITY_CURRENT in device.status_range
|
||||||
|
):
|
||||||
|
self._current_humidity_dpcode = DPCode.HUMIDITY_CURRENT
|
||||||
|
self._current_humidity_type = IntegerTypeData.from_json(
|
||||||
|
self.tuya_device.status_range[DPCode.HUMIDITY_CURRENT].values
|
||||||
|
)
|
||||||
|
|
||||||
|
# Determine dpcode to use for getting the current humidity
|
||||||
|
if (
|
||||||
|
DPCode.HUMIDITY_CURRENT in device.status
|
||||||
|
and DPCode.HUMIDITY_CURRENT in device.status_range
|
||||||
|
):
|
||||||
|
self._current_humidity_dpcode = DPCode.HUMIDITY_CURRENT
|
||||||
|
self._current_humidity_type = IntegerTypeData.from_json(
|
||||||
|
self.tuya_device.status_range[DPCode.HUMIDITY_CURRENT].values
|
||||||
|
)
|
||||||
|
|
||||||
|
# Determine fan modes
|
||||||
|
if (
|
||||||
|
DPCode.FAN_SPEED_ENUM in device.status
|
||||||
|
and DPCode.FAN_SPEED_ENUM in device.function
|
||||||
|
):
|
||||||
|
self._attr_supported_features |= SUPPORT_FAN_MODE
|
||||||
|
self._attr_fan_modes = EnumTypeData.from_json(
|
||||||
|
self.tuya_device.status_range[DPCode.FAN_SPEED_ENUM].values
|
||||||
|
).range
|
||||||
|
|
||||||
|
# Determine swing modes
|
||||||
|
if any(
|
||||||
|
dpcode in self.tuya_device.function
|
||||||
|
for dpcode in (
|
||||||
|
DPCode.SHAKE,
|
||||||
|
DPCode.SWING,
|
||||||
|
DPCode.SWITCH_HORIZONTAL,
|
||||||
|
DPCode.SWITCH_VERTICAL,
|
||||||
|
)
|
||||||
|
):
|
||||||
|
self._attr_supported_features |= SUPPORT_SWING_MODE
|
||||||
|
self._attr_swing_modes = [SWING_OFF]
|
||||||
|
if any(
|
||||||
|
dpcode in self.tuya_device.function
|
||||||
|
for dpcode in (DPCode.SHAKE, DPCode.SWING)
|
||||||
|
):
|
||||||
|
self._attr_swing_modes.append(SWING_ON)
|
||||||
|
|
||||||
|
if DPCode.SWITCH_HORIZONTAL in self.tuya_device.function:
|
||||||
|
self._attr_swing_modes.append(SWING_HORIZONTAL)
|
||||||
|
|
||||||
|
if DPCode.SWITCH_VERTICAL in self.tuya_device.function:
|
||||||
|
self._attr_swing_modes.append(SWING_VERTICAL)
|
||||||
|
|
||||||
def set_hvac_mode(self, hvac_mode: str) -> None:
|
def set_hvac_mode(self, hvac_mode: str) -> None:
|
||||||
"""Set new target hvac mode."""
|
"""Set new target hvac mode."""
|
||||||
commands = []
|
commands = [{"code": DPCode.SWITCH, "value": hvac_mode != HVAC_MODE_OFF}]
|
||||||
if hvac_mode == HVAC_MODE_OFF:
|
if hvac_mode in self._hvac_to_tuya:
|
||||||
commands.append({"code": DPCode.SWITCH, "value": False})
|
commands.append(
|
||||||
else:
|
{"code": DPCode.MODE, "value": self._hvac_to_tuya[hvac_mode]}
|
||||||
commands.append({"code": DPCode.SWITCH, "value": True})
|
)
|
||||||
|
|
||||||
for tuya_mode, ha_mode in TUYA_HVAC_TO_HA.items():
|
|
||||||
if ha_mode == hvac_mode:
|
|
||||||
commands.append({"code": DPCode.MODE, "value": tuya_mode})
|
|
||||||
break
|
|
||||||
|
|
||||||
self._send_command(commands)
|
self._send_command(commands)
|
||||||
|
|
||||||
def set_fan_mode(self, fan_mode: str) -> None:
|
def set_fan_mode(self, fan_mode: str) -> None:
|
||||||
|
@ -138,294 +315,177 @@ class TuyaHaClimate(TuyaHaEntity, ClimateEntity):
|
||||||
|
|
||||||
def set_humidity(self, humidity: float) -> None:
|
def set_humidity(self, humidity: float) -> None:
|
||||||
"""Set new target humidity."""
|
"""Set new target humidity."""
|
||||||
self._send_command([{"code": DPCode.HUMIDITY_SET, "value": int(humidity)}])
|
if self._set_humidity_dpcode is None or self._set_humidity_type is None:
|
||||||
|
raise RuntimeError(
|
||||||
def set_swing_mode(self, swing_mode: str) -> None:
|
"Cannot set humidity, device doesn't provide methods to set it"
|
||||||
"""Set new target swing operation."""
|
)
|
||||||
if swing_mode == SWING_BOTH:
|
|
||||||
commands = [
|
|
||||||
{"code": DPCode.SWITCH_VERTICAL, "value": True},
|
|
||||||
{"code": DPCode.SWITCH_HORIZONTAL, "value": True},
|
|
||||||
]
|
|
||||||
elif swing_mode == SWING_HORIZONTAL:
|
|
||||||
commands = [
|
|
||||||
{"code": DPCode.SWITCH_VERTICAL, "value": False},
|
|
||||||
{"code": DPCode.SWITCH_HORIZONTAL, "value": True},
|
|
||||||
]
|
|
||||||
elif swing_mode == SWING_VERTICAL:
|
|
||||||
commands = [
|
|
||||||
{"code": DPCode.SWITCH_VERTICAL, "value": True},
|
|
||||||
{"code": DPCode.SWITCH_HORIZONTAL, "value": False},
|
|
||||||
]
|
|
||||||
else:
|
|
||||||
commands = [
|
|
||||||
{"code": DPCode.SWITCH_VERTICAL, "value": False},
|
|
||||||
{"code": DPCode.SWITCH_HORIZONTAL, "value": False},
|
|
||||||
]
|
|
||||||
|
|
||||||
self._send_command(commands)
|
|
||||||
|
|
||||||
def set_temperature(self, **kwargs: Any) -> None:
|
|
||||||
"""Set new target temperature."""
|
|
||||||
_LOGGER.debug("climate temp-> %s", kwargs)
|
|
||||||
code = DPCode.TEMP_SET if self.is_celsius() else DPCode.TEMP_SET_F
|
|
||||||
temp_set_scale = self.get_temp_set_scale()
|
|
||||||
if not temp_set_scale:
|
|
||||||
return
|
|
||||||
|
|
||||||
self._send_command(
|
self._send_command(
|
||||||
[
|
[
|
||||||
{
|
{
|
||||||
"code": code,
|
"code": self._set_humidity_dpcode,
|
||||||
"value": int(kwargs["temperature"] * (10 ** temp_set_scale)),
|
"value": self.scale(humidity, self._set_humidity_type.scale),
|
||||||
}
|
}
|
||||||
]
|
]
|
||||||
)
|
)
|
||||||
|
|
||||||
def is_celsius(self) -> bool:
|
def set_swing_mode(self, swing_mode: str) -> None:
|
||||||
"""Return True if device reports in Celsius."""
|
"""Set new target swing operation."""
|
||||||
if (
|
# The API accepts these all at once and will ignore the codes
|
||||||
self.dp_temp_unit in self.tuya_device.status
|
# that don't apply to the device being controlled.
|
||||||
and self.tuya_device.status.get(self.dp_temp_unit).lower() == "c"
|
self._send_command(
|
||||||
):
|
[
|
||||||
return True
|
{
|
||||||
if (
|
"code": DPCode.SHAKE,
|
||||||
DPCode.TEMP_SET in self.tuya_device.status
|
"value": swing_mode == SWING_ON,
|
||||||
or DPCode.TEMP_CURRENT in self.tuya_device.status
|
},
|
||||||
):
|
{
|
||||||
return True
|
"code": DPCode.SWING,
|
||||||
return False
|
"value": swing_mode == SWING_ON,
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"code": DPCode.SWITCH_VERTICAL,
|
||||||
|
"value": swing_mode in (SWING_BOTH, SWING_VERTICAL),
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"code": DPCode.SWITCH_HORIZONTAL,
|
||||||
|
"value": swing_mode in (SWING_BOTH, SWING_HORIZONTAL),
|
||||||
|
},
|
||||||
|
]
|
||||||
|
)
|
||||||
|
|
||||||
@property
|
def set_temperature(self, **kwargs: Any) -> None:
|
||||||
def temperature_unit(self) -> str:
|
"""Set new target temperature."""
|
||||||
"""Return true if fan is on."""
|
if self._set_temperature_dpcode is None or self._set_temperature_type is None:
|
||||||
if self.is_celsius():
|
raise RuntimeError(
|
||||||
return TEMP_CELSIUS
|
"Cannot set target temperature, device doesn't provide methods to set it"
|
||||||
return TEMP_FAHRENHEIT
|
)
|
||||||
|
|
||||||
|
self._send_command(
|
||||||
|
[
|
||||||
|
{
|
||||||
|
"code": self._set_temperature_dpcode,
|
||||||
|
"value": round(
|
||||||
|
self.scale(
|
||||||
|
kwargs["temperature"], self._set_temperature_type.scale
|
||||||
|
)
|
||||||
|
),
|
||||||
|
}
|
||||||
|
]
|
||||||
|
)
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def current_temperature(self) -> float | None:
|
def current_temperature(self) -> float | None:
|
||||||
"""Return the current temperature."""
|
"""Return the current temperature."""
|
||||||
if (
|
if (
|
||||||
DPCode.TEMP_CURRENT not in self.tuya_device.status
|
self._current_temperature_dpcode is None
|
||||||
and DPCode.TEMP_CURRENT_F not in self.tuya_device.status
|
or self._current_temperature_type is None
|
||||||
):
|
):
|
||||||
return None
|
return None
|
||||||
|
|
||||||
temp_current_scale = self.get_temp_current_scale()
|
temperature = self.tuya_device.status.get(self._current_temperature_dpcode)
|
||||||
if not temp_current_scale:
|
if temperature is None:
|
||||||
return None
|
return None
|
||||||
|
|
||||||
if self.is_celsius():
|
return self.scale(temperature, self._current_temperature_type.scale)
|
||||||
temperature = self.tuya_device.status.get(DPCode.TEMP_CURRENT)
|
|
||||||
if not temperature:
|
|
||||||
return None
|
|
||||||
return temperature * 1.0 / (10 ** temp_current_scale)
|
|
||||||
|
|
||||||
temperature = self.tuya_device.status.get(DPCode.TEMP_CURRENT_F)
|
|
||||||
if not temperature:
|
|
||||||
return None
|
|
||||||
return temperature * 1.0 / (10 ** temp_current_scale)
|
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def current_humidity(self) -> int:
|
def current_humidity(self) -> int | None:
|
||||||
"""Return the current humidity."""
|
"""Return the current humidity."""
|
||||||
return int(self.tuya_device.status.get(DPCode.HUMIDITY_CURRENT, 0))
|
if self._current_humidity_dpcode is None or self._current_humidity_type is None:
|
||||||
|
return None
|
||||||
|
|
||||||
|
humidity = self.tuya_device.status.get(self._current_humidity_dpcode)
|
||||||
|
if humidity is None:
|
||||||
|
return None
|
||||||
|
|
||||||
|
return round(self.scale(humidity, self._current_humidity_type.scale))
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def target_temperature(self) -> float | None:
|
def target_temperature(self) -> float | None:
|
||||||
"""Return the temperature currently set to be reached."""
|
"""Return the temperature currently set to be reached."""
|
||||||
temp_set_scale = self.get_temp_set_scale()
|
if self._set_temperature_dpcode is None or self._set_temperature_type is None:
|
||||||
if temp_set_scale is None:
|
|
||||||
return None
|
return None
|
||||||
|
|
||||||
dpcode_temp_set = self.tuya_device.status.get(DPCode.TEMP_SET)
|
temperature = self.tuya_device.status.get(self._set_temperature_dpcode)
|
||||||
if dpcode_temp_set is None:
|
if temperature is None:
|
||||||
return None
|
return None
|
||||||
|
|
||||||
return dpcode_temp_set * 1.0 / (10 ** temp_set_scale)
|
return self.scale(temperature, self._set_temperature_type.scale)
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def max_temp(self) -> float:
|
def target_humidity(self) -> int | None:
|
||||||
"""Return the maximum temperature."""
|
"""Return the humidity currently set to be reached."""
|
||||||
scale = self.get_temp_set_scale()
|
if self._set_humidity_dpcode is None or self._set_humidity_type is None:
|
||||||
if scale is None:
|
|
||||||
return DEFAULT_MAX_TEMP
|
|
||||||
|
|
||||||
if self.is_celsius():
|
|
||||||
if DPCode.TEMP_SET not in self.tuya_device.function:
|
|
||||||
return DEFAULT_MAX_TEMP
|
|
||||||
|
|
||||||
function_item = self.tuya_device.function.get(DPCode.TEMP_SET)
|
|
||||||
if function_item is None:
|
|
||||||
return DEFAULT_MAX_TEMP
|
|
||||||
|
|
||||||
temp_value = json.loads(function_item.values)
|
|
||||||
|
|
||||||
temp_max = temp_value.get("max")
|
|
||||||
if temp_max is None:
|
|
||||||
return DEFAULT_MAX_TEMP
|
|
||||||
return temp_max * 1.0 / (10 ** scale)
|
|
||||||
if DPCode.TEMP_SET_F not in self.tuya_device.function:
|
|
||||||
return DEFAULT_MAX_TEMP
|
|
||||||
|
|
||||||
function_item_f = self.tuya_device.function.get(DPCode.TEMP_SET_F)
|
|
||||||
if function_item_f is None:
|
|
||||||
return DEFAULT_MAX_TEMP
|
|
||||||
|
|
||||||
temp_value_f = json.loads(function_item_f.values)
|
|
||||||
|
|
||||||
temp_max_f = temp_value_f.get("max")
|
|
||||||
if temp_max_f is None:
|
|
||||||
return DEFAULT_MAX_TEMP
|
|
||||||
return temp_max_f * 1.0 / (10 ** scale)
|
|
||||||
|
|
||||||
@property
|
|
||||||
def min_temp(self) -> float:
|
|
||||||
"""Return the minimum temperature."""
|
|
||||||
temp_set_scal = self.get_temp_set_scale()
|
|
||||||
if temp_set_scal is None:
|
|
||||||
return DEFAULT_MIN_TEMP
|
|
||||||
|
|
||||||
if self.is_celsius():
|
|
||||||
if DPCode.TEMP_SET not in self.tuya_device.function:
|
|
||||||
return DEFAULT_MIN_TEMP
|
|
||||||
|
|
||||||
function_temp_item = self.tuya_device.function.get(DPCode.TEMP_SET)
|
|
||||||
if function_temp_item is None:
|
|
||||||
return DEFAULT_MIN_TEMP
|
|
||||||
temp_value = json.loads(function_temp_item.values)
|
|
||||||
temp_min = temp_value.get("min")
|
|
||||||
if temp_min is None:
|
|
||||||
return DEFAULT_MIN_TEMP
|
|
||||||
return temp_min * 1.0 / (10 ** temp_set_scal)
|
|
||||||
|
|
||||||
if DPCode.TEMP_SET_F not in self.tuya_device.function:
|
|
||||||
return DEFAULT_MIN_TEMP
|
|
||||||
|
|
||||||
temp_value_temp_f = self.tuya_device.function.get(DPCode.TEMP_SET_F)
|
|
||||||
if temp_value_temp_f is None:
|
|
||||||
return DEFAULT_MIN_TEMP
|
|
||||||
temp_value_f = json.loads(temp_value_temp_f.values)
|
|
||||||
|
|
||||||
temp_min_f = temp_value_f.get("min")
|
|
||||||
if temp_min_f is None:
|
|
||||||
return DEFAULT_MIN_TEMP
|
|
||||||
|
|
||||||
return temp_min_f * 1.0 / (10 ** temp_set_scal)
|
|
||||||
|
|
||||||
@property
|
|
||||||
def target_temperature_step(self) -> float | None:
|
|
||||||
"""Return target temperature setp."""
|
|
||||||
if (
|
|
||||||
DPCode.TEMP_SET not in self.tuya_device.status_range
|
|
||||||
and DPCode.TEMP_SET_F not in self.tuya_device.status_range
|
|
||||||
):
|
|
||||||
return 1.0
|
|
||||||
temp_set_value_range = json.loads(
|
|
||||||
self.tuya_device.status_range.get(
|
|
||||||
DPCode.TEMP_SET if self.is_celsius() else DPCode.TEMP_SET_F
|
|
||||||
).values
|
|
||||||
)
|
|
||||||
step = temp_set_value_range.get("step")
|
|
||||||
if step is None:
|
|
||||||
return None
|
return None
|
||||||
|
|
||||||
temp_set_scale = self.get_temp_set_scale()
|
humidity = self.tuya_device.status.get(self._set_humidity_dpcode)
|
||||||
if temp_set_scale is None:
|
if humidity is None:
|
||||||
return None
|
return None
|
||||||
|
|
||||||
return step * 1.0 / (10 ** temp_set_scale)
|
return round(self.scale(humidity, self._set_humidity_type.scale))
|
||||||
|
|
||||||
@property
|
|
||||||
def target_humidity(self) -> int:
|
|
||||||
"""Return target humidity."""
|
|
||||||
return int(self.tuya_device.status.get(DPCode.HUMIDITY_SET, 0))
|
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def hvac_mode(self) -> str:
|
def hvac_mode(self) -> str:
|
||||||
"""Return hvac mode."""
|
"""Return hvac mode."""
|
||||||
if not self.tuya_device.status.get(DPCode.SWITCH, False):
|
# If the switch off, hvac mode is off as well. Unless the switch
|
||||||
|
# the switch is on or doesn't exists of course...
|
||||||
|
if not self.tuya_device.status.get(DPCode.SWITCH, True):
|
||||||
return HVAC_MODE_OFF
|
return HVAC_MODE_OFF
|
||||||
if DPCode.MODE not in self.tuya_device.status:
|
|
||||||
|
if DPCode.MODE not in self.tuya_device.function:
|
||||||
|
if self.tuya_device.status.get(DPCode.SWITCH, False):
|
||||||
|
return self.entity_description.switch_only_hvac_mode
|
||||||
return HVAC_MODE_OFF
|
return HVAC_MODE_OFF
|
||||||
|
|
||||||
if self.tuya_device.status.get(DPCode.MODE) is not None:
|
if self.tuya_device.status.get(DPCode.MODE) is not None:
|
||||||
return TUYA_HVAC_TO_HA[self.tuya_device.status[DPCode.MODE]]
|
return TUYA_HVAC_TO_HA[self.tuya_device.status[DPCode.MODE]]
|
||||||
return HVAC_MODE_OFF
|
return HVAC_MODE_OFF
|
||||||
|
|
||||||
@property
|
|
||||||
def hvac_modes(self) -> list[str]:
|
|
||||||
"""Return hvac modes for select."""
|
|
||||||
if DPCode.MODE not in self.tuya_device.function:
|
|
||||||
return []
|
|
||||||
modes = json.loads(self.tuya_device.function.get(DPCode.MODE, {}).values).get(
|
|
||||||
"range"
|
|
||||||
)
|
|
||||||
|
|
||||||
hvac_modes = [HVAC_MODE_OFF]
|
|
||||||
for tuya_mode, ha_mode in TUYA_HVAC_TO_HA.items():
|
|
||||||
if tuya_mode in modes:
|
|
||||||
hvac_modes.append(ha_mode)
|
|
||||||
|
|
||||||
return hvac_modes
|
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def fan_mode(self) -> str | None:
|
def fan_mode(self) -> str | None:
|
||||||
"""Return fan mode."""
|
"""Return fan mode."""
|
||||||
return self.tuya_device.status.get(DPCode.FAN_SPEED_ENUM)
|
return self.tuya_device.status.get(DPCode.FAN_SPEED_ENUM)
|
||||||
|
|
||||||
@property
|
|
||||||
def fan_modes(self) -> list[str]:
|
|
||||||
"""Return fan modes for select."""
|
|
||||||
fan_speed_device_function = self.tuya_device.function.get(DPCode.FAN_SPEED_ENUM)
|
|
||||||
if not fan_speed_device_function:
|
|
||||||
return []
|
|
||||||
return json.loads(fan_speed_device_function.values).get("range", [])
|
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def swing_mode(self) -> str:
|
def swing_mode(self) -> str:
|
||||||
"""Return swing mode."""
|
"""Return swing mode."""
|
||||||
mode = 0
|
if any(
|
||||||
if (
|
self.tuya_device.status.get(dpcode)
|
||||||
DPCode.SWITCH_HORIZONTAL in self.tuya_device.status
|
for dpcode in (DPCode.SHAKE, DPCode.SWING)
|
||||||
and self.tuya_device.status.get(DPCode.SWITCH_HORIZONTAL)
|
|
||||||
):
|
):
|
||||||
mode += 1
|
return SWING_ON
|
||||||
if (
|
|
||||||
DPCode.SWITCH_VERTICAL in self.tuya_device.status
|
|
||||||
and self.tuya_device.status.get(DPCode.SWITCH_VERTICAL)
|
|
||||||
):
|
|
||||||
mode += 2
|
|
||||||
|
|
||||||
if mode == 3:
|
horizontal = self.tuya_device.status.get(DPCode.SWITCH_HORIZONTAL)
|
||||||
|
vertical = self.tuya_device.status.get(DPCode.SWITCH_VERTICAL)
|
||||||
|
if horizontal and vertical:
|
||||||
return SWING_BOTH
|
return SWING_BOTH
|
||||||
if mode == 2:
|
if horizontal:
|
||||||
return SWING_VERTICAL
|
|
||||||
if mode == 1:
|
|
||||||
return SWING_HORIZONTAL
|
return SWING_HORIZONTAL
|
||||||
|
if vertical:
|
||||||
|
return SWING_VERTICAL
|
||||||
|
|
||||||
return SWING_OFF
|
return SWING_OFF
|
||||||
|
|
||||||
@property
|
def turn_on(self) -> None:
|
||||||
def swing_modes(self) -> list[str]:
|
"""Turn the device on, retaining current HVAC (if supported)."""
|
||||||
"""Return swing mode for select."""
|
if DPCode.SWITCH in self.tuya_device.function:
|
||||||
return [SWING_OFF, SWING_HORIZONTAL, SWING_VERTICAL, SWING_BOTH]
|
self._send_command([{"code": DPCode.SWITCH, "value": True}])
|
||||||
|
return
|
||||||
|
|
||||||
@property
|
# Fake turn on
|
||||||
def supported_features(self) -> int:
|
for mode in (HVAC_MODE_HEAT_COOL, HVAC_MODE_HEAT, HVAC_MODE_COOL):
|
||||||
"""Flag supported features."""
|
if mode not in self.hvac_modes:
|
||||||
supports = 0
|
continue
|
||||||
if (
|
self.set_hvac_mode(mode)
|
||||||
DPCode.TEMP_SET in self.tuya_device.status
|
break
|
||||||
or DPCode.TEMP_SET_F in self.tuya_device.status
|
|
||||||
):
|
def turn_off(self) -> None:
|
||||||
supports |= SUPPORT_TARGET_TEMPERATURE
|
"""Turn the device on, retaining current HVAC (if supported)."""
|
||||||
if DPCode.FAN_SPEED_ENUM in self.tuya_device.status:
|
if DPCode.SWITCH in self.tuya_device.function:
|
||||||
supports |= SUPPORT_FAN_MODE
|
self._send_command([{"code": DPCode.SWITCH, "value": False}])
|
||||||
if DPCode.HUMIDITY_SET in self.tuya_device.status:
|
return
|
||||||
supports |= SUPPORT_TARGET_HUMIDITY
|
|
||||||
if (
|
# Fake turn off
|
||||||
DPCode.SWITCH_HORIZONTAL in self.tuya_device.status
|
if HVAC_MODE_OFF in self.hvac_modes:
|
||||||
or DPCode.SWITCH_VERTICAL in self.tuya_device.status
|
self.set_hvac_mode(HVAC_MODE_OFF)
|
||||||
):
|
|
||||||
supports |= SUPPORT_SWING_MODE
|
|
||||||
return supports
|
|
||||||
|
|
|
@ -76,8 +76,10 @@ class DPCode(str, Enum):
|
||||||
LOCK = "lock" # Lock / Child lock
|
LOCK = "lock" # Lock / Child lock
|
||||||
MODE = "mode" # Working mode / Mode
|
MODE = "mode" # Working mode / Mode
|
||||||
PUMP_RESET = "pump_reset" # Water pump reset
|
PUMP_RESET = "pump_reset" # Water pump reset
|
||||||
|
SHAKE = "shake" # Oscillating
|
||||||
SPEED = "speed" # Speed level
|
SPEED = "speed" # Speed level
|
||||||
START = "start" # Start
|
START = "start" # Start
|
||||||
|
SWING = "swing" # Swing mode
|
||||||
SWITCH = "switch" # Switch
|
SWITCH = "switch" # Switch
|
||||||
SWITCH_HORIZONTAL = "switch_horizontal" # Horizontal swing flap switch
|
SWITCH_HORIZONTAL = "switch_horizontal" # Horizontal swing flap switch
|
||||||
SWITCH_LED = "switch_led" # Switch
|
SWITCH_LED = "switch_led" # Switch
|
||||||
|
|
Loading…
Add table
Add a link
Reference in a new issue