feat: Implement device capability metadata for status computed property fields (#875)

* feat: Implement device capability metadata for status computed property fields

* docs: Document metadata keys in field_metadata and is_field_supported

* test: Dynamically mock new_feature_info from device_info.new_feature_set

* test: Share default new_feature_info_str from APP_GET_INIT_STATUS and parse unconditionally

* test: Unconditionally retrieve new_feature_set and assert it exists

* fix: Gate dock_error_status on has_dock rather than is_collectable

Author: allenporter

roborock/data/containers.py

@@ -60,6 +60,25 @@ def _attr_repr(obj: Any) -> str:
     return f"{type(obj).__name__}({', '.join(parts)})"
 
 
+def field_metadata(**kwargs):
+    """Decorator to attach capability check metadata to a property.
+
+    This attaches a `_field_metadata` dictionary to the underlying getter function,
+    which is then preserved when decorated with `@property`.
+
+    Supported metadata keys:
+    - `feature` (str): Name of a capability property on `DeviceFeaturesTrait`.
+    - `dock_feature` (str): Name of a capability property on `RoborockDockFeatures`.
+    - `dps` (str/int): RoborockDataProtocol ID to check against supported schema IDs.
+    """
+
+    def decorator(func):
+        func._field_metadata = kwargs
+        return func
+
+    return decorator
+
+
 @dataclass(repr=False)
 class RoborockBase:
     """Base class for all Roborock data classes."""

roborock/data/v1/v1_containers.py

@@ -39,7 +39,7 @@
 from roborock.exceptions import RoborockException
 from roborock.roborock_message import RoborockDataProtocol
 
-from ..containers import NamedRoomMapping, RoborockBase, RoborockBaseTimer, _attr_repr
+from ..containers import NamedRoomMapping, RoborockBase, RoborockBaseTimer, _attr_repr, field_metadata
 from .v1_clean_modes import WashTowelModes
 from .v1_code_mappings import (
     CleanFluidStatus,
@@ -117,6 +117,15 @@ class StatusField(FieldNameBase):
     CHARGE_STATUS = "charge_status"
     DRY_STATUS = "dry_status"
     ERROR_CODE = "error_code"
+    WATER_BOX_CARRIAGE_STATUS = "water_box_carriage_status"
+    WATER_BOX_STATUS = "water_box_status"
+    WATER_SHORTAGE_STATUS = "water_shortage_status"
+    DIRTY_WATER_BOX_STATUS = "dirty_water_box_status"
+    CLEAR_WATER_BOX_STATUS = "clear_water_box_status"
+    CLEAN_FLUID_STATUS = "clean_fluid_status"
+    CLEAN_PERCENT = "clean_percent"
+    DOCK_ERROR_STATUS = "dock_error_status"
+    RDT = "rdt"
 
 
 @dataclass
@@ -246,12 +255,14 @@ def has_am(self) -> bool | None:
         return (self.dss & 3) == 2
 
     @property
+    @field_metadata(dock_feature="is_washable")
     def clear_water_box_status(self) -> ClearWaterBoxStatus | None:
         if self.dss:
             return ClearWaterBoxStatus((self.dss >> 2) & 3)
         return None
 
     @property
+    @field_metadata(dock_feature="is_washable")
     def dirty_water_box_status(self) -> DirtyWaterBoxStatus | None:
         if self.dss:
             return DirtyWaterBoxStatus((self.dss >> 4) & 3)
@@ -270,6 +281,9 @@ def water_box_filter_status(self) -> int | None:
         return None
 
     @property
+    @field_metadata(
+        dock_feature="is_clean_fluid_auto_delivery_supported",
+    )
     def clean_fluid_status(self) -> CleanFluidStatus | None:
         if self.dss:
             value = (self.dss >> 10) & 3
@@ -313,7 +327,7 @@ class StatusV2(RoborockBase):
     in_returning: int | None = None
     in_fresh_state: int | None = None
     lab_status: int | None = None
-    water_box_status: int | None = None
+    water_box_status: int | None = field(default=None, metadata={"feature": "is_support_water_mode"})
     back_type: int | None = None
     wash_phase: int | None = None
     wash_ready: int | None = None
@@ -323,14 +337,14 @@ class StatusV2(RoborockBase):
     is_locating: int | None = None
     lock_status: int | None = None
     water_box_mode: int | None = field(default=None, metadata={"dps": RoborockDataProtocol.WATER_BOX_MODE})
-    water_box_carriage_status: int | None = None
+    water_box_carriage_status: int | None = field(default=None, metadata={"feature": "is_support_water_mode"})
     mop_forbidden_enable: int | None = None
     camera_status: int | None = None
     is_exploring: int | None = None
     home_sec_status: int | None = None
     home_sec_enable_password: int | None = None
     adbumper_status: list[int] | None = None
-    water_shortage_status: int | None = None
+    water_shortage_status: int | None = field(default=None, metadata={"feature": "is_support_water_mode"})
     dock_type: RoborockDockTypeCode | None = None
     dust_collection_status: int | None = None
     auto_dust_collection: int | None = None
@@ -339,7 +353,8 @@ class StatusV2(RoborockBase):
     debug_mode: int | None = None
     collision_avoid_status: int | None = None
     switch_map_mode: int | None = None
-    dock_error_status: RoborockDockErrorCode | None = None
+    dock_error_status: RoborockDockErrorCode | None = field(default=None, metadata={"dock_feature": "has_dock"})
+
     charge_status: RoborockChargeStatus | None = field(
         default=None, metadata={"dps": RoborockDataProtocol.CHARGE_STATUS}
     )
@@ -349,8 +364,8 @@ class StatusV2(RoborockBase):
     distance_off: int | None = None
     in_warmup: int | None = None
     dry_status: int | None = field(default=None, metadata={"dps": RoborockDataProtocol.DRYING_STATUS})
-    rdt: int | None = None
-    clean_percent: int | None = None
+    rdt: int | None = field(default=None, metadata={"feature": "is_supported_drying"})
+    clean_percent: int | None = field(default=None, metadata={"feature": "is_support_clean_estimate"})
     rss: int | None = None
     dss: int | None = None
     common_status: int | None = None
@@ -389,12 +404,14 @@ def has_am(self) -> bool | None:
         return (self.dss & 3) == 2
 
     @property
+    @field_metadata(dock_feature="is_washable")
     def clear_water_box_status(self) -> ClearWaterBoxStatus | None:
         if self.dss:
             return ClearWaterBoxStatus((self.dss >> 2) & 3)
         return None
 
     @property
+    @field_metadata(dock_feature="is_washable")
     def dirty_water_box_status(self) -> DirtyWaterBoxStatus | None:
         if self.dss:
             return DirtyWaterBoxStatus((self.dss >> 4) & 3)
@@ -413,6 +430,9 @@ def water_box_filter_status(self) -> int | None:
         return None
 
     @property
+    @field_metadata(
+        dock_feature="is_clean_fluid_auto_delivery_supported",
+    )
     def clean_fluid_status(self) -> CleanFluidStatus | None:
         if self.dss:
             value = (self.dss >> 10) & 3

roborock/devices/traits/v1/__init__.py

@@ -264,6 +264,7 @@ async def discover_features(self) -> None:
         # Dock type also acts like a device feature for some traits.
         dock_type = await self._dock_type()
         dock_features = RoborockDockFeatures.from_dock_type(dock_type, has_am=self.status.has_am)
+        self.device_features.dock_features = dock_features
 
         # Initialize traits with special arguments before the generic loop
         if self.wash_towel_mode is None and self._is_supported(WashTowelModeTrait, "wash_towel_mode", dock_features):

roborock/devices/traits/v1/device_features.py

@@ -1,12 +1,33 @@
-from dataclasses import Field, fields
+from dataclasses import fields
+from typing import Any
 
 from roborock.data import AppInitStatus, HomeDataProduct, RoborockBase
+from roborock.data.v1 import RoborockDockTypeCode
 from roborock.data.v1.v1_containers import FieldNameBase
-from roborock.device_features import DeviceFeatures
+from roborock.device_features import DeviceFeatures, RoborockDockFeatures
 from roborock.devices.cache import DeviceCache
 from roborock.devices.traits.v1 import common
 from roborock.roborock_typing import RoborockCommand
 
+# Cache of metadata for each trait class
+_metadata_cache: dict[type[RoborockBase], dict[str, dict[str, Any]]] = {}
+
+
+def _get_field_metadata(cls: type[RoborockBase]) -> dict[str, Any]:
+    """Helper to get metadata from either class properties or dataclass fields."""
+    if cls not in _metadata_cache:
+        metadata_map = {}
+        # Inspect properties with @field_metadata
+        for name in dir(cls):
+            prop = getattr(cls, name, None)
+            if isinstance(prop, property):
+                metadata_map[name] = getattr(prop.fget, "_field_metadata", {})
+        # Inspect dataclass fields metadata
+        for f in fields(cls):
+            metadata_map[f.name] = f.metadata
+        _metadata_cache[cls] = metadata_map
+    return _metadata_cache[cls]
+
 
 class DeviceTraitsConverter(common.V1TraitDataConverter):
     """Converter for APP_GET_INIT_STATUS responses into DeviceFeatures."""
@@ -40,31 +61,37 @@ def __init__(self, product: HomeDataProduct, device_cache: DeviceCache) -> None:
         self.converter = DeviceTraitsConverter(product)
         self._product = product
         self._device_cache = device_cache
+        # Dock features are populated after device feature discovery
+        # is triggered.
+        self.dock_features: RoborockDockFeatures = RoborockDockFeatures.from_dock_type(RoborockDockTypeCode.o0_dock)
         # All fields of DeviceFeatures are required. Initialize them to False
         # so we have some known state.
         for field in fields(self):
             setattr(self, field.name, False)
 
-    @staticmethod
-    def _get_dataclass_field(cls: type[RoborockBase], field_name: FieldNameBase) -> Field:
-        """Look up a dataclass field by its FieldNameBase name."""
-        for f in fields(cls):
-            if f.name == field_name:
-                return f
-        raise ValueError(f"Field {field_name!r} not found in {cls}")
-
     def is_field_supported(self, cls: type[RoborockBase], field_name: FieldNameBase) -> bool:
         """Determines if the specified field is supported by this device.
 
-        We use the `dps` dataclass field metadata to get the `RoborockDataProtocol`
-        integer ID and check it against the set of supported schema IDs for the
-        device returned in the product information.
+        We inspect the metadata defined for the field (either via dataclass field metadata
+        or the `@field_metadata` decorator on properties). Supported checks include:
+
+        - `feature`: Maps to a boolean capability property on `DeviceFeatures` / `DeviceFeaturesTrait`
+          (e.g. `is_support_water_mode`).
+        - `dock_feature`: Maps to a boolean capability property on `RoborockDockFeatures` (e.g. `is_washable`).
+        - `dps`: Maps to a `RoborockDataProtocol` ID checked against the product's supported schema IDs.
         """
-        dataclass_field = self._get_dataclass_field(cls, field_name)
-        if (dps := dataclass_field.metadata.get("dps")) is None:
-            # No DPS metadata — field is assumed always supported
-            return True
-        return int(dps) in self._product.supported_schema_ids
+        if self.dock_features is None:
+            raise ValueError("DeviceFeaturesTrait was invoked but was not fully initialized")
+        metadata_map = _get_field_metadata(cls)
+        if (field_metadata := metadata_map.get(field_name)) is not None:
+            if (feature := field_metadata.get("feature")) is not None:
+                return getattr(self, feature, False)
+            if (dock_feature := field_metadata.get("dock_feature")) is not None:
+                return getattr(self.dock_features, dock_feature, False)
+            if (dps := field_metadata.get("dps")) is not None:
+                return int(dps) in self._product.supported_schema_ids
+        # No metadata, field is assumed always supported
+        return True
 
     async def refresh(self) -> None:
         """Refresh the contents of this trait.

tests/devices/traits/v1/__snapshots__/test_device_features.ambr

@@ -3,66 +3,120 @@
   dict({
     'battery': True,
     'charge_status': True,
+    'clean_fluid_status': True,
+    'clean_percent': True,
+    'clear_water_box_status': True,
+    'dirty_water_box_status': True,
+    'dock_error_status': True,
     'dry_status': True,
     'error_code': True,
     'fan_power': True,
+    'rdt': True,
     'state': True,
+    'water_box_carriage_status': True,
     'water_box_mode': True,
+    'water_box_status': True,
+    'water_shortage_status': True,
   })
 # ---
 # name: test_is_attribute_supported[home_data_device_q5_max.json]
   dict({
     'battery': True,
     'charge_status': True,
+    'clean_fluid_status': False,
+    'clean_percent': True,
+    'clear_water_box_status': False,
+    'dirty_water_box_status': False,
+    'dock_error_status': True,
     'dry_status': True,
     'error_code': True,
     'fan_power': True,
+    'rdt': False,
     'state': True,
+    'water_box_carriage_status': True,
     'water_box_mode': True,
+    'water_box_status': True,
+    'water_shortage_status': True,
   })
 # ---
 # name: test_is_attribute_supported[home_data_device_s5e.json]
   dict({
     'battery': True,
     'charge_status': True,
+    'clean_fluid_status': False,
+    'clean_percent': False,
+    'clear_water_box_status': False,
+    'dirty_water_box_status': False,
+    'dock_error_status': False,
     'dry_status': True,
     'error_code': True,
     'fan_power': True,
+    'rdt': False,
     'state': True,
+    'water_box_carriage_status': True,
     'water_box_mode': True,
+    'water_box_status': True,
+    'water_shortage_status': True,
   })
 # ---
 # name: test_is_attribute_supported[home_data_device_s7_maxv.json]
   dict({
     'battery': True,
     'charge_status': True,
+    'clean_fluid_status': False,
+    'clean_percent': False,
+    'clear_water_box_status': True,
+    'dirty_water_box_status': True,
+    'dock_error_status': True,
     'dry_status': True,
     'error_code': True,
     'fan_power': True,
+    'rdt': False,
     'state': True,
+    'water_box_carriage_status': True,
     'water_box_mode': True,
+    'water_box_status': True,
+    'water_shortage_status': True,
   })
 # ---
 # name: test_is_attribute_supported[home_data_device_saros.json]
   dict({
     'battery': True,
     'charge_status': True,
+    'clean_fluid_status': True,
+    'clean_percent': True,
+    'clear_water_box_status': True,
+    'dirty_water_box_status': True,
+    'dock_error_status': True,
     'dry_status': True,
     'error_code': True,
     'fan_power': True,
+    'rdt': True,
     'state': True,
+    'water_box_carriage_status': True,
     'water_box_mode': True,
+    'water_box_status': True,
+    'water_shortage_status': True,
   })
 # ---
 # name: test_is_attribute_supported[home_data_device_saros_10r.json]
   dict({
     'battery': True,
     'charge_status': True,
+    'clean_fluid_status': True,
+    'clean_percent': True,
+    'clear_water_box_status': True,
+    'dirty_water_box_status': True,
+    'dock_error_status': True,
     'dry_status': True,
     'error_code': True,
     'fan_power': True,
+    'rdt': True,
     'state': True,
+    'water_box_carriage_status': True,
     'water_box_mode': True,
+    'water_box_status': True,
+    'water_shortage_status': True,
   })
 # ---
 # name: test_is_consumable_field_supported[home_data_device_pearls.json]