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

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