🦎 docs: clarify B01 protobuf choice

Author: arduano

roborock/map/b01_map_parser.py

@@ -6,6 +6,12 @@
 - PKCS7 padded
 - ASCII hex for a zlib-compressed SCMap payload
 
+The inner SCMap blob appears to use protobuf wire encoding, but we do not have
+an upstream `.proto` schema for it. We intentionally keep a tiny schema-free
+wire parser here instead of introducing generated protobuf classes from a
+reverse-engineered schema, because that would add maintenance/guesswork without
+meaningfully reducing complexity for the small set of fields we actually use.
+
 This module keeps the decode path narrow and explicit to match the observed
 payload shape as closely as possible.
 """

tests/map/debug_b01_scmap.py

@@ -0,0 +1,142 @@
+"""Developer helper for inspecting B01/Q7 SCMap payloads.
+
+This script is intentionally kept outside the runtime package so it stays
+non-obtrusive. It is useful when reverse-engineering new payload samples or
+validating assumptions about the current parser.
+
+Why not generated protobuf classes here?
+- The inflated SCMap payload looks like protobuf wire format.
+- We do not have an upstream `.proto` schema.
+- For runtime code, reverse-engineering and committing guessed schema files
+  would imply more certainty than we actually have.
+
+So the library keeps a tiny schema-free parser for the fields it needs, while
+this script provides a convenient place to inspect unknown payloads during
+future debugging.
+"""
+
+from __future__ import annotations
+
+import argparse
+import gzip
+from pathlib import Path
+
+from roborock.map.b01_map_parser import (
+    _decode_b01_map_payload,
+    _parse_scmap_payload,
+    _read_len_delimited,
+    _read_varint,
+)
+
+
+def _looks_like_message(blob: bytes) -> bool:
+    """Return True if the blob plausibly looks like a protobuf-style message."""
+    if not blob or len(blob) > 4096:
+        return False
+
+    idx = 0
+    seen = 0
+    try:
+        while idx < len(blob):
+            key, idx = _read_varint(blob, idx)
+            wire = key & 0x07
+            seen += 1
+            if wire == 0:
+                _, idx = _read_varint(blob, idx)
+            elif wire == 1:
+                idx += 8
+            elif wire == 2:
+                _, idx = _read_len_delimited(blob, idx)
+            elif wire == 5:
+                idx += 4
+            else:
+                return False
+        return seen > 0 and idx == len(blob)
+    except Exception:
+        return False
+
+
+def _preview(blob: bytes, limit: int = 24) -> str:
+    text = blob[:limit].hex()
+    if len(blob) > limit:
+        return f"{text}... ({len(blob)} bytes)"
+    return f"{text} ({len(blob)} bytes)"
+
+
+def _dump_message(blob: bytes, *, indent: str = "", max_depth: int = 2, depth: int = 0) -> None:
+    idx = 0
+    while idx < len(blob):
+        start = idx
+        key, idx = _read_varint(blob, idx)
+        field_no = key >> 3
+        wire = key & 0x07
+
+        if wire == 0:
+            value, idx = _read_varint(blob, idx)
+            print(f"{indent}field {field_no} @ {start}: varint {value}")
+        elif wire == 1:
+            value = blob[idx : idx + 8]
+            idx += 8
+            print(f"{indent}field {field_no} @ {start}: fixed64 {_preview(value, 8)}")
+        elif wire == 2:
+            value, idx = _read_len_delimited(blob, idx)
+            print(f"{indent}field {field_no} @ {start}: len-delimited {_preview(value)}")
+            if depth < max_depth and _looks_like_message(value):
+                _dump_message(value, indent=indent + "  ", max_depth=max_depth, depth=depth + 1)
+        elif wire == 5:
+            value = blob[idx : idx + 4]
+            idx += 4
+            print(f"{indent}field {field_no} @ {start}: fixed32 {_preview(value, 4)}")
+        else:
+            print(f"{indent}field {field_no} @ {start}: unsupported wire type {wire}")
+            return
+
+
+def _load_payload(args: argparse.Namespace) -> bytes:
+    if args.inflated_gzip is not None:
+        return gzip.decompress(args.inflated_gzip.read_bytes())
+    if args.inflated_bin is not None:
+        return args.inflated_bin.read_bytes()
+    if args.raw_map_response is not None:
+        if not args.serial or not args.model:
+            raise SystemExit("--raw-map-response requires --serial and --model")
+        return _decode_b01_map_payload(
+            args.raw_map_response.read_bytes(),
+            serial=args.serial,
+            model=args.model,
+        )
+    raise SystemExit("one of --inflated-gzip, --inflated-bin, or --raw-map-response is required")
+
+
+def _build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(description=__doc__)
+    group = parser.add_mutually_exclusive_group(required=True)
+    group.add_argument("--inflated-gzip", type=Path, help="Path to gzipped inflated SCMap payload")
+    group.add_argument("--inflated-bin", type=Path, help="Path to raw inflated SCMap payload")
+    group.add_argument("--raw-map-response", type=Path, help="Path to raw MAP_RESPONSE payload bytes")
+    parser.add_argument("--serial", help="Device serial number (required for --raw-map-response)")
+    parser.add_argument("--model", help="Device model, e.g. roborock.vacuum.sc05 (required for --raw-map-response)")
+    parser.add_argument(
+        "--max-depth",
+        type=int,
+        default=2,
+        help="Maximum recursive dump depth for protobuf-like messages",
+    )
+    return parser
+
+
+def main() -> None:
+    args = _build_parser().parse_args()
+    payload = _load_payload(args)
+
+    size_x, size_y, grid, room_names = _parse_scmap_payload(payload)
+    print(f"Inflated payload: {len(payload)} bytes")
+    print(f"Map size: {size_x} x {size_y}")
+    print(f"Grid bytes: {len(grid)}")
+    print(f"Room names: {room_names}")
+    print("\nTop-level field dump:")
+    _dump_message(payload, max_depth=args.max_depth)
+
+
+if __name__ == "__main__":
+    main()