Add support for water heater devices and the Hydrotherm DYNAMIC/X8 heat pump hot water system.

custom_components/tuya_local/const.py

@@ -11,5 +11,6 @@ CONF_LIGHT = "light"
 CONF_LOCK = "lock"
 CONF_SWITCH = "switch"
 CONF_HUMIDIFIER = "humidifier"
+CONF_WATER_HEATER = "water_heater"
 API_PROTOCOL_VERSIONS = [3.3, 3.1, 3.2, 3.4]
 SCAN_INTERVAL = timedelta(seconds=30)

custom_components/tuya_local/devices/hydrotherm_dynamic_x8_water_heater.yaml

@@ -0,0 +1,79 @@
+name: Hydrotherm DYNAMIC/X8
+primary_entity:
+  entity: water_heater
+  dps:
+    - id: 1
+      type: boolean
+      name: power
+      hidden: true
+      mapping:
+        - dps_val: false
+          value: "off"
+    - id: 2
+      type: integer
+      name: temperature
+      range:
+        min: 60
+        max: 70
+    - id: 3
+      type: integer
+      name: current_temperature
+      mapping:
+        - scale: 1
+    - id: 4
+      type: string
+      name: operation_mode
+      mapping:
+        - dps_val: ECO
+          constraint: power
+          conditions:
+            - dps_val: false
+              value_redirect: power
+              value: "off"
+            - dps_val: true
+              value: eco
+        - dps_val: STANDARD
+          constraint: power
+          conditions:
+            - dps_val: false
+              value_redirect: power
+              value: "off"
+            - dps_val: true
+              value: heat_pump
+        - dps_val: HYBRID
+          constraint: power
+          conditions:
+            - dps_val: false
+              value_redirect: power
+              value: "off"
+            - dps_val: true
+              value: high_demand
+        - dps_val: HYBRID1
+          constraint: power
+          conditions:
+            - dps_val: false
+              value_redirect: power
+              value: "off"
+            - dps_val: true
+              value: performance
+        - dps_val: ELEMENT
+          constraint: power
+          conditions:
+            - dps_val: false
+              value_redirect: power
+              value: "off"
+            - dps_val: true
+              value: electric
+secondary_entities:
+  - entity: binary_sensor
+    class: problem
+    name: Fault
+    category: diagnostic
+    dps:
+      - id: 21
+        type: bitfield
+        name: sensor
+        mapping:
+          - dps_val: 0
+            value: false
+          - value: true

custom_components/tuya_local/generic/water_heater.py

@@ -0,0 +1,157 @@
+"""
+Platform to control tuya water heater devices.
+"""
+import logging
+
+from homeassistant.components.water_heater import (
+    WaterHeaterEntity,
+    WaterHeaterEntityFeature,
+)
+from homeassistant.components.climate.const import (
+    ATTR_CURRENT_TEMPERATURE,
+    ATTR_TARGET_TEMP_HIGH,
+    ATTR_TARGET_TEMP_LOW,
+)
+from homeassistant.const import (
+    ATTR_TEMPERATURE,
+    TEMP_CELSIUS,
+    TEMP_FAHRENHEIT,
+    TEMP_KELVIN,
+)
+
+from ..device import TuyaLocalDevice
+from ..helpers.device_config import TuyaEntityConfig
+from ..helpers.mixin import TuyaLocalEntity, unit_from_ascii
+
+_LOGGER = logging.getLogger(__name__)
+
+VALID_TEMP_UNIT = [TEMP_CELSIUS, TEMP_FAHRENHEIT, TEMP_KELVIN]
+
+
+def validate_temp_unit(unit):
+    unit = unit_from_ascii(unit)
+    return unit if unit in VALID_TEMP_UNIT else None
+
+
+class TuyaLocalWaterHeater(TuyaLocalEntity, WaterHeaterEntity):
+    """Representation of a Tuya water heater entity."""
+
+    def __init__(self, device: TuyaLocalDevice, config: TuyaEntityConfig):
+        """
+        Initialise the water heater device.
+        Args:
+           device (TuyaLocalDevice): The device API instance.
+           config (TuyaEntityConfig): The entity config.
+        """
+        dps_map = self._init_begin(device, config)
+
+        self._current_temperature_dps = dps_map.pop(ATTR_CURRENT_TEMPERATURE, None)
+        self._temperature_dps = dps_map.pop(ATTR_TEMPERATURE, None)
+        self._temp_high_dps = dps_map.pop(ATTR_TARGET_TEMP_HIGH, None)
+        self._temp_low_dps = dps_map.pop(ATTR_TARGET_TEMP_LOW, None)
+        self._unit_dps = dps_map.pop("temperature_unit", None)
+        self._mintemp_dps = dps_map.pop("min_temperature", None)
+        self._maxtemp_dps = dps_map.pop("max_temperature", None)
+        self._operation_mode_dps = dps_map.pop("operation_mode", None)
+        self._init_end(dps_map)
+        self._support_flags = 0
+
+        if self._operation_mode_dps:
+            self._support_flags |= WaterHeaterEntityFeature.OPERATION_MODE
+
+    @property
+    def supported_features(self):
+        """Return the features supported by this climate device."""
+        return self._support_flags
+
+    @property
+    def temperature_unit(self):
+        """Return the unit of measurement."""
+        # If there is a separate DPS that returns the units, use that
+        if self._unit_dps is not None:
+            unit = validate_temp_unit(self._unit_dps.get_value(self._device))
+            # Only return valid units
+            if unit is not None:
+                return unit
+        # If there unit attribute configured in the temperature dps, use that
+        if self._temperature_dps:
+            unit = validate_temp_unit(self._temperature_dps.unit)
+            if unit is not None:
+                return unit
+        # Return the default unit from the device
+        return self._device.temperature_unit
+
+    @property
+    def current_operation(self):
+        """Return current operation ie. eco, electric, performance, ..."""
+        return self._operation_mode_dps.get_value(self._device)
+
+    @property
+    def operation_list(self):
+        """Return the list of available operation modes."""
+        if self._operation_mode_dps is None:
+            return []
+        else:
+            return self._operation_mode_dps.values(self._device)
+
+    @property
+    def current_temperature(self):
+        """Return the current temperature."""
+        if self._current_temperature_dps is None:
+            return None
+        return self._current_temperature_dps.get_value(self._device)
+
+    @property
+    def target_temperature(self):
+        """Return the temperature we try to reach."""
+        if self._temperature_dps is None:
+            raise NotImplementedError()
+        return self._temperature_dps.get_value(self._device)
+
+    @property
+    def target_temperature_step(self):
+        """Return the supported step of target temperature."""
+        dps = self._temperature_dps
+        if dps is None:
+            return 1
+        return dps.step(self._device)
+
+    async def async_set_temperature(self, **kwargs):
+        """Disable changing temperature manually to protect certain devices (e.g., Hydrotherm DYNAMIC/X8)"""
+        raise NotImplementedError()
+
+    async def async_set_operation_mode(self, operation_mode):
+        """Set new target operation mode."""
+        if self._operation_mode_dps is None:
+            raise NotImplementedError()
+        await self._operation_mode_dps.async_set_value(self._device, operation_mode)
+
+    @property
+    def min_temp(self):
+        """Return the minimum supported target temperature."""
+        # if a separate min_temperature dps is specified, the device tells us.
+        if self._mintemp_dps is not None:
+            return self._mintemp_dps.get_value(self._device)
+
+        if self._temperature_dps is None:
+            if self._temp_low_dps is None:
+                return None
+            r = self._temp_low_dps.range(self._device)
+        else:
+            r = self._temperature_dps.range(self._device)
+        return DEFAULT_MIN_TEMP if r is None else r["min"]
+
+    @property
+    def max_temp(self):
+        """Return the maximum supported target temperature."""
+        # if a separate max_temperature dps is specified, the device tells us.
+        if self._maxtemp_dps is not None:
+            return self._maxtemp_dps.get_value(self._device)
+
+        if self._temperature_dps is None:
+            if self._temp_high_dps is None:
+                return None
+            r = self._temp_high_dps.range(self._device)
+        else:
+            r = self._temperature_dps.range(self._device)
+        return DEFAULT_MAX_TEMP if r is None else r["max"]

custom_components/tuya_local/water_heater.py

@@ -0,0 +1,16 @@
+"""
+Setup for different kinds of Tuya water heater devices
+"""
+from .generic.water_heater import TuyaLocalWaterHeater
+from .helpers.config import async_tuya_setup_platform
+
+
+async def async_setup_entry(hass, config_entry, async_add_entities):
+    config = {**config_entry.data, **config_entry.options}
+    await async_tuya_setup_platform(
+        hass,
+        async_add_entities,
+        config,
+        "water_heater",
+        TuyaLocalWaterHeater,
+    )