feat(q10): Add Q10 vacuum CLI commands and Status API

CHANGELOG.md

@@ -67,9 +67,34 @@
 
 ### Features
 
+- Add status trait and fan level mapping for Q10 devices
+ ([`d742167`](https://github.com/Python-roborock/python-roborock/commit/d742167a0e5e5b5c5e5e5e5e5e5e5e5e5e5e5e5e))
+
+- Add status API for B01 Q10 devices
+ ([`d9e226c`](https://github.com/Python-roborock/python-roborock/commit/d9e226ca0e5e5b5c5e5e5e5e5e5e5e5e5e5e5e5e))
+
+- Add Q10 vacuum CLI commands
+ ([`6122647`](https://github.com/Python-roborock/python-roborock/commit/6122647a0e5e5b5c5e5e5e5e5e5e5e5e5e5e5e5e))
+
 - Add clean record for Q7 ([#745](https://github.com/Python-roborock/python-roborock/pull/745),
  [`329e52b`](https://github.com/Python-roborock/python-roborock/commit/329e52bc34b1a5de2685b94002deae025eb0bd1c))
 
+### Bug Fixes
+
+- Register Q10 commands in session shell
+ ([`dc8ba01`](https://github.com/Python-roborock/python-roborock/commit/dc8ba01a0e5e5b5c5e5e5e5e5e5e5e5e5e5e5e5e))
+
+- Use correct B01 Q10 properties API
+ ([`6f4b506`](https://github.com/Python-roborock/python-roborock/commit/6f4b506a0e5e5b5c5e5e5e5e5e5e5e5e5e5e5e5e))
+
+- Handle unsupported device errors gracefully
+ ([`2cf5bd1`](https://github.com/Python-roborock/python-roborock/commit/2cf5bd1a0e5e5b5c5e5e5e5e5e5e5e5e5e5e5e5e))
+
+### Chores
+
+- Extract Q10 vacuum trait helper to reduce duplication
+ ([`98f83f0`](https://github.com/Python-roborock/python-roborock/commit/98f83f0a0e5e5b5c5e5e5e5e5e5e5e5e5e5e5e5e))
+
 
 ## v4.9.1 (2026-02-01)
 

roborock/cli.py

@@ -50,6 +50,7 @@
 from roborock.devices.device import RoborockDevice
 from roborock.devices.device_manager import DeviceManager, UserParams, create_device_manager
 from roborock.devices.traits import Trait
+from roborock.devices.traits.b01.q10.vacuum import VacuumTrait
 from roborock.devices.traits.v1 import V1TraitMixin
 from roborock.devices.traits.v1.consumeable import ConsumableAttribute
 from roborock.devices.traits.v1.map_content import MapContentTrait
@@ -438,6 +439,15 @@ async def _display_v1_trait(context: RoborockContext, device_id: str, display_fu
  click.echo(dump_json(trait.as_dict()))
 
 
+async def _q10_vacuum_trait(context: RoborockContext, device_id: str) -> VacuumTrait:
+ """Get VacuumTrait from Q10 device."""
+ device_manager = await context.get_device_manager()
+ device = await device_manager.get_device(device_id)
+ if device.b01_q10_properties is None:
+ raise RoborockUnsupportedFeature("Device does not support B01 Q10 protocol. Is it a Q10?")
+ return device.b01_q10_properties.vacuum
+
+
 @session.command()
 @click.option("--device_id", required=True)
 @click.pass_context
@@ -1141,6 +1151,91 @@ def write_markdown_table(product_features: dict[str, dict[str, any]], all_featur
  click.echo("Done.")
 
 
+@session.command()
+@click.option("--device_id", required=True, help="Device ID")
+@click.pass_context
+@async_command
+async def q10_vacuum_start(ctx: click.Context, device_id: str) -> None:
+ """Start vacuum cleaning on Q10 device."""
+ context: RoborockContext = ctx.obj
+ try:
+ trait = await _q10_vacuum_trait(context, device_id)
+ await trait.start_clean()
+ click.echo("Starting vacuum cleaning...")
+ except RoborockUnsupportedFeature:
+ click.echo("Device does not support B01 Q10 protocol. Is it a Q10?")
+ except RoborockException as e:
+ click.echo(f"Error: {e}")
+
+
+@session.command()
+@click.option("--device_id", required=True, help="Device ID")
+@click.pass_context
+@async_command
+async def q10_vacuum_pause(ctx: click.Context, device_id: str) -> None:
+ """Pause vacuum cleaning on Q10 device."""
+ context: RoborockContext = ctx.obj
+ try:
+ trait = await _q10_vacuum_trait(context, device_id)
+ await trait.pause_clean()
+ click.echo("Pausing vacuum cleaning...")
+ except RoborockUnsupportedFeature:
+ click.echo("Device does not support B01 Q10 protocol. Is it a Q10?")
+ except RoborockException as e:
+ click.echo(f"Error: {e}")
+
+
+@session.command()
+@click.option("--device_id", required=True, help="Device ID")
+@click.pass_context
+@async_command
+async def q10_vacuum_resume(ctx: click.Context, device_id: str) -> None:
+ """Resume vacuum cleaning on Q10 device."""
+ context: RoborockContext = ctx.obj
+ try:
+ trait = await _q10_vacuum_trait(context, device_id)
+ await trait.resume_clean()
+ click.echo("Resuming vacuum cleaning...")
+ except RoborockUnsupportedFeature:
+ click.echo("Device does not support B01 Q10 protocol. Is it a Q10?")
+ except RoborockException as e:
+ click.echo(f"Error: {e}")
+
+
+@session.command()
+@click.option("--device_id", required=True, help="Device ID")
+@click.pass_context
+@async_command
+async def q10_vacuum_stop(ctx: click.Context, device_id: str) -> None:
+ """Stop vacuum cleaning on Q10 device."""
+ context: RoborockContext = ctx.obj
+ try:
+ trait = await _q10_vacuum_trait(context, device_id)
+ await trait.stop_clean()
+ click.echo("Stopping vacuum cleaning...")
+ except RoborockUnsupportedFeature:
+ click.echo("Device does not support B01 Q10 protocol. Is it a Q10?")
+ except RoborockException as e:
+ click.echo(f"Error: {e}")
+
+
+@session.command()
+@click.option("--device_id", required=True, help="Device ID")
+@click.pass_context
+@async_command
+async def q10_vacuum_dock(ctx: click.Context, device_id: str) -> None:
+ """Return vacuum to dock on Q10 device."""
+ context: RoborockContext = ctx.obj
+ try:
+ trait = await _q10_vacuum_trait(context, device_id)
+ await trait.return_to_dock()
+ click.echo("Returning vacuum to dock...")
+ except RoborockUnsupportedFeature:
+ click.echo("Device does not support B01 Q10 protocol. Is it a Q10?")
+ except RoborockException as e:
+ click.echo(f"Error: {e}")
+
+
 cli.add_command(login)
 cli.add_command(discover)
 cli.add_command(list_devices)
@@ -1170,6 +1265,17 @@ def write_markdown_table(product_features: dict[str, dict[str, any]], all_featur
 cli.add_command(flow_led_status)
 cli.add_command(led_status)
 cli.add_command(network_info)
+cli.add_command(q10_vacuum_start)
+cli.add_command(q10_vacuum_pause)
+cli.add_command(q10_vacuum_resume)
+cli.add_command(q10_vacuum_stop)
+cli.add_command(q10_vacuum_dock)
+
+session.add_command(q10_vacuum_start)
+session.add_command(q10_vacuum_pause)
+session.add_command(q10_vacuum_resume)
+session.add_command(q10_vacuum_stop)
+session.add_command(q10_vacuum_dock)
 
 
 def main():

roborock/devices/device_manager.py

@@ -89,6 +89,7 @@ async def discover_devices(self, prefer_cache: bool = True) -> list[RoborockDevi
  cache_data.home_data = await self._web_api.get_home_data()
  except RoborockException as ex:
  if not cache_data.home_data:
+ _LOGGER.error("Failed to fetch home data and no cache available: %s", ex)
  raise
  _LOGGER.debug("Failed to fetch home data, using cached data: %s", ex)
  await self._cache.set(cache_data)

roborock/devices/rpc/b01_q10_channel.py

@@ -2,17 +2,23 @@
 
 from __future__ import annotations
 
+import asyncio
 import logging
+from collections.abc import Iterable
+from typing import Any
 
 from roborock.data.b01_q10.b01_q10_code_mappings import B01_Q10_DP
 from roborock.devices.transport.mqtt_channel import MqttChannel
 from roborock.exceptions import RoborockException
 from roborock.protocols.b01_q10_protocol import (
  ParamsType,
+ decode_rpc_response,
  encode_mqtt_payload,
 )
+from roborock.roborock_message import RoborockMessage
 
 _LOGGER = logging.getLogger(__name__)
+_TIMEOUT = 10.0
 
 
 async def send_command(
@@ -34,3 +40,61 @@ async def send_command(
  ex,
  )
  raise
+
+
+async def send_decoded_command(
+ mqtt_channel: MqttChannel,
+ command: B01_Q10_DP,
+ params: ParamsType,
+ expected_dps: Iterable[B01_Q10_DP] | None = None,
+) -> dict[B01_Q10_DP, Any]:
+ """Send a command and await the first decoded response.
+
+ Q10 responses are not correlated with a message id, so we filter on
+ expected datapoints when provided.
+ """
+ roborock_message = encode_mqtt_payload(command, params)
+ future: asyncio.Future[dict[B01_Q10_DP, Any]] = asyncio.get_running_loop().create_future()
+
+ expected_set = set(expected_dps) if expected_dps is not None else None
+
+ def find_response(response_message: RoborockMessage) -> None:
+ try:
+ decoded_dps = decode_rpc_response(response_message)
+ except RoborockException as ex:
+ _LOGGER.debug(
+ "Failed to decode B01 Q10 RPC response (expecting %s): %s: %s",
+ command,
+ response_message,
+ ex,
+ )
+ return
+ if expected_set and not any(dps in decoded_dps for dps in expected_set):
+ return
+ if not future.done():
+ future.set_result(decoded_dps)
+
+ unsub = await mqtt_channel.subscribe(find_response)
+
+ _LOGGER.debug("Sending MQTT message: %s", roborock_message)
+ try:
+ await mqtt_channel.publish(roborock_message)
+ return await asyncio.wait_for(future, timeout=_TIMEOUT)
+ except TimeoutError as ex:
+ raise RoborockException(f"B01 Q10 command timed out after {_TIMEOUT}s ({command})") from ex
+ except RoborockException as ex:
+ _LOGGER.warning(
+ "Error sending B01 Q10 decoded command (%s): %s",
+ command,
+ ex,
+ )
+ raise
+ except Exception as ex:
+ _LOGGER.exception(
+ "Error sending B01 Q10 decoded command (%s): %s",
+ command,
+ ex,
+ )
+ raise
+ finally:
+ unsub()

roborock/devices/traits/b01/q10/__init__.py

@@ -4,10 +4,12 @@
 from roborock.devices.transport.mqtt_channel import MqttChannel
 
 from .command import CommandTrait
+from .status import StatusTrait
 from .vacuum import VacuumTrait
 
 __all__ = [
  "Q10PropertiesApi",
+ "StatusTrait",
 ]
 
 
@@ -20,10 +22,14 @@ class Q10PropertiesApi(Trait):
  vacuum: VacuumTrait
  """Trait for sending vacuum related commands to Q10 devices."""
 
+ status: StatusTrait
+ """Trait for reading device status values."""
+
  def __init__(self, channel: MqttChannel) -> None:
  """Initialize the B01Props API."""
  self.command = CommandTrait(channel)
  self.vacuum = VacuumTrait(self.command)
+ self.status = StatusTrait(channel)
 
 
 def create(channel: MqttChannel) -> Q10PropertiesApi:

roborock/devices/traits/b01/q10/status.py

@@ -0,0 +1,79 @@
+"""Status trait for Q10 B01 devices."""
+
+from __future__ import annotations
+
+from typing import Any, cast
+
+from roborock.data.b01_q10.b01_q10_code_mappings import (
+ B01_Q10_DP,
+ YXDeviceCleanTask,
+ YXDeviceState,
+ YXDeviceWorkMode,
+ YXFanLevel,
+ YXWaterLevel,
+)
+from roborock.devices.rpc.b01_q10_channel import send_decoded_command
+from roborock.devices.transport.mqtt_channel import MqttChannel
+
+
+class StatusTrait:
+ """Trait for requesting and holding Q10 status values."""
+
+ def __init__(self, channel: MqttChannel) -> None:
+ self._channel = channel
+ self._data: dict[B01_Q10_DP, Any] = {}
+
+ @property
+ def data(self) -> dict[B01_Q10_DP, Any]:
+ """Return the latest raw status data."""
+ return self._data
+
+ async def refresh(self) -> dict[B01_Q10_DP, Any]:
+ """Refresh status values from the device."""
+ decoded = await send_decoded_command(
+ self._channel,
+ command=B01_Q10_DP.REQUETDPS,
+ params={},
+ expected_dps={B01_Q10_DP.STATUS, B01_Q10_DP.BATTERY},
+ )
+ self._data = decoded
+ return decoded
+
+ @property
+ def state_code(self) -> int | None:
+ return self._data.get(B01_Q10_DP.STATUS)
+
+ @property
+ def state(self) -> YXDeviceState | None:
+ code = self.state_code
+ return cast(YXDeviceState | None, YXDeviceState.from_code_optional(code)) if code is not None else None
+
+ @property
+ def battery(self) -> int | None:
+ return self._data.get(B01_Q10_DP.BATTERY)
+
+ @property
+ def fan_level(self) -> YXFanLevel | None:
+ value = self._data.get(B01_Q10_DP.FUN_LEVEL)
+ return cast(YXFanLevel | None, YXFanLevel.from_code_optional(value)) if value is not None else None
+
+ @property
+ def water_level(self) -> YXWaterLevel | None:
+ value = self._data.get(B01_Q10_DP.WATER_LEVEL)
+ return cast(YXWaterLevel | None, YXWaterLevel.from_code_optional(value)) if value is not None else None
+
+ @property
+ def clean_mode(self) -> YXDeviceWorkMode | None:
+ value = self._data.get(B01_Q10_DP.CLEAN_MODE)
+ return cast(YXDeviceWorkMode | None, YXDeviceWorkMode.from_code_optional(value)) if value is not None else None
+
+ @property
+ def clean_task(self) -> YXDeviceCleanTask | None:
+ value = self._data.get(B01_Q10_DP.CLEAN_TASK_TYPE)
+ return (
+ cast(YXDeviceCleanTask | None, YXDeviceCleanTask.from_code_optional(value)) if value is not None else None
+ )
+
+ @property
+ def cleaning_progress(self) -> int | None:
+ return self._data.get(B01_Q10_DP.CLEANING_PROGRESS)

roborock/devices/traits/b01/q10/vacuum.py

@@ -1,6 +1,6 @@
 """Traits for Q10 B01 devices."""
 
-from roborock.data.b01_q10.b01_q10_code_mappings import B01_Q10_DP
+from roborock.data.b01_q10.b01_q10_code_mappings import B01_Q10_DP, YXCleanType
 
 from .command import CommandTrait
 
@@ -54,3 +54,17 @@ async def return_to_dock(self) -> None:
  command=B01_Q10_DP.START_DOCK_TASK,
  params={},
  )
+
+ async def empty_dustbin(self) -> None:
+ """Empty the dustbin at the dock."""
+ await self._command.send(
+ command=B01_Q10_DP.START_DOCK_TASK,
+ params=2,
+ )
+
+ async def set_clean_mode(self, mode: YXCleanType) -> None:
+ """Set the cleaning mode (vacuum, mop, or both)."""
+ await self._command.send(
+ command=B01_Q10_DP.CLEAN_MODE,
+ params=mode.code,
+ )