roborock.data.v1.v1_dynamic_map

  1"""Wire models for the V1 dynamic map diff commands.
  2
  3These dataclasses describe the JSON results of ``get_dynamic_map_diff`` and
  4``get_dynamic_data`` after ordinary command-response decoding, plus the
  5snapshot metadata a full map result can carry. Field names match the wire keys
  6so ``RoborockBase.from_dict`` maps them directly.
  7
  8The models are deliberately passive. They decode base64 and answer the spec's
  9success predicate, but they hold no element state and make no merge decisions.
 10That logic lives with the dynamic element state machine.
 11"""
 12
 13import base64
 14import binascii
 15from dataclasses import dataclass
 16
 17from roborock.exceptions import RoborockException
 18
 19from ..code_mappings import RoborockEnum
 20from ..containers import RoborockBase
 21
 22__all__ = [
 23    "DynamicMapResult",
 24    "DynamicMapDiffEntry",
 25    "DynamicMapDiffResponse",
 26    "DynamicDataResponse",
 27    "DynamicMapSnapshotEntry",
 28    "DynamicMapSnapshot",
 29    "dynamic_map_diff_params",
 30    "dynamic_data_params",
 31]
 32
 33
 34class DynamicMapResult(RoborockEnum):
 35    """Application-level status returned by the dynamic map commands.
 36
 37    For ``get_dynamic_map_diff`` a ``nonce_mismatch`` means the requested base
 38    map is no longer current. For ``get_dynamic_data`` it means the requested
 39    element generation is no longer current. Neither establishes a new
 40    generation by itself; the accompanying nonce does.
 41    """
 42
 43    unknown = -1
 44    ok = 0
 45    nonce_mismatch = 1
 46    full_map_required = 2
 47
 48
 49def _decode_base64(value: str | None, field_name: str) -> bytes | None:
 50    """Decode standard or URL-safe base64, or return ``None`` when absent."""
 51    if value is None:
 52        return None
 53    try:
 54        return base64.b64decode(value, altchars=b"-_", validate=True)
 55    except (binascii.Error, ValueError) as err:
 56        raise RoborockException(f"Dynamic map field '{field_name}' is not valid base64") from err
 57
 58
 59def _validated_nonce(nonce: int) -> int:
 60    """Validate a 32-bit identity without changing its wire representation."""
 61    if not -(1 << 31) <= nonce <= 0xFFFFFFFF:
 62        raise ValueError("nonce must fit a signed or unsigned 32-bit integer")
 63    return nonce
 64
 65
 66@dataclass
 67class DynamicMapDiffEntry(RoborockBase):
 68    """One element entry inside a ``get_dynamic_map_diff`` result.
 69
 70    Refresh indicators (charger, map image, ...) carry only ``count``. Dynamic
 71    elements carry ``nonce`` and ``max_len``; the path element can also carry
 72    an inline page via ``start``, ``len``, and ``data``.
 73    """
 74
 75    count: int | None = None
 76    """Change indicator since the base-map snapshot. Not an item or byte count."""
 77
 78    result: DynamicMapResult | None = None
 79    """Optional page status; inline captures normally use ``count`` instead."""
 80
 81    nonce: int | None = None
 82    """Element generation identity."""
 83
 84    max_len: int | None = None
 85    """Total logical length currently available for this generation."""
 86
 87    start: int | None = None
 88    """Zero-based logical offset of an inline page."""
 89
 90    len: int | None = None
 91    """Logical item count of an inline page."""
 92
 93    data: str | None = None
 94    """Base64 block sequence of an inline page."""
 95
 96    obstacle: int | None = None
 97    """Obstacle component of the main map image count, when reported."""
 98
 99    space: int | None = None
100    """Space component of the main map image count, when reported."""
101
102    @property
103    def decoded_data(self) -> bytes | None:
104        """The inline page blocks, base64-decoded."""
105        return _decode_base64(self.data, "data")
106
107    @property
108    def has_inline_page(self) -> bool:
109        """Whether the inline range and data fields are present."""
110        return self.start is not None and self.len is not None and self.data is not None
111
112    @property
113    def is_success(self) -> bool:
114        """App acceptance predicate for a page: ``result == 0 or count == 0``."""
115        return self.result is DynamicMapResult.ok or self.count == 0
116
117
118@dataclass
119class DynamicMapDiffResponse(RoborockBase):
120    """Result object of ``get_dynamic_map_diff``."""
121
122    result: DynamicMapResult | None = None
123    """Application status. ``None`` when the device omitted it."""
124
125    nonce: int | None = None
126    """Map nonce the diff was computed against."""
127
128    diff: dict[int, DynamicMapDiffEntry] | None = None
129    """Element entries keyed by element id. An omitted element has no update."""
130
131    robot_data: str | None = None
132    """Base64 block sequence that updates the robot without adding path points."""
133
134    @property
135    def decoded_robot_data(self) -> bytes | None:
136        """The robot-only blocks, base64-decoded."""
137        return _decode_base64(self.robot_data, "robot_data")
138
139    def is_for_map(self, map_nonce: int) -> bool:
140        """Whether ``diff`` may be applied to the base map identified by ``map_nonce``."""
141        if self.result is not DynamicMapResult.ok or self.nonce is None or self.diff is None:
142            return False
143        try:
144            return (_validated_nonce(self.nonce) & 0xFFFFFFFF) == (_validated_nonce(map_nonce) & 0xFFFFFFFF)
145        except ValueError:
146            return False
147
148
149@dataclass
150class DynamicDataResponse(RoborockBase):
151    """Result object of ``get_dynamic_data``: one page of one element generation."""
152
153    result: DynamicMapResult | None = None
154    count: int | None = None
155    """Optional inline-page indicator accepted by the app's shared page handler."""
156
157    nonce: int | None = None
158    """Element nonce the page belongs to."""
159
160    data_id: int | None = None
161    start: int | None = None
162    len: int | None = None
163    """Actual logical range returned, which may be shorter than requested."""
164
165    max_len: int | None = None
166    """Total available length, which may have changed since it was announced."""
167
168    data: str | None = None
169
170    @property
171    def decoded_data(self) -> bytes | None:
172        """The page blocks, base64-decoded."""
173        return _decode_base64(self.data, "data")
174
175    @property
176    def is_success(self) -> bool:
177        """App acceptance predicate for a page: ``result == 0 or count == 0``."""
178        return self.result is DynamicMapResult.ok or self.count == 0
179
180
181@dataclass
182class DynamicMapSnapshotEntry(RoborockBase):
183    """Snapshot metadata for one element, supplied alongside a full map."""
184
185    nonce: int | None = None
186    limit_len: int | None = None
187    """Page limit for the element when it has no configured limit."""
188
189
190@dataclass
191class DynamicMapSnapshot(RoborockBase):
192    """Top-level ``nonce`` and per-element ``snapshot`` carried by a full map result."""
193
194    nonce: int | None = None
195    snapshot: dict[int, DynamicMapSnapshotEntry] | None = None
196
197
198def dynamic_map_diff_params(map_nonce: int, round_marker: int | None = None) -> dict[str, int]:
199    """Build ``get_dynamic_map_diff`` params for the cached base map.
200
201    ``round_marker`` is an opaque request marker, commonly the current time in
202    milliseconds. It can be omitted and is not a version, offset, or merge key.
203    Preserve signed or unsigned nonce representations supplied by the caller.
204    """
205    params = {"nonce": _validated_nonce(map_nonce)}
206    if round_marker is not None:
207        params["round"] = round_marker
208    return params
209
210
211def dynamic_data_params(element_nonce: int, data_id: int, start: int, length: int) -> dict[str, int]:
212    """Build ``get_dynamic_data`` params for one element range.
213
214    ``length`` of zero asks for the remaining available items, capped by the
215    element limit. Nonces retain the caller's signed or unsigned representation.
216    """
217    if start < 0 or length < 0:
218        raise ValueError("start and length must be nonnegative")
219    if data_id <= 0:
220        raise ValueError("data_id must be positive")
221    return {"nonce": _validated_nonce(element_nonce), "data_id": data_id, "start": start, "len": length}