roborock.devices.traits.b01

 25class Q7PropertiesApi(Trait):
 26    """API for interacting with B01 devices."""
 27
 28    def __init__(self, channel: MqttChannel) -> None:
 29        """Initialize the B01Props API."""
 30        self._channel = channel
 31
 32    async def query_values(self, props: list[RoborockB01Props]) -> B01Props | None:
 33        """Query the device for the values of the given Q7 properties."""
 34        result = await self.send(
 35            RoborockB01Q7Methods.GET_PROP,
 36            {"property": props},
 37        )
 38        if not isinstance(result, dict):
 39            raise TypeError(f"Unexpected response type for GET_PROP: {type(result).__name__}: {result!r}")
 40        return B01Props.from_dict(result)
 41
 42    async def set_prop(self, prop: RoborockB01Props, value: Any) -> None:
 43        """Set a property on the device."""
 44        await self.send(
 45            command=RoborockB01Q7Methods.SET_PROP,
 46            params={prop: value},
 47        )
 48
 49    async def set_fan_speed(self, fan_speed: SCWindMapping) -> None:
 50        """Set the fan speed (wind)."""
 51        await self.set_prop(RoborockB01Props.WIND, fan_speed.code)
 52
 53    async def set_water_level(self, water_level: WaterLevelMapping) -> None:
 54        """Set the water level (water)."""
 55        await self.set_prop(RoborockB01Props.WATER, water_level.code)
 56
 57    async def start_clean(self) -> None:
 58        """Start cleaning."""
 59        await self.send(
 60            command=RoborockB01Q7Methods.SET_ROOM_CLEAN,
 61            params={
 62                "clean_type": CleanTaskTypeMapping.ALL.code,
 63                "ctrl_value": SCDeviceCleanParam.START.code,
 64                "room_ids": [],
 65            },
 66        )
 67
 68    async def pause_clean(self) -> None:
 69        """Pause cleaning."""
 70        await self.send(
 71            command=RoborockB01Q7Methods.SET_ROOM_CLEAN,
 72            params={
 73                "clean_type": CleanTaskTypeMapping.ALL.code,
 74                "ctrl_value": SCDeviceCleanParam.PAUSE.code,
 75                "room_ids": [],
 76            },
 77        )
 78
 79    async def stop_clean(self) -> None:
 80        """Stop cleaning."""
 81        await self.send(
 82            command=RoborockB01Q7Methods.SET_ROOM_CLEAN,
 83            params={
 84                "clean_type": CleanTaskTypeMapping.ALL.code,
 85                "ctrl_value": SCDeviceCleanParam.STOP.code,
 86                "room_ids": [],
 87            },
 88        )
 89
 90    async def return_to_dock(self) -> None:
 91        """Return to dock."""
 92        await self.send(
 93            command=RoborockB01Q7Methods.START_RECHARGE,
 94            params={},
 95        )
 96
 97    async def find_me(self) -> None:
 98        """Locate the robot."""
 99        await self.send(
100            command=RoborockB01Q7Methods.FIND_DEVICE,
101            params={},
102        )
103
104    async def send(self, command: CommandType, params: ParamsType) -> Any:
105        """Send a command to the device."""
106        return await send_decoded_command(
107            self._channel,
108            dps=10000,
109            command=command,
110            params=params,
111        )