Skip to content

feat: add dynamic changing status #511

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
wants to merge 2 commits into
base: main
Choose a base branch
from
Draft

feat: add dynamic changing status #511

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
53b2676
feat: add dynamic changing status
Lash-L Oct 2, 2025
fc975c5
chore: remove unrelated code
Lash-L Oct 2, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
70 changes: 41 additions & 29 deletions roborock/containers.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,15 @@
from functools import cached_property
from typing import Any, NamedTuple, get_args, get_origin

from .clean_modes import (
CleanRoutes,
VacuumModes,
VacuumModesOld,
WaterModes,
get_clean_modes,
get_clean_routes,
get_water_modes,
)
from .code_mappings import (
SHORT_MODEL_TO_ENUM,
RoborockCategory,
Expand All @@ -19,7 +28,6 @@
RoborockDockTypeCode,
RoborockDockWashTowelModeCode,
RoborockErrorCode,
RoborockFanPowerCode,
RoborockFanSpeedP10,
RoborockFanSpeedQ7Max,
RoborockFanSpeedQRevoCurv,
Expand All @@ -34,7 +42,6 @@
RoborockFinishReason,
RoborockInCleaning,
RoborockModeEnum,
RoborockMopIntensityCode,
RoborockMopIntensityP10,
RoborockMopIntensityQ7Max,
RoborockMopIntensityQRevoCurv,
Expand All @@ -46,7 +53,6 @@
RoborockMopIntensityS8MaxVUltra,
RoborockMopIntensitySaros10,
RoborockMopIntensitySaros10R,
RoborockMopModeCode,
RoborockMopModeQRevoCurv,
RoborockMopModeQRevoMaster,
RoborockMopModeQRevoMaxV,
Expand Down Expand Up @@ -92,7 +98,6 @@
ROBOROCK_G20S_Ultra,
)
from .device_features import DeviceFeatures
from .exceptions import RoborockException

_LOGGER = logging.getLogger(__name__)

Expand Down Expand Up @@ -357,12 +362,12 @@ class Status(RoborockBase):
back_type: int | None = None
wash_phase: int | None = None
wash_ready: int | None = None
fan_power: RoborockFanPowerCode | 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: RoborockMopIntensityCode | 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
Expand All @@ -375,7 +380,7 @@ class Status(RoborockBase):
dust_collection_status: int | None = None
auto_dust_collection: int | None = None
avoid_count: int | None = None
mop_mode: RoborockMopModeCode | None = None
mop_mode: int | None = None
debug_mode: int | None = None
collision_avoid_status: int | None = None
switch_map_mode: int | None = None
Expand Down Expand Up @@ -406,28 +411,6 @@ def __post_init__(self) -> None:
self.error_code_name = self.error_code.name
if self.state is not None:
self.state_name = self.state.name
if self.water_box_mode is not None:
self.water_box_mode_name = self.water_box_mode.name
if self.fan_power is not None:
self.fan_power_options = self.fan_power.keys()
self.fan_power_name = self.fan_power.name
if self.mop_mode is not None:
self.mop_mode_name = self.mop_mode.name

def get_fan_speed_code(self, fan_speed: str) -> int:
if self.fan_power is None:
raise RoborockException("Attempted to get fan speed before status has been updated.")
return self.fan_power.as_dict().get(fan_speed)

def get_mop_intensity_code(self, mop_intensity: str) -> int:
if self.water_box_mode is None:
raise RoborockException("Attempted to get mop_intensity before status has been updated.")
return self.water_box_mode.as_dict().get(mop_intensity)

def get_mop_mode_code(self, mop_mode: str) -> int:
if self.mop_mode is None:
raise RoborockException("Attempted to get mop_mode before status has been updated.")
return self.mop_mode.as_dict().get(mop_mode)

@property
def current_map(self) -> int | None:
Expand Down Expand Up @@ -574,6 +557,34 @@ class Saros10Status(Status):
}


def get_custom_status(features: DeviceFeatures, region: str) -> "DeviceStatus":
_available_fan_speeds = get_clean_modes(features)
_available_fan_speed_mapping = {fan.code: fan.name for fan in _available_fan_speeds}
_available_water_modes = get_water_modes(features)
_available_water_modes_mapping = {mop.code: mop.name for mop in _available_water_modes}
_available_mop_routes = get_clean_routes(features, region)
_available_mop_routes_mapping = {route.code: route.name for route in _available_mop_routes}

class DeviceStatus(Status):
Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think this needs to be a protocol or something? Was hoping you may have some ideas on how to improve it

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I can see why this is set up this way.

Does it make sense for DeviceStatus to be a top level object with a DeviceFeatures + region as input to constructor?

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

So the issue is right now, every instance of status that we do we use from_dict()

So we'd need to add it to from_dict() and constructor. Theoretically fine, but potentially bloats from_dict() calls for status so i wasn't sure.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This seems similar to the traits that need to parse room names. We basically have a hacky dance to make this work:

  • RoomsTrait is constructed with home data passed to constructor and no values initially
  • On update, it cals _parse_data which parses the response object and merges with the home data
  • Parent class uses setattr to update the object

Perhaps this could just be added to StatusTrait?

available_fan_speeds: list[VacuumModes] | list[VacuumModesOld] = _available_fan_speeds
available_water_modes: list[WaterModes] = _available_water_modes
available_mop_routes: list[CleanRoutes] = _available_mop_routes

@property
def fan_speed(self) -> VacuumModes | None:
return _available_fan_speed_mapping.get(self.fan_power)

@property
def water_mode(self) -> WaterModes | None:
return _available_water_modes_mapping.get(self.water_box_mode)

@property
def mop_route(self) -> CleanRoutes | None:
return _available_mop_routes_mapping.get(self.mop_mode)

return DeviceStatus


@dataclass
class DnDTimer(RoborockBaseTimer):
"""DnDTimer"""
Expand Down Expand Up @@ -762,6 +773,7 @@ class DeviceData(RoborockBase):
host: str | None = None
product_nickname: RoborockProductNickname | None = None
device_features: DeviceFeatures | None = None
region: str | None = None

def __post_init__(self):
self.product_nickname = SHORT_MODEL_TO_ENUM.get(self.model.split(".")[-1], RoborockProductNickname.PEARLPLUS)
Expand Down
6 changes: 3 additions & 3 deletions roborock/devices/traits/v1/status.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
from typing import Self

from roborock.containers import HomeDataProduct, ModelStatus, S7MaxVStatus, Status
from roborock.containers import HomeDataProduct, Status, get_custom_status
from roborock.devices.traits.v1 import common
from roborock.roborock_typing import RoborockCommand

Expand All @@ -13,12 +13,12 @@ class StatusTrait(Status, common.V1TraitMixin):
def __init__(self, product_info: HomeDataProduct) -> None:
"""Initialize the StatusTrait."""
self._product_info = product_info
self._status_type = get_custom_status(self.device_info.device_features, self.device_info.region)
Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is where it gets weird - To determine the status type, we need the device features, which we get from init status, which we need to be connected for.

Hoping you may have some ideas here

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can device features be provided similar to how HomeDataProduct or network info are supplied? Do it up front, and we provide an interface for caching it.

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

That's what i thought about doing but I don't know if it is possible. The Trait objects are created synchronously before we are subscribed and we could send a message to get the init status.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Maybe making a separate trait for now can work, then once its loaded it can be used? This could override refresh to refresh it first before status is loaded the first time.

Separately: Is making it part of Status required?

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Separately: Is making it part of Status required?

Not explicitly, no, but i figured it would make sense. but open to doing it differently.

I could make a device features trait that gets the init status/whatever else it needs and other things rely on it? As we will need it to determine what is supported when we start doing more complicated traits? Thoughts on that?

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Seems worth a try? That way we can get it working then next figure out how to best use it?

this thing of combining traits together with dependencies is kinda weird but kind of simple too but not sure if it will bite us later, but may be an easy way to string things together.


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 self._status_type.from_dict(response)
raise ValueError(f"Unexpected status format: {response!r}")
26 changes: 18 additions & 8 deletions roborock/version_1_apis/roborock_client_v1.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,10 +4,12 @@
import time
from abc import ABC, abstractmethod
from collections.abc import Callable, Coroutine
from functools import cached_property
from typing import Any, TypeVar, final

from roborock import (
AppInitStatus,
DeviceFeatures,
DeviceProp,
DockSummary,
RoborockCommand,
Expand All @@ -33,17 +35,16 @@
DnDTimer,
DustCollectionMode,
FlowLedStatus,
ModelStatus,
MultiMapsList,
NetworkInfo,
RoborockBase,
RoomMapping,
S7MaxVStatus,
ServerTimer,
SmartWashParams,
Status,
ValleyElectricityTimer,
WashTowelMode,
get_custom_status,
)
from roborock.protocols.v1_protocol import MapResponse, SecurityData, create_map_response_decoder
from roborock.roborock_message import (
Expand Down Expand Up @@ -159,7 +160,7 @@ def __init__(self, device_info: DeviceData, security_data: SecurityData | None)
self._diagnostic_data.update({"misc_info": security_data.to_diagnostic_data()})
self._map_response_decoder = create_map_response_decoder(security_data)

self._status_type: type[Status] = ModelStatus.get(device_info.model, S7MaxVStatus)
self._status_type: Status | None = None
self.cache: dict[CacheableAttribute, AttributeCache] = {
cacheable_attribute: AttributeCache(attr, self._send_command)
for cacheable_attribute, attr in get_cache_map().items()
Expand All @@ -172,15 +173,17 @@ async def async_release(self) -> None:
await super().async_release()
[item.stop() for item in self.cache.values()]

@property
@cached_property
def status_type(self) -> type[Status]:
"""Gets the status type for this device"""
return self._status_type
if self.device_info.device_features is None or self.device_info.region is None:
raise RoborockException("Device features and region are required to get status type")
return get_custom_status(self.device_info.device_features, self.device_info.region)

async def get_status(self) -> Status:
data = self._status_type.from_dict(await self.cache[CacheableAttribute.status].async_value(force=True))
data = self.status_type.from_dict(await self.cache[CacheableAttribute.status].async_value(force=True))
if data is None:
return self._status_type()
return self.status_type()
return data

async def get_dnd_timer(self) -> DnDTimer | None:
Expand Down Expand Up @@ -345,7 +348,14 @@ async def load_multi_map(self, map_flag: int) -> None:

async def get_app_init_status(self) -> AppInitStatus:
"""Gets the app init status (needed for determining vacuum capabilities)."""
return await self.send_command(RoborockCommand.APP_GET_INIT_STATUS, return_type=AppInitStatus)
init_status = await self.send_command(RoborockCommand.APP_GET_INIT_STATUS, return_type=AppInitStatus)
self.device_info.device_features = DeviceFeatures.from_feature_flags(
init_status.new_feature_info,
init_status.new_feature_info_str,
init_status.feature_info,
self.device_info.product_nickname,
)
return init_status

@abstractmethod
async def _send_command(
Expand Down
26 changes: 25 additions & 1 deletion tests/conftest.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,7 @@
import pytest
from aioresponses import aioresponses

from roborock import HomeData, UserData
from roborock import SHORT_MODEL_TO_ENUM, DeviceFeatures, HomeData, UserData
from roborock.containers import DeviceData
from roborock.roborock_message import RoborockMessage
from roborock.version_1_apis.roborock_local_client_v1 import RoborockLocalClientV1
Expand Down Expand Up @@ -389,3 +389,27 @@ async def _subscribe(self, callback: Callable[[RoborockMessage], None]) -> Calla
"""Simulate subscribing to messages."""
self.subscribers.append(callback)
return lambda: self.subscribers.remove(callback)


@pytest.fixture(name="s7_device_features")
def s7_device_features_fixture() -> DeviceFeatures:
model = "roborock.vacuum.a15"
product_nickname = SHORT_MODEL_TO_ENUM.get(model.split(".")[-1])
return DeviceFeatures.from_feature_flags(
new_feature_info=636084721975295,
new_feature_info_str="0000000000002000",
feature_info=[111, 112, 113, 114, 115, 116, 117, 118, 119, 120, 122, 123, 124, 125],
product_nickname=product_nickname,
)


@pytest.fixture(name="qrevo_maxv_device_features")
def qrevo_maxv_device_features_fixture() -> DeviceFeatures:
model = "roborock.vacuum.a87"
product_nickname = SHORT_MODEL_TO_ENUM.get(model.split(".")[-1])
return DeviceFeatures.from_feature_flags(
new_feature_info=4499197267967999,
new_feature_info_str="508A977F7EFEFFFF",
feature_info=[111, 112, 113, 114, 115, 116, 117, 118, 119, 120, 121, 122, 123, 124, 125],
product_nickname=product_nickname,
)
3 changes: 1 addition & 2 deletions tests/devices/test_v1_device.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@
import pytest
from syrupy import SnapshotAssertion

from roborock.containers import HomeData, S7MaxVStatus, UserData
from roborock.containers import HomeData, UserData
from roborock.devices.device import RoborockDevice
from roborock.devices.traits import v1
from roborock.devices.traits.v1.common import V1TraitMixin
Expand All @@ -18,7 +18,6 @@

USER_DATA = UserData.from_dict(mock_data.USER_DATA)
HOME_DATA = HomeData.from_dict(mock_data.HOME_DATA_RAW)
STATUS = S7MaxVStatus.from_dict(mock_data.STATUS)

TESTDATA = pathlib.Path("tests/protocols/testdata/v1_protocol/")

Expand Down
49 changes: 32 additions & 17 deletions tests/test_containers.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@

from syrupy import SnapshotAssertion

from roborock import CleanRecord, CleanSummary, Consumable, DnDTimer, HomeData, S7MaxVStatus, UserData
from roborock import CleanRecord, CleanSummary, Consumable, DnDTimer, HomeData, UserData
from roborock.b01_containers import (
B01Fault,
B01Props,
Expand All @@ -17,12 +17,9 @@
RoborockDockErrorCode,
RoborockDockTypeCode,
RoborockErrorCode,
RoborockFanSpeedS7MaxV,
RoborockMopIntensityS7,
RoborockMopModeS7,
RoborockStateCode,
)
from roborock.containers import MultiMapsList, RoborockBase
from roborock.containers import MultiMapsList, RoborockBase, get_custom_status

from .mock_data import (
CLEAN_RECORD,
Expand Down Expand Up @@ -221,8 +218,12 @@ def test_consumable():
assert c.cleaning_brush_work_times == 66


def test_status():
s = S7MaxVStatus.from_dict(STATUS)
def test_status(s7_device_features):
status = get_custom_status(
s7_device_features,
"US",
)
s = status.from_dict(STATUS)
assert s.msg_ver == 2
assert s.msg_seq == 458
assert s.state == RoborockStateCode.charging
Expand Down Expand Up @@ -267,14 +268,19 @@ def test_status():
assert s.charge_status == 1
assert s.unsave_map_reason == 0
assert s.unsave_map_flag == 0
assert s.fan_power == RoborockFanSpeedS7MaxV.balanced
assert s.mop_mode == RoborockMopModeS7.standard
assert s.water_box_mode == RoborockMopIntensityS7.intense
assert s.fan_power == 102
assert s.fan_speed == "BALANCED"
assert s.mop_route == "STANDARD"
assert s.water_mode == "INTENSE"


def test_current_map() -> None:
def test_current_map(s7_device_features) -> None:
"""Test the current map logic based on map status."""
s = S7MaxVStatus.from_dict(STATUS)
status = get_custom_status(
s7_device_features,
"US",
)
s = status.from_dict(STATUS)
assert s.map_status == 3
assert s.current_map == 0

Expand Down Expand Up @@ -327,10 +333,15 @@ def test_clean_record():
assert cr.map_flag == 0


def test_no_value():
def test_no_value(s7_device_features):
modified_status = STATUS.copy()
modified_status["dock_type"] = 9999
s = S7MaxVStatus.from_dict(modified_status)
status = get_custom_status(
s7_device_features,
"US",
)

s = status.from_dict(modified_status)
assert s.dock_type == RoborockDockTypeCode.unknown
assert -9999 not in RoborockDockTypeCode.keys()
assert "missing" not in RoborockDockTypeCode.values()
Expand Down Expand Up @@ -483,11 +494,15 @@ def test_multi_maps_list_info(snapshot: SnapshotAssertion) -> None:
assert deserialized == snapshot


def test_accurate_map_flag() -> None:
def test_accurate_map_flag(s7_device_features) -> None:
"""Test that we parse the map flag accurately."""
s = S7MaxVStatus.from_dict(STATUS)
status = get_custom_status(
s7_device_features,
"US",
)
s = status.from_dict(STATUS)
assert s.current_map == 0
s = S7MaxVStatus.from_dict(
s = status.from_dict(
{
**STATUS,
"map_status": 252, # Code for no map
Expand Down
Loading
Loading