roborock.devices.traits.v1.device_features

  1from dataclasses import fields
  2from typing import Any
  3
  4from roborock.data import AppInitStatus, HomeDataProduct, RoborockBase
  5from roborock.data.v1 import RoborockDockTypeCode
  6from roborock.data.v1.v1_containers import FieldNameBase
  7from roborock.device_features import DeviceFeatures, RoborockDockFeatures
  8from roborock.devices.cache import DeviceCache
  9from roborock.devices.traits.v1 import common
 10from roborock.roborock_typing import RoborockCommand
 11
 12# Cache of metadata for each trait class
 13_metadata_cache: dict[type[RoborockBase], dict[str, dict[str, Any]]] = {}
 14
 15
 16def _get_field_metadata(cls: type[RoborockBase]) -> dict[str, Any]:
 17    """Helper to get metadata from either class properties or dataclass fields."""
 18    if cls not in _metadata_cache:
 19        metadata_map = {}
 20        # Inspect properties with @field_metadata
 21        for name in dir(cls):
 22            prop = getattr(cls, name, None)
 23            if isinstance(prop, property):
 24                metadata_map[name] = getattr(prop.fget, "_field_metadata", {})
 25        # Inspect dataclass fields metadata
 26        for f in fields(cls):
 27            metadata_map[f.name] = f.metadata
 28        _metadata_cache[cls] = metadata_map
 29    return _metadata_cache[cls]
 30
 31
 32class DeviceTraitsConverter(common.V1TraitDataConverter):
 33    """Converter for APP_GET_INIT_STATUS responses into DeviceFeatures."""
 34
 35    def __init__(self, product: HomeDataProduct) -> None:
 36        """Initialize DeviceTraitsConverter."""
 37        self._product = product
 38
 39    def convert(self, response: common.V1ResponseData) -> DeviceFeatures:
 40        """Parse an APP_GET_INIT_STATUS response into a DeviceFeatures instance."""
 41        if not isinstance(response, list):
 42            raise ValueError(f"Unexpected AppInitStatus response format: {type(response)}: {response!r}")
 43        app_status = AppInitStatus.from_dict(response[0])
 44        return DeviceFeatures.from_feature_flags(
 45            new_feature_info=app_status.new_feature_info,
 46            new_feature_info_str=app_status.new_feature_info_str,
 47            feature_info=app_status.feature_info,
 48            product_nickname=self._product.product_nickname,
 49        )
 50
 51
 52class DeviceFeaturesTrait(DeviceFeatures, common.V1TraitMixin):
 53    """Trait for managing supported features on Roborock devices."""
 54
 55    command = RoborockCommand.APP_GET_INIT_STATUS
 56    converter: DeviceTraitsConverter
 57
 58    def __init__(self, product: HomeDataProduct, device_cache: DeviceCache) -> None:  # pylint: disable=super-init-not-called
 59        """Initialize DeviceFeaturesTrait."""
 60        common.V1TraitMixin.__init__(self)
 61        self.converter = DeviceTraitsConverter(product)
 62        self._product = product
 63        self._device_cache = device_cache
 64        # Dock features are populated after device feature discovery
 65        # is triggered.
 66        self.dock_features: RoborockDockFeatures = RoborockDockFeatures.from_dock_type(RoborockDockTypeCode.o0_dock)
 67        # All fields of DeviceFeatures are required. Initialize them to False
 68        # so we have some known state.
 69        for field in fields(self):
 70            setattr(self, field.name, False)
 71
 72    def is_field_supported(self, cls: type[RoborockBase], field_name: FieldNameBase) -> bool:
 73        """Determines if the specified field is supported by this device.
 74
 75        We inspect the metadata defined for the field (either via dataclass field metadata
 76        or the `@field_metadata` decorator on properties). Supported checks include:
 77
 78        - `feature`: Maps to a boolean capability property on `DeviceFeatures` / `DeviceFeaturesTrait`
 79          (e.g. `is_support_water_mode`).
 80        - `dock_feature`: Maps to a boolean capability property on `RoborockDockFeatures` (e.g. `is_washable`).
 81        - `dps`: Maps to a `RoborockDataProtocol` ID checked against the product's supported schema IDs.
 82        """
 83        if self.dock_features is None:
 84            raise ValueError("DeviceFeaturesTrait was invoked but was not fully initialized")
 85        metadata_map = _get_field_metadata(cls)
 86        if (field_metadata := metadata_map.get(field_name)) is not None:
 87            if (feature := field_metadata.get("feature")) is not None:
 88                return getattr(self, feature, False)
 89            if (dock_feature := field_metadata.get("dock_feature")) is not None:
 90                return getattr(self.dock_features, dock_feature, False)
 91            if (dps := field_metadata.get("dps")) is not None:
 92                return int(dps) in self._product.supported_schema_ids
 93        # No metadata, field is assumed always supported
 94        return True
 95
 96    async def refresh(self) -> None:
 97        """Refresh the contents of this trait.
 98
 99        This will use cached device features if available since they do not
100        change often and this avoids unnecessary RPC calls. This would only
101        ever change with a firmware update, so caching is appropriate.
102        """
103        cache_data = await self._device_cache.get()
104        if cache_data.device_features is not None:
105            common.merge_trait_values(self, cache_data.device_features)
106            return
107        # Save cached device features
108        await super().refresh()
109        cache_data.device_features = self
110        await self._device_cache.set(cache_data)