roborock.devices.traits.v1.status

  1import logging
  2from functools import cached_property
  3from typing import Any
  4
  5from roborock import (
  6    CleanRoutes,
  7    StatusV2,
  8    VacuumModes,
  9    WaterModes,
 10    get_clean_modes,
 11    get_clean_routes,
 12    get_water_mode_mapping,
 13    get_water_modes,
 14)
 15from roborock.devices.traits.common import DpsDataConverter, TraitUpdateListener
 16from roborock.roborock_message import RoborockDataProtocol
 17from roborock.roborock_typing import RoborockCommand
 18
 19from . import common
 20from .device_features import DeviceFeaturesTrait
 21
 22_LOGGER = logging.getLogger(__name__)
 23
 24_DPS_CONVERTER = DpsDataConverter.from_dataclass(StatusV2)
 25
 26
 27class StatusTrait(StatusV2, common.V1TraitMixin, TraitUpdateListener):
 28    """Trait for managing the status of Roborock devices.
 29
 30    The StatusTrait gives you the access to the state of a Roborock vacuum.
 31    The various attribute options on state change per each device.
 32    Values like fan speed, mop mode, etc. have different options for every device
 33    and are dynamically determined.
 34
 35    Usage:
 36        Before accessing status properties, you should call `refresh()` to fetch
 37        the latest data from the device. You must pass in the device feature trait
 38        to this trait so that the dynamic attributes can be pre-determined.
 39
 40    The current dynamic attributes are:
 41    - Fan Speed
 42    - Water Mode
 43    - Mop Route
 44
 45    You should call the _options() version of the attribute to know which are supported for your device
 46    (i.e. fan_speed_options())
 47    Then you can call the _mapping to convert an int value to the actual Enum. (i.e. fan_speed_mapping())
 48    You can call the _name property to get the str value of the enum. (i.e. fan_speed_name)
 49
 50    """
 51
 52    command = RoborockCommand.GET_STATUS
 53    converter = common.DefaultConverter(StatusV2)
 54
 55    def __init__(self, device_feature_trait: DeviceFeaturesTrait, region: str | None = None) -> None:
 56        """Initialize the StatusTrait."""
 57        super().__init__()
 58        TraitUpdateListener.__init__(self, logger=_LOGGER)
 59        self._device_features_trait = device_feature_trait
 60        self._region = region
 61
 62    @cached_property
 63    def fan_speed_options(self) -> list[VacuumModes]:
 64        return get_clean_modes(self._device_features_trait)
 65
 66    @cached_property
 67    def fan_speed_mapping(self) -> dict[int, str]:
 68        return {fan.code: fan.value for fan in self.fan_speed_options}
 69
 70    @cached_property
 71    def water_mode_options(self) -> list[WaterModes]:
 72        return get_water_modes(self._device_features_trait)
 73
 74    @cached_property
 75    def water_mode_mapping(self) -> dict[int, str]:
 76        return get_water_mode_mapping(self._device_features_trait)
 77
 78    @cached_property
 79    def mop_route_options(self) -> list[CleanRoutes]:
 80        return get_clean_routes(self._device_features_trait, self._region or "us")
 81
 82    @cached_property
 83    def mop_route_mapping(self) -> dict[int, str]:
 84        return {route.code: route.value for route in self.mop_route_options}
 85
 86    @property
 87    def fan_speed_name(self) -> str | None:
 88        if self.fan_power is None:
 89            return None
 90        return self.fan_speed_mapping.get(self.fan_power)
 91
 92    @property
 93    def water_mode_name(self) -> str | None:
 94        if self.water_box_mode is None:
 95            return None
 96        return self.water_mode_mapping.get(self.water_box_mode)
 97
 98    @property
 99    def mop_route_name(self) -> str | None:
100        if self.mop_mode is None:
101            return None
102        return self.mop_route_mapping.get(self.mop_mode)
103
104    def update_from_dps(self, decoded_dps: dict[RoborockDataProtocol, Any]) -> None:
105        """Update the trait from data protocol push message data.
106
107        This handles unsolicited status updates pushed by the device
108        via RoborockDataProtocol codes (e.g. STATE=121, BATTERY=122).
109        """
110        if _DPS_CONVERTER.update_from_dps(self, decoded_dps):
111            self._notify_update()