roborock.devices.traits.b01.q7

  1"""Traits for Q7 B01 devices.
  2Potentially other devices may fall into this category in the future."""
  3
  4from typing import Any
  5
  6from roborock import B01Props
  7from roborock.data.b01_q7.b01_q7_code_mappings import (
  8    CleanTaskTypeMapping,
  9    SCDeviceCleanParam,
 10    SCWindMapping,
 11    WaterLevelMapping,
 12)
 13from roborock.devices.b01_channel import CommandType, ParamsType, send_decoded_command
 14from roborock.devices.mqtt_channel import MqttChannel
 15from roborock.devices.traits import Trait
 16from roborock.roborock_message import RoborockB01Props
 17from roborock.roborock_typing import RoborockB01Q7Methods
 18
 19__all__ = [
 20    "Q7PropertiesApi",
 21]
 22
 23
 24class Q7PropertiesApi(Trait):
 25    """API for interacting with B01 devices."""
 26
 27    def __init__(self, channel: MqttChannel) -> None:
 28        """Initialize the B01Props API."""
 29        self._channel = channel
 30
 31    async def query_values(self, props: list[RoborockB01Props]) -> B01Props | None:
 32        """Query the device for the values of the given Q7 properties."""
 33        result = await self.send(
 34            RoborockB01Q7Methods.GET_PROP,
 35            {"property": props},
 36        )
 37        if not isinstance(result, dict):
 38            raise TypeError(f"Unexpected response type for GET_PROP: {type(result).__name__}: {result!r}")
 39        return B01Props.from_dict(result)
 40
 41    async def set_prop(self, prop: RoborockB01Props, value: Any) -> None:
 42        """Set a property on the device."""
 43        await self.send(
 44            command=RoborockB01Q7Methods.SET_PROP,
 45            params={prop: value},
 46        )
 47
 48    async def set_fan_speed(self, fan_speed: SCWindMapping) -> None:
 49        """Set the fan speed (wind)."""
 50        await self.set_prop(RoborockB01Props.WIND, fan_speed.code)
 51
 52    async def set_water_level(self, water_level: WaterLevelMapping) -> None:
 53        """Set the water level (water)."""
 54        await self.set_prop(RoborockB01Props.WATER, water_level.code)
 55
 56    async def start_clean(self) -> None:
 57        """Start cleaning."""
 58        await self.send(
 59            command=RoborockB01Q7Methods.SET_ROOM_CLEAN,
 60            params={
 61                "clean_type": CleanTaskTypeMapping.ALL.code,
 62                "ctrl_value": SCDeviceCleanParam.START.code,
 63                "room_ids": [],
 64            },
 65        )
 66
 67    async def pause_clean(self) -> None:
 68        """Pause cleaning."""
 69        await self.send(
 70            command=RoborockB01Q7Methods.SET_ROOM_CLEAN,
 71            params={
 72                "clean_type": CleanTaskTypeMapping.ALL.code,
 73                "ctrl_value": SCDeviceCleanParam.PAUSE.code,
 74                "room_ids": [],
 75            },
 76        )
 77
 78    async def stop_clean(self) -> None:
 79        """Stop cleaning."""
 80        await self.send(
 81            command=RoborockB01Q7Methods.SET_ROOM_CLEAN,
 82            params={
 83                "clean_type": CleanTaskTypeMapping.ALL.code,
 84                "ctrl_value": SCDeviceCleanParam.STOP.code,
 85                "room_ids": [],
 86            },
 87        )
 88
 89    async def return_to_dock(self) -> None:
 90        """Return to dock."""
 91        await self.send(
 92            command=RoborockB01Q7Methods.START_RECHARGE,
 93            params={},
 94        )
 95
 96    async def find_me(self) -> None:
 97        """Locate the robot."""
 98        await self.send(
 99            command=RoborockB01Q7Methods.FIND_DEVICE,
100            params={},
101        )
102
103    async def send(self, command: CommandType, params: ParamsType) -> Any:
104        """Send a command to the device."""
105        return await send_decoded_command(
106            self._channel,
107            dps=10000,
108            command=command,
109            params=params,
110        )
111
112
113def create(channel: MqttChannel) -> Q7PropertiesApi:
114    """Create traits for B01 devices."""
115    return Q7PropertiesApi(channel)