roborock.devices.traits.b01

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

 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)