diff --git a/deebot_client/capabilities.py b/deebot_client/capabilities.py index e64ac596..088ce1c0 100644 --- a/deebot_client/capabilities.py +++ b/deebot_client/capabilities.py @@ -27,6 +27,7 @@ MapTraceEvent, MultimapStateEvent, NetworkInfoEvent, + OtaEvent, PositionsEvent, ReportStatsEvent, RoomsEvent, @@ -179,6 +180,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 diff --git a/deebot_client/commands/json/__init__.py b/deebot_client/commands/json/__init__.py index 26702288..cb0f30e2 100644 --- a/deebot_client/commands/json/__init__.py +++ b/deebot_client/commands/json/__init__.py @@ -30,6 +30,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 @@ -79,6 +80,8 @@ "GetMultimapState", "SetMultimapState", "GetNetInfo", + "GetOta", + "SetOta", "PlaySound", "GetPos", "SetRelocationState", @@ -153,6 +156,9 @@ GetNetInfo, + GetOta, + SetOta, + PlaySound, GetPos, diff --git a/deebot_client/commands/json/ota.py b/deebot_client/commands/json/ota.py new file mode 100644 index 00000000..833ff430 --- /dev/null +++ b/deebot_client/commands/json/ota.py @@ -0,0 +1,58 @@ +"""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 + """ + ota_event = OtaEvent( + support_auto=bool(data.get("supportAuto", False)), + auto_enabled=bool(data.get("autoSwitch", False)), + version=data["ver"], + status=data["status"], + progress=data["progress"], + ) + event_bus.notify(ota_event) + return HandlingResult.success() + + @classmethod + def handle_set_args( + cls, event_bus: EventBus, args: dict[str, Any] + ) -> HandlingResult: + """Handle arguments of set command.""" + event_bus.notify(OtaEvent(support_auto=True, auto_enabled=args["autoSwitch"])) + 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}) diff --git a/deebot_client/events/__init__.py b/deebot_client/events/__init__.py index 2ccfca44..de47656d 100644 --- a/deebot_client/events/__init__.py +++ b/deebot_client/events/__init__.py @@ -184,6 +184,17 @@ class AvailabilityEvent(Event): available: bool +@dataclass(frozen=True) +class OtaEvent(Event): + """Ota event.""" + + support_auto: bool + auto_enabled: bool | None = None + version: str | None = None + status: str | None = None + progress: int | None = None + + @dataclass(frozen=True) class StateEvent(Event): """State event representation.""" diff --git a/deebot_client/hardware/deebot/p95mgv.py b/deebot_client/hardware/deebot/p95mgv.py index 3e5ed915..c7cb7c78 100644 --- a/deebot_client/hardware/deebot/p95mgv.py +++ b/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 ), diff --git a/deebot_client/hardware/deebot/yna5xi.py b/deebot_client/hardware/deebot/yna5xi.py index 8fd97999..d0810c0b 100644 --- a/deebot_client/hardware/deebot/yna5xi.py +++ b/deebot_client/hardware/deebot/yna5xi.py @@ -41,6 +41,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 @@ -66,6 +67,7 @@ MapChangedEvent, MapTraceEvent, MultimapStateEvent, + OtaEvent, PositionsEvent, ReportStatsEvent, RoomsEvent, @@ -144,6 +146,7 @@ [GetCarpetAutoFanBoost()], SetCarpetAutoFanBoost, ), + ota=CapabilityEvent(OtaEvent, [GetOta()]), volume=CapabilitySet(VolumeEvent, [GetVolume()], SetVolume), ), state=CapabilityEvent(StateEvent, [GetChargeState(), GetCleanInfo()]), diff --git a/tests/commands/json/test_ota.py b/tests/commands/json/test_ota.py new file mode 100644 index 00000000..5afeab75 --- /dev/null +++ b/tests/commands/json/test_ota.py @@ -0,0 +1,52 @@ +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_enabled", "support_auto"), + [ + (False, True), + (True, True), + (False, False), + ], +) +async def test_GetOta(*, auto_enabled: bool, support_auto: bool) -> None: + json = get_request_json( + get_success_body( + { + "autoSwitch": 1 if auto_enabled else 0, + "supportAuto": 1 if support_auto else 0, + "ver": "1.7.2", + "status": "idle", + "progress": 0, + } + ) + ) + await assert_command( + GetOta(), + json, + OtaEvent( + auto_enabled=auto_enabled, + support_auto=support_auto, + version="1.7.2", + status="idle", + progress=0, + ), + ) + + +@pytest.mark.parametrize("auto_enabled", [False, True]) +async def test_SetOta(*, auto_enabled: bool) -> None: + args = {"autoSwitch": auto_enabled} + await assert_set_command( + SetOta(auto_enabled), + args, + OtaEvent(auto_enabled=auto_enabled, support_auto=True), + ) diff --git a/tests/hardware/test_init.py b/tests/hardware/test_init.py index 3cd7b2f3..c606bfd5 100644 --- a/tests/hardware/test_init.py +++ b/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()],