Implement XML Command GetCleanLogs (#910)

Author: nanomad

deebot_client/commands/xml/__init__.py

@@ -9,6 +9,7 @@
 from .battery import GetBatteryInfo
 from .charge import Charge
 from .charge_state import GetChargeState
+from .clean_logs import GetCleanLogs
 from .error import GetError
 from .fan_speed import GetFanSpeed
 from .life_span import GetLifeSpan
@@ -23,6 +24,7 @@
     "Charge",
     "GetBatteryInfo",
     "GetChargeState",
+    "GetCleanLogs",
     "GetCleanSum",
     "GetError",
     "GetFanSpeed",
@@ -35,6 +37,7 @@
 # ordered by file asc
 _COMMANDS: list[type[XmlCommand]] = [
     GetBatteryInfo,
+    GetCleanLogs,
     GetError,
     GetLifeSpan,
     PlaySound,

deebot_client/commands/xml/clean_logs.py

@@ -0,0 +1,105 @@
+"""Clean Logs commands."""
+
+from __future__ import annotations
+
+from enum import StrEnum, unique
+from typing import TYPE_CHECKING, Self
+
+from deebot_client.command import CommandResult
+from deebot_client.events import (
+    CleanJobStatus,
+    CleanLogEntry,
+    CleanLogEvent,
+)
+from deebot_client.logging_filter import get_logger
+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
+
+_LOGGER = get_logger(__name__)
+
+
+@unique
+class XmlStopReason(StrEnum):
+    """XML Reasons why cleaning has been stopped."""
+
+    clean_job_status: CleanJobStatus
+
+    def __new__(cls, value: str, clean_job_status: CleanJobStatus) -> Self:
+        """Create new XmlStopReason."""
+        obj = str.__new__(cls, value)
+        obj._value_ = value
+        obj.clean_job_status = clean_job_status
+        return obj
+
+    FINISHED = "s", CleanJobStatus.FINISHED
+    BATTERY_LOW = "r", CleanJobStatus.FINISHED_WITH_WARNINGS
+    STOPPED_BY_APP = "a", CleanJobStatus.MANUALLY_STOPPED
+    STOPPED_BY_REMOTE_CONTROL = "i", CleanJobStatus.MANUALLY_STOPPED
+    STOPPED_BY_BUTTON = "b", CleanJobStatus.MANUALLY_STOPPED
+    STOPPED_BY_WARNING = "w", CleanJobStatus.FINISHED_WITH_WARNINGS
+    STOPPED_BY_NO_DISTURB = "f", CleanJobStatus.FINISHED_WITH_WARNINGS
+    STOPPED_BY_CLEARMAP = "m", CleanJobStatus.FINISHED_WITH_WARNINGS
+    STOPPED_BY_NO_PATH = "n", CleanJobStatus.FINISHED_WITH_WARNINGS
+    STOPPED_BY_NOT_IN_MAP = "u", CleanJobStatus.FINISHED_WITH_WARNINGS
+    STOPPED_BY_VIRTUAL_WALL = "v", CleanJobStatus.FINISHED_WITH_WARNINGS
+
+    @classmethod
+    def from_value(cls, value: str) -> XmlStopReason:
+        """Fetch the right enum member given its string value."""
+        for elem in cls.__members__.values():
+            if elem.value == value:
+                return elem
+        raise ValueError(value)
+
+
+class GetCleanLogs(XmlCommandWithMessageHandling):
+    """GetCleanLogs command."""
+
+    NAME = "GetCleanLogs"
+
+    def __init__(self, count: int = 0) -> None:
+        super().__init__({"count": str(count)})
+
+    @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":
+            return HandlingResult.analyse()
+
+        resp_logs = xml.findall("CleanSt")
+        logs: list[CleanLogEntry] = []
+        for log in resp_logs:
+            xml_stop_reason_attrib = str(log.attrib["f"])
+            stop_reason = XmlStopReason.FINISHED
+            try:
+                stop_reason = XmlStopReason.from_value(xml_stop_reason_attrib)
+            except ValueError as e:
+                _LOGGER.error(
+                    "Could not decode stop reason: %s",
+                    xml_stop_reason_attrib,
+                    exc_info=e,
+                )
+            try:
+                logs.append(
+                    CleanLogEntry(
+                        timestamp=int(log.attrib["s"]),
+                        image_url="",  # Not available
+                        type=log.attrib["t"],
+                        area=int(log.attrib["a"]),
+                        stop_reason=stop_reason.clean_job_status,
+                        duration=int(log.attrib["l"]),
+                    )
+                )
+            except Exception:  # pylint: disable = broad-exception-caught
+                _LOGGER.warning("Skipping log entry: %s", log, exc_info=True)
+        event_bus.notify(CleanLogEvent(logs))
+        return CommandResult.success()

tests/commands/xml/test_clean_logs.py

@@ -0,0 +1,67 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import pytest
+
+from deebot_client.command import CommandResult
+from deebot_client.commands.xml import GetCleanLogs
+from deebot_client.events import CleanJobStatus, CleanLogEntry, CleanLogEvent
+from deebot_client.message import HandlingState
+from tests.commands import assert_command
+
+from . import get_request_xml
+
+if TYPE_CHECKING:
+    from deebot_client.events.base import Event
+
+
+@pytest.mark.parametrize(
+    ("payload", "expected_event"),
+    [
+        (
+            "<CleanSt a='0' s='1710433763' l='14' t='' f='s'/><CleanSt a='1' s='1710433269' l='22' t='' f='s'/>",
+            CleanLogEvent(
+                [
+                    CleanLogEntry(1710433763, "", "", 0, CleanJobStatus.FINISHED, 14),
+                    CleanLogEntry(1710433269, "", "", 1, CleanJobStatus.FINISHED, 22),
+                ]
+            ),
+        ),
+        (
+            "<CleanSt a='20' s='1710244976' l='1392' t='a' f='a'/>"
+            "<CleanSt a='wrong' s='1710083567' l='894' t='a' f='a'/>"
+            "<CleanSt a='21' s='1710244999' l='2392' t='a' f='????'/>",
+            CleanLogEvent(
+                [
+                    CleanLogEntry(
+                        1710244976, "", "a", 20, CleanJobStatus.MANUALLY_STOPPED, 1392
+                    ),
+                    CleanLogEntry(
+                        1710244999, "", "a", 21, CleanJobStatus.FINISHED, 2392
+                    ),
+                ]
+            ),
+        ),
+        ("", CleanLogEvent([])),
+    ],
+    ids=["finished_two_areas", "skipping_invalid_data", "no_data"],
+)
+async def test_get_clean_logs(payload: str, expected_event: Event | None) -> None:
+    json = get_request_xml(f"<ctl ret='ok'>{payload}</ctl>")
+    await assert_command(GetCleanLogs(count=3), json, expected_event)
+
+
+@pytest.mark.parametrize(
+    "xml",
+    ["<ctl ret='error'/>"],
+    ids=["error"],
+)
+async def test_get_clean_logs_error(xml: str) -> None:
+    json = get_request_xml(xml)
+    await assert_command(
+        GetCleanLogs(),
+        json,
+        None,
+        command_result=CommandResult(HandlingState.ANALYSE_LOGGED),
+    )