roborock.devices.traits.b01

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

24class Q10PropertiesApi(Trait):
25    """API for interacting with B01 devices."""
26
27    command: CommandTrait
28    """Trait for sending commands to Q10 devices."""
29
30    status: StatusTrait
31    """Trait for managing the status of Q10 devices."""
32
33    vacuum: VacuumTrait
34    """Trait for sending vacuum related commands to Q10 devices."""
35
36    def __init__(self, channel: MqttChannel) -> None:
37        """Initialize the B01Props API."""
38        self._channel = channel
39        self.command = CommandTrait(channel)
40        self.vacuum = VacuumTrait(self.command)
41        self.status = StatusTrait()
42        self._subscribe_task: asyncio.Task[None] | None = None
43
44    async def start(self) -> None:
45        """Start any necessary subscriptions for the trait."""
46        self._subscribe_task = asyncio.create_task(self._subscribe_loop())
47
48    async def close(self) -> None:
49        """Close any resources held by the trait."""
50        if self._subscribe_task is not None:
51            self._subscribe_task.cancel()
52            try:
53                await self._subscribe_task
54            except asyncio.CancelledError:
55                pass  # ignore cancellation errors
56            self._subscribe_task = None
57
58    async def refresh(self) -> None:
59        """Refresh all traits."""
60        # Sending the REQUEST_DPS will cause the device to send all DPS values
61        # to the device. Updates will be received by the subscribe loop below.
62        await self.command.send(B01_Q10_DP.REQUEST_DPS, params={})
63
64    async def _subscribe_loop(self) -> None:
65        """Persistent loop to listen for status updates."""
66        async for decoded_dps in stream_decoded_responses(self._channel):
67            _LOGGER.debug("Received Q10 status update: %s", decoded_dps)
68
69            # Notify all traits about a new message and each trait will
70            # only update what fields that it is responsible for.
71            # More traits can be added here below.
72            self.status.update_from_dps(decoded_dps)