roborock.devices.traits.b01.q10

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

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)

 72class MapContentTrait(MapContent, TraitUpdateListener):
 73    """Trait holding the most recently pushed parsed map content for Q10 devices.
 74
 75    The Q10 has no synchronous get-map request; the device pushes map and trace
 76    packets, which the protocol layer decodes and the ``Q10PropertiesApi``
 77    subscribe loop feeds into :meth:`update_from_map_packet` /
 78    :meth:`update_from_trace_packet`. Consumers read the cached fields and/or
 79    register a callback with :meth:`add_update_listener` to be notified when new
 80    map content arrives.
 81    """
 82
 83    def __init__(
 84        self,
 85        *,
 86        map_parser_config: B01Q10MapParserConfig | None = None,
 87    ) -> None:
 88        super().__init__()
 89        TraitUpdateListener.__init__(self, logger=_LOGGER)
 90        self._map_parser = B01Q10MapParser(map_parser_config)
 91
 92    def update_from_map_packet(self, packet: Q10MapPacket) -> None:
 93        """Render a pushed full-map packet into the cached image/rooms.
 94
 95        Rendering failures are logged and skipped (listeners are not notified) so
 96        a single bad push cannot tear down the subscribe loop.
 97        """
 98        parsed = self._map_parser.parse_packet(packet)
 99        if parsed.image_content is None:
100            _LOGGER.debug("Failed to render Q10 map image")
101            return
102        self.image_content = parsed.image_content
103        self.map_data = parsed.map_data
104        self.rooms = packet.rooms
105        self._notify_update()
106
107    def update_from_trace_packet(self, packet: Q10TracePacket) -> None:
108        """Cache the path/robot position from a pushed trace packet."""
109        self.path = packet.points
110        self.robot_position = packet.robot_position
111        self._notify_update()

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)