From 6e5bf5be71e9b055bf3a679c9286b96f32b9e0f2 Mon Sep 17 00:00:00 2001 From: Vincent Courcelle <2070309+tubededentifrice@users.noreply.github.com> Date: Sun, 19 Jul 2026 20:42:55 +0400 Subject: [PATCH 1/3] refactor(map): address Q10 decoding review feedback (#884) --- roborock/map/b01_grid_layers.py | 48 ++++++++++++++++++---------- roborock/map/b01_q10_map_parser.py | 14 ++++---- roborock/map/b01_q10_overlays.py | 27 ++++++++-------- tests/map/test_b01_grid_layers.py | 40 ++--------------------- tests/map/test_b01_q10_map_parser.py | 31 ++++++++++++++++++ tests/map/test_b01_q10_overlays.py | 22 +++++++------ 6 files changed, 97 insertions(+), 85 deletions(-) diff --git a/roborock/map/b01_grid_layers.py b/roborock/map/b01_grid_layers.py index 46026f17..bcc0f457 100644 --- a/roborock/map/b01_grid_layers.py +++ b/roborock/map/b01_grid_layers.py @@ -15,6 +15,7 @@ class (background / wall / per-room floor / ...). This module turns such a grid import io from collections.abc import Callable, Iterable from dataclasses import dataclass, field +from enum import IntEnum from math import ceil from PIL import Image @@ -28,6 +29,14 @@ class (background / wall / per-room floor / ...). This module turns such a grid _PNG = "PNG" +class _CalibrationCell(IntEnum): + """Cell categories used while scoring calibration candidates.""" + + OTHER = 0 + FLOOR = 1 + BLOCKED = 2 + + @dataclass class RoomLayer: """A single room (segment) and where its pixels sit in the grid.""" @@ -140,6 +149,21 @@ def pixel_to_world(self, px: float, py: float) -> tuple[float, float]: return ((px - self.origin_x) * self.resolution, self.y_sign * (self.origin_y - py) * self.resolution) +def _calibration_cells(layers: GridLayers) -> bytes: + """Classify the flat, row-major grid for calibration scoring.""" + cells = bytearray() + for value in layers.grid: + layer = layers.cell_class(value) + if layer == LAYER_FLOOR: + cell = _CalibrationCell.FLOOR + elif layer in (LAYER_WALL, LAYER_BACKGROUND): + cell = _CalibrationCell.BLOCKED + else: + cell = _CalibrationCell.OTHER + cells.append(cell) + return bytes(cells) + + def solve_calibration( layers: GridLayers, points: list[tuple[float, float]], @@ -162,11 +186,7 @@ def solve_calibration( if not points: return None w, h = layers.width, layers.height - classify = layers.classifier - # 1 = floor, 2 = wall/background (blocked), 0 = other. Index by cell. - klass = bytes( - 1 if (c := classify(v)) == LAYER_FLOOR else 2 if c in (LAYER_WALL, LAYER_BACKGROUND) else 0 for v in layers.grid - ) + cells = _calibration_cells(layers) best: tuple[float, GridCalibration] | None = None for resolution in resolutions: @@ -186,10 +206,10 @@ def solve_calibration( blocked = 0 for px_f, py_f in pts: cell = int(oy - py_f) * w + int(px_f + ox) - k = klass[cell] - if k == 1: + cell_type = cells[cell] + if cell_type == _CalibrationCell.FLOOR: on_floor += 1 - elif k == 2: + elif cell_type == _CalibrationCell.BLOCKED: blocked += 1 score = on_floor - 1.5 * blocked if best is None or score > best[0]: @@ -223,11 +243,7 @@ def solve_calibration_with_origin( return None w, h = layers.width, layers.height ox, oy = origin - classify = layers.classifier - # 1 = floor, 2 = wall/background (blocked), 0 = other. Index by cell. - klass = bytes( - 1 if (c := classify(v)) == LAYER_FLOOR else 2 if c in (LAYER_WALL, LAYER_BACKGROUND) else 0 for v in layers.grid - ) + cells = _calibration_cells(layers) best: tuple[float, GridCalibration] | None = None for resolution in resolutions: @@ -242,10 +258,10 @@ def solve_calibration_with_origin( if not (0 <= px < w and 0 <= py < h): blocked += 1 continue - k = klass[py * w + px] - if k == 1: + cell_type = cells[py * w + px] + if cell_type == _CalibrationCell.FLOOR: on_floor += 1 - elif k == 2: + elif cell_type == _CalibrationCell.BLOCKED: blocked += 1 score = on_floor - 1.5 * blocked if best is None or score > best[0]: diff --git a/roborock/map/b01_q10_map_parser.py b/roborock/map/b01_q10_map_parser.py index a3f7d1e7..062f85f3 100644 --- a/roborock/map/b01_q10_map_parser.py +++ b/roborock/map/b01_q10_map_parser.py @@ -64,13 +64,6 @@ def classify_q10_cell(value: int) -> str: return LAYER_FLOOR -def decompose_layers(packet: "Q10MapPacket") -> GridLayers: - """Split a parsed Q10 map packet into separable grid-pixel layers.""" - rooms = [(room.id, room.name, room.pixel_value, room.pixel_count) for room in packet.rooms] - # The ss07 grid is stored top-down (row 0 = top), so no display flip is applied. - return decompose_grid(packet.width, packet.height, packet.grid, rooms, classify_q10_cell, flip=False) - - MAP_PACKET_MARKER = b"\x01\x01" TRACE_PACKET_MARKER = b"\x02\x01" @@ -202,6 +195,13 @@ class Q10MapPacket: the same (top-down) pixel space as :attr:`grid`, where a non-zero cell is carpet (the value is the carpet kind). ``None`` if the packet carried none.""" + @property + def layers(self) -> GridLayers: + """Split the occupancy grid into separable grid-pixel layers.""" + rooms = [(room.id, room.name, room.pixel_value, room.pixel_count) for room in self.rooms] + # The ss07 grid is stored top-down (row 0 = top), so no display flip is applied. + return decompose_grid(self.width, self.height, self.grid, rooms, classify_q10_cell, flip=False) + @dataclass class Q10Point: diff --git a/roborock/map/b01_q10_overlays.py b/roborock/map/b01_q10_overlays.py index 34fcf71c..45b9bd80 100644 --- a/roborock/map/b01_q10_overlays.py +++ b/roborock/map/b01_q10_overlays.py @@ -27,6 +27,7 @@ """ import base64 +import binascii from dataclasses import dataclass, field _DEFAULT_RECORD_SIZE = 38 # 2-byte record header + up to 9 (x, y) int16 pairs @@ -40,25 +41,23 @@ class Q10Zone: vertices: list[tuple[int, int]] = field(default_factory=list) -def _as_bytes(data: bytes | str | None) -> bytes: +def _decode_blob(data: str | None) -> bytes: if data is None: return b"" - if isinstance(data, bytes): - return data try: return base64.b64decode(data + "=" * (-len(data) % 4)) - except (ValueError, base64.binascii.Error): # type: ignore[attr-defined] + except (ValueError, binascii.Error): return b"" -def parse_zone_blob(data: bytes | str | None) -> list[Q10Zone]: +def parse_zone_blob(data: str | None) -> list[Q10Zone]: """Decode a Q10 zone/wall overlay blob into a list of :class:`Q10Zone`. - Accepts the raw bytes or the base64 string straight from the data point. - Returns ``[]`` for empty/absent/unparsable blobs (the device sends a single - ``0x00`` byte when there are none). + Accepts the base64 string straight from the data point. Returns ``[]`` for + empty/absent/unparsable blobs (the device sends a single ``0x00`` byte when + there are none). """ - raw = _as_bytes(data) + raw = _decode_blob(data) if len(raw) < 2: return [] count = raw[1] @@ -94,7 +93,7 @@ def parse_zone_blob(data: bytes | str | None) -> list[Q10Zone]: _WALL_RECORD_SIZE = 8 # two (x, y) int16-BE endpoints -def parse_virtual_wall_blob(data: bytes | str | None) -> list[Q10Zone]: +def parse_virtual_wall_blob(data: str | None) -> list[Q10Zone]: """Decode a Q10 virtual-wall overlay blob (``dpVirtualWallUp`` 57). Virtual walls use a *different framing* from the restricted-zone DPs handled @@ -109,9 +108,9 @@ def parse_virtual_wall_blob(data: bytes | str | None) -> list[Q10Zone]: can place them onto the map through the same :class:`~roborock.map.b01_grid_layers.GridCalibration` as the zones. - Accepts raw bytes or the base64 string straight from the data point. Returns - ``[]`` for empty/absent/unparsable blobs (the device sends a single ``0x00`` - byte -- base64 ``AA==`` -- when there are none). + Accepts the base64 string straight from the data point. Returns ``[]`` for + empty/absent/unparsable blobs (the device sends a single ``0x00`` byte -- + base64 ``AA==`` -- when there are none). The axis order was confirmed against the app: a horizontal wall drawn below a room reads back with x varying and y constant (and the wide RDC no-go zone @@ -119,7 +118,7 @@ def parse_virtual_wall_blob(data: bytes | str | None) -> list[Q10Zone]: the wall axes to ``(y, x)`` -- following a misreading of PR #850's notes -- which placed every wall transposed 90 degrees from where it was drawn. """ - raw = _as_bytes(data) + raw = _decode_blob(data) if len(raw) < 1: return [] count = raw[0] diff --git a/tests/map/test_b01_grid_layers.py b/tests/map/test_b01_grid_layers.py index 52954203..85ab14f7 100644 --- a/tests/map/test_b01_grid_layers.py +++ b/tests/map/test_b01_grid_layers.py @@ -1,7 +1,6 @@ -"""Tests for the device-agnostic grid->layers decomposition + Q10 classifier.""" +"""Tests for the device-agnostic grid-to-layers decomposition.""" import io -from pathlib import Path import pytest from PIL import Image @@ -15,28 +14,6 @@ solve_calibration, solve_calibration_with_origin, ) -from roborock.map.b01_q10_map_parser import ( - classify_q10_cell, - decompose_layers, - parse_map_packet, -) - -FIXTURE = Path(__file__).resolve().parent / "testdata" / "b01_q10_map.bin" - - -@pytest.mark.parametrize( - ("value", "expected"), - [ - (0, "unknown"), - (8, LAYER_FLOOR), - (12, LAYER_FLOOR), - (240, LAYER_FLOOR), - (243, LAYER_BACKGROUND), - (249, LAYER_WALL), - ], -) -def test_classify_q10_cell(value: int, expected: str) -> None: - assert classify_q10_cell(value) == expected def test_decompose_grid_generic_classifier_and_bbox() -> None: @@ -97,21 +74,8 @@ def test_render_scale_upsamples() -> None: assert Image.open(io.BytesIO(png)).size == (6, 3) -def test_decompose_layers_on_q10_fixture() -> None: - """The Q10 synthetic fixture splits into floor + per-room layers.""" - layers = decompose_layers(parse_map_packet(FIXTURE.read_bytes())) - assert layers.class_counts.get(LAYER_FLOOR) == 26 - names = {room.id: room.name for room in layers.rooms} - assert names == {2: "Living Room", 3: "Bedroom"} - # Each room renders to a valid PNG and only its own pixels are opaque. - living = layers.render_room(2, (255, 0, 0, 255)) - img = Image.open(io.BytesIO(living)) - opaque = sum(1 for *_rgb, a in img.getdata() if a > 0) - assert opaque == next(r.pixel_count for r in layers.rooms if r.id == 2) - - def test_render_room_unknown_id_raises() -> None: - layers = decompose_layers(parse_map_packet(FIXTURE.read_bytes())) + layers = decompose_grid(1, 1, b"\x01", [(1, "Room", 1, 1)], lambda _: LAYER_FLOOR) with pytest.raises(KeyError): layers.render_room(999, (0, 0, 0, 255)) diff --git a/tests/map/test_b01_q10_map_parser.py b/tests/map/test_b01_q10_map_parser.py index 43095c57..9e815e29 100644 --- a/tests/map/test_b01_q10_map_parser.py +++ b/tests/map/test_b01_q10_map_parser.py @@ -1,13 +1,17 @@ """Tests for the Roborock Q10 (B01/ss07) map parser.""" +import io from pathlib import Path import pytest +from PIL import Image from roborock.exceptions import RoborockException +from roborock.map.b01_grid_layers import LAYER_BACKGROUND, LAYER_FLOOR, LAYER_WALL from roborock.map.b01_q10_map_parser import ( B01Q10MapParser, Q10Room, + classify_q10_cell, is_map_packet, is_trace_packet, lz4_block_decompress, @@ -118,6 +122,33 @@ def test_parse_map_packet() -> None: assert [(r.id, r.raw_name) for r in packet.rooms] == [(2, "rr_living_room"), (3, "bedroom")] +@pytest.mark.parametrize( + ("value", "expected"), + [ + (0, "unknown"), + (8, LAYER_FLOOR), + (12, LAYER_FLOOR), + (240, LAYER_FLOOR), + (243, LAYER_BACKGROUND), + (249, LAYER_WALL), + ], +) +def test_classify_q10_cell(value: int, expected: str) -> None: + assert classify_q10_cell(value) == expected + + +def test_packet_layers_decompose_q10_fixture() -> None: + """The Q10 synthetic fixture splits into floor + per-room layers.""" + layers = parse_map_packet(_payload()).layers + assert layers.class_counts.get(LAYER_FLOOR) == 26 + assert {room.id: room.name for room in layers.rooms} == {2: "Living Room", 3: "Bedroom"} + + living = layers.render_room(2, (255, 0, 0, 255)) + image = Image.open(io.BytesIO(living)) + opaque = sum(1 for *_rgb, alpha in image.getdata() if alpha > 0) + assert opaque == next(room.pixel_count for room in layers.rooms if room.id == 2) + + def test_parse_map_packet_allows_zero_room_metadata() -> None: """A map can be present before the robot has room segmentation records.""" grid = bytes([240, 240, 249, 243, 240, 240]) diff --git a/tests/map/test_b01_q10_overlays.py b/tests/map/test_b01_q10_overlays.py index 1d13eb62..9ed8481c 100644 --- a/tests/map/test_b01_q10_overlays.py +++ b/tests/map/test_b01_q10_overlays.py @@ -24,6 +24,10 @@ def _rect(zone_type: int, corners: list[tuple[int, int]]) -> bytes: return out +def _encoded(blob: bytes) -> str: + return base64.b64encode(blob).decode() + + def test_zone_type_constants() -> None: """ss07 + ioBroker: 0 no-go, 1 virtual wall, 2 no-mop, 3 threshold.""" assert (ZONE_TYPE_NO_GO, ZONE_TYPE_VIRTUAL_WALL, ZONE_TYPE_NO_MOP, ZONE_TYPE_THRESHOLD) == (0, 1, 2, 3) @@ -33,14 +37,14 @@ def test_parse_zone_blob_distinguishes_no_mop_and_threshold() -> None: """A no-mop (2) and a door-threshold (3) zone keep distinct types.""" no_mop = _rect(ZONE_TYPE_NO_MOP, [(0, 0), (10, 0), (10, 10), (0, 10)]) threshold = _rect(ZONE_TYPE_THRESHOLD, [(20, 20), (30, 20), (30, 22), (20, 22)]) - zones = parse_zone_blob(_blob(1, [no_mop, threshold])) + zones = parse_zone_blob(_encoded(_blob(1, [no_mop, threshold]))) assert [z.type for z in zones] == [ZONE_TYPE_NO_MOP, ZONE_TYPE_THRESHOLD] def test_parse_zone_blob_two_typed_rectangles() -> None: rect_a = _rect(ZONE_TYPE_NO_GO, [(0, 0), (10, 0), (10, 10), (0, 10)]) rect_b = _rect(ZONE_TYPE_NO_MOP, [(-5, -5), (5, -5), (5, 5), (-5, 5)]) - zones = parse_zone_blob(_blob(1, [rect_a, rect_b])) + zones = parse_zone_blob(_encoded(_blob(1, [rect_a, rect_b]))) assert [z.type for z in zones] == [ZONE_TYPE_NO_GO, ZONE_TYPE_NO_MOP] assert zones[0].vertices == [(0, 0), (10, 0), (10, 10), (0, 10)] assert zones[1].vertices == [(-5, -5), (5, -5), (5, 5), (-5, 5)] # signed coords @@ -48,29 +52,28 @@ def test_parse_zone_blob_two_typed_rectangles() -> None: def test_parse_zone_blob_accepts_base64() -> None: blob = _blob(1, [_rect(ZONE_TYPE_NO_GO, [(1, 2), (3, 4), (5, 6), (7, 8)])]) - zones = parse_zone_blob(base64.b64encode(blob).decode()) + zones = parse_zone_blob(_encoded(blob)) assert len(zones) == 1 and zones[0].vertices[2] == (5, 6) def test_parse_zone_blob_empty_variants() -> None: assert parse_zone_blob(None) == [] - assert parse_zone_blob(b"\x00") == [] # device's "no zones" sentinel assert parse_zone_blob("AA==") == [] # base64 of 0x00 - assert parse_zone_blob(bytes([1, 0, 0])) == [] # version=1, count=0 + assert parse_zone_blob("AQAA") == [] # version=1, count=0 def test_parse_zone_blob_skips_malformed_record() -> None: # vertex_count claims 9 verts (needs 38 bytes) but record is only 18 -> skipped. bad = bytes([ZONE_TYPE_NO_GO, 9]) + b"\x00" * 16 good = _rect(ZONE_TYPE_NO_GO, [(1, 1), (2, 2), (3, 3), (4, 4)]) - zones = parse_zone_blob(_blob(1, [bad, good])) + zones = parse_zone_blob(_encoded(_blob(1, [bad, good]))) assert len(zones) == 1 and zones[0].vertices[0] == (1, 1) def test_parse_zone_blob_real_record_size_inferred() -> None: """Record size is inferred from total/count (real device uses 38).""" rect = _rect(ZONE_TYPE_NO_GO, [(100, 200), (300, 200), (300, 50), (100, 50)]) - zones = parse_zone_blob(_blob(1, [rect], record_size=38)) + zones = parse_zone_blob(_encoded(_blob(1, [rect], record_size=38))) assert len(zones) == 1 and zones[0].vertices[0] == (100, 200) @@ -148,7 +151,6 @@ def test_parse_virtual_wall_blob_real_rdc_two_walls() -> None: def test_parse_virtual_wall_blob_empty_variants() -> None: assert parse_virtual_wall_blob(None) == [] - assert parse_virtual_wall_blob(b"\x00") == [] # device's "no walls" sentinel assert parse_virtual_wall_blob("AA==") == [] # base64 of 0x00 @@ -156,14 +158,14 @@ def test_parse_virtual_wall_blob_multiple_walls() -> None: """Two walls back-to-back; each is a separate 8-byte (x, y) record.""" wall_a = bytes.fromhex("000a0014001e0028") # (x,y)=(10,20)->(30,40) wall_b = bytes.fromhex("fffb0005fff6000a") # (x,y)=(-5,5)->(-10,10) - walls = parse_virtual_wall_blob(bytes([2]) + wall_a + wall_b) + walls = parse_virtual_wall_blob(_encoded(bytes([2]) + wall_a + wall_b)) assert [w.vertices for w in walls] == [[(10, 20), (30, 40)], [(-5, 5), (-10, 10)]] def test_parse_virtual_wall_blob_truncated_record_dropped() -> None: """A trailing record shorter than 8 bytes is dropped, not misread.""" blob = bytes([2]) + bytes([0x00, 0x0A, 0x00, 0x14, 0x00, 0x1E, 0x00, 0x28]) + b"\x00\x00" - walls = parse_virtual_wall_blob(blob) + walls = parse_virtual_wall_blob(_encoded(blob)) assert [w.vertices for w in walls] == [[(10, 20), (30, 40)]] From 01d0cc9722a66ed84390bae9e67f606d378123e4 Mon Sep 17 00:00:00 2001 From: Luke Lashley Date: Sun, 19 Jul 2026 12:43:07 -0400 Subject: [PATCH 2/3] chore: remove the old version of Status (#883) --- roborock/data/v1/v1_code_mappings.py | 307 ---------------------- roborock/data/v1/v1_containers.py | 375 +-------------------------- roborock/roborock_typing.py | 51 ---- tests/data/v1/test_v1_containers.py | 23 +- tests/devices/rpc/test_v1_channel.py | 10 +- tests/devices/test_v1_device.py | 4 +- tests/devices/traits/v1/fixtures.py | 4 +- 7 files changed, 19 insertions(+), 755 deletions(-) diff --git a/roborock/data/v1/v1_code_mappings.py b/roborock/data/v1/v1_code_mappings.py index d2a9b24d..82199186 100644 --- a/roborock/data/v1/v1_code_mappings.py +++ b/roborock/data/v1/v1_code_mappings.py @@ -281,209 +281,10 @@ class RoborockFanSpeedE2(RoborockFanPowerCode): turbo = 100 -class RoborockFanSpeedS7(RoborockFanPowerCode): - off = 105 - quiet = 101 - balanced = 102 - turbo = 103 - max = 104 - custom = 106 - - -class RoborockFanSpeedS7MaxV(RoborockFanPowerCode): - off = 105 - quiet = 101 - balanced = 102 - turbo = 103 - max = 104 - custom = 106 - max_plus = 108 - - -class RoborockFanSpeedS6Pure(RoborockFanPowerCode): - gentle = 105 - quiet = 101 - balanced = 102 - turbo = 103 - max = 104 - custom = 106 - - -class RoborockFanSpeedQ7Max(RoborockFanPowerCode): - quiet = 101 - balanced = 102 - turbo = 103 - max = 104 - - -class RoborockFanSpeedQRevoMaster(RoborockFanPowerCode): - off = 105 - quiet = 101 - balanced = 102 - turbo = 103 - max = 104 - custom = 106 - max_plus = 108 - smart_mode = 110 - - -class RoborockFanSpeedQRevoCurv(RoborockFanPowerCode): - quiet = 101 - balanced = 102 - turbo = 103 - max = 104 - off = 105 - custom = 106 - max_plus = 108 - smart_mode = 110 - - -class RoborockFanSpeedQRevoMaxV(RoborockFanPowerCode): - off = 105 - quiet = 101 - balanced = 102 - turbo = 103 - max = 104 - custom = 106 - max_plus = 108 - smart_mode = 110 - - -class RoborockFanSpeedP10(RoborockFanPowerCode): - off = 105 - quiet = 101 - balanced = 102 - turbo = 103 - max = 104 - custom = 106 - max_plus = 108 - smart_mode = 110 - - -class RoborockFanSpeedS8MaxVUltra(RoborockFanPowerCode): - off = 105 - quiet = 101 - balanced = 102 - turbo = 103 - max = 104 - custom = 106 - max_plus = 108 - smart_mode = 110 - - -class RoborockFanSpeedSaros10(RoborockFanPowerCode): - off = 105 - quiet = 101 - balanced = 102 - turbo = 103 - max = 104 - custom = 106 - max_plus = 108 - smart_mode = 110 - - -class RoborockFanSpeedSaros10R(RoborockFanPowerCode): - off = 105 - quiet = 101 - balanced = 102 - turbo = 103 - max = 104 - custom = 106 - max_plus = 108 - smart_mode = 110 - - -class RoborockMopModeCode(RoborockEnum): - """Describes the mop mode of the vacuum cleaner.""" - - -class RoborockMopModeQRevoCurv(RoborockMopModeCode): - standard = 300 - deep = 301 - custom = 302 - deep_plus = 303 - fast = 304 - smart_mode = 306 - - -class RoborockMopModeS7(RoborockMopModeCode): - """Describes the mop mode of the vacuum cleaner.""" - - standard = 300 - deep = 301 - custom = 302 - deep_plus = 303 - - -class RoborockMopModeS8ProUltra(RoborockMopModeCode): - standard = 300 - deep = 301 - deep_plus = 303 - fast = 304 - custom = 302 - smart_mode = 306 - - -class RoborockMopModeS8MaxVUltra(RoborockMopModeCode): - standard = 300 - deep = 301 - custom = 302 - deep_plus = 303 - fast = 304 - deep_plus_pearl = 305 - smart_mode = 306 - - -class RoborockMopModeSaros10R(RoborockMopModeCode): - standard = 300 - deep = 301 - custom = 302 - deep_plus = 303 - fast = 304 - smart_mode = 306 - - -class RoborockMopModeQRevoMaster(RoborockMopModeCode): - standard = 300 - deep = 301 - custom = 302 - deep_plus = 303 - fast = 304 - smart_mode = 306 - - -class RoborockMopModeQRevoMaxV(RoborockMopModeCode): - standard = 300 - deep = 301 - custom = 302 - deep_plus = 303 - fast = 304 - smart_mode = 306 - - -class RoborockMopModeSaros10(RoborockMopModeCode): - standard = 300 - deep = 301 - custom = 302 - deep_plus = 303 - fast = 304 - smart_mode = 306 - - class RoborockMopIntensityCode(RoborockEnum): """Describes the mop intensity of the vacuum cleaner.""" -class RoborockMopIntensityS7(RoborockMopIntensityCode): - """Describes the mop intensity of the vacuum cleaner.""" - - off = 200 - mild = 201 - moderate = 202 - intense = 203 - custom = 204 - - class RoborockMopIntensityV2(RoborockMopIntensityCode): """Describes the mop intensity of the vacuum cleaner.""" @@ -494,114 +295,6 @@ class RoborockMopIntensityV2(RoborockMopIntensityCode): custom = 207 -class RoborockMopIntensityQRevoMaster(RoborockMopIntensityCode): - """Describes the mop intensity of the vacuum cleaner.""" - - off = 200 - low = 201 - medium = 202 - high = 203 - custom = 204 - custom_water_flow = 207 - smart_mode = 209 - - -class RoborockMopIntensityQRevoCurv(RoborockMopIntensityCode): - off = 200 - low = 201 - medium = 202 - high = 203 - custom = 204 - custom_water_flow = 207 - smart_mode = 209 - - -class RoborockMopIntensityQRevoMaxV(RoborockMopIntensityCode): - off = 200 - low = 201 - medium = 202 - high = 203 - custom = 204 - custom_water_flow = 207 - smart_mode = 209 - - -class RoborockMopIntensityP10(RoborockMopIntensityCode): - """Describes the mop intensity of the vacuum cleaner.""" - - off = 200 - low = 201 - medium = 202 - high = 203 - custom = 204 - custom_water_flow = 207 - smart_mode = 209 - - -class RoborockMopIntensityS8MaxVUltra(RoborockMopIntensityCode): - off = 200 - low = 201 - medium = 202 - high = 203 - custom = 204 - max = 208 - smart_mode = 209 - custom_water_flow = 207 - - -class RoborockMopIntensitySaros10(RoborockMopIntensityCode): - off = 200 - mild = 201 - standard = 202 - intense = 203 - extreme = 208 - custom = 204 - smart_mode = 209 - - -class RoborockMopIntensitySaros10R(RoborockMopIntensityCode): - off = 200 - low = 201 - medium = 202 - high = 203 - custom = 204 - extreme = 250 - vac_followed_by_mop = 235 - smart_mode = 209 - - -class RoborockMopIntensityS5Max(RoborockMopIntensityCode): - """Describes the mop intensity of the vacuum cleaner.""" - - off = 200 - low = 201 - medium = 202 - high = 203 - custom = 204 - custom_water_flow = 207 - - -class RoborockMopIntensityS6MaxV(RoborockMopIntensityCode): - """Describes the mop intensity of the vacuum cleaner.""" - - off = 200 - low = 201 - medium = 202 - high = 203 - custom = 204 - custom_water_flow = 207 - - -class RoborockMopIntensityQ7Max(RoborockMopIntensityCode): - """Describes the mop intensity of the vacuum cleaner.""" - - off = 200 - low = 201 - medium = 202 - high = 203 - custom_water_flow = 207 - - class RoborockDockErrorCode(RoborockEnum): """Describes the error code of the dock.""" diff --git a/roborock/data/v1/v1_containers.py b/roborock/data/v1/v1_containers.py index 2f01870b..aaeb0800 100644 --- a/roborock/data/v1/v1_containers.py +++ b/roborock/data/v1/v1_containers.py @@ -11,32 +11,10 @@ MAIN_BRUSH_REPLACE_TIME, MOP_ROLLER_REPLACE_TIME, NO_MAP, - ROBOROCK_G10S_PRO, - ROBOROCK_P10, - ROBOROCK_Q7_MAX, - ROBOROCK_QREVO_CURV, - ROBOROCK_QREVO_MASTER, - ROBOROCK_QREVO_MAXV, - ROBOROCK_QREVO_PRO, - ROBOROCK_QREVO_S, - ROBOROCK_S4_MAX, - ROBOROCK_S5_MAX, - ROBOROCK_S6, - ROBOROCK_S6_MAXV, - ROBOROCK_S6_PURE, - ROBOROCK_S7, - ROBOROCK_S7_MAXV, - ROBOROCK_S8, - ROBOROCK_S8_MAXV_ULTRA, - ROBOROCK_S8_PRO_ULTRA, - ROBOROCK_SAROS_10, - ROBOROCK_SAROS_10R, SENSOR_DIRTY_REPLACE_TIME, SIDE_BRUSH_REPLACE_TIME, STRAINER_REPLACE_TIME, - ROBOROCK_G20S_Ultra, ) -from roborock.exceptions import RoborockException from roborock.roborock_message import RoborockDataProtocol from ..containers import NamedRoomMapping, RoborockBase, RoborockBaseTimer, _attr_repr, field_metadata @@ -53,41 +31,8 @@ RoborockDockState, RoborockDockTypeCode, RoborockErrorCode, - RoborockFanPowerCode, - RoborockFanSpeedP10, - RoborockFanSpeedQ7Max, - RoborockFanSpeedQRevoCurv, - RoborockFanSpeedQRevoMaster, - RoborockFanSpeedQRevoMaxV, - RoborockFanSpeedS6Pure, - RoborockFanSpeedS7, - RoborockFanSpeedS7MaxV, - RoborockFanSpeedS8MaxVUltra, - RoborockFanSpeedSaros10, - RoborockFanSpeedSaros10R, RoborockFinishReason, RoborockInCleaning, - RoborockMopIntensityCode, - RoborockMopIntensityP10, - RoborockMopIntensityQ7Max, - RoborockMopIntensityQRevoCurv, - RoborockMopIntensityQRevoMaster, - RoborockMopIntensityQRevoMaxV, - RoborockMopIntensityS5Max, - RoborockMopIntensityS6MaxV, - RoborockMopIntensityS7, - RoborockMopIntensityS8MaxVUltra, - RoborockMopIntensitySaros10, - RoborockMopIntensitySaros10R, - RoborockMopModeCode, - RoborockMopModeQRevoCurv, - RoborockMopModeQRevoMaster, - RoborockMopModeQRevoMaxV, - RoborockMopModeS7, - RoborockMopModeS8MaxVUltra, - RoborockMopModeS8ProUltra, - RoborockMopModeSaros10, - RoborockMopModeSaros10R, RoborockStartType, RoborockStateCode, ) @@ -128,192 +73,9 @@ class StatusField(FieldNameBase): RDT = "rdt" -@dataclass -class Status(RoborockBase): - """This status will be deprecated in favor of StatusV2.""" - - msg_ver: int | None = None - msg_seq: int | None = None - state: RoborockStateCode | None = field(default=None, metadata={"dps": RoborockDataProtocol.STATE}) - battery: int | None = field(default=None, metadata={"dps": RoborockDataProtocol.BATTERY}) - clean_time: int | None = None - clean_area: int | None = None - error_code: RoborockErrorCode | None = field(default=None, metadata={"dps": RoborockDataProtocol.ERROR_CODE}) - map_present: int | None = None - in_cleaning: RoborockInCleaning | None = None - in_returning: int | None = None - in_fresh_state: int | None = None - lab_status: int | None = None - water_box_status: int | None = None - back_type: int | None = None - wash_phase: int | None = None - wash_ready: int | None = None - fan_power: RoborockFanPowerCode | None = field(default=None, metadata={"dps": RoborockDataProtocol.FAN_POWER}) - dnd_enabled: int | None = None - map_status: int | None = None - is_locating: int | None = None - lock_status: int | None = None - water_box_mode: RoborockMopIntensityCode | None = field( - default=None, metadata={"dps": RoborockDataProtocol.WATER_BOX_MODE} - ) - water_box_carriage_status: int | None = None - mop_forbidden_enable: int | None = None - camera_status: int | None = None - is_exploring: int | None = None - home_sec_status: int | None = None - home_sec_enable_password: int | None = None - adbumper_status: list[int] | None = None - water_shortage_status: int | None = None - dock_type: RoborockDockTypeCode | None = None - dust_collection_status: int | None = None - auto_dust_collection: int | None = None - avoid_count: int | None = None - mop_mode: RoborockMopModeCode | None = None - debug_mode: int | None = None - collision_avoid_status: int | None = None - switch_map_mode: int | None = None - dock_error_status: RoborockDockErrorCode | None = None - charge_status: RoborockChargeStatus | None = field( - default=None, metadata={"dps": RoborockDataProtocol.CHARGE_STATUS} - ) - unsave_map_reason: int | None = None - unsave_map_flag: int | None = None - wash_status: int | None = None - distance_off: int | None = None - in_warmup: int | None = None - dry_status: int | None = field(default=None, metadata={"dps": RoborockDataProtocol.DRYING_STATUS}) - rdt: int | None = None - clean_percent: int | None = None - rss: int | None = None - dss: int | None = None - common_status: int | None = None - corner_clean_mode: int | None = None - last_clean_t: int | None = None - replenish_mode: int | None = None - repeat: int | None = None - kct: int | None = None - subdivision_sets: int | None = None - - @property - def square_meter_clean_area(self) -> float | None: - return round(self.clean_area / 1000000, 1) if self.clean_area is not None else None - - @property - def error_code_name(self) -> str | None: - return self.error_code.name if self.error_code is not None else None - - @property - def state_name(self) -> str | None: - return self.state.name if self.state is not None else None - - @property - def water_box_mode_name(self) -> str | None: - return self.water_box_mode.name if self.water_box_mode is not None else None - - @property - def fan_power_options(self) -> list[str]: - if self.fan_power is None: - return [] - return list(self.fan_power.keys()) - - @property - def fan_power_name(self) -> str | None: - return self.fan_power.name if self.fan_power is not None else None - - @property - def mop_mode_name(self) -> str | None: - return self.mop_mode.name if self.mop_mode is not None else None - - def get_fan_speed_code(self, fan_speed: str) -> int: - if self.fan_power is None: - raise RoborockException("Attempted to get fan speed before status has been updated.") - return self.fan_power.as_dict().get(fan_speed) - - def get_mop_intensity_code(self, mop_intensity: str) -> int: - if self.water_box_mode is None: - raise RoborockException("Attempted to get mop_intensity before status has been updated.") - return self.water_box_mode.as_dict().get(mop_intensity) - - def get_mop_mode_code(self, mop_mode: str) -> int: - if self.mop_mode is None: - raise RoborockException("Attempted to get mop_mode before status has been updated.") - return self.mop_mode.as_dict().get(mop_mode) - - @property - def current_map(self) -> int | None: - """Returns the current map ID if the map is present.""" - if self.map_status is not None: - map_flag = self.map_status >> 2 - if map_flag != NO_MAP: - return map_flag - return None - - @property - def has_am(self) -> bool | None: - if self.dss is None: - return None - return (self.dss & 3) == 2 - - @property - @field_metadata(dock_feature="is_washable") - def clear_water_box_status(self) -> ClearWaterBoxStatus | None: - if self.dss: - return ClearWaterBoxStatus((self.dss >> 2) & 3) - return None - - @property - @field_metadata(dock_feature="is_washable") - def dirty_water_box_status(self) -> DirtyWaterBoxStatus | None: - if self.dss: - return DirtyWaterBoxStatus((self.dss >> 4) & 3) - return None - - @property - def dust_bag_status(self) -> DustBagStatus | None: - if self.dss: - return DustBagStatus((self.dss >> 6) & 3) - return None - - @property - def water_box_filter_status(self) -> int | None: - if self.dss: - return (self.dss >> 8) & 3 - return None - - @property - @field_metadata( - dock_feature="is_clean_fluid_auto_delivery_supported", - ) - def clean_fluid_status(self) -> CleanFluidStatus | None: - if self.dss: - value = (self.dss >> 10) & 3 - if value == 0: - return None # Feature not supported by this device - return CleanFluidStatus(value) - return None - - @property - def hatch_door_status(self) -> int | None: - if self.dss: - return (self.dss >> 12) & 7 - return None - - @property - def dock_cool_fan_status(self) -> int | None: - if self.dss: - return (self.dss >> 15) & 3 - return None - - def __repr__(self) -> str: - return _attr_repr(self) - - @dataclass class StatusV2(RoborockBase): - """ - This is a new version of the Status object. - This is the result of GET_STATUS from the api. - """ + """The result of a GET_STATUS API request.""" msg_ver: int | None = None msg_seq: int | None = None @@ -492,141 +254,6 @@ def __repr__(self) -> str: return _attr_repr(self) -@dataclass -class S4MaxStatus(Status): - fan_power: RoborockFanSpeedS6Pure | None = None - water_box_mode: RoborockMopIntensityS7 | None = None - mop_mode: RoborockMopModeS7 | None = None - - -@dataclass -class S5MaxStatus(Status): - fan_power: RoborockFanSpeedS6Pure | None = None - water_box_mode: RoborockMopIntensityS5Max | None = None - - -@dataclass -class Q7MaxStatus(Status): - fan_power: RoborockFanSpeedQ7Max | None = None - water_box_mode: RoborockMopIntensityQ7Max | None = None - - -@dataclass -class QRevoMasterStatus(Status): - fan_power: RoborockFanSpeedQRevoMaster | None = None - water_box_mode: RoborockMopIntensityQRevoMaster | None = None - mop_mode: RoborockMopModeQRevoMaster | None = None - - -@dataclass -class QRevoCurvStatus(Status): - fan_power: RoborockFanSpeedQRevoCurv | None = None - water_box_mode: RoborockMopIntensityQRevoCurv | None = None - mop_mode: RoborockMopModeQRevoCurv | None = None - - -@dataclass -class QRevoMaxVStatus(Status): - fan_power: RoborockFanSpeedQRevoMaxV | None = None - water_box_mode: RoborockMopIntensityQRevoMaxV | None = None - mop_mode: RoborockMopModeQRevoMaxV | None = None - - -@dataclass -class S6MaxVStatus(Status): - fan_power: RoborockFanSpeedS7MaxV | None = None - water_box_mode: RoborockMopIntensityS6MaxV | None = None - - -@dataclass -class S6PureStatus(Status): - fan_power: RoborockFanSpeedS6Pure | None = None - - -@dataclass -class S7MaxVStatus(Status): - fan_power: RoborockFanSpeedS7MaxV | None = None - water_box_mode: RoborockMopIntensityS7 | None = None - mop_mode: RoborockMopModeS7 | None = None - - -@dataclass -class S7Status(Status): - fan_power: RoborockFanSpeedS7 | None = None - water_box_mode: RoborockMopIntensityS7 | None = None - mop_mode: RoborockMopModeS7 | None = None - - -@dataclass -class S8ProUltraStatus(Status): - fan_power: RoborockFanSpeedS7MaxV | None = None - water_box_mode: RoborockMopIntensityS7 | None = None - mop_mode: RoborockMopModeS8ProUltra | None = None - - -@dataclass -class S8Status(Status): - fan_power: RoborockFanSpeedS7MaxV | None = None - water_box_mode: RoborockMopIntensityS7 | None = None - mop_mode: RoborockMopModeS8ProUltra | None = None - - -@dataclass -class P10Status(Status): - fan_power: RoborockFanSpeedP10 | None = None - water_box_mode: RoborockMopIntensityP10 | None = None - mop_mode: RoborockMopModeS8ProUltra | None = None - - -@dataclass -class S8MaxvUltraStatus(Status): - fan_power: RoborockFanSpeedS8MaxVUltra | None = None - water_box_mode: RoborockMopIntensityS8MaxVUltra | None = None - mop_mode: RoborockMopModeS8MaxVUltra | None = None - - -@dataclass -class Saros10RStatus(Status): - fan_power: RoborockFanSpeedSaros10R | None = None - water_box_mode: RoborockMopIntensitySaros10R | None = None - mop_mode: RoborockMopModeSaros10R | None = None - - -@dataclass -class Saros10Status(Status): - fan_power: RoborockFanSpeedSaros10 | None = None - water_box_mode: RoborockMopIntensitySaros10 | None = None - mop_mode: RoborockMopModeSaros10 | None = None - - -ModelStatus: dict[str, type[Status]] = { - ROBOROCK_S4_MAX: S4MaxStatus, - ROBOROCK_S5_MAX: S5MaxStatus, - ROBOROCK_Q7_MAX: Q7MaxStatus, - ROBOROCK_QREVO_MASTER: QRevoMasterStatus, - ROBOROCK_QREVO_CURV: QRevoCurvStatus, - ROBOROCK_S6: S6PureStatus, - ROBOROCK_S6_MAXV: S6MaxVStatus, - ROBOROCK_S6_PURE: S6PureStatus, - ROBOROCK_S7_MAXV: S7MaxVStatus, - ROBOROCK_S7: S7Status, - ROBOROCK_S8: S8Status, - ROBOROCK_S8_PRO_ULTRA: S8ProUltraStatus, - ROBOROCK_G10S_PRO: S7MaxVStatus, - ROBOROCK_G20S_Ultra: QRevoMasterStatus, - ROBOROCK_P10: P10Status, - # These likely are not correct, - # but i am currently unable to do my typical reverse engineering/ get any data from users on this, - # so this will be here in the mean time. - ROBOROCK_QREVO_S: P10Status, - ROBOROCK_QREVO_MAXV: QRevoMaxVStatus, - ROBOROCK_QREVO_PRO: P10Status, - ROBOROCK_S8_MAXV_ULTRA: S8MaxvUltraStatus, - ROBOROCK_SAROS_10R: Saros10RStatus, - ROBOROCK_SAROS_10: Saros10Status, -} - - @dataclass class DnDTimer(RoborockBaseTimer): """DnDTimer""" diff --git a/roborock/roborock_typing.py b/roborock/roborock_typing.py index 71251324..ea97235b 100644 --- a/roborock/roborock_typing.py +++ b/roborock/roborock_typing.py @@ -1,17 +1,4 @@ -from dataclasses import dataclass, field from enum import Enum, StrEnum -from typing import Self - -from .data import ( - CleanRecord, - CleanSummary, - Consumable, - DustCollectionMode, - RoborockBase, - SmartWashParams, - Status, - WashTowelMode, -) class RoborockCommand(str, Enum): @@ -341,41 +328,3 @@ class RoborockB01Q7Methods(StrEnum): GET_RECORD_LIST = "service.get_record_list" GET_ORDER = "service.get_order" POST_PROP = "prop.post" - - -@dataclass -class DockSummary(RoborockBase): - dust_collection_mode: DustCollectionMode | None = None - wash_towel_mode: WashTowelMode | None = None - smart_wash_params: SmartWashParams | None = None - - -@dataclass -class DeviceProp(RoborockBase): - status: Status = field(default_factory=Status) - clean_summary: CleanSummary = field(default_factory=CleanSummary) - consumable: Consumable = field(default_factory=Consumable) - last_clean_record: CleanRecord | None = None - dock_summary: DockSummary | None = None - dust_collection_mode_name: str | None = None - - def __post_init__(self) -> None: - if ( - self.dock_summary - and self.dock_summary.dust_collection_mode is not None - and self.dock_summary.dust_collection_mode.mode is not None - ): - self.dust_collection_mode_name = self.dock_summary.dust_collection_mode.mode.name - - def update(self, device_prop: Self) -> None: - if device_prop.status: - self.status = device_prop.status - if device_prop.clean_summary: - self.clean_summary = device_prop.clean_summary - if device_prop.consumable: - self.consumable = device_prop.consumable - if device_prop.last_clean_record: - self.last_clean_record = device_prop.last_clean_record - if device_prop.dock_summary: - self.dock_summary = device_prop.dock_summary - self.__post_init__() diff --git a/tests/data/v1/test_v1_containers.py b/tests/data/v1/test_v1_containers.py index 34b671e7..ca7c91d4 100644 --- a/tests/data/v1/test_v1_containers.py +++ b/tests/data/v1/test_v1_containers.py @@ -12,9 +12,6 @@ RoborockDockState, RoborockDockTypeCode, RoborockErrorCode, - RoborockFanSpeedS7MaxV, - RoborockMopIntensityS7, - RoborockMopModeS7, RoborockStateCode, ) from roborock.data.v1.v1_code_mappings import ClearWaterBoxStatus, DirtyWaterBoxStatus, DustBagStatus @@ -24,7 +21,6 @@ CleanSummary, Consumable, DnDTimer, - S7MaxVStatus, StatusV2, ) from tests.mock_data import ( @@ -49,7 +45,7 @@ def test_consumable(): def test_status(): - s = S7MaxVStatus.from_dict(STATUS) + s = StatusV2.from_dict(STATUS) assert s.msg_ver == 2 assert s.msg_seq == 458 assert s.state == RoborockStateCode.charging @@ -94,9 +90,9 @@ def test_status(): assert s.charge_status == RoborockChargeStatus.charging assert s.unsave_map_reason == 0 assert s.unsave_map_flag == 0 - assert s.fan_power == RoborockFanSpeedS7MaxV.balanced - assert s.mop_mode == RoborockMopModeS7.standard - assert s.water_box_mode == RoborockMopIntensityS7.intense + assert s.fan_power == 102 + assert s.mop_mode == 300 + assert s.water_box_mode == 203 assert s.dss == 169 assert s.clear_water_box_status == ClearWaterBoxStatus.okay assert s.dirty_water_box_status == DirtyWaterBoxStatus.okay @@ -129,7 +125,7 @@ def test_dss_status( def test_current_map() -> None: """Test the current map logic based on map status.""" - s = S7MaxVStatus.from_dict(STATUS) + s = StatusV2.from_dict(STATUS) assert s.map_status == 3 assert s.current_map == 0 @@ -257,7 +253,7 @@ def test_clean_record(): def test_no_value(): modified_status = STATUS.copy() modified_status["dock_type"] = 9999 - s = S7MaxVStatus.from_dict(modified_status) + s = StatusV2.from_dict(modified_status) assert s.dock_type == RoborockDockTypeCode.unknown assert -9999 not in RoborockDockTypeCode.keys() assert "missing" not in RoborockDockTypeCode.values() @@ -267,7 +263,7 @@ def test_qrevo_s5v_dock_type(): """Test that dock type code 22 (Qrevo S5V dock) is properly recognized.""" modified_status = STATUS.copy() modified_status["dock_type"] = 22 - s = S7MaxVStatus.from_dict(modified_status) + s = StatusV2.from_dict(modified_status) assert s.dock_type == RoborockDockTypeCode.shell_2e_dock assert s.dock_type.value == 22 @@ -276,7 +272,6 @@ def test_has_am_dss_zero_is_not_missing(): modified_status = STATUS.copy() modified_status["dss"] = 0 - assert S7MaxVStatus.from_dict(modified_status).has_am is False assert StatusV2.from_dict(modified_status).has_am is False @@ -334,9 +329,9 @@ def test_multi_maps_list_info(snapshot: SnapshotAssertion) -> None: def test_accurate_map_flag() -> None: """Test that we parse the map flag accurately.""" - s = S7MaxVStatus.from_dict(STATUS) + s = StatusV2.from_dict(STATUS) assert s.current_map == 0 - s = S7MaxVStatus.from_dict( + s = StatusV2.from_dict( { **STATUS, "map_status": 252, # Code for no map diff --git a/tests/devices/rpc/test_v1_channel.py b/tests/devices/rpc/test_v1_channel.py index 278a796b..301bf581 100644 --- a/tests/devices/rpc/test_v1_channel.py +++ b/tests/devices/rpc/test_v1_channel.py @@ -11,7 +11,7 @@ import pytest -from roborock.data import NetworkInfo, RoborockStateCode, S5MaxStatus, UserData +from roborock.data import NetworkInfo, RoborockStateCode, StatusV2, UserData from roborock.devices.cache import DeviceCache, DeviceCacheData, InMemoryCache from roborock.devices.rpc.v1_channel import V1Channel from roborock.devices.transport.local_channel import LocalSession @@ -341,7 +341,7 @@ async def test_v1_channel_send_command_local_preferred( mock_local_channel.response_queue.append(TEST_RESPONSE) result = await rpc_channel.send_command( RoborockCommand.GET_STATUS, - response_type=S5MaxStatus, + response_type=StatusV2, ) # Verify local response was parsed @@ -370,7 +370,7 @@ async def test_v1_channel_send_command_local_fails( # Send command result = await rpc_channel.send_command( RoborockCommand.GET_STATUS, - response_type=S5MaxStatus, + response_type=StatusV2, ) # Verify result @@ -419,7 +419,7 @@ async def test_v1_channel_send_pick_first_available( mock_local_channel.response_queue.extend(local_channel_responses) result = await rpc_channel.send_command( RoborockCommand.GET_STATUS, - response_type=S5MaxStatus, + response_type=StatusV2, ) # Verify only MQTT was used @@ -442,7 +442,7 @@ async def test_v1_channel_send_decoded_command_with_params( test_params = {"volume": 80} await rpc_channel.send_command( RoborockCommand.CHANGE_SOUND_VOLUME, - response_type=S5MaxStatus, + response_type=StatusV2, params=test_params, ) diff --git a/tests/devices/test_v1_device.py b/tests/devices/test_v1_device.py index c327bdf0..3e387086 100644 --- a/tests/devices/test_v1_device.py +++ b/tests/devices/test_v1_device.py @@ -8,7 +8,7 @@ import pytest from syrupy import SnapshotAssertion -from roborock.data import HomeData, NetworkInfo, S7MaxVStatus, UserData +from roborock.data import HomeData, NetworkInfo, StatusV2, UserData from roborock.devices.cache import DeviceCache, DeviceCacheData, InMemoryCache, NoCache from roborock.devices.device import RoborockDevice from roborock.devices.rpc.v1_channel import V1Channel @@ -23,7 +23,7 @@ USER_DATA = UserData.from_dict(mock_data.USER_DATA) HOME_DATA = HomeData.from_dict(mock_data.HOME_DATA_RAW) -STATUS = S7MaxVStatus.from_dict(mock_data.STATUS) +STATUS = StatusV2.from_dict(mock_data.STATUS) TESTDATA = pathlib.Path("tests/protocols/testdata/v1_protocol/") diff --git a/tests/devices/traits/v1/fixtures.py b/tests/devices/traits/v1/fixtures.py index f73a9b11..d35cc0eb 100644 --- a/tests/devices/traits/v1/fixtures.py +++ b/tests/devices/traits/v1/fixtures.py @@ -5,7 +5,7 @@ import pytest -from roborock.data import HomeData, HomeDataDevice, HomeDataProduct, RoborockDockTypeCode, S7MaxVStatus, UserData +from roborock.data import HomeData, HomeDataDevice, HomeDataProduct, RoborockDockTypeCode, StatusV2, UserData from roborock.devices.cache import Cache, DeviceCache, InMemoryCache from roborock.devices.device import RoborockDevice from roborock.devices.traits import v1 @@ -13,7 +13,7 @@ USER_DATA = UserData.from_dict(mock_data.USER_DATA) HOME_DATA = HomeData.from_dict(mock_data.HOME_DATA_RAW) -STATUS = S7MaxVStatus.from_dict(mock_data.STATUS) +STATUS = StatusV2.from_dict(mock_data.STATUS) @pytest.fixture(autouse=True, name="channel") From 303fe71f68d6317baacebeab313f6e5aa0636c0d Mon Sep 17 00:00:00 2001 From: Vincent <2070309+tubededentifrice@users.noreply.github.com> Date: Sun, 19 Jul 2026 19:14:53 +0400 Subject: [PATCH 3/3] feat: compose Q10 maps in a pure renderer --- roborock/map/b01_q10_render.py | 337 +++++++++++++++++++++++++++++++ tests/map/test_b01_q10_render.py | 206 +++++++++++++++++++ 2 files changed, 543 insertions(+) create mode 100644 roborock/map/b01_q10_render.py create mode 100644 tests/map/test_b01_q10_render.py diff --git a/roborock/map/b01_q10_render.py b/roborock/map/b01_q10_render.py new file mode 100644 index 00000000..fd8c99d8 --- /dev/null +++ b/roborock/map/b01_q10_render.py @@ -0,0 +1,337 @@ +"""Compose a Q10 (B01/ss07) map into a single rendered result. + +The :class:`~roborock.map.b01_q10_map_parser.B01Q10MapParser` turns wire bytes +into a :class:`~roborock.map.b01_q10_map_parser.Q10MapPacket`; this module takes +that packet plus the *other* inputs the device streams separately -- the cleaning +path, the vector overlays (no-go / no-mop zones, virtual walls) and a solved +world<->pixel calibration -- and composes them into one :class:`Q10MapRender` result object +(image + ``MapData`` + layers). + +It exists so the map trait stays about state management: the trait accumulates +the pushed inputs and calls :func:`render_q10_map` once per change, holding the +returned object rather than mutating a pile of derived fields itself. All the +low-level pixel work (erase-zone blanking, world->pixel overlay placement, path +drawing) and the calibration policy live here, next to the rest of the map code. +""" + +import io +import math +from collections.abc import Sequence +from dataclasses import dataclass + +from PIL import Image, ImageDraw +from vacuum_map_parser_base.map_data import Area, MapData, Path, Point, Wall + +from roborock.exceptions import RoborockException + +from .b01_grid_layers import ( + GridCalibration, + GridLayers, + solve_calibration, + solve_calibration_with_origin, +) +from .b01_q10_map_parser import ( + B01Q10MapParser, + B01Q10MapParserConfig, + Q10HeaderCalibration, + Q10MapPacket, + Q10Point, + Q10Room, + erased_packet, +) +from .b01_q10_overlays import ZONE_TYPE_NO_GO, ZONE_TYPE_NO_MOP, Q10Zone + +# Path-units-per-pixel candidates for calibration. A dense ss07 path lands a +# best fit of 20.0 around the header origin -- ground-truthed June 2026 on the +# R1: a corridor drive registered at 20 (matching the format author's +# independent "20 path-units/px"), and the dock->corridor span lined up with the +# ruler-measured 8.81 m corridor. With the header resolution=5 (50 mm/px grid) +# that makes one path-unit exactly 50/20 = 2.5 mm -- so a path-unit is NOT a +# millimetre (the open scale question). An earlier [10.0..18.0] range couldn't +# reach 20 (it railed at the bound), biasing the fit. A dense cleaning path +# selects the best fit within this bracket. +_Q10_RESOLUTIONS = [step * 0.5 for step in range(24, 53)] # 12.0 .. 26.0 +# A path needs enough shape to constrain a full (origin + resolution) fit; a few +# points cannot. +_MIN_CALIBRATION_POINTS = 20 +# When the grid-frame header supplies the origin, only the resolution is fit, so +# a much shorter path suffices to confirm it (early in a clean, not just a dense +# one). See :func:`solve_calibration_with_origin`. +_MIN_HEADER_CALIBRATION_POINTS = 4 + + +@dataclass +class Q10MapRender: + """The fully composed result of rendering a Q10 map packet. + + Built by :func:`render_q10_map` from the packet plus the current path, + overlays and calibration, so every derived field is consistent with one set + of inputs. Analogous to :class:`~roborock.map.map_parser.ParsedMapData`, but + also carrying the separable :attr:`layers` and the :attr:`calibration` used + to place the vector overlays. + """ + + image_content: bytes + """The rendered base map (PNG) with erase zones blanked, path not drawn.""" + + map_data: MapData + """Parsed map data: image metadata, room names, and -- once a calibration is + known -- the path / robot position / zones / walls placed in pixel space.""" + + layers: GridLayers + """Separable map layers (background / wall / floor / per-room) in grid-pixel + space, each renderable to a transparent PNG for frontend compositing.""" + + rooms: list[Q10Room] + """Rooms (segments) reported by the device, with ids and names.""" + + calibration: GridCalibration | None + """World<->pixel transform used to place the overlays, or ``None`` if no + calibration was available (the overlays are then absent from ``map_data``).""" + + +def render_q10_map( + packet: Q10MapPacket, + *, + calibration: GridCalibration | None, + path: Sequence[Q10Point], + robot_position: Q10Point | None, + zones: Sequence[Q10Zone], + virtual_walls: Sequence[Q10Zone], + config: B01Q10MapParserConfig, +) -> Q10MapRender: + """Compose a Q10 map packet and its overlays into a :class:`Q10MapRender`. + + With a ``calibration`` the erase zones are blanked out of the raster and the + path / robot position / restricted zones / virtual walls are placed onto + ``map_data`` in pixel space; without one only the base raster is rendered + (the overlays are world-coordinate only and can't be placed yet). Raises + :class:`RoborockException` if the packet fails to render. + """ + parser = B01Q10MapParser(config) + layers = packet.layers + + render_packet = packet + if calibration is not None: + cells = _erased_cells(layers, packet.erase_zones, calibration) + if cells: + # Blank the erase-zone cells and re-derive the raster/layers from the + # modified packet so the phantom areas disappear (as the app shows). + render_packet = erased_packet(packet, cells) + layers = render_packet.layers + + parsed = parser.parsed_from_packet(render_packet) + if parsed.image_content is None or parsed.map_data is None: + raise RoborockException("Failed to render Q10 map image") + map_data = parsed.map_data + + if calibration is not None: + _place_path(map_data, calibration, path, robot_position) + _place_zones(map_data, calibration, path, zones, virtual_walls) + + return Q10MapRender( + image_content=parsed.image_content, + map_data=map_data, + layers=layers, + rooms=packet.rooms, + calibration=calibration, + ) + + +def solve_q10_calibration( + layers: GridLayers, + header_calibration: Q10HeaderCalibration | None, + path: Sequence[Q10Point], +) -> GridCalibration | None: + """Fit the world<->pixel calibration from the current cleaning path. + + When the map packet's grid-frame header carries a calibration origin (ss07), + only the resolution is fit -- around that fixed origin -- so a short path + suffices and the origin is exact rather than recovered by a slide. Otherwise + the full origin + resolution fit is used, which needs a reasonably dense + cleaning path. Returns ``None`` if the path is too short/featureless to fit. + """ + points: list[tuple[float, float]] = [(point.x, point.y) for point in path] + return _calibration_from_header(layers, header_calibration, points) or _calibration_from_fit(layers, points) + + +def _calibration_from_header( + layers: GridLayers, + header_calibration: Q10HeaderCalibration | None, + points: list[tuple[float, float]], +) -> GridCalibration | None: + """Calibrate around the header-supplied origin (resolution fit to a path).""" + if header_calibration is None or len(points) < _MIN_HEADER_CALIBRATION_POINTS: + return None + origin = header_calibration.origin_pixels() + if origin is None: # keepalive frame -- no usable origin + return None + return solve_calibration_with_origin(layers, points, origin, resolutions=_Q10_RESOLUTIONS) + + +def _calibration_from_fit(layers: GridLayers, points: list[tuple[float, float]]) -> GridCalibration | None: + """Full origin + resolution fit; needs a reasonably dense path.""" + if len(points) < _MIN_CALIBRATION_POINTS: + return None + return solve_calibration(layers, points, resolutions=_Q10_RESOLUTIONS) + + +def _erased_cells(layers: GridLayers, erase_zones: Sequence, calibration: GridCalibration) -> set[int]: + """Grid-cell indices covered by the erase zones (axis-aligned bbox fill).""" + if not erase_zones: + return set() + width, height = layers.width, layers.height + cells: set[int] = set() + for zone in erase_zones: + pixels = [calibration.world_to_pixel(x, y) for x, y in zone.vertices] + xs = [p[0] for p in pixels] + ys = [p[1] for p in pixels] + x0, x1 = int(min(xs)), int(max(xs)) + y0, y1 = int(min(ys)), int(max(ys)) + for py in range(max(0, y0), min(height, y1 + 1)): + for px in range(max(0, x0), min(width, x1 + 1)): + cells.add(py * width + px) + return cells + + +def _place_path( + map_data: MapData, + calibration: GridCalibration, + path: Sequence[Q10Point], + robot_position: Q10Point | None, +) -> None: + """Fill ``MapData.path`` / ``vacuum_position`` in grid-pixel coords. + + Points are stored in grid-pixel space (origin top-left), matching the Q10's + top-down, un-flipped raster so they line up with the rendered image. + """ + pixels = [Point(*calibration.world_to_pixel(point.x, point.y)) for point in path] + map_data.path = Path(len(pixels), 1, 0, [pixels]) + if robot_position is not None: + px, py = calibration.world_to_pixel(robot_position.x, robot_position.y) + map_data.vacuum_position = Point(px, py) + + +def _place_zones( + map_data: MapData, + calibration: GridCalibration, + path: Sequence[Q10Point], + zones: Sequence[Q10Zone], + virtual_walls: Sequence[Q10Zone], +) -> None: + """Convert world-coordinate zones/walls into pixel-space ``MapData`` layers.""" + + def to_area(zone: Q10Zone) -> Area | None: + if len(zone.vertices) != 4: + return None # MapData.Area is a quad + pts = [calibration.world_to_pixel(x, y) for x, y in zone.vertices] + return Area(pts[0][0], pts[0][1], pts[1][0], pts[1][1], pts[2][0], pts[2][1], pts[3][0], pts[3][1]) + + no_go = [area for zone in zones if zone.type == ZONE_TYPE_NO_GO and (area := to_area(zone))] + no_mop = [area for zone in zones if zone.type == ZONE_TYPE_NO_MOP and (area := to_area(zone))] + map_data.no_go_areas = no_go or None + map_data.no_mopping_areas = no_mop or None + + walls: list[Wall] = [] + for zone in virtual_walls: + if len(zone.vertices) >= 2: + (x0, y0), (x1, y1) = zone.vertices[0], zone.vertices[1] + p0 = calibration.world_to_pixel(x0, y0) + p1 = calibration.world_to_pixel(x1, y1) + walls.append(Wall(p0[0], p0[1], p1[0], p1[1])) + map_data.walls = walls or None + + # The robot starts a session at its dock, so the path origin is the charger. + if path: + cx, cy = calibration.world_to_pixel(path[0].x, path[0].y) + map_data.charger = Point(cx, cy) + + +def draw_path_on_map( + render: Q10MapRender, + *, + config: B01Q10MapParserConfig, + path: Sequence[Q10Point], + robot_position: Q10Point | None, + robot_heading: int | None, + zones: Sequence[Q10Zone], + virtual_walls: Sequence[Q10Zone], + line_color: tuple[int, int, int, int] = (235, 64, 52, 255), + position_color: tuple[int, int, int, int] = (255, 211, 0, 255), +) -> bytes: + """Draw the session path + robot position + overlays onto the base map (PNG). + + ``render`` must carry a calibration (its :attr:`Q10MapRender.calibration`) -- + the caller is responsible for solving one first. Returns a fresh PNG; the + ``render.image_content`` base raster is left untouched. + """ + calibration = render.calibration + if calibration is None: + raise RoborockException("No calibration available; a cleaning path must be captured during a clean") + + scale = config.map_scale + base = Image.open(io.BytesIO(render.image_content)).convert("RGBA") + + def world_to_image(x: float, y: float) -> tuple[float, float]: + px, py = calibration.world_to_pixel(x, y) + # The ss07 grid renders top-down (no flip), so grid-pixel (px, py) maps + # straight to image space, only upscaled by ``scale``. + return (px * scale, py * scale) + + def to_image(point: Q10Point) -> tuple[float, float]: + return world_to_image(point.x, point.y) + + draw = ImageDraw.Draw(base, "RGBA") + + # Erase zones are applied to the raster itself (cells blanked), so they are + # not drawn here -- the base image already reflects them. + + # No-go (blue) and no-mop (magenta) zones beneath the path. + for zone in zones: + if len(zone.vertices) < 3: + continue + polygon = [world_to_image(x, y) for x, y in zone.vertices] + fill = (0, 120, 255, 70) if zone.type == ZONE_TYPE_NO_GO else (255, 0, 200, 70) + outline = (0, 80, 200, 255) if zone.type == ZONE_TYPE_NO_GO else (200, 0, 160, 255) + draw.polygon(polygon, fill=fill, outline=outline) + + # Virtual walls (line segments, not polygons) drawn over the zones. + for wall in virtual_walls: + if len(wall.vertices) < 2: + continue + draw.line( + [world_to_image(x, y) for x, y in wall.vertices[:2]], + fill=(255, 64, 64, 255), + width=max(2, scale), + ) + + if len(path) >= 2: + draw.line([to_image(point) for point in path], fill=line_color, width=max(1, scale // 2)) + if path: # path origin == dock / charger + dx, dy = to_image(path[0]) + draw.ellipse([dx - scale, dy - scale, dx + scale, dy + scale], outline=(40, 200, 40, 255), width=2) + if robot_position is not None: + cx, cy = to_image(robot_position) + radius = scale + draw.ellipse([cx - radius, cy - radius, cx + radius, cy + radius], fill=position_color) + if robot_heading is not None: + # Heading is world-space degrees (0 = +x, +90 = +y). Map a unit + # world-space facing vector through the same transform (so the + # Y-flip/scale match the marker), then normalize to a fixed + # pixel-length tick so it reads at any calibration resolution. + angle = math.radians(robot_heading) + hx, hy = world_to_image( + robot_position.x + math.cos(angle), + robot_position.y + math.sin(angle), + ) + norm = math.hypot(hx - cx, hy - cy) + if norm > 0: + tick = 4 * radius + draw.line( + [cx, cy, cx + (hx - cx) / norm * tick, cy + (hy - cy) / norm * tick], + fill=position_color, + width=max(1, scale // 2), + ) + buffer = io.BytesIO() + base.save(buffer, format="PNG") + return buffer.getvalue() diff --git a/tests/map/test_b01_q10_render.py b/tests/map/test_b01_q10_render.py new file mode 100644 index 00000000..927aa2bc --- /dev/null +++ b/tests/map/test_b01_q10_render.py @@ -0,0 +1,206 @@ +"""Tests for composing a Q10 map packet + overlays into a rendered result. + +The pixel-level machinery (erase blanking, world->pixel overlay placement, path +drawing, calibration fitting) lives in ``b01_q10_render``; these exercise it with +explicit calibrations so the geometry is deterministic. The map trait's own tests +cover the state management that drives this module. +""" + +import io +from dataclasses import replace +from pathlib import Path + +from PIL import Image + +from roborock.map.b01_grid_layers import GridCalibration +from roborock.map.b01_q10_map_parser import ( + B01Q10MapParserConfig, + Q10EraseZone, + Q10HeaderCalibration, + Q10MapPacket, + Q10Point, + parse_map_packet, +) +from roborock.map.b01_q10_overlays import ZONE_TYPE_NO_GO, ZONE_TYPE_NO_MOP, Q10Zone +from roborock.map.b01_q10_render import ( + _Q10_RESOLUTIONS, + Q10MapRender, + draw_path_on_map, + render_q10_map, + solve_q10_calibration, +) + +FIXTURE = Path("tests/map/testdata/b01_q10_map.bin") +CONFIG = B01Q10MapParserConfig() + +# identity-ish calibration used across the geometry tests: world (x, y) -> grid +# pixel (x, 5 - y) over the fixture's 8x6 grid (top-down, no flip). +IDENTITY = GridCalibration(resolution=1.0, origin_x=0.0, origin_y=5.0, y_sign=1) + + +def _packet() -> Q10MapPacket: + return parse_map_packet(FIXTURE.read_bytes()) + + +def _render( + packet: Q10MapPacket | None = None, + *, + calibration: GridCalibration | None = None, + path: list[Q10Point] | None = None, + robot_position: Q10Point | None = None, + zones: list[Q10Zone] | None = None, + virtual_walls: list[Q10Zone] | None = None, +) -> Q10MapRender: + return render_q10_map( + packet if packet is not None else _packet(), + calibration=calibration, + path=path or [], + robot_position=robot_position, + zones=zones or [], + virtual_walls=virtual_walls or [], + config=CONFIG, + ) + + +def _floor_world_points(layers, cal: GridCalibration, count: int) -> list[Q10Point]: + """``count`` world points lying on the map's floor under ``cal``.""" + floor = [ + (px, py) + for py in range(layers.height) + for px in range(layers.width) + if layers.cell_class(layers.grid[py * layers.width + px]) == "floor" + ] + return [Q10Point(*(int(v) for v in cal.pixel_to_world(px, py))) for px, py in floor[:count]] + + +def test_render_base_map_without_calibration() -> None: + """Without a calibration only the base raster/layers/rooms are produced.""" + render = _render() + assert render.image_content[:8] == b"\x89PNG\r\n\x1a\n" + assert render.map_data is not None + assert render.calibration is None + assert {room.id: room.name for room in render.rooms} == {2: "Living Room", 3: "Bedroom"} + assert render.layers.class_counts.get("floor") == 26 + # Overlays are world-coordinate only, so nothing is placed yet. + assert render.map_data.path is None + + +def test_render_places_path_and_position() -> None: + """A calibration places the path + robot position onto MapData in pixels.""" + path = [Q10Point(1, 2), Q10Point(3, 2)] + render = _render(calibration=IDENTITY, path=path, robot_position=Q10Point(3, 2)) + assert render.map_data.path is not None + assert render.map_data.vacuum_position is not None + # world (3, 2) -> grid pixel (3, 5 - 2) = (3, 3) + assert (render.map_data.vacuum_position.x, render.map_data.vacuum_position.y) == (3.0, 3.0) + + +def test_render_places_zones_and_charger() -> None: + """Decoded no-go / no-mop zones become pixel-space MapData areas + charger.""" + zones = [ + Q10Zone(type=ZONE_TYPE_NO_GO, vertices=[(0, 0), (4, 0), (4, 4), (0, 4)]), + Q10Zone(type=ZONE_TYPE_NO_MOP, vertices=[(1, 1), (2, 1), (2, 2), (1, 2)]), + ] + render = _render(calibration=IDENTITY, path=[Q10Point(1, 1)], zones=zones) + assert len(render.map_data.no_go_areas or []) == 1 + assert len(render.map_data.no_mopping_areas or []) == 1 + # charger = path origin in pixels: (1, 5 - 1) = (1, 4) + assert render.map_data.charger is not None + assert (render.map_data.charger.x, render.map_data.charger.y) == (1.0, 4.0) + + +def test_render_applies_erase_zones() -> None: + """With a calibration, erase-zone cells are blanked from layers + image.""" + base = _render() + before_floor = base.layers.class_counts.get("floor") + assert before_floor and before_floor > 0 + + # A rectangle covering the whole grid in world coords erases every cell. + packet = replace(_packet(), erase_zones=[Q10EraseZone(vertices=[(0, 0), (7, 0), (7, 5), (0, 5)])]) + render = _render(packet, calibration=IDENTITY) + + assert render.layers.class_counts.get("floor", 0) == 0 # all floor erased + assert render.image_content != base.image_content # re-rendered + + +def test_render_partial_erase() -> None: + """An erase rectangle only blanks the cells it covers, leaving the rest.""" + before_floor = _render().layers.class_counts.get("floor", 0) + + # Cover only the top two grid rows (pixel y 0..1 -> world y 4..5). + packet = replace(_packet(), erase_zones=[Q10EraseZone(vertices=[(0, 4), (7, 4), (7, 5), (0, 5)])]) + render = _render(packet, calibration=IDENTITY) + + after_floor = render.layers.class_counts.get("floor", 0) + assert 0 < after_floor < before_floor # some, not all, floor removed + + +def test_draw_path_on_map_draws_position() -> None: + """The robot position is drawn at the mapped pixel.""" + path = [Q10Point(1, 2), Q10Point(3, 2)] + render = _render(calibration=IDENTITY, path=path, robot_position=Q10Point(3, 2)) + png = draw_path_on_map( + render, + config=CONFIG, + path=path, + robot_position=Q10Point(3, 2), + robot_heading=None, + zones=[], + virtual_walls=[], + position_color=(255, 211, 0, 255), + ) + img = Image.open(io.BytesIO(png)).convert("RGBA") + # world (3, 2) -> grid pixel (3, 3) -> image (12, 12) at scale 4 (no flip). + assert img.size == (8 * 4, 6 * 4) + assert img.getpixel((12, 12)) == (255, 211, 0, 255) + + +def test_draw_path_on_map_draws_heading_indicator() -> None: + """A known heading draws a facing tick from the robot marker. + + With heading 0 (= +x world) and the identity-ish calibration, the tick + extends to the right of the robot pixel; with the marker at image (12, 12) + the tick covers pixels at x > 12 along y == 12. + """ + path = [Q10Point(1, 2), Q10Point(3, 2)] + render = _render(calibration=IDENTITY, path=path, robot_position=Q10Point(3, 2)) + png = draw_path_on_map( + render, + config=CONFIG, + path=path, + robot_position=Q10Point(3, 2), + robot_heading=0, # facing +x + zones=[], + virtual_walls=[], + position_color=(255, 211, 0, 255), + ) + img = Image.open(io.BytesIO(png)).convert("RGBA") + # tick runs +x from the marker (4 * radius = 16 px at scale 4) + assert img.getpixel((20, 12)) == (255, 211, 0, 255) + # ...and not behind it (the marker is a small disc; sample well to the left) + assert img.getpixel((4, 12)) != (255, 211, 0, 255) + + +def test_solve_q10_calibration_uses_header_origin_with_short_path() -> None: + """A grid-frame header origin lets a short path calibrate (origin is exact).""" + layers = _render().layers + # Header origin in 5 mm units -> pixel origin (0, 5); not a keepalive frame. + header = Q10HeaderCalibration(origin_x=0, origin_y=50, resolution=5, charger_x=0, charger_y=0, charger_phi=0) + true = GridCalibration(resolution=20.0, origin_x=0.0, origin_y=5.0, y_sign=1) + path = _floor_world_points(layers, true, 6) + assert len(path) < 20 # far too short for the full origin+resolution fit + + cal = solve_q10_calibration(layers, header, path) + assert cal is not None + # Origin comes straight from the header (exact); only the resolution is fit, + # so it lands on one of the candidates (the exact pick is grid-quantized). + assert (cal.origin_x, cal.origin_y) == (0.0, 5.0) + assert cal.resolution in _Q10_RESOLUTIONS + + +def test_solve_q10_calibration_short_path_without_header_returns_none() -> None: + """Without a header origin a short path is too sparse for the full fit.""" + layers = _render().layers + true = GridCalibration(resolution=10.0, origin_x=0.0, origin_y=5.0, y_sign=1) + path = _floor_world_points(layers, true, 6) + assert solve_q10_calibration(layers, None, path) is None