roborock.devices.traits.b01

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

 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)