Add "Ota" commands

deebot_client/capabilities.py

@@ -27,6 +27,7 @@
     MapTraceEvent,
     MultimapStateEvent,
     NetworkInfoEvent,
+    OtaEvent,
     PositionsEvent,
     ReportStatsEvent,
     RoomsEvent,
@@ -178,6 +179,7 @@ class CapabilitySettings:
     efficiency_mode: (
         CapabilitySetTypes[EfficiencyModeEvent, EfficiencyMode] | None
     ) = None
+    ota: CapabilitySetEnable[OtaEvent] | CapabilityEvent[OtaEvent] | None = None
     sweep_mode: CapabilitySetEnable[SweepModeEvent] | None = None
     true_detect: CapabilitySetEnable[TrueDetectEvent] | None = None
     voice_assistant: CapabilitySetEnable[VoiceAssistantStateEvent] | None = None

deebot_client/commands/json/__init__.py

@@ -29,6 +29,7 @@
 )
 from .multimap_state import GetMultimapState, SetMultimapState
 from .network import GetNetInfo
+from .ota import GetOta, SetOta
 from .play_sound import PlaySound
 from .pos import GetPos
 from .relocation import SetRelocationState
@@ -77,6 +78,8 @@
     "GetMultimapState",
     "SetMultimapState",
     "GetNetInfo",
+    "GetOta",
+    "SetOta",
     "PlaySound",
     "GetPos",
     "SetRelocationState",
@@ -149,6 +152,9 @@
 
     GetNetInfo,
 
+    GetOta,
+    SetOta,
+
     PlaySound,
 
     GetPos,

deebot_client/commands/json/ota.py

@@ -0,0 +1,51 @@
+"""Ota command module."""
+from __future__ import annotations
+
+from types import MappingProxyType
+from typing import TYPE_CHECKING, Any
+
+from deebot_client.command import InitParam
+from deebot_client.events import OtaEvent
+from deebot_client.message import HandlingResult
+
+from .common import JsonGetCommand, JsonSetCommand
+
+if TYPE_CHECKING:
+    from deebot_client.event_bus import EventBus
+
+
+class GetOta(JsonGetCommand):
+    """Get ota command."""
+
+    name = "getOta"
+
+    @classmethod
+    def _handle_body_data_dict(
+        cls, event_bus: EventBus, data: dict[str, Any]
+    ) -> HandlingResult:
+        """Handle message->body->data and notify the correct event subscribers.
+
+        :return: A message response
+        """
+        support_auto = data.get("supportAuto")
+        ota_event = OtaEvent(
+            enable=bool(data.get("autoSwitch", False)),
+            support_auto=None if support_auto is None else bool(support_auto),
+            version=data.get("ver"),
+            status=data.get("status"),
+            progress=data.get("progress"),
+        )
+        event_bus.notify(ota_event)
+        return HandlingResult.success()
+
+
+class SetOta(JsonSetCommand):
+    """Set ota command."""
+
+    name = "setOta"
+    get_command = GetOta
+
+    _mqtt_params = MappingProxyType({"autoSwitch": InitParam(bool, "auto_enabled")})
+
+    def __init__(self, auto_enabled: bool) -> None:  # noqa: FBT001
+        super().__init__({"autoSwitch": 1 if auto_enabled else 0})

deebot_client/events/__init__.py

@@ -231,6 +231,16 @@ class MultimapStateEvent(EnableEvent):
     """Multimap state event."""
 
 
+@dataclass(frozen=True)
+class OtaEvent(EnableEvent):
+    """Ota event."""
+
+    support_auto: bool | None
+    version: str | None
+    status: str | None
+    progress: int | None
+
+
 @dataclass(frozen=True)
 class TrueDetectEvent(EnableEvent):
     """TrueDetect event."""

deebot_client/hardware/deebot/p95mgv.py

@@ -51,6 +51,7 @@
     SetMultimapState,
 )
 from deebot_client.commands.json.network import GetNetInfo
+from deebot_client.commands.json.ota import GetOta, SetOta
 from deebot_client.commands.json.play_sound import PlaySound
 from deebot_client.commands.json.pos import GetPos
 from deebot_client.commands.json.relocation import SetRelocationState
@@ -84,6 +85,7 @@
     MapTraceEvent,
     MultimapStateEvent,
     NetworkInfoEvent,
+    OtaEvent,
     PositionsEvent,
     ReportStatsEvent,
     RoomsEvent,
@@ -191,6 +193,7 @@
                     EfficiencyMode.STANDART_MODE,
                 ),
             ),
+            ota=CapabilitySetEnable(OtaEvent, [GetOta()], SetOta),
             true_detect=CapabilitySetEnable(
                 TrueDetectEvent, [GetTrueDetect()], SetTrueDetect
             ),

deebot_client/hardware/deebot/yna5xi.py

@@ -40,6 +40,7 @@
     SetMultimapState,
 )
 from deebot_client.commands.json.network import GetNetInfo
+from deebot_client.commands.json.ota import GetOta
 from deebot_client.commands.json.play_sound import PlaySound
 from deebot_client.commands.json.pos import GetPos
 from deebot_client.commands.json.relocation import SetRelocationState
@@ -65,6 +66,7 @@
     MapChangedEvent,
     MapTraceEvent,
     MultimapStateEvent,
+    OtaEvent,
     PositionsEvent,
     ReportStatsEvent,
     RoomsEvent,
@@ -142,6 +144,7 @@
                 [GetCarpetAutoFanBoost()],
                 SetCarpetAutoFanBoost,
             ),
+            ota=CapabilityEvent(OtaEvent, [GetOta()]),
             volume=CapabilitySet(VolumeEvent, [GetVolume()], SetVolume),
         ),
         state=CapabilityEvent(StateEvent, [GetChargeState(), GetCleanInfo()]),

tests/commands/json/test_ota.py

@@ -0,0 +1,38 @@
+from __future__ import annotations
+
+import pytest
+
+from deebot_client.commands.json import GetOta, SetOta
+from deebot_client.events import OtaEvent
+from tests.helpers import get_request_json, get_success_body
+
+from . import assert_command, assert_set_command
+
+
+@pytest.mark.parametrize(
+    ("auto_switch", "support_auto"),
+    [
+        (False, True),
+        (True, False),
+    ],
+)
+async def test_GetOta(auto_switch: bool, support_auto: bool) -> None:  # noqa: FBT001
+    json = get_request_json(
+        get_success_body(
+            {
+                "autoSwitch": 1 if auto_switch else 0,
+                "supportAuto": 1 if support_auto else 0,
+            }
+        )
+    )
+    await assert_command(
+        GetOta(), json, OtaEvent(auto_switch, support_auto, None, None, None)
+    )
+
+
+@pytest.mark.parametrize("auto_switch", [False, True])
+async def test_SetOta(auto_switch: bool) -> None:  # noqa: FBT001
+    args = {"autoSwitch": auto_switch}
+    await assert_set_command(
+        SetOta(auto_switch), args, OtaEvent(auto_switch, None, None, None, None)
+    )

tests/hardware/test_init.py

@@ -21,6 +21,7 @@
 from deebot_client.commands.json.map import GetCachedMapInfo, GetMajorMap, GetMapTrace
 from deebot_client.commands.json.multimap_state import GetMultimapState
 from deebot_client.commands.json.network import GetNetInfo
+from deebot_client.commands.json.ota import GetOta
 from deebot_client.commands.json.pos import GetPos
 from deebot_client.commands.json.stats import GetStats, GetTotalStats
 from deebot_client.commands.json.true_detect import GetTrueDetect
@@ -41,6 +42,7 @@
     LifeSpan,
     LifeSpanEvent,
     MultimapStateEvent,
+    OtaEvent,
     ReportStatsEvent,
     RoomsEvent,
     StateEvent,
@@ -145,6 +147,7 @@ def test_get_static_device_info(
                 MapTraceEvent: [GetMapTrace()],
                 MultimapStateEvent: [GetMultimapState()],
                 NetworkInfoEvent: [GetNetInfo()],
+                OtaEvent: [GetOta()],
                 PositionsEvent: [GetPos()],
                 ReportStatsEvent: [],
                 RoomsEvent: [GetCachedMapInfo()],
@@ -185,6 +188,7 @@ def test_get_static_device_info(
                 MapTraceEvent: [GetMapTrace()],
                 MultimapStateEvent: [GetMultimapState()],
                 NetworkInfoEvent: [GetNetInfo()],
+                OtaEvent: [GetOta()],
                 PositionsEvent: [GetPos()],
                 ReportStatsEvent: [],
                 RoomsEvent: [GetCachedMapInfo()],