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 DeviceFeaturesTrait(DeviceFeatures, common.V1TraitMixin):
12    """Trait for managing supported features on Roborock devices."""
13
14    command = RoborockCommand.APP_GET_INIT_STATUS
15
16    def __init__(self, product: HomeDataProduct, device_cache: DeviceCache) -> None:  # pylint: disable=super-init-not-called
17        """Initialize DeviceFeaturesTrait."""
18        self._product = product
19        self._nickname = product.product_nickname
20        self._device_cache = device_cache
21        # All fields of DeviceFeatures are required. Initialize them to False
22        # so we have some known state.
23        for field in fields(self):
24            setattr(self, field.name, False)
25
26    def is_field_supported(self, cls: type[RoborockBase], field_name: FieldNameBase) -> bool:
27        """Determines if the specified field is supported by this device.
28
29        We use dataclass attributes on the field to specify the schema code that is required
30        for the field to be supported and it is compared against the list of
31        supported schema codes for the device returned in the product information.
32        """
33        dataclass_field: Field | None = None
34        for field in fields(cls):
35            if field.name == field_name:
36                dataclass_field = field
37                break
38        if dataclass_field is None:
39            raise ValueError(f"Field {field_name} not found in {cls}")
40
41        requires_schema_code = dataclass_field.metadata.get("requires_schema_code", None)
42        if requires_schema_code is None:
43            # We assume the field is supported
44            return True
45        # If the field requires a protocol that is not supported, we return False
46        return requires_schema_code in self._product.supported_schema_codes
47
48    async def refresh(self) -> None:
49        """Refresh the contents of this trait.
50
51        This will use cached device features if available since they do not
52        change often and this avoids unnecessary RPC calls. This would only
53        ever change with a firmware update, so caching is appropriate.
54        """
55        cache_data = await self._device_cache.get()
56        if cache_data.device_features is not None:
57            self._update_trait_values(cache_data.device_features)
58            return
59        # Save cached device features
60        await super().refresh()
61        cache_data.device_features = self
62        await self._device_cache.set(cache_data)
63
64    def _parse_response(self, response: common.V1ResponseData) -> DeviceFeatures:
65        """Parse the response from the device into a MapContentTrait instance."""
66        if not isinstance(response, list):
67            raise ValueError(f"Unexpected AppInitStatus response format: {type(response)}")
68        app_status = AppInitStatus.from_dict(response[0])
69        return DeviceFeatures.from_feature_flags(
70            new_feature_info=app_status.new_feature_info,
71            new_feature_info_str=app_status.new_feature_info_str,
72            feature_info=app_status.feature_info,
73            product_nickname=self._nickname,
74        )