diff --git a/deebot_client/commands/xml/__init__.py b/deebot_client/commands/xml/__init__.py
index df7488032..a1cb45c68 100644
--- a/deebot_client/commands/xml/__init__.py
+++ b/deebot_client/commands/xml/__init__.py
@@ -16,6 +16,7 @@
from .play_sound import PlaySound
from .pos import GetChargerPos, GetPos
from .stats import GetCleanSum
+from .water_info import GetWaterBoxInfo, GetWaterPermeability, SetWaterPermeability
if TYPE_CHECKING:
from deebot_client.command import Command
@@ -40,11 +41,14 @@
"GetMapSt",
"GetPos",
"GetTrM",
+ "GetWaterBoxInfo",
+ "GetWaterPermeability",
"PlaySound",
"PullM",
"PullMP",
"ResetLifeSpan",
"SetCleanSpeed",
+ "SetWaterPermeability",
]
# fmt: off
@@ -83,6 +87,10 @@
GetPos,
GetCleanSum,
+
+ GetWaterPermeability,
+ GetWaterBoxInfo,
+ SetWaterPermeability,
]
# fmt: on
diff --git a/deebot_client/commands/xml/water_info.py b/deebot_client/commands/xml/water_info.py
new file mode 100644
index 000000000..7105b7806
--- /dev/null
+++ b/deebot_client/commands/xml/water_info.py
@@ -0,0 +1,71 @@
+"""WaterBox command module."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from deebot_client.commands.xml.common import (
+ ExecuteCommand,
+ XmlCommandWithMessageHandling,
+)
+from deebot_client.events.water_info import (
+ WaterAmount,
+ WaterAmountEvent,
+)
+from deebot_client.message import HandlingResult
+from deebot_client.messages.xml import WaterBoxInfo
+from deebot_client.util import get_enum
+
+if TYPE_CHECKING:
+ from xml.etree.ElementTree import Element
+
+ from deebot_client.event_bus import EventBus
+
+
+class GetWaterPermeability(XmlCommandWithMessageHandling):
+ """GetWaterPermeability command."""
+
+ NAME = "GetWaterPermeability"
+
+ @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 not (value := xml.attrib.get("v")):
+ return HandlingResult.analyse()
+
+ if value.isdecimal() and (value_int := int(value)) >= 0:
+ event_bus.notify(WaterAmountEvent(WaterAmount(value_int)))
+ return HandlingResult.success()
+
+ return HandlingResult.analyse()
+
+
+class SetWaterPermeability(ExecuteCommand):
+ """SetWaterPermeability command."""
+
+ NAME = "SetWaterPermeability"
+
+ def __init__(self, amount: WaterAmount | str) -> None:
+ if isinstance(amount, str):
+ amount = get_enum(WaterAmount, amount)
+ super().__init__({"v": str(amount.value)})
+
+
+class GetWaterBoxInfo(XmlCommandWithMessageHandling, WaterBoxInfo):
+ """GetWaterBoxInfo command."""
+
+ NAME = "GetWaterBoxInfo"
+
+ @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()
+
+ return cls._parse_xml(event_bus, xml)
diff --git a/deebot_client/messages/xml/__init__.py b/deebot_client/messages/xml/__init__.py
index 02406ca95..95575295f 100644
--- a/deebot_client/messages/xml/__init__.py
+++ b/deebot_client/messages/xml/__init__.py
@@ -7,13 +7,14 @@
from deebot_client.messages.xml.battery import BatteryInfo
from deebot_client.messages.xml.charge_state import ChargeState
from deebot_client.messages.xml.pos import Pos
+from deebot_client.messages.xml.water_info import WaterBoxInfo
if TYPE_CHECKING:
from collections.abc import Sequence
from deebot_client.message import Message
-__all__: Sequence[str] = ["BatteryInfo", "ChargeState", "Pos"]
+__all__: Sequence[str] = ["BatteryInfo", "ChargeState", "Pos", "WaterBoxInfo"]
# fmt: off
# ordered by file asc
_MESSAGES: list[type[Message]] = [
@@ -21,7 +22,9 @@
ChargeState,
- Pos
+ Pos,
+
+ WaterBoxInfo
]
# fmt: on
diff --git a/deebot_client/messages/xml/water_info.py b/deebot_client/messages/xml/water_info.py
new file mode 100644
index 000000000..1cfc97624
--- /dev/null
+++ b/deebot_client/messages/xml/water_info.py
@@ -0,0 +1,40 @@
+"""Water messages."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from deebot_client.events.water_info import MopAttachedEvent
+from deebot_client.message import HandlingResult
+from deebot_client.messages.xml.common import XmlMessage
+
+if TYPE_CHECKING:
+ from xml.etree.ElementTree import Element
+
+ from deebot_client.event_bus import EventBus
+
+
+class WaterBoxInfo(XmlMessage):
+ """WaterBoxInfo message."""
+
+ NAME = "WaterBoxInfo"
+
+ @classmethod
+ def _handle_xml(cls, event_bus: EventBus, xml: Element) -> HandlingResult:
+ """Handle xml message and notify the correct event subscribers.
+
+ :return: A message response.
+ """
+ return cls._parse_xml(event_bus, xml)
+
+ @classmethod
+ def _parse_xml(cls, event_bus: EventBus, xml: Element) -> HandlingResult:
+ """Handle xml message and notify the correct event subscribers.
+
+ :return: A message response.
+ """
+ if (on := xml.attrib.get("on")) is None:
+ return HandlingResult.analyse()
+
+ event_bus.notify(MopAttachedEvent(on != "0"))
+ return HandlingResult.success()
diff --git a/tests/commands/xml/test_water_info.py b/tests/commands/xml/test_water_info.py
new file mode 100644
index 000000000..ab48d7aa6
--- /dev/null
+++ b/tests/commands/xml/test_water_info.py
@@ -0,0 +1,50 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import pytest
+
+from deebot_client.command import CommandResult
+from deebot_client.commands.xml import GetWaterBoxInfo
+from deebot_client.events.water_info import MopAttachedEvent
+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(
+ ("state", "expected_event"),
+ [
+ (1, MopAttachedEvent(True)),
+ (0, MopAttachedEvent(False)),
+ ],
+ ids=["mop_attached", "mop_not_attached"],
+)
+async def test_get_water_box_info(state: int, expected_event: Event) -> None:
+ xml_message = get_request_xml(f"")
+ await assert_command(GetWaterBoxInfo(), xml_message, expected_event)
+
+
+@pytest.mark.parametrize(
+ "payload",
+ [
+ '',
+ '',
+ ],
+ ids=[
+ "error",
+ "no_state",
+ ],
+)
+async def test_get_water_box_info_error(payload: str) -> None:
+ xml_message = get_request_xml(payload)
+ await assert_command(
+ GetWaterBoxInfo(),
+ xml_message,
+ None,
+ command_result=CommandResult(HandlingState.ANALYSE_LOGGED),
+ )