roborock.devices.traits.v1.status

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