Add XML command "GetLifeSpan"

deebot_client/commands/xml/__init__.py

@@ -9,6 +9,7 @@
 from .charge_state import GetChargeState
 from .error import GetError
 from .fan_speed import GetFanSpeed
+from .life_span import GetLifeSpan
 from .pos import GetPos
 from .stats import GetCleanSum
 
@@ -20,13 +21,15 @@
     "GetCleanSum",
     "GetError",
     "GetFanSpeed",
+    "GetLifeSpan",
     "GetPos",
 ]
 
 # fmt: off
 # ordered by file asc
 _COMMANDS: list[type[XmlCommand]] = [
     GetError,
+    GetLifeSpan,
 ]
 # fmt: on
 

deebot_client/commands/xml/life_span.py

@@ -0,0 +1,49 @@
+"""Life span commands."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from deebot_client.events import LifeSpan, LifeSpanEvent
+from deebot_client.message import HandlingResult
+
+from .common import XmlCommandWithMessageHandling
+
+if TYPE_CHECKING:
+    from xml.etree.ElementTree import Element
+
+    from deebot_client.event_bus import EventBus
+
+
+class GetLifeSpan(XmlCommandWithMessageHandling):
+    """GetLifeSpan command."""
+
+    NAME = "GetLifeSpan"
+
+    def __init__(self, life_span: LifeSpan) -> None:
+        super().__init__({"type": life_span.xml_value})
+
+    @classmethod
+    def _handle_xml(cls, event_bus: EventBus, xml: Element) -> HandlingResult:
+        """Handle xml message and notify the correct event subscribers.
+
+        :return: A message response
+        """
+        if (
+            xml.attrib.get("ret") != "ok"
+            or (component_type := xml.attrib.get("type")) is None
+            or (left_str := xml.attrib.get("left")) is None
+            or (total_str := xml.attrib.get("total")) is None
+        ):
+            return HandlingResult.analyse()
+
+        percent = 0.0
+        left = int(left_str)
+        total = int(total_str)
+        if total > 0:
+            percent = round((left / total) * 100, 2)
+
+        event_bus.notify(
+            LifeSpanEvent(LifeSpan.from_xml(component_type), percent, left)
+        )
+        return HandlingResult.success()

deebot_client/events/__init__.py

@@ -3,8 +3,8 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
-from enum import Enum, IntEnum, unique
-from typing import TYPE_CHECKING, Any
+from enum import IntEnum, StrEnum, unique
+from typing import TYPE_CHECKING, Any, Self
 
 from deebot_client.events.base import Event
 
@@ -122,24 +122,43 @@ class ErrorEvent(Event):
 
 
 @unique
-class LifeSpan(str, Enum):
+class LifeSpan(StrEnum):
     """Enum class for all possible life span components."""
 
-    BRUSH = "brush"
-    FILTER = "heap"
-    SIDE_BRUSH = "sideBrush"
-    UNIT_CARE = "unitCare"
-    ROUND_MOP = "roundMop"
-    AIR_FRESHENER = "dModule"
-    UV_SANITIZER = "uv"
-    HUMIDIFY = "humidify"
-    HUMIDIFY_MAINTENANCE = "wbCare"
-    BLADE = "blade"
-    LENS_BRUSH = "lensBrush"
-    DUST_BAG = "dustBag"
-    CLEANING_FLUID = "autoWater_cleaningFluid"
-    STRAINER = "strainer"
-    HAND_FILTER = "handFilter"
+    xml_value: str
+
+    def __new__(cls, value: str, xml_value: str = "") -> Self:
+        obj = str.__new__(cls)
+        obj._value_ = value
+        obj.xml_value = xml_value
+        return obj
+
+    @classmethod
+    def from_xml(cls, value: str) -> LifeSpan:
+        """Get LifeSpan from xml value."""
+        for life_span in LifeSpan:
+            if life_span.xml_value == value:
+                return life_span
+
+        msg = f"{value} is not a valid {cls.__name__}"
+        raise ValueError(msg)
+
+    BRUSH = "brush", "Brush"
+    FILTER = "heap", "Heap"
+    SIDE_BRUSH = "sideBrush", "SideBrush"
+    UNIT_CARE = "unitCare", "UnitCare"
+    ROUND_MOP = "roundMop", "RoundMop"
+    AIR_FRESHENER = "dModule", "DModule"
+    UV_SANITIZER = "uv", "Uv"
+    HUMIDIFY = "humidify", "Humidify"
+    HUMIDIFY_MAINTENANCE = "wbCare", "WbCare"
+    BLADE = "blade", "Blade"
+    LENS_BRUSH = "lensBrush", "LensBrush"
+    DUST_BAG = "dustBag", "DustBag"
+    CLEANING_FLUID = "autoWater_cleaningFluid", "AutoWater_cleaningFluid"
+    STRAINER = "strainer", "Strainer"
+    HAND_FILTER = "handFilter", "HandFilter"
+    DUST_CASE_HEAP = "dustCaseHeap", "DustCaseHeap"
 
 
 @dataclass(frozen=True)

tests/commands/xml/test_life_span.py

@@ -0,0 +1,59 @@
+from __future__ import annotations
+
+import pytest
+
+from deebot_client.command import CommandResult
+from deebot_client.commands.xml import GetLifeSpan
+from deebot_client.events import LifeSpan, LifeSpanEvent
+from deebot_client.message import HandlingState
+from tests.commands import assert_command
+
+from . import get_request_xml
+
+
+@pytest.mark.parametrize(
+    ("component_type", "lifespan_type", "left", "total", "expected_event"),
+    [
+        ("Brush", LifeSpan.BRUSH, 50, 100, LifeSpanEvent(LifeSpan.BRUSH, 50, 50)),
+        (
+            "DustCaseHeap",
+            LifeSpan.DUST_CASE_HEAP,
+            50,
+            200,
+            LifeSpanEvent(LifeSpan.DUST_CASE_HEAP, 25, 50),
+        ),
+        (
+            "SideBrush",
+            LifeSpan.SIDE_BRUSH,
+            25,
+            200,
+            LifeSpanEvent(LifeSpan.SIDE_BRUSH, 12.5, 25),
+        ),
+    ],
+)
+async def test_get_life_span(
+    component_type: str,
+    lifespan_type: LifeSpan,
+    left: int,
+    total: int,
+    expected_event: LifeSpanEvent,
+) -> None:
+    json = get_request_xml(
+        f"<ctl ret='ok' type='{component_type}' left='{left}' total='{total}'/>"
+    )
+    await assert_command(GetLifeSpan(lifespan_type), json, expected_event)
+
+
+@pytest.mark.parametrize(
+    "xml",
+    ["<ctl ret='error'/>", "<ctl ret='ok' type='SideBrush' left='123' />"],
+    ids=["error", "no_state"],
+)
+async def test_get_life_span_error(xml: str) -> None:
+    json = get_request_xml(xml)
+    await assert_command(
+        GetLifeSpan(LifeSpan.BRUSH),
+        json,
+        None,
+        command_result=CommandResult(HandlingState.ANALYSE_LOGGED),
+    )