roborock.devices.traits.v1.device_features

 1from dataclasses import Field, fields
 2
 3from roborock.data import AppInitStatus, HomeDataProduct, RoborockBase
 4from roborock.data.v1.v1_containers import FieldNameBase
 5from roborock.device_features import DeviceFeatures
 6from roborock.devices.cache import DeviceCache
 7from roborock.devices.traits.v1 import common
 8from roborock.roborock_typing import RoborockCommand
 9
10
11class DeviceTraitsConverter(common.V1TraitDataConverter):
12    """Converter for APP_GET_INIT_STATUS responses into DeviceFeatures."""
13
14    def __init__(self, product: HomeDataProduct) -> None:
15        """Initialize DeviceTraitsConverter."""
16        self._product = product
17
18    def convert(self, response: common.V1ResponseData) -> DeviceFeatures:
19        """Parse an APP_GET_INIT_STATUS response into a DeviceFeatures instance."""
20        if not isinstance(response, list):
21            raise ValueError(f"Unexpected AppInitStatus response format: {type(response)}: {response!r}")
22        app_status = AppInitStatus.from_dict(response[0])
23        return DeviceFeatures.from_feature_flags(
24            new_feature_info=app_status.new_feature_info,
25            new_feature_info_str=app_status.new_feature_info_str,
26            feature_info=app_status.feature_info,
27            product_nickname=self._product.product_nickname,
28        )
29
30
31class DeviceFeaturesTrait(DeviceFeatures, common.V1TraitMixin):
32    """Trait for managing supported features on Roborock devices."""
33
34    command = RoborockCommand.APP_GET_INIT_STATUS
35    converter: DeviceTraitsConverter
36
37    def __init__(self, product: HomeDataProduct, device_cache: DeviceCache) -> None:  # pylint: disable=super-init-not-called
38        """Initialize DeviceFeaturesTrait."""
39        common.V1TraitMixin.__init__(self)
40        self.converter = DeviceTraitsConverter(product)
41        self._product = product
42        self._device_cache = device_cache
43        # All fields of DeviceFeatures are required. Initialize them to False
44        # so we have some known state.
45        for field in fields(self):
46            setattr(self, field.name, False)
47
48    @staticmethod
49    def _get_dataclass_field(cls: type[RoborockBase], field_name: FieldNameBase) -> Field:
50        """Look up a dataclass field by its FieldNameBase name."""
51        for f in fields(cls):
52            if f.name == field_name:
53                return f
54        raise ValueError(f"Field {field_name!r} not found in {cls}")
55
56    def is_field_supported(self, cls: type[RoborockBase], field_name: FieldNameBase) -> bool:
57        """Determines if the specified field is supported by this device.
58
59        We use the `dps` dataclass field metadata to get the `RoborockDataProtocol`
60        integer ID and check it against the set of supported schema IDs for the
61        device returned in the product information.
62        """
63        dataclass_field = self._get_dataclass_field(cls, field_name)
64        if (dps := dataclass_field.metadata.get("dps")) is None:
65            # No DPS metadata — field is assumed always supported
66            return True
67        return int(dps) in self._product.supported_schema_ids
68
69    async def refresh(self) -> None:
70        """Refresh the contents of this trait.
71
72        This will use cached device features if available since they do not
73        change often and this avoids unnecessary RPC calls. This would only
74        ever change with a firmware update, so caching is appropriate.
75        """
76        cache_data = await self._device_cache.get()
77        if cache_data.device_features is not None:
78            common.merge_trait_values(self, cache_data.device_features)
79            return
80        # Save cached device features
81        await super().refresh()
82        cache_data.device_features = self
83        await self._device_cache.set(cache_data)