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    RoborockDockState,
  9    StatusV2,
 10    VacuumModes,
 11    WaterModes,
 12    get_clean_modes,
 13    get_clean_routes,
 14    get_cleaning_mode_options,
 15    get_cleaning_mode_parameters,
 16    get_current_cleaning_mode,
 17    get_dock_state,
 18    get_water_mode_mapping,
 19    get_water_modes,
 20    get_water_slide_mode_name,
 21    resolve_cleaning_mode,
 22)
 23from roborock.devices.traits.common import DpsDataConverter, TraitUpdateListener
 24from roborock.roborock_message import RoborockDataProtocol
 25from roborock.roborock_typing import RoborockCommand
 26
 27from . import common
 28from .device_features import DeviceFeaturesTrait
 29
 30_LOGGER = logging.getLogger(__name__)
 31
 32_DPS_CONVERTER = DpsDataConverter.from_dataclass(StatusV2)
 33
 34
 35class StatusTrait(StatusV2, common.V1TraitMixin, TraitUpdateListener):
 36    """Trait for managing the status of Roborock devices.
 37
 38    The StatusTrait gives you the access to the state of a Roborock vacuum.
 39    The various attribute options on state change per each device.
 40    Values like fan speed, mop mode, etc. have different options for every device
 41    and are dynamically determined.
 42
 43    Usage:
 44        Before accessing status properties, you should call `refresh()` to fetch
 45        the latest data from the device. You must pass in the device feature trait
 46        to this trait so that the dynamic attributes can be pre-determined.
 47
 48    The current dynamic attributes are:
 49    - Fan Speed
 50    - Water Mode
 51    - Mop Route
 52
 53    You should use the _options version of the attribute to know which are
 54    supported for your device (i.e. fan_speed_options)
 55    Then you can use the _mapping to convert an int value to the actual Enum.
 56    (i.e. fan_speed_mapping)
 57    You can use the _name property to get the str value of the enum. (i.e. fan_speed_name)
 58
 59    """
 60
 61    command = RoborockCommand.GET_STATUS
 62    converter = common.DefaultConverter(StatusV2)
 63
 64    def __init__(self, device_feature_trait: DeviceFeaturesTrait, region: str | None = None) -> None:
 65        """Initialize the StatusTrait."""
 66        super().__init__()
 67        TraitUpdateListener.__init__(self, logger=_LOGGER)
 68        self._device_features_trait = device_feature_trait
 69        self._region = region
 70
 71    @property
 72    def dock_state(self) -> RoborockDockState:
 73        """Return the feature-aware dock state used by the Roborock app."""
 74        return get_dock_state(
 75            state=self.state,
 76            battery=self.battery,
 77            charge_status=self.charge_status,
 78            is_supported_valley_electricity=self._device_features_trait.is_supported_valley_electricity,
 79        )
 80
 81    @property
 82    def is_battery_charging(self) -> bool:
 83        """Return whether the battery is actively charging."""
 84        return self.dock_state == RoborockDockState.charging
 85
 86    @cached_property
 87    def fan_speed_options(self) -> list[VacuumModes]:
 88        return get_clean_modes(self._device_features_trait)
 89
 90    @cached_property
 91    def fan_speed_mapping(self) -> dict[int, str]:
 92        return {fan.code: fan.value for fan in self.fan_speed_options}
 93
 94    @cached_property
 95    def water_mode_options(self) -> list[WaterModes]:
 96        return get_water_modes(self._device_features_trait)
 97
 98    @cached_property
 99    def water_mode_mapping(self) -> dict[int, str]:
100        return get_water_mode_mapping(self._device_features_trait)
101
102    @cached_property
103    def mop_route_options(self) -> list[CleanRoutes]:
104        return get_clean_routes(self._device_features_trait, self._region or "us")
105
106    @cached_property
107    def mop_route_mapping(self) -> dict[int, str]:
108        return {route.code: route.display_name for route in self.mop_route_options}
109
110    @cached_property
111    def cleaning_mode_options(self) -> list[CleaningMode]:
112        return get_cleaning_mode_options(self._device_features_trait)
113
114    @property
115    def fan_speed_name(self) -> str | None:
116        if self.fan_power is None:
117            return None
118        return self.fan_speed_mapping.get(self.fan_power)
119
120    @property
121    def water_mode_name(self) -> str | None:
122        if self.water_box_mode is None:
123            return None
124        if self._device_features_trait.is_water_slide_mode_supported:
125            return get_water_slide_mode_name(self.water_box_mode)
126        return self.water_mode_mapping.get(self.water_box_mode)
127
128    @property
129    def mop_route_name(self) -> str | None:
130        if self.mop_mode is None:
131            return None
132        return self.mop_route_mapping.get(self.mop_mode)
133
134    @property
135    def current_cleaning_mode(self) -> CleaningMode | None:
136        return get_current_cleaning_mode(
137            clean_mode=self.fan_power,
138            water_mode=self.water_box_mode,
139            mop_mode=self.mop_mode,
140            features=self._device_features_trait,
141        )
142
143    @property
144    def current_cleaning_mode_name(self) -> str | None:
145        if (cleaning_mode := self.current_cleaning_mode) is None:
146            return None
147        return cleaning_mode.value
148
149    async def set_cleaning_mode(self, cleaning_mode: str | CleaningMode) -> None:
150        """Set the preferred high-level cleaning mode for the device."""
151        await self.rpc_channel.send_command(
152            RoborockCommand.SET_CLEAN_MOTOR_MODE,
153            params=get_cleaning_mode_parameters(resolve_cleaning_mode(cleaning_mode), self._device_features_trait),
154        )
155
156    async def resolve_error(self, error_code: int | None = None) -> None:
157        """Resolve an error, like tapping "Resolved" in the Roborock app.
158
159        Without an explicit code this resolves the current dock error if there is one,
160        otherwise the current robot error. A dock error such as water_empty stays latched
161        on the device until it is resolved, even after the cause has been fixed.
162        """
163        if error_code is None:
164            if self.dock_error_status:
165                error_code = int(self.dock_error_status)
166            elif self.error_code:
167                error_code = int(self.error_code)
168            else:
169                return
170        await self.rpc_channel.send_command(RoborockCommand.RESOLVE_ERROR, params={"error_code": error_code})
171        await self.refresh()
172
173    def update_from_dps(self, decoded_dps: dict[RoborockDataProtocol, Any]) -> None:
174        """Update the trait from data protocol push message data.
175
176        This handles unsolicited status updates pushed by the device
177        via RoborockDataProtocol codes (e.g. STATE=121, BATTERY=122).
178        """
179        if _DPS_CONVERTER.update_from_dps(self, decoded_dps):
180            self._notify_update()