roborock.devices.traits.b01.q7

 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        )

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

15class MapTrait(Q7MapList, Trait):
16    """Map retrieval + map metadata helpers for Q7 devices."""
17
18    def __init__(self, channel: MqttChannel) -> None:
19        super().__init__()
20        self._channel = channel
21        # Map uploads are serialized per-device to avoid response cross-wiring.
22        self._map_command_lock = asyncio.Lock()
23        self._loaded = False
24
25    async def refresh(self) -> None:
26        """Refresh cached map list metadata from the device."""
27        response = await send_decoded_command(
28            self._channel,
29            Q7RequestMessage(dps=B01_Q7_DPS, command=RoborockB01Q7Methods.GET_MAP_LIST, params={}),
30        )
31        if not isinstance(response, dict):
32            raise RoborockException(
33                f"Unexpected response type for GET_MAP_LIST: {type(response).__name__}: {response!r}"
34            )
35
36        if (parsed := Q7MapList.from_dict(response)) is None:
37            raise RoborockException(f"Failed to decode map list response: {response!r}")
38
39        self.map_list = parsed.map_list
40        self._loaded = True
41
42    async def _get_map_payload(self, *, map_id: int) -> bytes:
43        """Fetch raw map payload bytes for the given map id."""
44        request = Q7RequestMessage(
45            dps=B01_Q7_DPS,
46            command=RoborockB01Q7Methods.UPLOAD_BY_MAPID,
47            params={"map_id": map_id},
48        )
49        async with self._map_command_lock:
50            return await send_map_command(self._channel, request)
51
52    async def get_current_map_payload(self) -> bytes:
53        """Fetch raw map payload bytes for the currently selected map."""
54        if not self._loaded:
55            await self.refresh()
56
57        map_id = self.current_map_id
58        if map_id is None:
59            raise RoborockException(f"Unable to determine map_id from map list response: {self!r}")
60        return await self._get_map_payload(map_id=map_id)

 87@dataclass
 88class Q7MapList(RoborockBase):
 89    """Map list response returned by `service.get_map_list`."""
 90
 91    map_list: list[Q7MapListEntry] = field(default_factory=list)
 92
 93    @property
 94    def current_map_id(self) -> int | None:
 95        """Current map id, preferring the entry marked current."""
 96        if not self.map_list:
 97            return None
 98
 99        ordered = sorted(self.map_list, key=lambda entry: entry.cur or False, reverse=True)
100        first = next(iter(ordered), None)
101        if first is None or not isinstance(first.id, int):
102            return None
103        return first.id

79@dataclass
80class Q7MapListEntry(RoborockBase):
81    """Single map list entry returned by `service.get_map_list`."""
82
83    id: int | None = None
84    cur: bool | None = None