roborock.devices.traits.b01.q10

 54class Q10PropertiesApi(Trait):
 55    """API for interacting with B01 devices."""
 56
 57    command: CommandTrait
 58    """Trait for sending commands to Q10 devices."""
 59
 60    status: StatusTrait
 61    """Trait for managing the core status of Q10 devices."""
 62
 63    vacuum: VacuumTrait
 64    """Trait for sending vacuum related commands to Q10 devices."""
 65
 66    remote: RemoteTrait
 67    """Trait for sending remote control related commands to Q10 devices."""
 68
 69    volume: SoundVolumeTrait
 70    """Trait for reading / setting the speaker volume."""
 71
 72    child_lock: ChildLockTrait
 73    """Trait for reading / controlling the child lock."""
 74
 75    do_not_disturb: DoNotDisturbTrait
 76    """Trait for reading / controlling Do Not Disturb."""
 77
 78    dust_collection: DustCollectionTrait
 79    """Trait for reading / controlling dock auto-empty (dust collection)."""
 80
 81    button_light: ButtonLightTrait
 82    """Trait for controlling the indicator / button light (LED)."""
 83
 84    network_info: NetworkInfoTrait
 85    """Trait exposing the device's network information."""
 86
 87    consumable: ConsumableTrait
 88    """Trait exposing remaining life of consumables."""
 89
 90    map: MapContentTrait
 91    """Composed map image plus caller-facing map and trace data."""
 92
 93    maps: MapsTrait
 94    """Saved-map list metadata."""
 95
 96    _map_dps: MapDpsTrait
 97    """Private source of restricted zones and virtual walls received through DPS."""
 98
 99    clean_history: CleanHistoryTrait
100    """Trait for fetching the device clean-record history (``dpCleanRecord``)."""
101
102    def __init__(
103        self,
104        channel: B01Q10Channel,
105        *,
106        map_parser_config: B01Q10MapParserConfig | None = None,
107    ) -> None:
108        """Initialize the B01Props API."""
109        self._channel = channel
110        self.command = CommandTrait(channel)
111        self.remote = RemoteTrait(self.command)
112        self.status = StatusTrait()
113        self.volume = SoundVolumeTrait(self.command)
114        self.child_lock = ChildLockTrait(self.command)
115        self.do_not_disturb = DoNotDisturbTrait(self.command)
116        self.dust_collection = DustCollectionTrait(self.command)
117        self.button_light = ButtonLightTrait(self.command)
118        self.network_info = NetworkInfoTrait()
119        self.consumable = ConsumableTrait()
120        self._map_dps = MapDpsTrait()
121        self.maps = MapsTrait(self.command)
122        self.map = MapContentTrait(
123            self._map_dps,
124            self.command,
125            map_parser_config=map_parser_config,
126        )
127        self.clean_history = CleanHistoryTrait(self.command)
128        self.vacuum = VacuumTrait(self.command, self.status, self.map)
129        # Read-model traits updated from the device's DPS push stream.
130        self._updatable_traits = [
131            self.status,
132            self.volume,
133            self.child_lock,
134            self.do_not_disturb,
135            self.dust_collection,
136            self.network_info,
137            self.consumable,
138            self.clean_history,
139            self._map_dps,
140            self.maps,
141        ]
142        self._subscribe_task: asyncio.Task[None] | None = None
143
144    async def start(self) -> None:
145        """Start any necessary subscriptions for the trait."""
146        self._subscribe_task = asyncio.create_task(self._subscribe_loop())
147
148    async def close(self) -> None:
149        """Close any resources held by the trait."""
150        await self.vacuum.close()
151        if self._subscribe_task is not None:
152            self._subscribe_task.cancel()
153            try:
154                await self._subscribe_task
155            except asyncio.CancelledError:
156                pass  # ignore cancellation errors
157            self._subscribe_task = None
158
159    async def refresh(self) -> None:
160        """Refresh all traits."""
161        # Sending REQUEST_DPS causes the device to publish its ordinary status
162        # values. Map-list and map-content refreshes have separate schedules.
163        await self.command.send(B01_Q10_DP.REQUEST_DPS, params={})
164
165    async def _subscribe_loop(self) -> None:
166        """Persistent loop dispatching decoded messages to the read-model traits."""
167        async for message in self._channel.subscribe_stream():
168            self._handle_message(message)
169
170    def _handle_message(self, message: Q10Message) -> None:
171        """Route a single decoded message to the trait responsible for it.
172
173        Map and trace packets arrive as protocol-301 ``MAP_RESPONSE`` pushes.
174        Map-list DPS responses and other DPS updates feed the read-model traits.
175        """
176        if isinstance(message, Q10MapPacket):
177            if message.kind is Q10MapPacketKind.CURRENT:
178                self.map.update_from_map_packet(message)
179            elif message.kind is Q10MapPacketKind.SAVED_MAP_DETAIL:
180                self.maps.update_from_map_packet(message)
181        elif isinstance(message, Q10CleanRecordDetail):
182            self.clean_history.update_from_detail(message)
183        elif isinstance(message, Q10TracePacket):
184            self.map.update_from_trace_packet(message)
185        elif isinstance(message, Q10DpsUpdate):
186            _LOGGER.debug("Received Q10 status update: %s", message.dps)
187            # Notify all read-model traits about the new message; each trait
188            # only updates the fields that it is responsible for.
189            for trait in self._updatable_traits:
190                trait.update_from_dps(message.dps)
191
192    def as_dict(self) -> dict[str, Any]:
193        """Return the trait data as a dictionary."""
194        result: dict[str, Any] = {}
195        for name, value in self.__dict__.items():
196            if isinstance(value, RoborockBase) and not name.startswith("_"):
197                result[name] = value.as_dict()
198        if hasattr(self, "map") and hasattr(self.map, "as_dict"):
199            result["map"] = self.map.as_dict()
200        return result

 9class ButtonLightTrait:
10    """Trait for controlling the indicator / button light (LED) of a Q10 device.
11
12    The device does not report the button-light state in its status dump, so
13    this trait is write-only (no read-back).
14    """
15
16    def __init__(self, command: CommandTrait) -> None:
17        """Initialize the button light trait."""
18        self._command = command
19
20    async def _write(self, value: int) -> None:
21        """Write the button-light data point via the dpCommon (101) wrapper."""
22        await self._command.send(B01_Q10_DP.COMMON, {str(B01_Q10_DP.BUTTON_LIGHT_SWITCH.code): value})
23
24    async def enable(self) -> None:
25        """Turn the indicator light on."""
26        await self._write(1)
27
28    async def disable(self) -> None:
29        """Turn the indicator light off."""
30        await self._write(0)

16class ChildLockTrait(ChildLock, UpdatableTrait):
17    """Trait for reading and controlling the child lock of a Q10 device."""
18
19    _CONVERTER = DpsDataConverter.from_dataclass(ChildLock)
20
21    def __init__(self, command: CommandTrait) -> None:
22        """Initialize the child lock trait."""
23        ChildLock.__init__(self)
24        UpdatableTrait.__init__(self, command, _LOGGER)
25
26    @property
27    def is_on(self) -> bool:
28        """Return whether the child lock is enabled."""
29        return bool(self.child_lock)
30
31    async def enable(self) -> None:
32        """Enable the child lock."""
33        await self._write(B01_Q10_DP.CHILD_LOCK, 1)
34
35    async def disable(self) -> None:
36        """Disable the child lock."""
37        await self._write(B01_Q10_DP.CHILD_LOCK, 0)

152class CleanHistoryTrait(CleanHistory, UpdatableTrait):
153    """Browse past cleaning runs and retrieve their parsed maps and historical paths.
154
155    Subscribe with ``add_update_listener`` to receive requested or unsolicited
156    archives. Archive pushes contain no record ID, so the latest detail cannot
157    be attributed to a particular selection.
158    """
159
160    _command: CommandTrait
161
162    def __init__(self, command: CommandTrait) -> None:
163        CleanHistory.__init__(self)
164        UpdatableTrait.__init__(self, command, _LOGGER)
165        self._command = command
166        self._converter = CleanRecordConverter()
167
168    async def refresh(self) -> None:
169        """Request the clean-record list from the device.
170
171        This sends the query and returns immediately; the records arrive
172        asynchronously on the device stream and populate :attr:`records` once
173        :meth:`update_from_dps` processes the ``dpCleanRecord`` push.
174        """
175        await self._command.send(
176            B01_Q10_DP.COMMON,
177            params={str(B01_Q10_DP.CLEAN_RECORD.code): {"op": "list"}},
178        )
179
180    async def refresh_detail(self, record: Q10CleanRecord) -> None:
181        """Request the saved map and path for a past cleaning run.
182
183        Returns after sending; listeners receive the archive separately. The
184        push contains no record ID to correlate it with this selection.
185        """
186        if not record.raw or not record.map_len:
187            raise RoborockException("The Q10 clean record has no saved map detail")
188        await self._command.send(
189            B01_Q10_DP.COMMON,
190            {str(B01_Q10_DP.CLEAN_RECORD.code): {"op": "select", "id": record.raw}},
191        )
192
193    def update_from_dps(self, decoded_dps: dict[B01_Q10_DP, Any]) -> None:
194        """Apply a ``dpCleanRecord`` push (a full list reply or a single notify)."""
195        envelope = decoded_dps.get(B01_Q10_DP.CLEAN_RECORD)
196        if not isinstance(envelope, dict):
197            return
198        push = self._converter.parse(envelope)
199        if push is None:
200            return
201        self._apply(push)
202
203    def update_from_detail(self, detail: Q10CleanRecordDetail) -> None:
204        """Store a pushed clean-record map and path and notify listeners."""
205        if detail.map.kind is not Q10MapPacketKind.CLEAN_RECORD_DETAIL:
206            raise ValueError(f"Expected a Q10 clean-record detail packet, got {detail.map.kind.value}")
207        self.detail = detail
208        self.detail_path = list(detail.trace.points) if detail.trace else []
209        self._notify_update()
210
211    def _apply(self, push: CleanRecordPush) -> None:
212        """Merge or replace the records from ``push``, then sort newest-first and notify."""
213        if push.replace:
214            records = list(push.records)
215        else:
216            updated_ids = {record.record_id for record in push.records}
217            records = [record for record in self.records if record.record_id not in updated_ids]
218            records.extend(push.records)
219        records.sort(key=lambda record: record.start_time or 0, reverse=True)
220        self.records = records
221        self._notify_update()

14class ConsumableTrait(Q10Consumable, UpdatableTrait):
15    """Trait exposing remaining life of consumables (brushes, filter, sensors)."""
16
17    _CONVERTER = DpsDataConverter.from_dataclass(Q10Consumable)
18
19    def __init__(self) -> None:
20        """Initialize the consumable trait."""
21        Q10Consumable.__init__(self)
22        UpdatableTrait.__init__(self, command=None, logger=_LOGGER)

16class DoNotDisturbTrait(DoNotDisturb, UpdatableTrait):
17    """Trait for reading and controlling Do Not Disturb on a Q10 device."""
18
19    _CONVERTER = DpsDataConverter.from_dataclass(DoNotDisturb)
20
21    def __init__(self, command: CommandTrait) -> None:
22        """Initialize the Do Not Disturb trait."""
23        DoNotDisturb.__init__(self)
24        UpdatableTrait.__init__(self, command, _LOGGER)
25
26    @property
27    def is_on(self) -> bool:
28        """Return whether Do Not Disturb is enabled."""
29        return bool(self.not_disturb)
30
31    async def enable(self) -> None:
32        """Enable Do Not Disturb."""
33        await self._write(B01_Q10_DP.NOT_DISTURB, 1)
34
35    async def disable(self) -> None:
36        """Disable Do Not Disturb."""
37        await self._write(B01_Q10_DP.NOT_DISTURB, 0)

16class DustCollectionTrait(DustCollection, UpdatableTrait):
17    """Trait for reading and controlling automatic dust collection at the dock."""
18
19    _CONVERTER = DpsDataConverter.from_dataclass(DustCollection)
20
21    def __init__(self, command: CommandTrait) -> None:
22        """Initialize the dust collection trait."""
23        DustCollection.__init__(self)
24        UpdatableTrait.__init__(self, command, _LOGGER)
25
26    @property
27    def is_on(self) -> bool:
28        """Return whether automatic dust collection is enabled."""
29        return bool(self.dust_switch)
30
31    async def enable(self) -> None:
32        """Enable automatic dust collection at the dock."""
33        await self._write(B01_Q10_DP.DUST_SWITCH, 1)
34
35    async def disable(self) -> None:
36        """Disable automatic dust collection at the dock."""
37        await self._write(B01_Q10_DP.DUST_SWITCH, 0)
38
39    async def set_frequency(self, frequency: YXDeviceDustCollectionFrequency) -> None:
40        """Set how often the dock empties the bin.
41
42        The value is the interval in cleans, with ``DAILY`` (0) meaning daily.
43        """
44        await self._write(B01_Q10_DP.DUST_SETTING, frequency.code)

 90class MapContentTrait(TraitUpdateListener):
 91    """High-level composed Q10 map view.
 92
 93    The latest map and trace packets are combined with the injected
 94    :class:`MapDpsTrait` whenever a source changes. Current-map acquisition is
 95    independent of the saved-map list.
 96    """
 97
 98    def __init__(
 99        self,
100        map_dps: MapDpsTrait,
101        command: CommandTrait,
102        *,
103        map_parser_config: B01Q10MapParserConfig | None = None,
104    ) -> None:
105        TraitUpdateListener.__init__(self, logger=_LOGGER)
106        self._config = map_parser_config or B01Q10MapParserConfig()
107        self._map_dps = map_dps
108        self._command = command
109        self._map_packet: Q10MapPacket | None = None
110        self._trace_packet: Q10TracePacket | None = None
111        self._image_content: bytes | None = None
112        self._map_dps.add_update_listener(self._map_dps_updated)
113
114    async def refresh(self) -> None:
115        """Request a safe asynchronous current-map/status push.
116
117        Some ss07 firmware treats ``dpMultiMap op:get`` as an active
118        cleaning/relocation command. ``REQUEST_DPS`` is the device's read-only
119        current-map request and does not depend on a saved-map ID.
120        """
121        await self._command.send(B01_Q10_DP.REQUEST_DPS, params={})
122
123    @property
124    def image_content(self) -> bytes | None:
125        """The composed map PNG, if the latest map rendered successfully."""
126        return self._image_content
127
128    @property
129    def rooms(self) -> list[Q10Room]:
130        """Rooms reported by the device."""
131        return self._map_packet.rooms if self._map_packet else []
132
133    @property
134    def path(self) -> list[Q10Point]:
135        """Full path in the Q10 trace coordinate space used by the map renderer."""
136        return self._trace_packet.points if self._trace_packet else []
137
138    @property
139    def restricted_zones(self) -> tuple[Q10RestrictedZone, ...]:
140        """Restrictions reported by the device in common Roborock coordinates."""
141        restrictions: list[Q10RestrictedZone] = []
142        for zone in self._map_dps.overlays.zones:
143            if len(zone.vertices) != 4:
144                continue
145            vertices = (
146                _to_roborock_point(zone.vertices[0]),
147                _to_roborock_point(zone.vertices[1]),
148                _to_roborock_point(zone.vertices[2]),
149                _to_roborock_point(zone.vertices[3]),
150            )
151            restriction_type = Q10RestrictionType(zone.type)
152            restrictions.append(Q10RestrictedZone(restriction_type, vertices))
153        return tuple(restrictions)
154
155    @property
156    def virtual_walls(self) -> tuple[Q10VirtualWall, ...]:
157        """Virtual walls reported by the device in common Roborock coordinates."""
158        walls: list[Q10VirtualWall] = []
159        for wall in self._map_dps.overlays.virtual_walls:
160            if len(wall.vertices) != 2:
161                continue
162            walls.append(
163                Q10VirtualWall(
164                    start=_to_roborock_point(wall.vertices[0]),
165                    end=_to_roborock_point(wall.vertices[1]),
166                )
167            )
168        return tuple(walls)
169
170    async def set_restricted_zones(self, zones: Sequence[Q10RestrictedZone]) -> None:
171        """Replace the complete restricted-zone collection."""
172        payload = encode_restricted_zones(zones)
173        await self._command.send(
174            B01_Q10_DP.COMMON,
175            {str(B01_Q10_DP.RESTRICTED_ZONE.code): payload},
176        )
177
178    async def set_virtual_walls(self, walls: Sequence[Q10VirtualWall]) -> None:
179        """Replace the complete virtual-wall collection."""
180        payload = encode_virtual_walls(walls)
181        await self._command.send(
182            B01_Q10_DP.COMMON,
183            {str(B01_Q10_DP.VIRTUAL_WALL.code): payload},
184        )
185
186    async def set_room_name(self, room_id: int, name: str) -> None:
187        """Set the name of a room in the current map."""
188        if room_id not in {room.id for room in self.rooms}:
189            raise RoborockException(f"Unknown Q10 room ID: {room_id}")
190        payload = encode_room_name(room_id, name)
191        await self._command.send(
192            B01_Q10_DP.COMMON,
193            {str(B01_Q10_DP.RESET_ROOM_NAME.code): payload},
194        )
195
196    @property
197    def robot_position(self) -> Q10RoborockPoint | None:
198        """Current position in the common Roborock millimetre coordinate space."""
199        if self._trace_packet is None or (position := self._trace_packet.robot_position) is None:
200            return None
201        return position.to_roborock()
202
203    @property
204    def trace_sequence(self) -> int | None:
205        """Current cleaning-session sequence from the trace stream."""
206        return self._trace_packet.sequence if self._trace_packet else None
207
208    @property
209    def robot_heading(self) -> int | None:
210        """Current heading for orienting a robot marker on a caller-rendered map."""
211        return self._trace_packet.heading if self._trace_packet else None
212
213    @property
214    def current_room(self) -> Q10Room | None:
215        """Room inferred from the latest live robot position and segmented map.
216
217        Returns ``None`` when position or calibration is unavailable, the pose
218        lies outside the map, or nearby room labels are ambiguous.
219        """
220        if self._map_packet is None or self._trace_packet is None:
221            return None
222        return resolve_q10_current_room(self._map_packet, self._trace_packet)
223
224    def update_from_map_packet(self, packet: Q10MapPacket) -> None:
225        """Store a map-protocol update and render the latest sources."""
226        if packet.kind is not Q10MapPacketKind.CURRENT:
227            raise ValueError(f"Expected a current Q10 map packet, got {packet.kind.value}")
228        self._map_packet = packet
229        self._render()
230        self._notify_update()
231
232    def update_from_trace_packet(self, packet: Q10TracePacket) -> None:
233        """Store a trace-protocol update and render the latest sources."""
234        self._trace_packet = None if self._map_dps.robot_at_dock else packet
235        self._render()
236        self._notify_update()
237
238    def _map_dps_updated(self) -> None:
239        """Render after the low-level map DPS source changes."""
240        if self._map_dps.robot_at_dock:
241            # A completed path no longer identifies the current pose at dock.
242            self._trace_packet = None
243        if self._map_packet is None:
244            return
245        self._render()
246        self._notify_update()
247
248    def _render(self) -> None:
249        """Render the required map with the latest optional trace and overlays."""
250        if self._map_packet is None:
251            return
252        try:
253            self._image_content = render_q10_map(
254                self._map_packet,
255                self._trace_packet if not self._map_dps.robot_at_dock else None,
256                self._map_dps.overlays,
257                config=self._config,
258                robot_at_dock=self._map_dps.robot_at_dock,
259            )
260        except RoborockException as ex:
261            _LOGGER.debug("Failed to render Q10 map packet: %s", ex)
262            self._image_content = None
263
264    def as_dict(self, exclude: set[str] | None = None) -> dict[str, Any]:
265        """Return the trait data as a dictionary, excluding large binary data."""
266        exclude_set = exclude or set()
267        current_room = self.current_room
268        data = {
269            "rooms": [room.as_dict() for room in self.rooms],
270            "path": [point.as_dict() for point in self.path],
271            "robotPosition": (
272                {"x": position.x, "y": position.y} if (position := self.robot_position) is not None else None
273            ),
274            "robotHeading": self.robot_heading,
275            "currentRoom": current_room.as_dict() if current_room is not None else None,
276        }
277        for key in exclude_set:
278            data.pop(key, None)
279        return data