roborock.devices.traits.b01.q10

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

111class CleanHistoryTrait(UpdatableTrait):
112    """Access to the Q10 clean-record history (``dpCleanRecord``, DP 52).
113
114    A read-model trait updated from the DPS stream like the others, but it overrides
115    :meth:`update_from_dps` because the payload is a structured push (a record list,
116    or a single ``op:"notify"`` record) rather than a flat data-point-to-field map.
117    """
118
119    def __init__(self, command: CommandTrait) -> None:
120        """Initialize the clean history trait."""
121        UpdatableTrait.__init__(self, command, _LOGGER)
122        self._converter = CleanRecordConverter()
123        self.records: list[Q10CleanRecord] = []
124        """Decoded clean records, most recent first."""
125
126    @property
127    def last_record(self) -> Q10CleanRecord | None:
128        """The most recent clean record, or ``None`` if there are none."""
129        return self.records[0] if self.records else None
130
131    async def refresh(self) -> None:
132        """Request the clean-record list from the device.
133
134        This sends the query and returns immediately; the records arrive
135        asynchronously on the device stream and populate :attr:`records` once
136        :meth:`update_from_dps` processes the ``dpCleanRecord`` push.
137        """
138        if self._command is None:
139            raise ValueError("Trait is read-only; no command channel was provided")
140        await self._command.send(
141            B01_Q10_DP.COMMON,
142            params={str(B01_Q10_DP.CLEAN_RECORD.code): {"op": "list"}},
143        )
144
145    def update_from_dps(self, decoded_dps: dict[B01_Q10_DP, Any]) -> None:
146        """Apply a ``dpCleanRecord`` push (a full list reply or a single notify)."""
147        envelope = decoded_dps.get(B01_Q10_DP.CLEAN_RECORD)
148        if not isinstance(envelope, dict):
149            return
150        push = self._converter.parse(envelope)
151        if push is None:
152            return
153        self._apply(push)
154
155    def _apply(self, push: CleanRecordPush) -> None:
156        """Merge or replace the records from ``push``, then sort newest-first and notify."""
157        if push.replace:
158            records = list(push.records)
159        else:
160            updated_ids = {record.record_id for record in push.records}
161            records = [record for record in self.records if record.record_id not in updated_ids]
162            records.extend(push.records)
163        records.sort(key=lambda record: record.start_time or 0, reverse=True)
164        self.records = records
165        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)

 84class MapContentTrait(TraitUpdateListener):
 85    """High-level composed Q10 map view.
 86
 87    The latest map and trace packets are combined with the injected
 88    :class:`MapDpsTrait` whenever a source changes. The
 89    :class:`MapsTrait` supplies a stored ID only when this trait requests
 90    content.
 91    """
 92
 93    def __init__(
 94        self,
 95        map_dps: MapDpsTrait,
 96        maps: MapsTrait,
 97        command: CommandTrait,
 98        *,
 99        map_parser_config: B01Q10MapParserConfig | None = None,
100    ) -> None:
101        TraitUpdateListener.__init__(self, logger=_LOGGER)
102        self._config = map_parser_config or B01Q10MapParserConfig()
103        self._map_dps = map_dps
104        self._maps = maps
105        self._command = command
106        self._map_packet: Q10MapPacket | None = None
107        self._trace_packet: Q10TracePacket | None = None
108        self._image_content: bytes | None = None
109        self._map_dps.add_update_listener(self._map_dps_updated)
110
111    async def refresh(self) -> None:
112        """Request a safe asynchronous current-map/status push.
113
114        Some ss07 firmware treats ``dpMultiMap op:get`` as an active
115        cleaning/relocation command. ``REQUEST_DPS`` is the device's read-only
116        current-map request and does not depend on a saved-map ID.
117        """
118        await self._command.send(B01_Q10_DP.REQUEST_DPS, params={})
119
120    @property
121    def image_content(self) -> bytes | None:
122        """The composed map PNG, if the latest map rendered successfully."""
123        return self._image_content
124
125    @property
126    def rooms(self) -> list[Q10Room]:
127        """Rooms reported by the device."""
128        return self._map_packet.rooms if self._map_packet else []
129
130    @property
131    def path(self) -> list[Q10Point]:
132        """Full path in the Q10 trace coordinate space used by the map renderer."""
133        return self._trace_packet.points if self._trace_packet else []
134
135    @property
136    def robot_position(self) -> Q10RoborockPoint | None:
137        """Current position in the common Roborock millimetre coordinate space."""
138        if self._trace_packet is None or (position := self._trace_packet.robot_position) is None:
139            return None
140        return position.to_roborock()
141
142    @property
143    def trace_sequence(self) -> int | None:
144        """Current cleaning-session sequence from the trace stream."""
145        return self._trace_packet.sequence if self._trace_packet else None
146
147    @property
148    def robot_heading(self) -> int | None:
149        """Current heading for orienting a robot marker on a caller-rendered map."""
150        return self._trace_packet.heading if self._trace_packet else None
151
152    def update_from_map_packet(self, packet: Q10MapPacket) -> None:
153        """Store a map-protocol update and render the latest sources."""
154        self._map_packet = packet
155        self._render()
156        self._notify_update()
157
158    def update_from_trace_packet(self, packet: Q10TracePacket) -> None:
159        """Store a trace-protocol update and render the latest sources."""
160        self._trace_packet = packet
161        self._render()
162        self._notify_update()
163
164    def _map_dps_updated(self) -> None:
165        """Render after the low-level map DPS source changes."""
166        if self._map_packet is None:
167            return
168        self._render()
169        self._notify_update()
170
171    def _render(self) -> None:
172        """Render the required map with the latest optional trace and overlays."""
173        if self._map_packet is None:
174            return
175        try:
176            self._image_content = render_q10_map(
177                self._map_packet,
178                self._trace_packet if not self._map_dps.robot_at_dock else None,
179                self._map_dps.overlays,
180                config=self._config,
181                robot_at_dock=self._map_dps.robot_at_dock,
182            )
183        except RoborockException as ex:
184            _LOGGER.debug("Failed to render Q10 map packet: %s", ex)
185            self._image_content = None
186
187    def as_dict(self, exclude: set[str] | None = None) -> dict[str, Any]:
188        """Return the trait data as a dictionary, excluding large binary data."""
189        exclude_set = exclude or set()
190        data = {
191            "rooms": [room.as_dict() for room in self.rooms],
192            "path": [point.as_dict() for point in self.path],
193            "robotPosition": (
194                {"x": position.x, "y": position.y} if (position := self.robot_position) is not None else None
195            ),
196            "robotHeading": self.robot_heading,
197        }
198        for key in exclude_set:
199            data.pop(key, None)
200        return data

33class MapsTrait(Maps, UpdatableTrait):
34    """Request and store the Q10 saved-map list."""
35
36    _CONVERTER = DpsDataConverter.from_dataclass(Maps)
37    _command: CommandTrait
38
39    def __init__(self, command: CommandTrait) -> None:
40        """Initialize the saved-map list trait."""
41        Maps.__init__(self)
42        UpdatableTrait.__init__(self, command, _LOGGER)
43        self._command = command
44
45    async def refresh(self) -> None:
46        """Request a new saved-map list from the device."""
47        await self._command.send(
48            B01_Q10_DP.COMMON,
49            {str(B01_Q10_DP.MULTI_MAP.code): {"op": "list"}},
50        )
51
52    def update_from_dps(self, decoded_dps: dict[B01_Q10_DP, Any]) -> None:
53        """Store a successful saved-map list response."""
54        response = decoded_dps.get(B01_Q10_DP.MULTI_MAP)
55        # DP 61 also carries map-content acknowledgements. Ignore them so they
56        # cannot replace a usable map list with an unrelated response.
57        if not isinstance(response, dict) or response.get("op") != "list" or response.get("result") != 1:
58            return
59        super().update_from_dps(decoded_dps)