roborock.devices.traits.b01.q7

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

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