Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 19 additions & 0 deletions roborock/data/containers.py
Original file line number Diff line number Diff line change
Expand Up @@ -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."""
Expand Down
34 changes: 27 additions & 7 deletions roborock/data/v1/v1_containers.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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)
Expand All @@ -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
Expand Down Expand Up @@ -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
Expand All @@ -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
Expand All @@ -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}
)
Expand All @@ -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
Expand Down Expand Up @@ -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)
Expand All @@ -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
Expand Down
1 change: 1 addition & 0 deletions roborock/devices/traits/v1/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Expand Down
63 changes: 45 additions & 18 deletions roborock/devices/traits/v1/device_features.py
Original file line number Diff line number Diff line change
@@ -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."""
Expand Down Expand Up @@ -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.
Expand Down
54 changes: 54 additions & 0 deletions tests/devices/traits/v1/__snapshots__/test_device_features.ambr
Original file line number Diff line number Diff line change
Expand Up @@ -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]
Expand Down
Loading
Loading