roborock.devices.traits.b01.q10

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

 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)

 81class MapContentTrait(TraitUpdateListener):
 82    """High-level composed Q10 map view.
 83
 84    The latest map and trace packets are combined with the injected map DPS
 85    whenever any source changes.
 86    """
 87
 88    def __init__(
 89        self,
 90        map_dps: MapDpsTrait,
 91        *,
 92        map_parser_config: B01Q10MapParserConfig | None = None,
 93    ) -> None:
 94        TraitUpdateListener.__init__(self, logger=_LOGGER)
 95        self._config = map_parser_config or B01Q10MapParserConfig()
 96        self._map_dps = map_dps
 97        self._map_packet: Q10MapPacket | None = None
 98        self._trace_packet: Q10TracePacket | None = None
 99        self._image_content: bytes | None = None
100        self._map_dps.add_update_listener(self._map_dps_updated)
101
102    @property
103    def image_content(self) -> bytes | None:
104        """The composed map PNG, if the latest map rendered successfully."""
105        return self._image_content
106
107    @property
108    def rooms(self) -> list[Q10Room]:
109        """Rooms reported by the device."""
110        return self._map_packet.rooms if self._map_packet else []
111
112    @property
113    def path(self) -> list[Q10Point]:
114        """Full path for live status and callers drawing their own map overlay."""
115        return self._trace_packet.points if self._trace_packet else []
116
117    @property
118    def robot_position(self) -> Q10Point | None:
119        """Current position for live status and caller-rendered map overlays."""
120        return self._trace_packet.robot_position if self._trace_packet else None
121
122    @property
123    def robot_heading(self) -> int | None:
124        """Current heading for orienting a robot marker on a caller-rendered map."""
125        return self._trace_packet.heading if self._trace_packet else None
126
127    def update_from_map_packet(self, packet: Q10MapPacket) -> None:
128        """Store a map-protocol update and render the latest sources."""
129        self._map_packet = packet
130        self._render()
131        self._notify_update()
132
133    def update_from_trace_packet(self, packet: Q10TracePacket) -> None:
134        """Store a trace-protocol update and render the latest sources."""
135        self._trace_packet = packet
136        self._render()
137        self._notify_update()
138
139    def _map_dps_updated(self) -> None:
140        """Render after the low-level map DPS source changes."""
141        if self._map_packet is None:
142            return
143        self._render()
144        self._notify_update()
145
146    def _render(self) -> None:
147        """Render the required map with the latest optional trace and overlays."""
148        if self._map_packet is None:
149            return
150        try:
151            self._image_content = render_q10_map(
152                self._map_packet,
153                self._trace_packet if not self._map_dps.robot_at_dock else None,
154                self._map_dps.overlays,
155                config=self._config,
156                robot_at_dock=self._map_dps.robot_at_dock,
157            )
158        except RoborockException as ex:
159            _LOGGER.debug("Failed to render Q10 map packet: %s", ex)
160            self._image_content = None

14class NetworkInfoTrait(Q10NetworkInfo, UpdatableTrait):
15    """Trait exposing the device's network information (read-only)."""
16
17    _CONVERTER = DpsDataConverter.from_dataclass(Q10NetworkInfo)
18
19    def __init__(self) -> None:
20        """Initialize the network info trait."""
21        Q10NetworkInfo.__init__(self)
22        UpdatableTrait.__init__(self, command=None, logger=_LOGGER)

16class SoundVolumeTrait(SoundVolume, UpdatableTrait):
17    """Trait for reading and setting the speaker volume of a Q10 device."""
18
19    _CONVERTER = DpsDataConverter.from_dataclass(SoundVolume)
20
21    def __init__(self, command: CommandTrait) -> None:
22        """Initialize the volume trait."""
23        SoundVolume.__init__(self)
24        UpdatableTrait.__init__(self, command, _LOGGER)
25
26    async def set_volume(self, volume: int) -> None:
27        """Set the speaker volume (0-100)."""
28        if not 0 <= volume <= 100:
29            raise ValueError("volume must be between 0 and 100")
30        await self._write(B01_Q10_DP.VOLUME, volume)

14class StatusTrait(Q10Status, UpdatableTrait):
15    """Trait for managing the core status of Q10 Roborock devices.
16
17    This is a thin wrapper around Q10Status that provides the Trait interface.
18    The current values reflect the most recently received data from the device.
19    New values can be requested through the `Q10PropertiesApi`'s `refresh` method.
20    """
21
22    _CONVERTER = DpsDataConverter.from_dataclass(Q10Status)
23
24    def __init__(self) -> None:
25        """Initialize the status trait."""
26        Q10Status.__init__(self)
27        UpdatableTrait.__init__(self, command=None, logger=_LOGGER)