roborock.devices.traits.b01.q7

 44class Q7PropertiesApi(Trait):
 45    """API for interacting with B01 Q7 devices."""
 46
 47    clean_summary: CleanSummaryTrait
 48    """Trait for clean records / clean summary (Q7 `service.get_record_list`)."""
 49
 50    map: MapTrait
 51    """Trait for map list metadata + raw map payload retrieval."""
 52
 53    map_content: MapContentTrait
 54    """Trait for fetching parsed current map content."""
 55
 56    def __init__(
 57        self, channel: MqttChannel, map_rpc_channel: MapRpcChannel, device: HomeDataDevice, product: HomeDataProduct
 58    ) -> None:
 59        """Initialize the Q7 API."""
 60        self._channel = channel
 61        self._map_rpc_channel = map_rpc_channel
 62        self._device = device
 63        self._product = product
 64
 65        if not device.sn or not product.model:
 66            raise ValueError("B01 Q7 map content requires device serial number and product model metadata")
 67
 68        self.clean_summary = CleanSummaryTrait(channel)
 69        self.map = MapTrait(channel)
 70        self.map_content = MapContentTrait(
 71            self._map_rpc_channel,
 72            self.map,
 73        )
 74
 75    async def query_values(self, props: list[RoborockB01Props]) -> B01Props | None:
 76        """Query the device for the values of the given Q7 properties."""
 77        result = await self.send(
 78            RoborockB01Q7Methods.GET_PROP,
 79            {"property": props},
 80        )
 81        if not isinstance(result, dict):
 82            raise TypeError(f"Unexpected response type for GET_PROP: {type(result).__name__}: {result!r}")
 83        return B01Props.from_dict(result)
 84
 85    async def set_prop(self, prop: RoborockB01Props, value: Any) -> None:
 86        """Set a property on the device."""
 87        await self.send(
 88            command=RoborockB01Q7Methods.SET_PROP,
 89            params={prop: value},
 90        )
 91
 92    async def set_fan_speed(self, fan_speed: SCWindMapping) -> None:
 93        """Set the fan speed (wind)."""
 94        await self.set_prop(RoborockB01Props.WIND, fan_speed.code)
 95
 96    async def set_water_level(self, water_level: WaterLevelMapping) -> None:
 97        """Set the water level (water)."""
 98        await self.set_prop(RoborockB01Props.WATER, water_level.code)
 99
100    async def set_mode(self, mode: CleanTypeMapping) -> None:
101        """Set the cleaning mode (vacuum, mop, or vacuum and mop)."""
102        await self.set_prop(RoborockB01Props.MODE, mode.code)
103
104    async def set_clean_path_preference(self, preference: CleanPathPreferenceMapping) -> None:
105        """Set the cleaning path preference (route)."""
106        await self.set_prop(RoborockB01Props.CLEAN_PATH_PREFERENCE, preference.code)
107
108    async def set_repeat_state(self, repeat: CleanRepeatMapping) -> None:
109        """Set the cleaning repeat state (cycles)."""
110        await self.set_prop(RoborockB01Props.REPEAT_STATE, repeat.code)
111
112    async def set_volume(self, volume: int) -> None:
113        """Set the robot voice volume (0-100)."""
114        await self.set_prop(RoborockB01Props.VOLUME, volume)
115
116    async def set_child_lock(self, enabled: bool) -> None:
117        """Enable or disable the child lock."""
118        await self.set_prop(RoborockB01Props.CHILD_LOCK, int(enabled))
119
120    async def set_do_not_disturb(self, enabled: bool, begin_time: int, end_time: int) -> None:
121        """Configure do-not-disturb.
122
123        The device expects all three values together via ``service.set_quiet_time``
124        (individual ``prop.set`` calls are ignored). ``begin_time``/``end_time`` are
125        minutes since midnight and must be in the range 0-1439 (inclusive).
126
127        Ranges that cross midnight are supported by passing a ``begin_time`` that is
128        greater than ``end_time`` (e.g. 22:00-07:00 is ``begin_time=1320``,
129        ``end_time=420``).
130        """
131        for name, value in (("begin_time", begin_time), ("end_time", end_time)):
132            if not 0 <= value <= 1439:
133                raise ValueError(f"{name} must be between 0 and 1439 minutes since midnight, got {value}")
134        await self.send(
135            RoborockB01Q7Methods.SET_QUIET_TIME,
136            {
137                "is_open": int(enabled),
138                "quiet_begin_time": begin_time,
139                "quiet_end_time": end_time,
140            },
141        )
142
143    async def start_clean(self) -> None:
144        """Start cleaning."""
145        await self.send(
146            command=RoborockB01Q7Methods.SET_ROOM_CLEAN,
147            params={
148                "clean_type": CleanTaskTypeMapping.ALL.code,
149                "ctrl_value": SCDeviceCleanParam.START.code,
150                "room_ids": [],
151            },
152        )
153
154    async def clean_segments(self, segment_ids: list[int]) -> None:
155        """Start segment cleaning for the given ids (Q7 uses room ids)."""
156        await self.send(
157            command=RoborockB01Q7Methods.SET_ROOM_CLEAN,
158            params={
159                "clean_type": CleanTaskTypeMapping.ROOM.code,
160                "ctrl_value": SCDeviceCleanParam.START.code,
161                "room_ids": segment_ids,
162            },
163        )
164
165    async def pause_clean(self) -> None:
166        """Pause cleaning."""
167        await self.send(
168            command=RoborockB01Q7Methods.SET_ROOM_CLEAN,
169            params={
170                "clean_type": CleanTaskTypeMapping.ALL.code,
171                "ctrl_value": SCDeviceCleanParam.PAUSE.code,
172                "room_ids": [],
173            },
174        )
175
176    async def stop_clean(self) -> None:
177        """Stop cleaning."""
178        await self.send(
179            command=RoborockB01Q7Methods.SET_ROOM_CLEAN,
180            params={
181                "clean_type": CleanTaskTypeMapping.ALL.code,
182                "ctrl_value": SCDeviceCleanParam.STOP.code,
183                "room_ids": [],
184            },
185        )
186
187    async def return_to_dock(self) -> None:
188        """Return to dock."""
189        await self.send(
190            command=RoborockB01Q7Methods.START_RECHARGE,
191            params={},
192        )
193
194    async def find_me(self) -> None:
195        """Locate the robot."""
196        await self.send(
197            command=RoborockB01Q7Methods.FIND_DEVICE,
198            params={},
199        )
200
201    async def send(self, command: CommandType, params: ParamsType) -> Any:
202        """Send a command to the device."""
203        return await send_decoded_command(
204            self._channel,
205            Q7RequestMessage(dps=B01_Q7_DPS, command=command, params=params),
206        )

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

13class MapTrait(Q7MapList, Trait):
14    """Map trait for B01/Q7 devices, responsible for fetching and caching map list metadata.
15
16    The MapContent is fetched from the MapContent trait, which relies on this trait to determine the
17    current map ID to fetch.
18    """
19
20    def __init__(self, channel: MqttChannel) -> None:
21        super().__init__()
22        self._channel = channel
23
24    async def refresh(self) -> None:
25        """Refresh cached map list metadata from the device."""
26        response = await send_decoded_command(
27            self._channel,
28            Q7RequestMessage(dps=B01_Q7_DPS, command=RoborockB01Q7Methods.GET_MAP_LIST, params={}),
29        )
30        if not isinstance(response, dict):
31            raise RoborockException(
32                f"Unexpected response type for GET_MAP_LIST: {type(response).__name__}: {response!r}"
33            )
34
35        if (parsed := Q7MapList.from_dict(response)) is None:
36            raise RoborockException(f"Failed to decode map list response: {response!r}")
37
38        self.map_list = parsed.map_list

 54class MapContentTrait(MapContent, Trait):
 55    """Trait for fetching parsed map content for Q7 devices."""
 56
 57    def __init__(
 58        self,
 59        map_rpc_channel: MapRpcChannel,
 60        map_trait: MapTrait,
 61        *,
 62        map_parser_config: B01MapParserConfig | None = None,
 63    ) -> None:
 64        super().__init__()
 65        self._map_rpc_channel = map_rpc_channel
 66        self._map_trait = map_trait
 67        self._map_parser = B01MapParser(map_parser_config)
 68        # Map uploads are serialized per-device to avoid response cross-wiring.
 69        self._map_command_lock = asyncio.Lock()
 70
 71    async def refresh(self) -> None:
 72        """Fetch, decode, and parse the current map payload.
 73
 74        This relies on the Map Trait already having fetched the map list metadata
 75        so it can determine the current map_id.
 76        """
 77        # Users must call first
 78        if (map_id := self._map_trait.current_map_id) is None:
 79            raise RoborockException("Unable to determine current map ID")
 80
 81        request = Q7RequestMessage(
 82            dps=B01_Q7_DPS,
 83            command=RoborockB01Q7Methods.UPLOAD_BY_MAPID,
 84            params={"map_id": map_id},
 85        )
 86        async with self._map_command_lock:
 87            raw_payload = await self._map_rpc_channel.send_map_command(request)
 88
 89        try:
 90            parsed_data = self._map_parser.parse(raw_payload)
 91        except RoborockException:
 92            raise
 93        except Exception as ex:
 94            raise RoborockException("Failed to parse B01 map data") from ex
 95
 96        if parsed_data.image_content is None:
 97            raise RoborockException("Failed to render B01 map image")
 98
 99        self.image_content = parsed_data.image_content
100        self.map_data = parsed_data.map_data
101        self.raw_api_response = raw_payload

 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