roborock.devices.traits.b01.q7

 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        )

27class CleanSummaryTrait(CleanRecordSummary, Trait):
28    """B01/Q7 clean summary + clean record access (via record list service)."""
29
30    def __init__(self, channel: MqttChannel) -> None:
31        """Initialize the clean summary trait.
32
33        Args:
34            channel: MQTT channel used to communicate with the device.
35        """
36        super().__init__()
37        self._channel = channel
38
39    async def refresh(self) -> None:
40        """Refresh totals and last record detail from the device."""
41        record_list = await self._get_record_list()
42
43        self.total_time = record_list.total_time
44        self.total_area = record_list.total_area
45        self.total_count = record_list.total_count
46
47        details = await self._get_clean_record_details(record_list=record_list)
48        self.last_record_detail = details[0] if details else None
49
50    async def _get_record_list(self) -> CleanRecordList:
51        """Fetch the raw device clean record list (`service.get_record_list`)."""
52        result = await send_decoded_command(
53            self._channel,
54            Q7RequestMessage(dps=10000, command=RoborockB01Q7Methods.GET_RECORD_LIST, params={}),
55        )
56
57        if not isinstance(result, dict):
58            raise TypeError(f"Unexpected response type for GET_RECORD_LIST: {type(result).__name__}: {result!r}")
59        return CleanRecordList.from_dict(result)
60
61    async def _get_clean_record_details(self, *, record_list: CleanRecordList) -> list[CleanRecordDetail]:
62        """Return parsed record detail objects (newest-first)."""
63        details: list[CleanRecordDetail] = []
64        for item in record_list.record_list:
65            try:
66                parsed = item.detail_parsed
67            except RoborockException as ex:
68                # Rather than failing if something goes wrong here, we should fail and log to tell the user.
69                _LOGGER.debug("Failed to parse record detail: %s", ex)
70                continue
71            if parsed is not None:
72                details.append(parsed)
73
74        # The server returns the newest record at the end of record_list; reverse so newest is first (index 0).
75        details.reverse()
76        return details