|
1 | | -"""Map content trait for B01 Q10 devices. |
2 | | -
|
3 | | -Unlike the v1 / Q7 maps, the Q10 has no synchronous "get map" command, so this |
4 | | -trait is purely push-driven and mirrors the Q10 ``StatusTrait`` contract: |
5 | | -
|
6 | | -- The device pushes its current map/path as protocol-301 ``MAP_RESPONSE`` |
7 | | - messages (a ``dpRequestDps`` nudges it to do so). The protocol layer decodes |
8 | | - those into :class:`Q10MapPacket` / :class:`Q10TracePacket` objects and the |
9 | | - ``Q10PropertiesApi`` subscribe loop routes them to |
10 | | - :meth:`MapContentTrait.update_from_map_packet` / |
11 | | - :meth:`MapContentTrait.update_from_trace_packet`. |
12 | | -- Those methods render/cache the content and notify update listeners (register |
13 | | - via :meth:`add_update_listener`). |
14 | | -- ``image_content``, ``map_data``, ``rooms``, ``path`` and ``robot_position`` |
15 | | - are readable and reflect the most recently pushed map. |
16 | | -
|
17 | | -Unlike the Q7, the Q10 map payload is unencrypted, so no map key is required. |
| 1 | +"""Push-driven map traits for B01 Q10 devices. |
| 2 | +
|
| 3 | +Map-related state arrives on three independent streams: |
| 4 | +
|
| 5 | +* map packets are decoded from map-protocol responses; |
| 6 | +* trace packets are decoded from trace-protocol responses; |
| 7 | +* restricted zones and virtual walls arrive as ordinary DPS values. |
| 8 | +
|
| 9 | +``MapDpsTrait`` owns the low-level DPS read model. ``MapContentTrait`` depends |
| 10 | +on it and combines that state with the latest map/trace packets through the pure |
| 11 | +functions in :mod:`roborock.map.b01_q10_render`. The high-level trait keeps only |
| 12 | +the latest value from each source and one replace-whole rendered image; |
| 13 | +calibration, path placement and overlay placement remain inside the renderer. |
18 | 14 | """ |
19 | 15 |
|
20 | 16 | import logging |
21 | 17 | from dataclasses import dataclass, field |
22 | | - |
23 | | -from vacuum_map_parser_base.map_data import MapData |
| 18 | +from typing import Any |
24 | 19 |
|
25 | 20 | from roborock.data import RoborockBase |
26 | | -from roborock.devices.traits.common import TraitUpdateListener |
| 21 | +from roborock.data.b01_q10.b01_q10_code_mappings import B01_Q10_DP |
| 22 | +from roborock.devices.traits.common import DpsDataConverter, TraitUpdateListener |
| 23 | +from roborock.exceptions import RoborockException |
27 | 24 | from roborock.map.b01_q10_map_parser import ( |
28 | | - B01Q10MapParser, |
29 | 25 | B01Q10MapParserConfig, |
30 | 26 | Q10MapPacket, |
31 | 27 | Q10Point, |
32 | 28 | Q10Room, |
33 | 29 | Q10TracePacket, |
34 | 30 | ) |
| 31 | +from roborock.map.b01_q10_overlays import parse_virtual_wall_blob, parse_zone_blob |
| 32 | +from roborock.map.b01_q10_render import Q10MapOverlays, render_q10_map |
35 | 33 |
|
36 | | -_LOGGER = logging.getLogger(__name__) |
| 34 | +from .common import UpdatableTrait |
37 | 35 |
|
38 | | -_TRUNCATE_LENGTH = 20 |
| 36 | +_LOGGER = logging.getLogger(__name__) |
39 | 37 |
|
40 | 38 |
|
41 | 39 | @dataclass |
42 | | -class MapContent(RoborockBase): |
43 | | - """Dataclass representing Q10 map content.""" |
| 40 | +class MapDps(RoborockBase): |
| 41 | + """Low-level map values delivered in the Q10 DPS stream.""" |
44 | 42 |
|
45 | | - image_content: bytes | None = None |
46 | | - """The rendered image of the map in PNG format.""" |
| 43 | + restricted_zone_up: str | None = field(default=None, metadata={"dps": B01_Q10_DP.RESTRICTED_ZONE_UP}) |
| 44 | + virtual_wall_up: str | None = field(default=None, metadata={"dps": B01_Q10_DP.VIRTUAL_WALL_UP}) |
47 | 45 |
|
48 | | - map_data: MapData | None = None |
49 | | - """Parsed map data (image metadata + room names).""" |
50 | 46 |
|
51 | | - rooms: list[Q10Room] = field(default_factory=list) |
52 | | - """Rooms (segments) reported by the device, with ids and names.""" |
| 47 | +class MapDpsTrait(MapDps, UpdatableTrait): |
| 48 | + """Private read model for map-related DPS values and decoded overlays.""" |
53 | 49 |
|
54 | | - path: list[Q10Point] = field(default_factory=list) |
55 | | - """Full path of the current cleaning session (oldest point first). |
| 50 | + _CONVERTER = DpsDataConverter.from_dataclass(MapDps) |
56 | 51 |
|
57 | | - The robot accumulates this server-side and serves the whole trajectory so |
58 | | - far in one packet, so it is complete even if we connect mid-session. Only |
59 | | - populated while a cleaning session is active.""" |
| 52 | + def __init__(self) -> None: |
| 53 | + MapDps.__init__(self) |
| 54 | + UpdatableTrait.__init__(self, command=None, logger=_LOGGER) |
| 55 | + self._overlays = Q10MapOverlays() |
60 | 56 |
|
61 | | - robot_position: Q10Point | None = None |
62 | | - """Current robot position (the most recent path point), if known.""" |
| 57 | + @property |
| 58 | + def overlays(self) -> Q10MapOverlays: |
| 59 | + """Overlays decoded once from the latest relevant DPS update.""" |
| 60 | + return self._overlays |
63 | 61 |
|
64 | | - def __repr__(self) -> str: |
65 | | - img = self.image_content |
66 | | - if img and len(img) > _TRUNCATE_LENGTH: |
67 | | - img = img[: _TRUNCATE_LENGTH - 3] + b"..." |
68 | | - return f"MapContent(image_content={img!r}, rooms={self.rooms!r})" |
| 62 | + def update_from_dps(self, decoded_dps: dict[B01_Q10_DP, Any]) -> None: |
| 63 | + """Decode overlay blobs when they arrive, then notify dependents.""" |
| 64 | + if not self._CONVERTER.update_from_dps(self, decoded_dps): |
| 65 | + return |
| 66 | + self._overlays = Q10MapOverlays( |
| 67 | + zones=tuple(parse_zone_blob(self.restricted_zone_up)), |
| 68 | + virtual_walls=tuple(parse_virtual_wall_blob(self.virtual_wall_up)), |
| 69 | + ) |
| 70 | + self._notify_update() |
69 | 71 |
|
70 | 72 |
|
71 | | -class MapContentTrait(MapContent, TraitUpdateListener): |
72 | | - """Trait holding the most recently pushed parsed map content for Q10 devices. |
| 73 | +class MapContentTrait(TraitUpdateListener): |
| 74 | + """High-level composed Q10 map view. |
73 | 75 |
|
74 | | - The Q10 has no synchronous get-map request; the device pushes map and trace |
75 | | - packets, which the protocol layer decodes and the ``Q10PropertiesApi`` |
76 | | - subscribe loop feeds into :meth:`update_from_map_packet` / |
77 | | - :meth:`update_from_trace_packet`. Consumers read the cached fields and/or |
78 | | - register a callback with :meth:`add_update_listener` to be notified when new |
79 | | - map content arrives. |
| 76 | + The latest map and trace packets are combined with the injected |
| 77 | + :class:`MapDpsTrait` whenever any of those three sources changes. |
80 | 78 | """ |
81 | 79 |
|
82 | 80 | def __init__( |
83 | 81 | self, |
| 82 | + map_dps: MapDpsTrait, |
84 | 83 | *, |
85 | 84 | map_parser_config: B01Q10MapParserConfig | None = None, |
86 | 85 | ) -> None: |
87 | | - super().__init__() |
88 | 86 | TraitUpdateListener.__init__(self, logger=_LOGGER) |
89 | | - self._map_parser = B01Q10MapParser(map_parser_config) |
| 87 | + self._config = map_parser_config or B01Q10MapParserConfig() |
| 88 | + self._map_dps = map_dps |
| 89 | + self._map_packet: Q10MapPacket | None = None |
| 90 | + self._trace_packet: Q10TracePacket | None = None |
| 91 | + self._image_content: bytes | None = None |
| 92 | + self._map_dps.add_update_listener(self._map_dps_updated) |
| 93 | + |
| 94 | + @property |
| 95 | + def image_content(self) -> bytes | None: |
| 96 | + """The composed map PNG, if the latest map rendered successfully.""" |
| 97 | + return self._image_content |
| 98 | + |
| 99 | + @property |
| 100 | + def rooms(self) -> list[Q10Room]: |
| 101 | + """Rooms reported by the device.""" |
| 102 | + return self._map_packet.rooms if self._map_packet else [] |
| 103 | + |
| 104 | + @property |
| 105 | + def path(self) -> list[Q10Point]: |
| 106 | + """Full path for live status and callers drawing their own map overlay.""" |
| 107 | + return self._trace_packet.points if self._trace_packet else [] |
| 108 | + |
| 109 | + @property |
| 110 | + def robot_position(self) -> Q10Point | None: |
| 111 | + """Current position for live status and caller-rendered map overlays.""" |
| 112 | + return self._trace_packet.robot_position if self._trace_packet else None |
| 113 | + |
| 114 | + @property |
| 115 | + def robot_heading(self) -> int | None: |
| 116 | + """Current heading for orienting a robot marker on a caller-rendered map.""" |
| 117 | + return self._trace_packet.heading if self._trace_packet else None |
90 | 118 |
|
91 | 119 | def update_from_map_packet(self, packet: Q10MapPacket) -> None: |
92 | | - """Render a pushed full-map packet into the cached image/rooms. |
93 | | -
|
94 | | - Rendering failures are logged and skipped (listeners are not notified) so |
95 | | - a single bad push cannot tear down the subscribe loop. |
96 | | - """ |
97 | | - parsed = self._map_parser.parse_packet(packet) |
98 | | - if parsed.image_content is None: |
99 | | - _LOGGER.debug("Failed to render Q10 map image") |
100 | | - return |
101 | | - self.image_content = parsed.image_content |
102 | | - self.map_data = parsed.map_data |
103 | | - self.rooms = packet.rooms |
| 120 | + """Store a map-protocol update and render the latest sources.""" |
| 121 | + self._map_packet = packet |
| 122 | + self._render() |
104 | 123 | self._notify_update() |
105 | 124 |
|
106 | 125 | def update_from_trace_packet(self, packet: Q10TracePacket) -> None: |
107 | | - """Cache the path/robot position from a pushed trace packet.""" |
108 | | - self.path = packet.points |
109 | | - self.robot_position = packet.robot_position |
| 126 | + """Store a trace-protocol update and render the latest sources.""" |
| 127 | + self._trace_packet = packet |
| 128 | + self._render() |
110 | 129 | self._notify_update() |
| 130 | + |
| 131 | + def _map_dps_updated(self) -> None: |
| 132 | + """Render after the low-level DPS source changes.""" |
| 133 | + if self._map_packet is None: |
| 134 | + return |
| 135 | + self._render() |
| 136 | + self._notify_update() |
| 137 | + |
| 138 | + def _render(self) -> None: |
| 139 | + """Render the required map with the latest optional trace and overlays.""" |
| 140 | + if self._map_packet is None: |
| 141 | + return |
| 142 | + try: |
| 143 | + self._image_content = render_q10_map( |
| 144 | + self._map_packet, |
| 145 | + self._trace_packet, |
| 146 | + self._map_dps.overlays, |
| 147 | + config=self._config, |
| 148 | + ) |
| 149 | + except RoborockException as ex: |
| 150 | + _LOGGER.debug("Failed to render Q10 map packet: %s", ex) |
| 151 | + self._image_content = None |
0 commit comments