Add vacuum platform support.

New device: Lefant M213 Vacuum Cleaner, using it.
Issue #123

custom_components/tuya_local/devices/lefant_m213_vacuum.yaml

@@ -0,0 +1,107 @@
+name: Lefant M213 Vacuum
+primary_entity:
+  entity: vacuum
+  dps:
+    - id: 1
+      type: boolean
+      name: power
+    - id: 2
+      type: boolean
+      name: activate
+    - id: 3
+      type: string
+      name: status
+      mapping:
+        - dps_val: standby
+          value: standby
+        - dps_val: smart
+          value: smart
+        - dps_val: chargego
+          value: return_to_base
+        - dps_val: random
+          value: random
+        - dps_val: wall_follow
+          value: wall_follow
+        - dps_val: spiral
+          value: clean_spot
+    - id: 4
+      type: string
+      name: direction_control
+      mapping:
+        - dps_val: backward
+          value: reverse
+        - dps_val: turn_left
+          value: left
+        - dps_val: turn_right
+          value: right
+        - dps_val: stop
+          value: stop
+    - id: 5
+      type: string
+      name: unknown_5
+    - id: 6
+      name: battery
+      type: integer
+      readonly: true
+    - id: 13
+      type: boolean
+      name: locate
+#    - id: 15
+#      type: string
+#      name: clean_record
+#      readonly: true
+    - id: 18
+      type: bitfield
+      name: error
+      mapping:
+        - dps_val: 1
+          value: edge_sweep
+        - dps_val: 2
+          value: middle_sweep
+        - dps_val: 4
+          value: left_wheel
+        - dps_val: 8
+          value: right_wheel
+        - dps_val: 16
+          value: garbage_box
+        - dps_val: 32
+          value: land_check
+        - dps_val: 64
+          value: collision
+    - id: 101
+      type: string
+      name: unknown_101
+    - id: 102
+      type: integer
+      name: unknown_102
+    - id: 103
+      type: integer
+      name: unknown_103
+    - id: 104
+      type: integer
+      name: unknown_104
+    - id: 106
+      type: string
+      name: unknown_106
+    - id: 108
+      type: string
+      name: unknown_108
+secondary_entities:
+  - entity: sensor
+    name: Clean Area
+    category: diagnostic
+    icon: "mdi:floor-plan"
+    dps:
+      - id: 16
+        type: integer
+        name: sensor
+        unit: m2
+  - entity: sensor
+    name: Clean Time
+    category: diagnostic
+    icon: "mdi:clock-outline"
+    dps:
+      - id: 17
+        type: integer
+        name: sensor
+        unit: min

custom_components/tuya_local/generic/vacuum.py

@@ -0,0 +1,153 @@
+"""
+Platform to control Tuya robot vacuums.
+"""
+from homeassistant.components.vacuum import (
+    SERVICE_CLEAN_SPOT,
+    SERVICE_RETURN_TO_BASE,
+    STATE_CLEANING,
+    STATE_DOCKED,
+    STATE_RETURNING,
+    STATE_ERROR,
+    SUPPORT_BATTERY,
+    SUPPORT_CLEAN_SPOT,
+    SUPPORT_LOCATE,
+    SUPPORT_PAUSE,
+    SUPPORT_RETURN_HOME,
+    SUPPORT_SEND_COMMAND,
+    SUPPORT_START,
+    SUPPORT_STATE,
+    SUPPORT_STATUS,
+    SUPPORT_TURN_ON,
+    SUPPORT_TURN_OFF,
+    StateVacuumEntity,
+)
+from ..device import TuyaLocalDevice
+from ..helpers.device_config import TuyaEntityConfig
+from ..helpers.mixin import TuyaLocalEntity
+
+
+class TuyaLocalVacuum(TuyaLocalEntity, StateVacuumEntity):
+    """Representation of a Tuya Vacuum Cleaner"""
+
+    def __init__(self, device: TuyaLocalDevice, config: TuyaEntityConfig):
+        """
+        Initialise the sensor.
+        Args:
+            device (TuyaLocalDevice): the device API instance.
+            config (TuyaEntityConfig): the configuration for this entity
+        """
+        dps_map = self._init_begin(device, config)
+        self._status_dps = dps_map.get("status")
+        self._locate_dps = dps_map.get("locate")
+        self._power_dps = dps_map.get("power")
+        self._active_dps = dps_map.get("activate")
+        self._battery_dps = dps_map.pop("battery", None)
+        self._direction_dps = dps_map.get("direction_control")
+        self._error_dps = dps_map.get("error")
+
+        if self._status_dps is None:
+            raise AttributeError(f"{config.name} is missing a status dps")
+        self._init_end(dps_map)
+
+    @property
+    def supported_features(self):
+        """Return the features supported by this vacuum cleaner."""
+        support = SUPPORT_STATE | SUPPORT_STATUS | SUPPORT_SEND_COMMAND
+        if self._battery_dps:
+            support |= SUPPORT_BATTERY
+        if self._power_dps:
+            support |= SUPPORT_TURN_ON | SUPPORT_TURN_OFF
+        if self._active_dps:
+            support |= SUPPORT_START | SUPPORT_PAUSE
+        if self._locate_dps:
+            support |= SUPPORT_LOCATE
+        status_support = self._status_dps.values(self._device)
+        if SERVICE_RETURN_TO_BASE in status_support:
+            support |= SUPPORT_RETURN_HOME
+        if SERVICE_CLEAN_SPOT in status_support:
+            support |= SUPPORT_CLEAN_SPOT
+        return support
+
+    @property
+    def battery_level(self):
+        """Return the battery level of the vacuum cleaner."""
+        if self._battery_dps:
+            return self._battery_dps.get_value(self._device)
+
+    @property
+    def status(self):
+        """Return the status of the vacuum cleaner."""
+        return self._status_dps.get_value(self._device)
+
+    @property
+    def state(self):
+        """Return the state of the vacuum cleaner."""
+        status = self._status_dps.get_value(self._device)
+        if self._error_dps and self._error_dps.get_value(self._device) != 0:
+            return STATE_ERROR
+        elif status == SERVICE_RETURN_TO_BASE:
+            return STATE_RETURNING
+        elif status == "standby":
+            return STATE_DOCKED
+        elif self._power_dps and not self._power_dps.get_value(self._device):
+            return STATE_DOCKED
+        elif self._active_dps and not self._active_dps.get_value(self._device):
+            return STATE_DOCKED
+        else:
+            return STATE_CLEANING
+
+    async def async_turn_on(self, **kwargs):
+        """Turn on the vacuum cleaner."""
+        if self._power_dps:
+            await self._power_dps.async_set_value(self._device, True)
+
+    async def async_turn_off(self, **kwargs):
+        """Turn off the vacuum cleaner."""
+        if self._power_dps:
+            await self._power_dps.async_set_value(self._device, False)
+
+    async def async_toggle(self, **kwargs):
+        """Toggle the vacuum cleaner."""
+        dps = self._power_dps
+        if not dps:
+            dps = self._activate_dps
+        if dps:
+            switch_to = not dps.get_value(self._device)
+            await dps.async_set_value(self._device, switch_to)
+
+    async def async_start(self):
+        if self._active_dps:
+            await self._active_dps.async_set_value(self._device, True)
+
+    async def async_pause(self):
+        """Pause the vacuum cleaner."""
+        if self._active_dps:
+            await self._active_dps.async_set_value(self._device, False)
+
+    async def async_return_to_base(self, **kwargs):
+        """Tell the vacuum cleaner to return to its base."""
+        if self._status_dps and SERVICE_RETURN_TO_BASE in self._status_dps.values(
+            self._device
+        ):
+            await self._status_dps.async_set_value(self._device, SERVICE_RETURN_TO_BASE)
+
+    async def async_clean_spot(self, **kwargs):
+        """Tell the vacuum cleaner do a spot clean."""
+        if self._status_dps and SERVICE_CLEAN_SPOT in self._status_dps.values(
+            self._device
+        ):
+            await self._status_dps.async_set_value(self._device, SERVICE_CLEAN_SPOT)
+
+    async def async_locate(self, **kwargs):
+        """Locate the vacuum cleaner."""
+        if self._locate_dps:
+            await self._locate_dps.async_set_value(self._device, True)
+
+    async def async_send_command(self, command, params=None, **kwargs):
+        """Send a command to the vacuum cleaner."""
+        if command in self._status_dps.values(self._device):
+            await self._status_dps.async_set_value(self._device, command)
+        elif self._direction_dps and command in self._direction_dps.values(
+            self._device
+        ):
+            await self._direction_dps.async_set_value(self._device, command)

custom_components/tuya_local/helpers/mixin.py

@@ -3,6 +3,7 @@ Mixins to make writing new platforms easier
 """
 import logging
 from homeassistant.const import (
+    AREA_SQUARE_METERS,
     CONCENTRATION_MICROGRAMS_PER_CUBIC_METER,
     TEMP_CELSIUS,
     TEMP_FAHRENHEIT,
@@ -78,6 +79,7 @@ UNIT_ASCII_MAP = {
     "C": TEMP_CELSIUS,
     "F": TEMP_FAHRENHEIT,
     "ugm3": CONCENTRATION_MICROGRAMS_PER_CUBIC_METER,
+    "m2": AREA_SQUARE_METERS,
 }
 
 

custom_components/tuya_local/translations/en.json

@@ -104,6 +104,8 @@
                     "sensor_air_quality": "Include air quality as a sensor entity",
 		    "sensor_balance_energy": "Include balance energy as a sensor_entity",
                     "sensor_charcoal_filter_life": "Include charcoal filter life as a sensor entity",
+		    "sensor_clean_area": "Include clean area as a sensor entity",
+		    "sensor_clean_time": "Include clean time as a sensor entity",
                     "sensor_current": "Include current as a sensor entity",
                     "sensor_current_humidity": "Include current humidity as a sensor entity",
                     "sensor_current_temperature": "Include current temperature as a sensor entity",
@@ -249,6 +251,8 @@
                     "sensor_air_quality": "Include air quality as a sensor entity",
 		    "sensor_balance_energy": "Include balance energy as a sensor_entity",
                     "sensor_charcoal_filter_life": "Include charcoal filter life as a sensor entity",
+		    "sensor_clean_area": "Include clean area as a sensor entity",
+		    "sensor_clean_time": "Include clean time as a sensor entity",
                     "sensor_current": "Include current as a sensor entity",
                     "sensor_current_humidity": "Include current humidity as a sensor entity",
                     "sensor_current_temperature": "Include current temperature as a sensor entity",

custom_components/tuya_local/vacuum.py

@@ -0,0 +1,49 @@
+"""
+Setup for different kinds of Tuya vacuum cleaners
+"""
+import logging
+
+from . import DOMAIN
+from .const import (
+    CONF_DEVICE_ID,
+    CONF_TYPE,
+)
+from .generic.vacuum import TuyaLocalVacuum
+from .helpers.device_config import get_config
+
+_LOGGER = logging.getLogger(__name__)
+
+
+async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
+    """Set up the vacuum device according to its type."""
+    data = hass.data[DOMAIN][discovery_info[CONF_DEVICE_ID]]
+    device = data["device"]
+    vacs = []
+
+    cfg = get_config(discovery_info[CONF_TYPE])
+    if cfg is None:
+        raise ValueError(f"No device config found for {discovery_info}")
+    ecfg = cfg.primary_entity
+    if ecfg.entity == "vacuum" and discovery_info.get(ecfg.config_id, False):
+        data[ecfg.config_id] = TuyaLocalVacuum(device, ecfg)
+        vacs.append(data[ecfg.config_id])
+        if ecfg.deprecated:
+            _LOGGER.warning(ecfg.deprecation_message)
+        _LOGGER.debug(f"Adding vacuum for {device.name}/{ecfg.name}")
+
+    for ecfg in cfg.secondary_entities():
+        if ecfg.entity == "vacuum" and discovery_info.get(ecfg.config_id, False):
+            data[ecfg.config_id] = TuyaLocalVacuum(device, ecfg)
+            vacs.append(data[ecfg.config_id])
+            if ecfg.deprecated:
+                _LOGGER.warning(ecfg.deprecation_message)
+            _LOGGER.debug(f"Adding vacuum for {ecfg.name}")
+
+    if not vacs:
+        raise ValueError(f"{device.name} does not support use as a vacuum device.")
+    async_add_entities(vacs)
+
+
+async def async_setup_entry(hass, config_entry, async_add_entities):
+    config = {**config_entry.data, **config_entry.options}
+    await async_setup_platform(hass, {}, async_add_entities, config)

tests/const.py

@@ -965,3 +965,22 @@ SMARTMCB_SMT006_METER_PAYLOAD = {
     "105": False,
     "106": False,
 }
+
+LEFANT_M213_VACUUM_PAYLOAD = {
+    "1": True,
+    "2": False,
+    "3": "standby",
+    "4": "forward",
+    "5": "0",
+    "6": 91,
+    "13": False,
+    "16": 0,
+    "17": 0,
+    "18": 0,
+    "101": "nar",
+    "102": -23,
+    "103": 27,
+    "104": 0,
+    "106": "ChargeStage:DETSWITCGH",
+    "108": "BatVol:13159",
+}

tests/devices/base_device_tests.py

@@ -13,6 +13,7 @@ from custom_components.tuya_local.generic.number import TuyaLocalNumber
 from custom_components.tuya_local.generic.select import TuyaLocalSelect
 from custom_components.tuya_local.generic.sensor import TuyaLocalSensor
 from custom_components.tuya_local.generic.switch import TuyaLocalSwitch
+from custom_components.tuya_local.generic.vacuum import TuyaLocalVacuum
 
 from custom_components.tuya_local.helpers.device_config import (
     TuyaDeviceConfig,
@@ -33,6 +34,7 @@ DEVICE_TYPES = {
     "switch": TuyaLocalSwitch,
     "select": TuyaLocalSelect,
     "sensor": TuyaLocalSensor,
+    "vacuum": TuyaLocalVacuum,
 }
 
 

tests/devices/test_lefant_m213_vacuum.py

@@ -0,0 +1,231 @@
+from homeassistant.components.vacuum import (
+    STATE_CLEANING,
+    STATE_DOCKED,
+    STATE_ERROR,
+    STATE_RETURNING,
+    SUPPORT_BATTERY,
+    SUPPORT_CLEAN_SPOT,
+    SUPPORT_LOCATE,
+    SUPPORT_PAUSE,
+    SUPPORT_RETURN_HOME,
+    SUPPORT_SEND_COMMAND,
+    SUPPORT_START,
+    SUPPORT_STATE,
+    SUPPORT_STATUS,
+    SUPPORT_TURN_OFF,
+    SUPPORT_TURN_ON,
+)
+from homeassistant.const import (
+    AREA_SQUARE_METERS,
+    TIME_MINUTES,
+)
+
+from ..const import LEFANT_M213_VACUUM_PAYLOAD
+from ..helpers import assert_device_properties_set
+from ..mixins.sensor import MultiSensorTests
+from .base_device_tests import TuyaDeviceTestCase
+
+POWER_DPS = "1"
+SWITCH_DPS = "2"
+STATUS_DPS = "3"
+DIRECTION_DPS = "4"
+UNKNOWN5_DPS = "5"
+BATTERY_DPS = "6"
+LOCATE_DPS = "13"
+AREA_DPS = "16"
+TIME_DPS = "17"
+ERROR_DPS = "18"
+UNKNOWN101_DPS = "101"
+UNKNOWN102_DPS = "102"
+UNKNOWN103_DPS = "103"
+UNKNOWN104_DPS = "104"
+UNKNOWN106_DPS = "106"
+UNKNOWN108_DPS = "108"
+
+
+class TestLefantM213Vacuum(MultiSensorTests, TuyaDeviceTestCase):
+    __test__ = True
+
+    def setUp(self):
+        self.setUpForConfig("lefant_m213_vacuum.yaml", LEFANT_M213_VACUUM_PAYLOAD)
+        self.subject = self.entities.get("vacuum")
+        self.setUpMultiSensors(
+            [
+                {
+                    "dps": AREA_DPS,
+                    "name": "sensor_clean_area",
+                    "unit": AREA_SQUARE_METERS,
+                },
+                {
+                    "dps": TIME_DPS,
+                    "name": "sensor_clean_time",
+                    "unit": TIME_MINUTES,
+                },
+            ],
+        )
+        self.mark_secondary(["sensor_clean_area", "sensor_clean_time"])
+
+    def test_supported_features(self):
+        self.assertEqual(
+            self.subject.supported_features,
+            SUPPORT_STATE
+            | SUPPORT_STATUS
+            | SUPPORT_SEND_COMMAND
+            | SUPPORT_BATTERY
+            | SUPPORT_TURN_ON
+            | SUPPORT_TURN_OFF
+            | SUPPORT_START
+            | SUPPORT_PAUSE
+            | SUPPORT_LOCATE
+            | SUPPORT_RETURN_HOME
+            | SUPPORT_CLEAN_SPOT,
+        )
+
+    def test_battery_level(self):
+        self.dps[BATTERY_DPS] = 50
+        self.assertEqual(self.subject.battery_level, 50)
+
+    def test_status(self):
+        self.dps[STATUS_DPS] = "standby"
+        self.assertEqual(self.subject.status, "standby")
+        self.dps[STATUS_DPS] = "smart"
+        self.assertEqual(self.subject.status, "smart")
+        self.dps[STATUS_DPS] = "chargego"
+        self.assertEqual(self.subject.status, "return_to_base")
+        self.dps[STATUS_DPS] = "random"
+        self.assertEqual(self.subject.status, "random")
+        self.dps[STATUS_DPS] = "wall_follow"
+        self.assertEqual(self.subject.status, "wall_follow")
+        self.dps[STATUS_DPS] = "spiral"
+        self.assertEqual(self.subject.status, "clean_spot")
+
+    def test_state(self):
+        self.dps[POWER_DPS] = True
+        self.dps[SWITCH_DPS] = True
+        self.dps[ERROR_DPS] = 0
+        self.dps[STATUS_DPS] = "return_to_base"
+        self.assertEqual(self.subject.state, STATE_RETURNING)
+        self.dps[STATUS_DPS] = "standby"
+        self.assertEqual(self.subject.state, STATE_DOCKED)
+        self.dps[STATUS_DPS] = "random"
+        self.assertEqual(self.subject.state, STATE_CLEANING)
+        self.dps[POWER_DPS] = False
+        self.assertEqual(self.subject.state, STATE_DOCKED)
+        self.dps[POWER_DPS] = True
+        self.dps[SWITCH_DPS] = False
+        self.assertEqual(self.subject.state, STATE_DOCKED)
+        self.dps[ERROR_DPS] = 1
+        self.assertEqual(self.subject.state, STATE_ERROR)
+
+    async def test_async_turn_on(self):
+        async with assert_device_properties_set(
+            self.subject._device,
+            {POWER_DPS: True},
+        ):
+            await self.subject.async_turn_on()
+
+    async def test_async_turn_off(self):
+        async with assert_device_properties_set(
+            self.subject._device,
+            {POWER_DPS: False},
+        ):
+            await self.subject.async_turn_off()
+
+    async def test_async_toggle(self):
+        self.dps[POWER_DPS] = False
+        async with assert_device_properties_set(
+            self.subject._device,
+            {POWER_DPS: True},
+        ):
+            await self.subject.async_toggle()
+
+    async def test_async_start(self):
+        async with assert_device_properties_set(
+            self.subject._device,
+            {SWITCH_DPS: True},
+        ):
+            await self.subject.async_start()
+
+    async def test_async_pause(self):
+        async with assert_device_properties_set(
+            self.subject._device,
+            {SWITCH_DPS: False},
+        ):
+            await self.subject.async_pause()
+
+    async def test_async_return_to_base(self):
+        async with assert_device_properties_set(
+            self.subject._device,
+            {STATUS_DPS: "chargego"},
+        ):
+            await self.subject.async_return_to_base()
+
+    async def test_async_clean_spot(self):
+        async with assert_device_properties_set(
+            self.subject._device,
+            {STATUS_DPS: "spiral"},
+        ):
+            await self.subject.async_clean_spot()
+
+    async def test_async_locate(self):
+        async with assert_device_properties_set(
+            self.subject._device,
+            {LOCATE_DPS: True},
+        ):
+            await self.subject.async_locate()
+
+    async def test_async_send_standby_command(self):
+        async with assert_device_properties_set(
+            self.subject._device,
+            {STATUS_DPS: "standby"},
+        ):
+            await self.subject.async_send_command("standby")
+
+    async def test_async_send_smart_command(self):
+        async with assert_device_properties_set(
+            self.subject._device,
+            {STATUS_DPS: "smart"},
+        ):
+            await self.subject.async_send_command("smart")
+
+    async def test_async_send_random_command(self):
+        async with assert_device_properties_set(
+            self.subject._device,
+            {STATUS_DPS: "random"},
+        ):
+            await self.subject.async_send_command("random")
+
+    async def test_async_send_wall_follow_command(self):
+        async with assert_device_properties_set(
+            self.subject._device,
+            {STATUS_DPS: "wall_follow"},
+        ):
+            await self.subject.async_send_command("wall_follow")
+
+    async def test_async_send_reverse_command(self):
+        async with assert_device_properties_set(
+            self.subject._device,
+            {DIRECTION_DPS: "backward"},
+        ):
+            await self.subject.async_send_command("reverse")
+
+    async def test_async_send_left_command(self):
+        async with assert_device_properties_set(
+            self.subject._device,
+            {DIRECTION_DPS: "turn_left"},
+        ):
+            await self.subject.async_send_command("left")
+
+    async def test_async_send_right_command(self):
+        async with assert_device_properties_set(
+            self.subject._device,
+            {DIRECTION_DPS: "turn_right"},
+        ):
+            await self.subject.async_send_command("right")
+
+    async def test_async_send_stop_command(self):
+        async with assert_device_properties_set(
+            self.subject._device,
+            {DIRECTION_DPS: "stop"},
+        ):
+            await self.subject.async_send_command("stop")

tests/test_vacuum.py

@@ -0,0 +1,79 @@
+"""Tests for the vacuum entity."""
+from pytest_homeassistant_custom_component.common import MockConfigEntry
+from unittest.mock import AsyncMock, Mock
+from custom_components.tuya_local.const import (
+    CONF_DEVICE_ID,
+    CONF_TYPE,
+    DOMAIN,
+)
+from custom_components.tuya_local.generic.vacuum import TuyaLocalVacuum
+from custom_components.tuya_local.vacuum import async_setup_entry
+
+
+async def test_init_entry(hass):
+    """Test the initialisation."""
+    entry = MockConfigEntry(
+        domain=DOMAIN,
+        data={
+            CONF_TYPE: "lefant_m213_vacuum",
+            CONF_DEVICE_ID: "dummy",
+            "vacuum": True,
+        },
+    )
+    m_add_entities = Mock()
+    m_device = AsyncMock()
+
+    hass.data[DOMAIN] = {
+        "dummy": {
+            "device": m_device,
+        },
+    }
+
+    await async_setup_entry(hass, entry, m_add_entities)
+    assert type(hass.data[DOMAIN]["dummy"]["vacuum"]) == TuyaLocalVacuum
+    m_add_entities.assert_called_once()
+
+
+async def test_init_entry_fails_if_device_has_no_vacuum(hass):
+    """Test initialisation when device has no matching entity"""
+    entry = MockConfigEntry(
+        domain=DOMAIN,
+        data={CONF_TYPE: "kogan_heater", CONF_DEVICE_ID: "dummy", "vacuum": True},
+    )
+    m_add_entities = Mock()
+    m_device = AsyncMock()
+
+    hass.data[DOMAIN] = {
+        "dummy": {
+            "device": m_device,
+        },
+    }
+    try:
+        await async_setup_entry(hass, entry, m_add_entities)
+        assert False
+    except ValueError:
+        pass
+    m_add_entities.assert_not_called()
+
+
+async def test_init_entry_fails_if_config_is_missing(hass):
+    """Test initialisation when device has no matching entity"""
+    entry = MockConfigEntry(
+        domain=DOMAIN,
+        data={CONF_TYPE: "non_existing", CONF_DEVICE_ID: "dummy", "vacuum": True},
+    )
+    # although async, the async_add_entities function passed to
+    # async_setup_entry is called truly asynchronously. If we use
+    # AsyncMock, it expects us to await the result.
+    m_add_entities = Mock()
+    m_device = AsyncMock()
+
+    hass.data[DOMAIN] = {}
+    hass.data[DOMAIN]["dummy"] = {}
+    hass.data[DOMAIN]["dummy"]["device"] = m_device
+    try:
+        await async_setup_entry(hass, entry, m_add_entities)
+        assert False
+    except ValueError:
+        pass
+    m_add_entities.assert_not_called()