roborock.devices.traits.b01

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

 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