python-kasa/kasa/smart/modules/clean.py

"""Implementation of vacuum clean module."""

from __future__ import annotations

import logging
from datetime import timedelta
from enum import IntEnum, StrEnum
from typing import Annotated, Literal

from ...feature import Feature
from ...module import FeatureAttribute
from ..smartmodule import SmartModule

_LOGGER = logging.getLogger(__name__)


class Status(IntEnum):
    """Status of vacuum."""

    Idle = 0
    Cleaning = 1
    Mapping = 2
    GoingHome = 4
    Charging = 5
    Charged = 6
    Paused = 7
    Undocked = 8
    Error = 100

    UnknownInternal = -1000


class ErrorCode(IntEnum):
    """Error codes for vacuum."""

    Ok = 0
    SideBrushStuck = 2
    MainBrushStuck = 3
    WheelBlocked = 4
    Trapped = 6
    TrappedCliff = 7
    DustBinRemoved = 14
    UnableToMove = 15
    LidarBlocked = 16
    UnableToFindDock = 21
    BatteryLow = 22

    UnknownInternal = -1000


class FanSpeed(IntEnum):
    """Fan speed level."""

    Quiet = 1
    Standard = 2
    Turbo = 3
    Max = 4
    Ultra = 5


class CarpetCleanMode(StrEnum):
    """Carpet clean mode."""

    Normal = "normal"
    Boost = "boost"


class AreaUnit(IntEnum):
    """Area unit."""

    #: Square meter
    Sqm = 0
    #: Square feet
    Sqft = 1
    #: Taiwanese unit: https://en.wikipedia.org/wiki/Taiwanese_units_of_measurement#Area
    Ping = 2


class Clean(SmartModule):
    """Implementation of vacuum clean module."""

    REQUIRED_COMPONENT = "clean"
    _error_code = ErrorCode.Ok
    _logged_error_code_warnings: set | None = None
    _logged_status_code_warnings: set

    def _initialize_features(self) -> None:
        """Initialize features."""
        self._add_feature(
            Feature(
                self._device,
                id="vacuum_return_home",
                name="Return home",
                container=self,
                attribute_setter="return_home",
                category=Feature.Category.Primary,
                type=Feature.Action,
            )
        )
        self._add_feature(
            Feature(
                self._device,
                id="vacuum_start",
                name="Start cleaning",
                container=self,
                attribute_setter="start",
                category=Feature.Category.Primary,
                type=Feature.Action,
            )
        )
        self._add_feature(
            Feature(
                self._device,
                id="vacuum_pause",
                name="Pause",
                container=self,
                attribute_setter="pause",
                category=Feature.Category.Primary,
                type=Feature.Action,
            )
        )
        self._add_feature(
            Feature(
                self._device,
                id="vacuum_status",
                name="Vacuum status",
                container=self,
                attribute_getter="status",
                category=Feature.Category.Primary,
                type=Feature.Type.Sensor,
            )
        )
        self._add_feature(
            Feature(
                self._device,
                id="vacuum_error",
                name="Error",
                container=self,
                attribute_getter="error",
                category=Feature.Category.Info,
                type=Feature.Type.Sensor,
            )
        )
        self._add_feature(
            Feature(
                self._device,
                id="battery_level",
                name="Battery level",
                container=self,
                attribute_getter="battery",
                icon="mdi:battery",
                unit_getter=lambda: "%",
                category=Feature.Category.Info,
                type=Feature.Type.Sensor,
            )
        )
        self._add_feature(
            Feature(
                self._device,
                id="vacuum_fan_speed",
                name="Fan speed",
                container=self,
                attribute_getter="fan_speed_preset",
                attribute_setter="set_fan_speed_preset",
                icon="mdi:fan",
                choices_getter=lambda: list(FanSpeed.__members__),
                category=Feature.Category.Primary,
                type=Feature.Type.Choice,
            )
        )
        self._add_feature(
            Feature(
                self._device,
                id="clean_count",
                name="Clean count",
                container=self,
                attribute_getter="clean_count",
                attribute_setter="set_clean_count",
                range_getter=lambda: (1, 3),
                category=Feature.Category.Config,
                type=Feature.Type.Number,
            )
        )
        self._add_feature(
            Feature(
                self._device,
                id="carpet_clean_mode",
                name="Carpet clean mode",
                container=self,
                attribute_getter="carpet_clean_mode",
                attribute_setter="set_carpet_clean_mode",
                icon="mdi:rug",
                choices_getter=lambda: list(CarpetCleanMode.__members__),
                category=Feature.Category.Config,
                type=Feature.Type.Choice,
            )
        )
        self._add_feature(
            Feature(
                self._device,
                id="clean_area",
                name="Cleaning area",
                container=self,
                attribute_getter="clean_area",
                unit_getter="area_unit",
                category=Feature.Category.Info,
                type=Feature.Type.Sensor,
            )
        )
        self._add_feature(
            Feature(
                self._device,
                id="clean_time",
                name="Cleaning time",
                container=self,
                attribute_getter="clean_time",
                category=Feature.Category.Info,
                type=Feature.Type.Sensor,
            )
        )
        self._add_feature(
            Feature(
                self._device,
                id="clean_progress",
                name="Cleaning progress",
                container=self,
                attribute_getter="clean_progress",
                unit_getter=lambda: "%",
                category=Feature.Category.Info,
                type=Feature.Type.Sensor,
            )
        )

    async def _post_update_hook(self) -> None:
        """Set error code after update."""
        if self._logged_error_code_warnings is None:
            self._logged_error_code_warnings = set()
            self._logged_status_code_warnings = set()

        errors = self._vac_status.get("err_status")
        if errors is None or not errors:
            self._error_code = ErrorCode.Ok
            return

        if len(errors) > 1 and "multiple" not in self._logged_error_code_warnings:
            self._logged_error_code_warnings.add("multiple")
            _LOGGER.warning(
                "Multiple error codes, using the first one only: %s", errors
            )

        error = errors.pop(0)
        try:
            self._error_code = ErrorCode(error)
        except ValueError:
            if error not in self._logged_error_code_warnings:
                self._logged_error_code_warnings.add(error)
                _LOGGER.warning(
                    "Unknown error code, please create an issue "
                    "describing the error: %s",
                    error,
                )
            self._error_code = ErrorCode.UnknownInternal

    def query(self) -> dict:
        """Query to execute during the update cycle."""
        return {
            "getVacStatus": {},
            "getCleanInfo": {},
            "getCarpetClean": {},
            "getAreaUnit": {},
            "getBatteryInfo": {},
            "getCleanStatus": {},
            "getCleanAttr": {"type": "global"},
        }

    async def start(self) -> dict:
        """Start cleaning."""
        # If we are paused, do not restart cleaning

        if self.status is Status.Paused:
            return await self.resume()

        return await self.call(
            "setSwitchClean",
            {
                "clean_mode": 0,
                "clean_on": True,
                "clean_order": True,
                "force_clean": False,
            },
        )

    async def pause(self) -> dict:
        """Pause cleaning."""
        if self.status is Status.GoingHome:
            return await self.set_return_home(False)

        return await self.set_pause(True)

    async def resume(self) -> dict:
        """Resume cleaning."""
        return await self.set_pause(False)

    async def set_pause(self, enabled: bool) -> dict:
        """Pause or resume cleaning."""
        return await self.call("setRobotPause", {"pause": enabled})

    async def return_home(self) -> dict:
        """Return home."""
        return await self.set_return_home(True)

    async def set_return_home(self, enabled: bool) -> dict:
        """Return home / pause returning."""
        return await self.call("setSwitchCharge", {"switch_charge": enabled})

    @property
    def error(self) -> ErrorCode:
        """Return error."""
        return self._error_code

    @property
    def fan_speed_preset(self) -> Annotated[str, FeatureAttribute()]:
        """Return fan speed preset."""
        return FanSpeed(self._settings["suction"]).name

    async def set_fan_speed_preset(
        self, speed: str
    ) -> Annotated[dict, FeatureAttribute]:
        """Set fan speed preset."""
        name_to_value = {x.name: x.value for x in FanSpeed}
        if speed not in name_to_value:
            raise ValueError("Invalid fan speed %s, available %s", speed, name_to_value)
        return await self._change_setting("suction", name_to_value[speed])

    async def _change_setting(
        self, name: str, value: int, *, scope: Literal["global", "pose"] = "global"
    ) -> dict:
        """Change device setting."""
        params = {
            name: value,
            "type": scope,
        }
        return await self.call("setCleanAttr", params)

    @property
    def battery(self) -> int:
        """Return battery level."""
        return self.data["getBatteryInfo"]["battery_percentage"]

    @property
    def _vac_status(self) -> dict:
        """Return vac status container."""
        return self.data["getVacStatus"]

    @property
    def _info(self) -> dict:
        """Return current cleaning info."""
        return self.data["getCleanInfo"]

    @property
    def _settings(self) -> dict:
        """Return cleaning settings."""
        return self.data["getCleanAttr"]

    @property
    def status(self) -> Status:
        """Return current status."""
        if self._error_code is not ErrorCode.Ok:
            return Status.Error

        status_code = self._vac_status["status"]
        try:
            return Status(status_code)
        except ValueError:
            if status_code not in self._logged_status_code_warnings:
                self._logged_status_code_warnings.add(status_code)
                _LOGGER.warning(
                    "Got unknown status code: %s (%s)", status_code, self.data
                )
            return Status.UnknownInternal

    @property
    def carpet_clean_mode(self) -> Annotated[str, FeatureAttribute()]:
        """Return carpet clean mode."""
        return CarpetCleanMode(self.data["getCarpetClean"]["carpet_clean_prefer"]).name

    async def set_carpet_clean_mode(
        self, mode: str
    ) -> Annotated[dict, FeatureAttribute()]:
        """Set carpet clean mode."""
        name_to_value = {x.name: x.value for x in CarpetCleanMode}
        if mode not in name_to_value:
            raise ValueError(
                "Invalid carpet clean mode %s, available %s", mode, name_to_value
            )
        return await self.call(
            "setCarpetClean", {"carpet_clean_prefer": name_to_value[mode]}
        )

    @property
    def area_unit(self) -> AreaUnit:
        """Return area unit."""
        return AreaUnit(self.data["getAreaUnit"]["area_unit"])

    @property
    def clean_area(self) -> Annotated[int, FeatureAttribute()]:
        """Return currently cleaned area."""
        return self._info["clean_area"]

    @property
    def clean_time(self) -> timedelta:
        """Return current cleaning time."""
        return timedelta(minutes=self._info["clean_time"])

    @property
    def clean_progress(self) -> int:
        """Return amount of currently cleaned area."""
        return self._info["clean_percent"]

    @property
    def clean_count(self) -> Annotated[int, FeatureAttribute()]:
        """Return number of times to clean."""
        return self._settings["clean_number"]

    async def set_clean_count(self, count: int) -> Annotated[dict, FeatureAttribute()]:
        """Set number of times to clean."""
        return await self._change_setting("clean_number", count)