roborock.devices.traits.b01.q7

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

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

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

 61class MapContentTrait(MapContent, Trait, TraitUpdateListener):
 62    """Trait for fetching parsed map content for Q7 devices."""
 63
 64    def __init__(
 65        self,
 66        map_rpc_channel: Q7MapRpcChannel,
 67        map_trait: MapTrait,
 68        *,
 69        map_parser_config: B01MapParserConfig | None = None,
 70    ) -> None:
 71        MapContent.__init__(self)
 72        TraitUpdateListener.__init__(self, logger=_LOGGER)
 73        self._map_rpc_channel = map_rpc_channel
 74        self._map_trait = map_trait
 75        self._map_parser = B01MapParser(map_parser_config)
 76        # Map uploads are serialized per-device to avoid response cross-wiring.
 77        self._map_command_lock = asyncio.Lock()
 78
 79    async def refresh(self) -> None:
 80        """Fetch, decode, and parse the current map payload.
 81
 82        This relies on the Map Trait already having fetched the map list metadata
 83        so it can determine the current map_id.
 84        """
 85        # Users must call first
 86        if (map_id := self._map_trait.current_map_id) is None:
 87            raise RoborockException("Unable to determine current map ID")
 88
 89        async with self._map_command_lock:
 90            raw_payload = await self._map_rpc_channel.send_map_command(
 91                RoborockB01Q7Methods.UPLOAD_BY_MAPID,
 92                {"map_id": map_id},
 93            )
 94
 95        self._parse_and_store(raw_payload)
 96
 97    def update_from_push(self, raw_payload: bytes) -> None:
 98        """Store an unsolicited SCMap frame pushed by the device during cleaning.
 99
100        Pushed frames carry the live robot pose and cleaning path, so the
101        rendered image stays current without polling. Frames for other maps
102        than the live one are ignored to not overwrite the current map.
103        """
104        try:
105            map_type = parse_map_type(raw_payload)
106        except RoborockException as ex:
107            _LOGGER.debug("Failed to parse pushed B01 map frame: %s", ex)
108            return
109
110        if map_type != _LIVE_MAP_TYPE:
111            _LOGGER.debug("Ignoring pushed B01 map frame of type %s", map_type)
112            return
113
114        try:
115            self._parse_and_store(raw_payload)
116        except RoborockException as ex:
117            _LOGGER.debug("Failed to parse pushed B01 map frame: %s", ex)
118            return
119        self._notify_update()
120
121    def _parse_and_store(self, raw_payload: bytes) -> None:
122        """Parse decoded SCMap bytes and update the cached fields."""
123        try:
124            parsed_data = self._map_parser.parse(raw_payload)
125        except RoborockException:
126            raise
127        except Exception as ex:
128            raise RoborockException("Failed to parse B01 map data") from ex
129
130        if parsed_data.image_content is None:
131            raise RoborockException("Failed to render B01 map image")
132
133        self.image_content = parsed_data.image_content
134        self.map_data = parsed_data.map_data
135        self.raw_api_response = raw_payload

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

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