Add vacuum platform to Tuya (#57996)

.coveragerc

@@ -1123,6 +1123,7 @@ omit =
     homeassistant/components/tuya/sensor.py
     homeassistant/components/tuya/siren.py
     homeassistant/components/tuya/switch.py
+    homeassistant/components/tuya/vacuum.py
     homeassistant/components/twentemilieu/const.py
     homeassistant/components/twentemilieu/sensor.py
     homeassistant/components/twilio_call/notify.py

homeassistant/components/tuya/const.py

@@ -107,6 +107,7 @@ TUYA_SUPPORTED_PRODUCT_CATEGORIES = (
     "pc",  # Power Strip
     "pir",  # PIR Detector
     "qn",  # Heater
+    "sd",  # Robot vacuum
     "sgbj",  # Siren Alarm
     "sos",  # SOS Button
     "sp",  # Smart Camera
@@ -133,6 +134,7 @@ PLATFORMS = [
     "sensor",
     "siren",
     "switch",
+    "vacuum",
 ]
 
 
@@ -175,6 +177,7 @@ class DPCode(str, Enum):
     CUR_POWER = "cur_power"  # Actual power
     CUR_VOLTAGE = "cur_voltage"  # Actual voltage
     DOORCONTACT_STATE = "doorcontact_state"  # Status of door window sensor
+    ELECTRICITY_LEFT = "electricity_left"
     FAN_DIRECTION = "fan_direction"  # Fan direction
     FAN_SPEED_ENUM = "fan_speed_enum"  # Speed mode
     FAN_SPEED_PERCENT = "fan_speed_percent"  # Stepless speed
@@ -188,10 +191,13 @@ class DPCode(str, Enum):
     MODE = "mode"  # Working mode / Mode
     MOTION_SWITCH = "motion_switch"  # Motion switch
     MUFFLING = "muffling"  # Muffling
+    PAUSE = "pause"
     PIR = "pir"  # Motion sensor
     POWDER_SET = "powder_set"  # Powder
+    POWER_GO = "power_go"
     PUMP_RESET = "pump_reset"  # Water pump reset
     RECORD_SWITCH = "record_switch"  # Recording switch
+    SEEK = "seek"
     SENSITIVITY = "sensitivity"  # Sensitivity
     SHAKE = "shake"  # Oscillating
     SHOCK_STATE = "shock_state"  # Vibration status
@@ -199,6 +205,8 @@ class DPCode(str, Enum):
     SOS_STATE = "sos_state"  # Emergency mode
     SPEED = "speed"  # Speed level
     START = "start"  # Start
+    STATUS = "status"
+    SUCTION = "suction"
     SWING = "swing"  # Swing mode
     SWITCH = "switch"  # Switch
     SWITCH_1 = "switch_1"  # Switch 1
@@ -208,6 +216,7 @@ class DPCode(str, Enum):
     SWITCH_5 = "switch_5"  # Switch 5
     SWITCH_6 = "switch_6"  # Switch 6
     SWITCH_BACKLIGHT = "switch_backlight"  # Backlight switch
+    SWITCH_CHARGE = "switch_charge"
     SWITCH_CONTROLLER = "switch_controller"
     SWITCH_HORIZONTAL = "switch_horizontal"  # Horizontal swing flap switch
     SWITCH_LED = "switch_led"  # Switch

homeassistant/components/tuya/vacuum.py

@@ -0,0 +1,167 @@
+"""Support for Tuya Vacuums."""
+from __future__ import annotations
+
+from typing import Any
+
+from tuya_iot import TuyaDevice, TuyaDeviceManager
+
+from homeassistant.components.vacuum import (
+    STATE_CLEANING,
+    STATE_DOCKED,
+    STATE_IDLE,
+    STATE_PAUSED,
+    STATE_RETURNING,
+    SUPPORT_BATTERY,
+    SUPPORT_FAN_SPEED,
+    SUPPORT_PAUSE,
+    SUPPORT_RETURN_HOME,
+    SUPPORT_START,
+    SUPPORT_STATE,
+    SUPPORT_STATUS,
+    SUPPORT_STOP,
+    StateVacuumEntity,
+)
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant, callback
+from homeassistant.helpers.dispatcher import async_dispatcher_connect
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+
+from . import HomeAssistantTuyaData
+from .base import EnumTypeData, IntegerTypeData, TuyaEntity
+from .const import DOMAIN, TUYA_DISCOVERY_NEW, DPCode
+
+TUYA_STATUS_TO_HA = {
+    "charge_done": STATE_DOCKED,
+    "chargecompleted": STATE_DOCKED,
+    "charging": STATE_DOCKED,
+    "cleaning": STATE_CLEANING,
+    "docking": STATE_RETURNING,
+    "goto_charge": STATE_RETURNING,
+    "goto_pos": STATE_CLEANING,
+    "mop_clean": STATE_CLEANING,
+    "part_clean": STATE_CLEANING,
+    "paused": STATE_PAUSED,
+    "pick_zone_clean": STATE_CLEANING,
+    "pos_arrived": STATE_CLEANING,
+    "pos_unarrive": STATE_CLEANING,
+    "sleep": STATE_IDLE,
+    "smart_clean": STATE_CLEANING,
+    "spot_clean": STATE_CLEANING,
+    "standby": STATE_IDLE,
+    "wall_clean": STATE_CLEANING,
+    "zone_clean": STATE_CLEANING,
+}
+
+
+async def async_setup_entry(
+    hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
+) -> None:
+    """Set up Tuya vacuum dynamically through Tuya discovery."""
+    hass_data: HomeAssistantTuyaData = hass.data[DOMAIN][entry.entry_id]
+
+    @callback
+    def async_discover_device(device_ids: list[str]) -> None:
+        """Discover and add a discovered Tuya vacuum."""
+        entities: list[TuyaVacuumEntity] = []
+        for device_id in device_ids:
+            device = hass_data.device_manager.device_map[device_id]
+            if device.category == "sd":
+                entities.append(TuyaVacuumEntity(device, hass_data.device_manager))
+        async_add_entities(entities)
+
+    async_discover_device([*hass_data.device_manager.device_map])
+
+    entry.async_on_unload(
+        async_dispatcher_connect(hass, TUYA_DISCOVERY_NEW, async_discover_device)
+    )
+
+
+class TuyaVacuumEntity(TuyaEntity, StateVacuumEntity):
+    """Tuya Vacuum Device."""
+
+    _fan_speed_type: EnumTypeData | None = None
+    _battery_level_type: IntegerTypeData | None = None
+    _supported_features = 0
+
+    def __init__(self, device: TuyaDevice, device_manager: TuyaDeviceManager) -> None:
+        """Init Tuya vacuum."""
+        super().__init__(device, device_manager)
+
+        if DPCode.PAUSE in self.device.status:
+            self._supported_features |= SUPPORT_PAUSE
+
+        if DPCode.SWITCH_CHARGE in self.device.status:
+            self._supported_features |= SUPPORT_RETURN_HOME
+
+        if DPCode.STATUS in self.device.status:
+            self._supported_features |= SUPPORT_STATE | SUPPORT_STATUS
+
+        if DPCode.POWER_GO in self.device.status:
+            self._supported_features |= SUPPORT_STOP | SUPPORT_START
+
+        if function := device.function.get(DPCode.SUCTION):
+            self._supported_features |= SUPPORT_FAN_SPEED
+            self._fan_speed_type = EnumTypeData.from_json(function.values)
+
+        if function := device.function.get(DPCode.ELECTRICITY_LEFT):
+            self._supported_features |= SUPPORT_BATTERY
+            self._battery_level_type = IntegerTypeData.from_json(function.values)
+
+    @property
+    def battery_level(self) -> int | None:
+        """Return Tuya device state."""
+        if self._battery_level_type is None or not (
+            status := self.device.status.get(DPCode.ELECTRICITY_LEFT)
+        ):
+            return None
+        return round(self._battery_level_type.scale_value(status))
+
+    @property
+    def fan_speed(self) -> str | None:
+        """Return the fan speed of the vacuum cleaner."""
+        return self.device.status.get(DPCode.SUCTION)
+
+    @property
+    def fan_speed_list(self) -> list[str]:
+        """Get the list of available fan speed steps of the vacuum cleaner."""
+        if self._fan_speed_type is None:
+            return []
+        return self._fan_speed_type.range
+
+    @property
+    def state(self) -> str | None:
+        """Return Tuya vacuum device state."""
+        if self.device.status.get(DPCode.PAUSE):
+            return STATE_PAUSED
+        if not (status := self.device.status.get(DPCode.STATUS)):
+            return None
+        return TUYA_STATUS_TO_HA.get(status)
+
+    @property
+    def supported_features(self) -> int:
+        """Flag supported features."""
+        return self._supported_features
+
+    def start(self, **kwargs: Any) -> None:
+        """Turn the device on."""
+        self._send_command([{"code": DPCode.POWER_GO, "value": True}])
+
+    def stop(self, **kwargs: Any) -> None:
+        """Turn the device off."""
+        self._send_command([{"code": DPCode.POWER_GO, "value": False}])
+
+    def pause(self, **kwargs: Any) -> None:
+        """Pause the device."""
+        self._send_command([{"code": DPCode.POWER_GO, "value": True}])
+
+    def return_to_base(self, **kwargs: Any) -> None:
+        """Return device to dock."""
+        self._send_command([{"code": DPCode.MODE, "value": "chargego"}])
+
+    def locate(self, **kwargs: Any) -> None:
+        """Return device to dock."""
+        self._send_command([{"code": DPCode.SEEK, "value": True}])
+
+    def set_fan_speed(self, fan_speed: str, **kwargs: Any) -> None:
+        """Set fan speed."""
+        self._send_command([{"code": DPCode.SUCTION, "value": fan_speed}])