Add VacuumDeviceStatus and VacuumState (#1560)

This provides a common, vendor agnostic API for obtaining vacuum state
information,
to be used by downstreams like homeassistant.
This also converts roborock vacuum to use these common facilities.

Added common API:
* `VacuumDeviceStatus` is an interface extending `DeviceStatus` to
enforce common API for devices implementing `VacuumInterface`.
* `VacuumState` provides a generic interface to map different vacuum
statuses.

These interfaces are bound to change and are introduced to make it
simpler to make other vacuum integrations available to downstream users.
This is related to #1495.

.pre-commit-config.yaml

@@ -51,7 +51,7 @@ repos:
   rev: v0.961
   hooks:
   - id: mypy
-    additional_dependencies: [types-attrs, types-PyYAML, types-requests, types-pytz, types-croniter]
+    additional_dependencies: [types-attrs, types-PyYAML, types-requests, types-pytz, types-croniter, types-freezegun]
 
 - repo: https://github.com/asottile/pyupgrade
   rev: v2.37.1

miio/integrations/vacuum/roborock/vacuumcontainers.py

@@ -7,14 +7,55 @@
 
 from miio.device import DeviceStatus
 from miio.devicestatus import sensor, setting
+from miio.interfaces.vacuuminterface import VacuumDeviceStatus, VacuumState
 from miio.utils import pretty_seconds, pretty_time
 
 
 def pretty_area(x: float) -> float:
     return int(x) / 1000000
 
 
-error_codes = {  # from vacuum_cleaner-EN.pdf
+STATE_CODE_TO_STRING = {
+    1: "Starting",
+    2: "Charger disconnected",
+    3: "Idle",
+    4: "Remote control active",
+    5: "Cleaning",
+    6: "Returning home",
+    7: "Manual mode",
+    8: "Charging",
+    9: "Charging problem",
+    10: "Paused",
+    11: "Spot cleaning",
+    12: "Error",
+    13: "Shutting down",
+    14: "Updating",
+    15: "Docking",
+    16: "Going to target",
+    17: "Zoned cleaning",
+    18: "Segment cleaning",
+    22: "Emptying the bin",  # on s7+, see #1189
+    23: "Washing the mop",  # on a46, #1435
+    26: "Going to wash the mop",  # on a46, #1435
+    100: "Charging complete",
+    101: "Device offline",
+}
+
+VACUUMSTATE_TO_STATE_CODES = {
+    VacuumState.Idle: [1, 2, 3, 13],
+    VacuumState.Paused: [10],
+    VacuumState.Cleaning: [4, 5, 7, 11, 16, 17, 18],
+    VacuumState.Docked: [8, 14, 22, 100],
+    VacuumState.Returning: [6, 15],
+    VacuumState.Error: [9, 12, 101],
+}
+STATE_CODE_TO_VACUUMSTATE = {}
+for state, codes in VACUUMSTATE_TO_STATE_CODES.items():
+    for code in codes:
+        STATE_CODE_TO_VACUUMSTATE[code] = state
+
+
+ERROR_CODES = {  # from vacuum_cleaner-EN.pdf
     0: "No error",
     1: "Laser distance sensor error",
     2: "Collision sensor error",
@@ -42,7 +83,7 @@ def pretty_area(x: float) -> float:
 }
 
 
-class VacuumStatus(DeviceStatus):
+class VacuumStatus(VacuumDeviceStatus):
     """Container for status reports from the vacuum."""
 
     def __init__(self, data: Dict[str, Any]) -> None:
@@ -94,38 +135,17 @@ def state_code(self) -> int:
         return int(self.data["state"])
 
     @property
-    @sensor("State")
+    @sensor("State message")
     def state(self) -> str:
         """Human readable state description, see also :func:`state_code`."""
-        states = {
-            1: "Starting",
-            2: "Charger disconnected",
-            3: "Idle",
-            4: "Remote control active",
-            5: "Cleaning",
-            6: "Returning home",
-            7: "Manual mode",
-            8: "Charging",
-            9: "Charging problem",
-            10: "Paused",
-            11: "Spot cleaning",
-            12: "Error",
-            13: "Shutting down",
-            14: "Updating",
-            15: "Docking",
-            16: "Going to target",
-            17: "Zoned cleaning",
-            18: "Segment cleaning",
-            22: "Emptying the bin",  # on s7+, see #1189
-            23: "Washing the mop",  # on a46, #1435
-            26: "Going to wash the mop",  # on a46, #1435
-            100: "Charging complete",
-            101: "Device offline",
-        }
-        try:
-            return states[int(self.state_code)]
-        except KeyError:
-            return "Definition missing for state %s" % self.state_code
+        return STATE_CODE_TO_STRING.get(
+            self.state_code, f"Unknown state (code: {self.state_code})"
+        )
+
+    @sensor("Vacuum state")
+    def vacuum_state(self) -> VacuumState:
+        """Return vacuum state."""
+        return STATE_CODE_TO_VACUUMSTATE.get(self.state_code, VacuumState.Unknown)
 
     @property
     @sensor("Error Code", icon="mdi:alert")
@@ -138,7 +158,7 @@ def error_code(self) -> int:
     def error(self) -> str:
         """Human readable error description, see also :func:`error_code`."""
         try:
-            return error_codes[self.error_code]
+            return ERROR_CODES[self.error_code]
         except KeyError:
             return "Definition missing for error %s" % self.error_code
 
@@ -342,7 +362,7 @@ def error_code(self) -> int:
     @property
     def error(self) -> str:
         """Error state of this cleaning run."""
-        return error_codes[self.data["error"]]
+        return ERROR_CODES[self.data["error"]]
 
     @property
     def complete(self) -> bool:

miio/integrations/viomidishwasher/test_viomidishwasher.py

@@ -2,6 +2,7 @@
 from unittest import TestCase
 
 import pytest
+from freezegun import freeze_time
 
 from miio import ViomiDishwasher
 from miio.tests.dummies import DummyDevice
@@ -145,6 +146,7 @@ def test_program(self):
         self.device.start(Program.Intensive)
         assert self.state().program == Program.Intensive
 
+    @freeze_time()
     def test_schedule(self):
         self.device.on()  # ensure on
         assert self.is_on() is True

miio/interfaces/vacuuminterface.py

@@ -1,12 +1,46 @@
 """`VacuumInterface` is an interface (abstract class) with shared API for all vacuum
 devices."""
 from abc import abstractmethod
-from typing import Dict
+from enum import Enum, auto
+from typing import Dict, Optional
+
+from miio import DeviceStatus
 
 # Dictionary of predefined fan speeds
 FanspeedPresets = Dict[str, int]
 
 
+class VacuumState(Enum):
+    """Vacuum state enum.
+
+    This offers a simplified API to the vacuum state.
+    """
+
+    Unknown = auto()
+    Cleaning = auto()
+    Returning = auto()
+    Idle = auto()
+    Docked = auto()
+    Paused = auto()
+    Error = auto()
+
+
+class VacuumDeviceStatus(DeviceStatus):
+    """Status container for vacuums."""
+
+    @abstractmethod
+    def vacuum_state(self) -> VacuumState:
+        """Return vacuum state."""
+
+    @abstractmethod
+    def error(self) -> Optional[str]:
+        """Return error message, if errored."""
+
+    @abstractmethod
+    def battery(self) -> Optional[int]:
+        """Return current battery charge, if available."""
+
+
 class VacuumInterface:
     """Vacuum API interface."""