feat: parse getRTK for mower devices

deebot_client/capabilities.py

@@ -41,6 +41,7 @@
     PositionsEvent,
     ReportStatsEvent,
     RoomsEvent,
+    RtkEvent,
     SafeProtectEvent,
     StateEvent,
     StationEvent,
@@ -276,6 +277,7 @@ class Capabilities(ABC):
     map: CapabilityMap | None = None
     network: CapabilityEvent[NetworkInfoEvent]
     play_sound: CapabilityExecute[[]]
+    rtk: CapabilityEvent[RtkEvent] | None = None
     settings: CapabilitySettings
     state: CapabilityEvent[StateEvent]
     station: CapabilityStation | None = None

deebot_client/commands/json/__init__.py

@@ -44,6 +44,7 @@
 from .play_sound import PlaySound
 from .pos import GetPos
 from .relocation import SetRelocationState
+from .rtk import GetRtk
 from .safe_protect import GetSafeProtect, SetSafeProtect
 from .stats import GetStats, GetTotalStats
 from .sweep_mode import GetSweepMode, SetSweepMode
@@ -98,6 +99,7 @@
     "GetNetInfoLegacy",
     "GetOta",
     "GetPos",
+    "GetRtk",
     "GetSafeProtect",
     "GetStats",
     "GetSweepMode",
@@ -226,6 +228,8 @@
 
     GetPos,
 
+    GetRtk,
+
     SetRelocationState,
 
     GetSafeProtect,

deebot_client/commands/json/rtk.py

@@ -0,0 +1,84 @@
+"""RTK status command (mowers only)."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any
+
+from deebot_client.events.rtk import RtkBaseStation, RtkEvent
+from deebot_client.message import HandlingResult, MessageBodyDataDict
+
+from .common import JsonCommandWithMessageHandling
+
+if TYPE_CHECKING:
+    from deebot_client.event_bus import EventBus
+
+
+def _coerce_base_station(raw: dict[str, Any]) -> RtkBaseStation:
+    return RtkBaseStation(
+        serial_number=str(raw.get("sn", "")),
+        satellites_visible=int(raw.get("star", 0) or 0),
+        firmware=raw.get("version"),
+        state=raw.get("state"),
+        mode=raw.get("mode"),
+    )
+
+
+class GetRtk(JsonCommandWithMessageHandling, MessageBodyDataDict):
+    r"""Get RTK status command for mower devices.
+
+    Sends ``getRTK`` to the device and notifies a :class:`RtkEvent` with
+    the parsed response. The cloud payload is shaped like this (sample
+    captured from a GOAT A1600 RTK on firmware 1.15.13)::
+
+        {
+            "result": 0,
+            "rtks": [
+                {"sn": "908276", "star": 30, "state": 0, "mode": 0,
+                 "version": "...,QD302 1.3.8,...,QD302 1.3.1"}
+            ],
+            "observations": {
+                "solStat": 0, "poseType": 50,
+                "roverId": "908336", "roverSvs": 35, "roverSolnSvs": 30,
+                "roverSignalRate": 44, "roverSignalScore": 90,
+                "roverOcclusionRate": 8,
+                "baseStnId": "\\"1544\\"", "baseSolnSvs": 29,
+                "baseSignalRate": 45, "baseSignalScore": 94,
+                "baseOcclusionRate": 24
+            }
+        }
+    """
+
+    NAME = "getRTK"
+
+    @classmethod
+    def _handle_body_data_dict(
+        cls, event_bus: EventBus, data: dict[str, Any]
+    ) -> HandlingResult:
+        observations = data.get("observations") or {}
+        if not observations:
+            return HandlingResult.analyse()
+
+        rtks_raw = data.get("rtks") or []
+        base_stations = [
+            _coerce_base_station(r) for r in rtks_raw if isinstance(r, dict)
+        ]
+
+        event_bus.notify(
+            RtkEvent(
+                rover_serial_number=str(observations.get("roverId", "")),
+                rover_satellites_visible=int(observations.get("roverSvs", 0) or 0),
+                rover_satellites_used=int(observations.get("roverSolnSvs", 0) or 0),
+                base_satellites_used=int(observations.get("baseSolnSvs", 0) or 0),
+                rover_signal_score=int(observations.get("roverSignalScore", 0) or 0),
+                base_signal_score=int(observations.get("baseSignalScore", 0) or 0),
+                rover_occlusion_rate=int(
+                    observations.get("roverOcclusionRate", 0) or 0
+                ),
+                base_occlusion_rate=int(observations.get("baseOcclusionRate", 0) or 0),
+                base_stations=base_stations,
+                base_station_id=observations.get("baseStnId"),
+                solution_status=observations.get("solStat"),
+                pose_type=observations.get("poseType"),
+            )
+        )
+        return HandlingResult.success()

deebot_client/events/__init__.py

@@ -28,6 +28,7 @@
     PositionsEvent,
 )
 from .network import NetworkInfoEvent
+from .rtk import RtkBaseStation, RtkEvent
 from .station import StationEvent
 from .work_mode import WorkMode, WorkModeEvent
 
@@ -58,6 +59,8 @@
     "NetworkInfoEvent",
     "Position",
     "PositionsEvent",
+    "RtkBaseStation",
+    "RtkEvent",
     "StationEvent",
     "SweepModeEvent",
     "WorkMode",

deebot_client/events/rtk.py

@@ -0,0 +1,74 @@
+"""RTK status event for mowers (e.g. Ecovacs GOAT family)."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+
+from .base import Event
+
+
+@dataclass(frozen=True, kw_only=True)
+class RtkBaseStation:
+    """RTK reference (base) station info as reported by the mower."""
+
+    serial_number: str
+    """Serial number of the base station, e.g. ``"908276"``."""
+
+    satellites_visible: int
+    """Number of satellites currently visible by the base station."""
+
+    firmware: str | None = None
+    """Firmware version string reported by the base station, if any."""
+
+    state: int | None = None
+    """Raw state code as reported by the device (meaning device-defined)."""
+
+    mode: int | None = None
+    """Raw mode code as reported by the device (meaning device-defined)."""
+
+
+@dataclass(frozen=True)
+class RtkEvent(Event):
+    """RTK status snapshot for a mower.
+
+    The Ecovacs GOAT family pairs a reference (base) station with a
+    rover (the mower itself) and a Real-Time-Kinematic correction loop.
+    The official Ecovacs Home app surfaces the relevant counters under
+    "Settings → RTK" of the device. This event mirrors that screen.
+    """
+
+    rover_serial_number: str
+    """Serial number of the rover (mower), e.g. ``"908336"``."""
+
+    rover_satellites_visible: int
+    """Total number of satellites the rover currently sees."""
+
+    rover_satellites_used: int
+    """Satellites actually used in the rover's RTK solution."""
+
+    base_satellites_used: int
+    """Satellites actually used in the base station's RTK solution."""
+
+    rover_signal_score: int
+    """Quality score (0-100) of the rover's GNSS signal."""
+
+    base_signal_score: int
+    """Quality score (0-100) of the base station's GNSS signal."""
+
+    rover_occlusion_rate: int
+    """Estimated rover sky-occlusion rate (percent)."""
+
+    base_occlusion_rate: int
+    """Estimated base sky-occlusion rate (percent)."""
+
+    base_stations: list[RtkBaseStation] = field(default_factory=list)
+    """One entry per known reference station (usually exactly one)."""
+
+    base_station_id: str | None = None
+    """ID of the base station currently providing corrections, if any."""
+
+    solution_status: int | None = None
+    """Raw ``solStat`` from the device (meaning device-defined)."""
+
+    pose_type: int | None = None
+    """Raw ``poseType`` from the device (meaning device-defined)."""

deebot_client/hardware/xmp9ds.py

@@ -40,6 +40,7 @@
 from deebot_client.commands.json.life_span import GetLifeSpan, ResetLifeSpan
 from deebot_client.commands.json.network import GetNetInfo
 from deebot_client.commands.json.play_sound import PlaySound
+from deebot_client.commands.json.rtk import GetRtk
 from deebot_client.commands.json.stats import GetStats, GetTotalStats
 from deebot_client.commands.json.true_detect import GetTrueDetect, SetTrueDetect
 from deebot_client.commands.json.volume import GetVolume, SetVolume
@@ -59,6 +60,7 @@
     MoveUpWarningEvent,
     NetworkInfoEvent,
     ReportStatsEvent,
+    RtkEvent,
     SafeProtectEvent,
     StateEvent,
     StatsEvent,
@@ -102,6 +104,7 @@ def get_device_info() -> StaticDeviceInfo:
             ),
             network=CapabilityEvent(NetworkInfoEvent, [GetNetInfo()]),
             play_sound=CapabilityExecute(PlaySound),
+            rtk=CapabilityEvent(RtkEvent, [GetRtk()]),
             settings=CapabilitySettings(
                 advanced_mode=CapabilitySetEnable(
                     AdvancedModeEvent, [GetAdvancedMode()], SetAdvancedMode

tests/commands/json/test_rtk.py

@@ -0,0 +1,126 @@
+"""Tests for the GetRtk command."""
+
+from __future__ import annotations
+
+from deebot_client.commands.json import GetRtk
+from deebot_client.events import RtkBaseStation, RtkEvent
+from tests.commands.json import assert_command
+from tests.helpers import get_request_json, get_success_body
+
+
+async def test_GetRtk() -> None:
+    """Parse a real GOAT A1600 RTK getRTK response."""
+    json, firmware_event = get_request_json(
+        get_success_body(
+            {
+                "result": 0,
+                "rtks": [
+                    {
+                        "x": -5332,
+                        "y": -518,
+                        "sn": "908276",
+                        "state": 0,
+                        "mode": 0,
+                        "star": 30,
+                        "version": (
+                            "630ZG-B23A5-1,QD302 1.3.8,630ZG-B23A5-1,QD302 1.3.1"
+                        ),
+                    }
+                ],
+                "observations": {
+                    "solStat": 0,
+                    "poseType": 50,
+                    "roverId": "908336",
+                    "roverSvs": 35,
+                    "roverSolnSvs": 30,
+                    "roverSignalRate": 44,
+                    "roverSignalScore": 90,
+                    "roverOcclusionRate": 8,
+                    "baseStnId": '"1544"',
+                    "baseSolnSvs": 29,
+                    "baseSignalRate": 45,
+                    "baseSignalScore": 94,
+                    "baseOcclusionRate": 24,
+                },
+            }
+        )
+    )
+    await assert_command(
+        GetRtk(),
+        json,
+        (
+            firmware_event,
+            RtkEvent(
+                rover_serial_number="908336",
+                rover_satellites_visible=35,
+                rover_satellites_used=30,
+                base_satellites_used=29,
+                rover_signal_score=90,
+                base_signal_score=94,
+                rover_occlusion_rate=8,
+                base_occlusion_rate=24,
+                base_stations=[
+                    RtkBaseStation(
+                        serial_number="908276",
+                        satellites_visible=30,
+                        firmware=(
+                            "630ZG-B23A5-1,QD302 1.3.8,630ZG-B23A5-1,QD302 1.3.1"
+                        ),
+                        state=0,
+                        mode=0,
+                    )
+                ],
+                base_station_id='"1544"',
+                solution_status=0,
+                pose_type=50,
+            ),
+        ),
+    )
+
+
+async def test_GetRtk_no_base_stations() -> None:
+    """A response with `rtks` empty still notifies the rover-side event."""
+    json, firmware_event = get_request_json(
+        get_success_body(
+            {
+                "result": 0,
+                "rtks": [],
+                "observations": {
+                    "solStat": 0,
+                    "poseType": 0,
+                    "roverId": "908336",
+                    "roverSvs": 12,
+                    "roverSolnSvs": 8,
+                    "roverSignalRate": 30,
+                    "roverSignalScore": 50,
+                    "roverOcclusionRate": 20,
+                    "baseStnId": "",
+                    "baseSolnSvs": 0,
+                    "baseSignalRate": 0,
+                    "baseSignalScore": 0,
+                    "baseOcclusionRate": 0,
+                },
+            }
+        )
+    )
+    await assert_command(
+        GetRtk(),
+        json,
+        (
+            firmware_event,
+            RtkEvent(
+                rover_serial_number="908336",
+                rover_satellites_visible=12,
+                rover_satellites_used=8,
+                base_satellites_used=0,
+                rover_signal_score=50,
+                base_signal_score=0,
+                rover_occlusion_rate=20,
+                base_occlusion_rate=0,
+                base_stations=[],
+                base_station_id="",
+                solution_status=0,
+                pose_type=0,
+            ),
+        ),
+    )