fix(messages): skip map-related legacy fallback for mower devices (#1565)

* fix(messages): skip map-related legacy fallback for mower devices

When a mower (e.g. Ecovacs GOAT G1, A1600 RTK, A3000 LiDAR Pro) is
mowing, its firmware pushes spontaneous `onMapTrace` MQTT messages.
They flow through `messages.get_message()` which falls back via
`_LEGACY_USE_GET_COMMAND` to `GetMapTrace`, whose parser does not
support the on-wire format used by recent mower firmware
(observed: 1.15.13). The parse fails inside
`_handle_error_or_analyse` and emits a WARNING per push.

In one user setup this produced 217 520 `Could not parse getMapTrace`
warnings over 3 days (~70 k/day, ~50/min during mowing), bloating the
HA `homeassistant.log` and the recorder DB by ~100 MB.

Mowers do not expose a `map=` capability in their hardware
definitions, so successfully parsed map-trace data would have nowhere
to land anyway — the fallback is pure noise on these devices.

This adds an optional `device_type` argument to `get_message` and
`get_legacy_message`. When the device is a `MOWER`, map-related
legacy messages (`onMapTrace`, `onMapSet`, `onMapSubSet`,
`onMinorMap`, `onCachedMapInfo`, `onMultiMapState` and their
`get/off/report` variants) are skipped silently before reaching the
parser. Vacuums and devices with no specified type fall through
unchanged.

`Device._handle_message` reads `device_type` defensively via `getattr`
so existing test mocks that don't expose the dataclass field on
`spec_set=Capabilities` continue to work — they fall back to the
no-filter path.

Tests:

- `test_get_messages` (existing 7 cases) — pass unchanged.
- `test_get_messages_device_type_filter` (10 new cases) — verify
  mowers skip map-related legacy fallbacks while non-map messages
  and other device types remain unaffected.
- Full suite: 707/707 pass, no regression.

Refs: #1376 (Disable getMapTrace for Goat)
Refs: #852 (GOAT A1600 RTK Support)

* review: gate on map capability instead of mower device_type

Address review feedback on #1565:
- Drop the explanatory comment block above ``_MAP_LEGACY_COMMANDS``
- Replace the ``device_type == DeviceType.MOWER`` check with a
  capability-based one. ``get_message`` now takes ``has_map: bool``
  (keyword-only) and ``get_legacy_message`` skips map-related fallbacks
  when the device does not expose a ``map`` capability — independent of
  whether the device is a mower, vacuum, or future device class.
- ``Device._handle_message`` derives ``has_map`` from
  ``capabilities.map`` via ``getattr`` so test mocks unaffected.

* review: pass StaticDeviceInfo to get_message, remove getattr

Address review feedback from edenhaus:
- Pass the whole `static: StaticDeviceInfo` object to `get_message()`
  instead of extracting `data_type` and `has_map` separately at the
  call site.
- Remove `getattr` in device.py — access `capabilities.map` directly
  since the attribute is defined on the dataclass.
- Remove `lru_cache` on `get_message` since `StaticDeviceInfo` contains
  non-trivially-hashable `Capabilities` objects. The function is cheap
  enough without caching (string lookups + one regex).
- Update tests to use real `StaticDeviceInfo` fixtures (yna5xi with map,
  xmp9ds without map) instead of passing primitive params.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* Minor sorting

* refactor tests

* forgotten prek

* Use realistic bot for test

---------

Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Co-authored-by: Robert Resch <robert@resch.dev>

Author: 3 people

deebot_client/device.py

@@ -228,7 +228,10 @@ def _handle_message(
         try:
             _LOGGER.debug("Try to handle message %s: %s", message_name, message_data)
 
-            if message := get_message(message_name, self._device_info.static.data_type):
+            if message := get_message(
+                message_name,
+                self._device_info.static,
+            ):
                 result = message.handle(self.events, message_data)
                 if result.state == HandlingState.SUCCESS and result.requested_commands:
                     _LOGGER.debug(

deebot_client/messages/__init__.py

@@ -2,7 +2,6 @@
 
 from __future__ import annotations
 
-from functools import lru_cache
 from typing import TYPE_CHECKING
 
 from deebot_client.const import DataType
@@ -13,6 +12,7 @@
 
 if TYPE_CHECKING:
     from deebot_client.message import Message
+    from deebot_client.models import StaticDeviceInfo
 
 _LOGGER = get_logger(__name__)
 
@@ -22,15 +22,17 @@
 }
 
 
-@lru_cache(maxsize=256)
-def get_message(message_name: str, data_type: DataType) -> type[Message] | None:
+def get_message(
+    message_name: str,
+    static: StaticDeviceInfo,
+) -> type[Message] | None:
     """Try to find the message for the given name.
 
     If there exists no exact match, some conversations are performed on the name to get message object similar to the name.
     """
-    messages = MESSAGES.get(data_type)
+    messages = MESSAGES.get(static.data_type)
     if messages is None:
-        _LOGGER.warning("Datatype %s is not supported.", data_type)
+        _LOGGER.warning("Datatype %s is not supported.", static.data_type)
         return None
 
     if message_type := messages.get(message_name, None):
@@ -43,8 +45,10 @@ def get_message(message_name: str, data_type: DataType) -> type[Message] | None:
     if message_type := messages.get(converted_name, None):
         return message_type
 
-    if data_type == DataType.JSON and (
-        found_message := get_legacy_message(message_name, converted_name)
+    if static.data_type == DataType.JSON and (
+        found_message := get_legacy_message(
+            message_name, converted_name, has_map=static.capabilities.map is not None
+        )
     ):
         return found_message
 

deebot_client/messages/json/__init__.py

@@ -54,37 +54,49 @@
 
 MESSAGES: dict[str, type[Message]] = {message.NAME: message for message in _MESSAGES}
 
-_LEGACY_USE_GET_COMMAND = [
-    "getAdvancedMode",
-    "getBreakPoint",
-    "getCachedMapInfo",
-    "getCarpertPressure",
-    "getChargeState",
-    "getCleanCount",
-    "getCleanInfo",
-    "getCleanPreference",
-    "getEfficiency",
-    "getError",
-    "getLifeSpan",
-    "getMapSet",
-    "getMapSubSet",
-    "getMapTrace",
-    "getMinorMap",
-    "getMultiMapState",
-    "getNetInfo",
-    "getPos",
-    "getSpeed",
-    "getSweepMode",
-    "getTotalStats",
-    "getTrueDetect",
-    "getVoiceAssistantState",
-    "getVolume",
-    "getWaterInfo",
-    "getWorkMode",
-]
-
-
-def get_legacy_message(message_name: str, converted_name: str) -> type[Message] | None:
+_MAP_LEGACY_COMMANDS = frozenset(
+    {
+        "getCachedMapInfo",
+        "getMapSet",
+        "getMapSubSet",
+        "getMapTrace",
+        "getMinorMap",
+        "getMultiMapState",
+    }
+)
+
+_LEGACY_USE_GET_COMMAND = _MAP_LEGACY_COMMANDS | frozenset(
+    {
+        "getAdvancedMode",
+        "getBreakPoint",
+        "getCarpertPressure",
+        "getChargeState",
+        "getCleanCount",
+        "getCleanInfo",
+        "getCleanPreference",
+        "getEfficiency",
+        "getError",
+        "getLifeSpan",
+        "getNetInfo",
+        "getPos",
+        "getSpeed",
+        "getSweepMode",
+        "getTotalStats",
+        "getTrueDetect",
+        "getVoiceAssistantState",
+        "getVolume",
+        "getWaterInfo",
+        "getWorkMode",
+    }
+)
+
+
+def get_legacy_message(
+    message_name: str,
+    converted_name: str,
+    *,
+    has_map: bool = True,
+) -> type[Message] | None:
     """Try to find the message for the given name using legacy way."""
     # Handle message starting with "on","off","report" the same as "get" commands
     converted_name = re.sub(
@@ -97,6 +109,13 @@ def get_legacy_message(message_name: str, converted_name: str) -> type[Message]
         _LOGGER.debug('Unknown message "%s"', message_name)
         return None
 
+    if not has_map and converted_name in _MAP_LEGACY_COMMANDS:
+        _LOGGER.debug(
+            'Skipping legacy map fallback for "%s" on device without map capability',
+            message_name,
+        )
+        return None
+
     from deebot_client.commands.json import (  # noqa: PLC0415
         COMMANDS,
     )

tests/conftest.py

@@ -135,7 +135,6 @@ def device_class() -> str:
 async def static_device_info(device_class: str) -> StaticDeviceInfo:
     info = await get_static_device_info(device_class)
     assert info is not None
-    assert info.capabilities.map is not None
     return info
 
 

tests/messages/test_get_messages.py

@@ -5,29 +5,33 @@
 import pytest
 
 from deebot_client.commands.json.error import GetError
-from deebot_client.const import DataType
 from deebot_client.messages import get_message
 from deebot_client.messages.json.battery import OnBattery
 from deebot_client.messages.json.stats import OnStats
 
 if TYPE_CHECKING:
     from deebot_client.message import Message
+    from deebot_client.models import StaticDeviceInfo
 
 
 @pytest.mark.parametrize(
-    ("name", "data_type", "expected"),
+    ("device_class", "name", "expected"),
     [
-        ("onBattery", DataType.JSON, OnBattery),
-        ("onBattery_V2", DataType.JSON, OnBattery),
-        ("onError", DataType.JSON, GetError),
-        ("onStats", DataType.JSON, OnStats),
-        ("GetCleanLogs", DataType.JSON, None),
-        ("unknown", DataType.JSON, None),
-        ("unknown", DataType.XML, None),
+        ("yna5xi", "onBattery", OnBattery),
+        ("qhe2o2", "onBattery_V2", OnBattery),
+        ("yna5xi", "onError", GetError),
+        ("yna5xi", "onStats", OnStats),
+        ("yna5xi", "GetCleanLogs", None),
+        ("yna5xi", "unknown", None),
+        ("2pv572", "unknown", None),
+        ("xmp9ds", "onMapTrace", None),
+        ("xmp9ds", "onMapSet", None),
+        ("xmp9ds", "onMinorMap", None),
+        ("xmp9ds", "onBattery", OnBattery),
     ],
 )
 def test_get_messages(
-    name: str, data_type: DataType, expected: type[Message] | None
+    static_device_info: StaticDeviceInfo, name: str, expected: type[Message] | None
 ) -> None:
     """Test get messages."""
-    assert get_message(name, data_type) == expected
+    assert get_message(name, static_device_info) == expected