roborock.devices.traits.b01

 32class Q7PropertiesApi(Trait):
 33    """API for interacting with B01 devices."""
 34
 35    clean_summary: CleanSummaryTrait
 36    """Trait for clean records / clean summary (Q7 `service.get_record_list`)."""
 37
 38    def __init__(self, channel: MqttChannel) -> None:
 39        """Initialize the B01Props API."""
 40        self._channel = channel
 41        self.clean_summary = CleanSummaryTrait(channel)
 42
 43    async def query_values(self, props: list[RoborockB01Props]) -> B01Props | None:
 44        """Query the device for the values of the given Q7 properties."""
 45        result = await self.send(
 46            RoborockB01Q7Methods.GET_PROP,
 47            {"property": props},
 48        )
 49        if not isinstance(result, dict):
 50            raise TypeError(f"Unexpected response type for GET_PROP: {type(result).__name__}: {result!r}")
 51        return B01Props.from_dict(result)
 52
 53    async def set_prop(self, prop: RoborockB01Props, value: Any) -> None:
 54        """Set a property on the device."""
 55        await self.send(
 56            command=RoborockB01Q7Methods.SET_PROP,
 57            params={prop: value},
 58        )
 59
 60    async def set_fan_speed(self, fan_speed: SCWindMapping) -> None:
 61        """Set the fan speed (wind)."""
 62        await self.set_prop(RoborockB01Props.WIND, fan_speed.code)
 63
 64    async def set_water_level(self, water_level: WaterLevelMapping) -> None:
 65        """Set the water level (water)."""
 66        await self.set_prop(RoborockB01Props.WATER, water_level.code)
 67
 68    async def set_mode(self, mode: CleanTypeMapping) -> None:
 69        """Set the cleaning mode (vacuum, mop, or vacuum and mop)."""
 70        await self.set_prop(RoborockB01Props.MODE, mode.code)
 71
 72    async def set_clean_path_preference(self, preference: CleanPathPreferenceMapping) -> None:
 73        """Set the cleaning path preference (route)."""
 74        await self.set_prop(RoborockB01Props.CLEAN_PATH_PREFERENCE, preference.code)
 75
 76    async def set_repeat_state(self, repeat: CleanRepeatMapping) -> None:
 77        """Set the cleaning repeat state (cycles)."""
 78        await self.set_prop(RoborockB01Props.REPEAT_STATE, repeat.code)
 79
 80    async def start_clean(self) -> None:
 81        """Start cleaning."""
 82        await self.send(
 83            command=RoborockB01Q7Methods.SET_ROOM_CLEAN,
 84            params={
 85                "clean_type": CleanTaskTypeMapping.ALL.code,
 86                "ctrl_value": SCDeviceCleanParam.START.code,
 87                "room_ids": [],
 88            },
 89        )
 90
 91    async def pause_clean(self) -> None:
 92        """Pause cleaning."""
 93        await self.send(
 94            command=RoborockB01Q7Methods.SET_ROOM_CLEAN,
 95            params={
 96                "clean_type": CleanTaskTypeMapping.ALL.code,
 97                "ctrl_value": SCDeviceCleanParam.PAUSE.code,
 98                "room_ids": [],
 99            },
100        )
101
102    async def stop_clean(self) -> None:
103        """Stop cleaning."""
104        await self.send(
105            command=RoborockB01Q7Methods.SET_ROOM_CLEAN,
106            params={
107                "clean_type": CleanTaskTypeMapping.ALL.code,
108                "ctrl_value": SCDeviceCleanParam.STOP.code,
109                "room_ids": [],
110            },
111        )
112
113    async def return_to_dock(self) -> None:
114        """Return to dock."""
115        await self.send(
116            command=RoborockB01Q7Methods.START_RECHARGE,
117            params={},
118        )
119
120    async def find_me(self) -> None:
121        """Locate the robot."""
122        await self.send(
123            command=RoborockB01Q7Methods.FIND_DEVICE,
124            params={},
125        )
126
127    async def send(self, command: CommandType, params: ParamsType) -> Any:
128        """Send a command to the device."""
129        return await send_decoded_command(
130            self._channel,
131            Q7RequestMessage(dps=10000, command=command, params=params),
132        )

15class Q10PropertiesApi(Trait):
16    """API for interacting with B01 devices."""
17
18    command: CommandTrait
19    """Trait for sending commands to Q10 devices."""
20
21    vacuum: VacuumTrait
22    """Trait for sending vacuum related commands to Q10 devices."""
23
24    def __init__(self, channel: MqttChannel) -> None:
25        """Initialize the B01Props API."""
26        self.command = CommandTrait(channel)
27        self.vacuum = VacuumTrait(self.command)