chore: do dynamic status

Author: Lash-L

roborock/data/v1/v1_containers.py

@@ -92,6 +92,8 @@
 
 @dataclass
 class Status(RoborockBase):
+    """This status will be depreciated in favor of StatusV2."""
+
     msg_ver: int | None = None
     msg_seq: int | None = None
     state: RoborockStateCode | None = None
@@ -250,6 +252,137 @@ def __repr__(self) -> str:
         return _attr_repr(self)
 
 
+class StatusV2(RoborockBase):
+    """
+    This is a new version of the Status object.
+    This is the result of GET_STATUS from the api.
+    """
+
+    msg_ver: int | None = None
+    msg_seq: int | None = None
+    state: RoborockStateCode | None = None
+    battery: int | None = None
+    clean_time: int | None = None
+    clean_area: int | None = None
+    error_code: RoborockErrorCode | None = None
+    map_present: int | None = None
+    in_cleaning: RoborockInCleaning | None = None
+    in_returning: int | None = None
+    in_fresh_state: int | None = None
+    lab_status: int | None = None
+    water_box_status: int | None = None
+    back_type: int | None = None
+    wash_phase: int | None = None
+    wash_ready: int | None = None
+    fan_power: int | None = None
+    dnd_enabled: int | None = None
+    map_status: int | None = None
+    is_locating: int | None = None
+    lock_status: int | None = None
+    water_box_mode: int | None = None
+    water_box_carriage_status: int | None = None
+    mop_forbidden_enable: int | None = None
+    camera_status: int | None = None
+    is_exploring: int | None = None
+    home_sec_status: int | None = None
+    home_sec_enable_password: int | None = None
+    adbumper_status: list[int] | None = None
+    water_shortage_status: int | None = None
+    dock_type: RoborockDockTypeCode | None = None
+    dust_collection_status: int | None = None
+    auto_dust_collection: int | None = None
+    avoid_count: int | None = None
+    mop_mode: int | None = None
+    debug_mode: int | None = None
+    collision_avoid_status: int | None = None
+    switch_map_mode: int | None = None
+    dock_error_status: RoborockDockErrorCode | None = None
+    charge_status: int | None = None
+    unsave_map_reason: int | None = None
+    unsave_map_flag: int | None = None
+    wash_status: int | None = None
+    distance_off: int | None = None
+    in_warmup: int | None = None
+    dry_status: int | None = None
+    rdt: int | None = None
+    clean_percent: int | None = None
+    rss: int | None = None
+    dss: int | None = None
+    common_status: int | None = None
+    corner_clean_mode: int | None = None
+    last_clean_t: int | None = None
+    replenish_mode: int | None = None
+    repeat: int | None = None
+    kct: int | None = None
+    subdivision_sets: int | None = None
+
+    @property
+    def square_meter_clean_area(self) -> float | None:
+        return round(self.clean_area / 1000000, 1) if self.clean_area is not None else None
+
+    @property
+    def error_code_name(self) -> str | None:
+        return self.error_code.name if self.error_code is not None else None
+
+    @property
+    def state_name(self) -> str | None:
+        return self.state.name if self.state is not None else None
+
+    @property
+    def current_map(self) -> int | None:
+        """Returns the current map ID if the map is present."""
+        if self.map_status is not None:
+            map_flag = self.map_status >> 2
+            if map_flag != NO_MAP:
+                return map_flag
+        return None
+
+    @property
+    def clear_water_box_status(self) -> ClearWaterBoxStatus | None:
+        if self.dss:
+            return ClearWaterBoxStatus((self.dss >> 2) & 3)
+        return None
+
+    @property
+    def dirty_water_box_status(self) -> DirtyWaterBoxStatus | None:
+        if self.dss:
+            return DirtyWaterBoxStatus((self.dss >> 4) & 3)
+        return None
+
+    @property
+    def dust_bag_status(self) -> DustBagStatus | None:
+        if self.dss:
+            return DustBagStatus((self.dss >> 6) & 3)
+        return None
+
+    @property
+    def water_box_filter_status(self) -> int | None:
+        if self.dss:
+            return (self.dss >> 8) & 3
+        return None
+
+    @property
+    def clean_fluid_status(self) -> int | None:
+        if self.dss:
+            return (self.dss >> 10) & 3
+        return None
+
+    @property
+    def hatch_door_status(self) -> int | None:
+        if self.dss:
+            return (self.dss >> 12) & 7
+        return None
+
+    @property
+    def dock_cool_fan_status(self) -> int | None:
+        if self.dss:
+            return (self.dss >> 15) & 3
+        return None
+
+    def __repr__(self) -> str:
+        return _attr_repr(self)
+
+
 @dataclass
 class S4MaxStatus(Status):
     fan_power: RoborockFanSpeedS6Pure | None = None

roborock/devices/traits/v1/__init__.py

@@ -141,14 +141,13 @@ def __init__(
         self._map_rpc_channel = map_rpc_channel
         self._web_api = web_api
         self._cache = cache
-
-        self.status = StatusTrait(product)
+        self.device_features = DeviceFeaturesTrait(product.product_nickname, cache)
+        self.status = StatusTrait(self.device_features)
         self.consumables = ConsumableTrait()
         self.rooms = RoomsTrait(home_data)
         self.maps = MapsTrait(self.status)
         self.map_content = MapContentTrait(map_parser_config)
         self.home = HomeTrait(self.status, self.maps, self.map_content, self.rooms, cache)
-        self.device_features = DeviceFeaturesTrait(product.product_nickname, cache)
         self.network_info = NetworkInfoTrait(device_uid, cache)
         self.routines = RoutinesTrait(device_uid, web_api)
 

roborock/devices/traits/v1/status.py

@@ -1,24 +1,66 @@
+from functools import cached_property
 from typing import Self
 
-from roborock.data import HomeDataProduct, ModelStatus, S7MaxVStatus, Status
-from roborock.devices.traits.v1 import common
+from roborock import CleanRoutes, StatusV2, VacuumModes, WaterModes, get_clean_modes, get_clean_routes, get_water_modes
+from roborock.devices.traits.v1 import DeviceFeaturesTrait, common
 from roborock.roborock_typing import RoborockCommand
 
 
-class StatusTrait(Status, common.V1TraitMixin):
+class StatusTrait(StatusV2, common.V1TraitMixin):
     """Trait for managing the status of Roborock devices."""
 
     command = RoborockCommand.GET_STATUS
 
-    def __init__(self, product_info: HomeDataProduct) -> None:
+    def __init__(self, device_feature_trait: DeviceFeaturesTrait) -> None:
         """Initialize the StatusTrait."""
-        self._product_info = product_info
+        self._device_features_trait = device_feature_trait
+
+    @cached_property
+    def fan_speed_options(self) -> list[VacuumModes]:
+        return get_clean_modes(self._device_features_trait)
+
+    @cached_property
+    def fan_speed_mapping(self) -> dict[int, str]:
+        return {fan.code: fan.name for fan in self.fan_speed_options}
+
+    @cached_property
+    def water_mode_options(self) -> list[WaterModes]:
+        return get_water_modes(self._device_features_trait)
+
+    @cached_property
+    def water_mode_mapping(self) -> dict[int, str]:
+        return {mop.code: mop.name for mop in self.water_mode_options}
+
+    @cached_property
+    def mop_route_options(self) -> list[CleanRoutes]:
+        return get_clean_routes(self._device_features_trait, "US")  # TODO: find best place to get region
+
+    @cached_property
+    def mop_route_mapping(self) -> dict[int, str]:
+        return {route.code: route.name for route in self.mop_route_options}
+
+    @property
+    def fan_speed_name(self) -> str | None:
+        if self.fan_power is None:
+            return None
+        return self.fan_speed_mapping.get(self.fan_power)
+
+    @property
+    def water_mode_name(self) -> str | None:
+        if self.water_box_mode is None:
+            return None
+        return self.water_mode_mapping.get(self.water_box_mode)
+
+    @property
+    def mop_route_name(self) -> str | None:
+        if self.mop_mode is None:
+            return None
+        return self.mop_route_mapping.get(self.mop_mode)
 
     def _parse_response(self, response: common.V1ResponseData) -> Self:
         """Parse the response from the device into a CleanSummary."""
-        status_type: type[Status] = ModelStatus.get(self._product_info.model, S7MaxVStatus)
         if isinstance(response, list):
             response = response[0]
         if isinstance(response, dict):
-            return status_type.from_dict(response)
+            return StatusV2.from_dict(response)
         raise ValueError(f"Unexpected status format: {response!r}")