From 4da996c07dc70c70bac58ec86dc14356ba7b0488 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 1/7] feat: compose Q10 maps in a pure renderer --- roborock/map/b01_q10_render.py | 334 +++++++++++++++++++++++++++++++ tests/map/test_b01_q10_render.py | 229 +++++++++++++++++++++ 2 files changed, 563 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..48d1ffa9 --- /dev/null +++ b/roborock/map/b01_q10_render.py @@ -0,0 +1,334 @@ +"""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 +combines that map-protocol packet with the latest trace-protocol packet and DPS +overlay snapshot. Calibration, path and position are derived from those source +objects rather than managed independently by callers. + +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, + Q10EraseZone, + Q10MapPacket, + Q10Room, + Q10TracePacket, + 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(frozen=True) +class Q10MapOverlays: + """Latest decoded map-overlay values from the Q10 DPS stream.""" + + zones: Sequence[Q10Zone] = () + virtual_walls: Sequence[Q10Zone] = () + + +@dataclass +class Q10MapRender: + """The fully composed result of rendering a Q10 map packet. + + Built by :func:`render_q10_map` from one map packet, trace packet and DPS + overlay snapshot, so every derived field is consistent with one set of + source inputs. Analogous to + :class:`~roborock.map.map_parser.ParsedMapData`, but also carrying the + separable :attr:`layers` and derived :attr:`calibration`. + """ + + 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, + trace: Q10TracePacket | None, + overlays: Q10MapOverlays, + *, + config: B01Q10MapParserConfig, +) -> Q10MapRender: + """Compose the latest map, trace and DPS inputs into a render. + + Calibration is derived from ``packet`` (layers + header calibration) and + ``trace`` (path points). Once calibrated, erase zones are blanked out of the + raster and trace/overlay data is projected into ``map_data`` pixel space. + Without a usable trace only the base raster is rendered. Raises + :class:`RoborockException` if map rendering fails. + """ + parser = B01Q10MapParser(config) + layers = packet.layers + calibration = solve_q10_calibration(packet, trace) + + 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 and trace is not None: + _place_trace(map_data, calibration, trace) + _place_overlays(map_data, calibration, overlays) + + return Q10MapRender( + image_content=parsed.image_content, + map_data=map_data, + layers=layers, + rooms=packet.rooms, + calibration=calibration, + ) + + +def solve_q10_calibration( + packet: Q10MapPacket, + trace: Q10TracePacket | None, +) -> GridCalibration | None: + """Derive world-to-pixel calibration from a map and its current trace. + + 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. + """ + if trace is None: + return None + points: list[tuple[float, float]] = [(point.x, point.y) for point in trace.points] + return _calibration_from_header(packet, points) or _calibration_from_fit(packet.layers, points) + + +def _calibration_from_header( + packet: Q10MapPacket, + points: list[tuple[float, float]], +) -> GridCalibration | None: + """Calibrate around the header-supplied origin (resolution fit to a path).""" + header_calibration = packet.header_calibration + 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(packet.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[Q10EraseZone], + 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_trace( + map_data: MapData, + calibration: GridCalibration, + trace: Q10TracePacket, +) -> None: + """Project trace path, position, heading and charger into pixel space. + + 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 trace.points] + map_data.path = Path(len(pixels), 1, 0, [pixels]) + robot_position = trace.robot_position + 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, trace.heading) + if pixels: + map_data.charger = pixels[0] + + +def _place_overlays( + map_data: MapData, + calibration: GridCalibration, + overlays: Q10MapOverlays, +) -> 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 overlays.zones if zone.type == ZONE_TYPE_NO_GO and (area := to_area(zone))] + no_mop = [area for zone in overlays.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 overlays.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 + + +def draw_path_on_map( + render: Q10MapRender, + *, + config: B01Q10MapParserConfig, + 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 projected ``MapData`` content onto the base map PNG. + + ``render`` must carry its derived calibration. Returns a fresh PNG; the base + raster in :attr:`Q10MapRender.image_content` 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 to_image(point: Point) -> tuple[float, float]: + return (point.x * scale, point.y * scale) + + 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 areas, fill, outline in ( + (render.map_data.no_go_areas or [], (0, 120, 255, 70), (0, 80, 200, 255)), + (render.map_data.no_mopping_areas or [], (255, 0, 200, 70), (200, 0, 160, 255)), + ): + for area in areas: + polygon = [ + (area.x0 * scale, area.y0 * scale), + (area.x1 * scale, area.y1 * scale), + (area.x2 * scale, area.y2 * scale), + (area.x3 * scale, area.y3 * scale), + ] + draw.polygon(polygon, fill=fill, outline=outline) + + # Virtual walls (line segments, not polygons) drawn over the zones. + for wall in render.map_data.walls or []: + draw.line( + [(wall.x0 * scale, wall.y0 * scale), (wall.x1 * scale, wall.y1 * scale)], + fill=(255, 64, 64, 255), + width=max(2, scale), + ) + + for path in render.map_data.path.path if render.map_data.path else []: + if len(path) >= 2: + draw.line([to_image(point) for point in path], fill=line_color, width=max(1, scale // 2)) + if render.map_data.charger is not None: + dx, dy = to_image(render.map_data.charger) + draw.ellipse([dx - scale, dy - scale, dx + scale, dy + scale], outline=(40, 200, 40, 255), width=2) + robot_position = render.map_data.vacuum_position + 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) + robot_heading = robot_position.a + 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) + dx = math.cos(angle) / calibration.resolution + dy = -calibration.y_sign * math.sin(angle) / calibration.resolution + norm = math.hypot(dx, dy) + if norm > 0: + tick = 4 * radius + draw.line( + [cx, cy, cx + dx / norm * tick, cy + dy / 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..97a17049 --- /dev/null +++ b/tests/map/test_b01_q10_render.py @@ -0,0 +1,229 @@ +"""Tests for composing a Q10 map packet + overlays into a rendered result. + +The pixel-level machinery (erase blanking, world-to-pixel overlay placement, +path drawing and calibration fitting) lives in ``b01_q10_render``. 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, + Q10TracePacket, + parse_map_packet, +) +from roborock.map.b01_q10_overlays import ( + ZONE_TYPE_NO_GO, + ZONE_TYPE_NO_MOP, + ZONE_TYPE_VIRTUAL_WALL, + Q10Zone, +) +from roborock.map.b01_q10_render import ( + _Q10_RESOLUTIONS, + Q10MapOverlays, + 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) +HEADER = Q10HeaderCalibration(origin_x=0, origin_y=50, resolution=5, charger_x=0, charger_y=0, charger_phi=0) +TRACE_CALIBRATION = GridCalibration(resolution=20.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, + *, + trace: Q10TracePacket | None = None, + overlays: Q10MapOverlays | None = None, +) -> Q10MapRender: + return render_q10_map( + packet if packet is not None else _packet(), + trace, + overlays or Q10MapOverlays(), + 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 _world_vertices(calibration: GridCalibration, pixels: list[tuple[int, int]]) -> list[tuple[int, int]]: + vertices = [] + for px, py in pixels: + world_x, world_y = calibration.pixel_to_world(px, py) + vertices.append((int(world_x), int(world_y))) + return vertices + + +def _calibrated_inputs(*, heading: int = 0) -> tuple[Q10MapPacket, Q10TracePacket]: + packet = replace(_packet(), header_calibration=HEADER) + pixels = [(1, 1), (6, 1), (1, 2), (6, 2), (2, 3), (5, 3)] + points = [Q10Point(*(int(value) for value in TRACE_CALIBRATION.pixel_to_world(px, py))) for px, py in pixels] + trace = Q10TracePacket(points=points, heading=heading) + return packet, trace + + +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: + """The map and trace derive calibration, path and position together.""" + packet, trace = _calibrated_inputs(heading=45) + render = _render(packet, trace=trace) + assert render.calibration is not None + assert render.map_data.path is not None + assert render.map_data.vacuum_position is not None + assert trace.robot_position is not None + expected = render.calibration.world_to_pixel(trace.robot_position.x, trace.robot_position.y) + assert (render.map_data.vacuum_position.x, render.map_data.vacuum_position.y) == expected + assert render.map_data.vacuum_position.a == 45 + + +def test_render_places_zones_and_charger() -> None: + """Decoded no-go / no-mop zones become pixel-space MapData areas + charger.""" + packet, trace = _calibrated_inputs() + 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)]), + ] + walls = [Q10Zone(type=ZONE_TYPE_VIRTUAL_WALL, vertices=[(0, 0), (4, 0)])] + render = _render(packet, trace=trace, overlays=Q10MapOverlays(zones=zones, virtual_walls=walls)) + assert len(render.map_data.no_go_areas or []) == 1 + assert len(render.map_data.no_mopping_areas or []) == 1 + assert len(render.map_data.walls or []) == 1 + assert render.calibration is not None + assert render.map_data.charger is not None + expected = render.calibration.world_to_pixel(trace.points[0].x, trace.points[0].y) + assert (render.map_data.charger.x, render.map_data.charger.y) == expected + + +def test_render_applies_erase_zones() -> None: + """With a calibration, erase-zone cells are blanked from layers + image.""" + packet, trace = _calibrated_inputs() + base = _render(packet, trace=trace) + before_floor = base.layers.class_counts.get("floor") + assert before_floor and before_floor > 0 + assert base.calibration is not None + + # A rectangle covering the whole grid in world coords erases every cell. + corners = [(-1, -1), (8, -1), (8, 6), (-1, 6)] + erase_zone = Q10EraseZone(vertices=_world_vertices(base.calibration, corners)) + render = _render(replace(packet, erase_zones=[erase_zone]), trace=trace) + + 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.""" + packet, trace = _calibrated_inputs() + base = _render(packet, trace=trace) + before_floor = base.layers.class_counts.get("floor", 0) + assert base.calibration is not None + + # Cover only the top two grid rows. + corners = [(-1, -1), (8, -1), (8, 2), (-1, 2)] + erase_zone = Q10EraseZone(vertices=_world_vertices(base.calibration, corners)) + render = _render(replace(packet, erase_zones=[erase_zone]), trace=trace) + + 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.""" + packet, trace = _calibrated_inputs() + render = _render(packet, trace=trace) + png = draw_path_on_map( + render, + config=CONFIG, + position_color=(255, 211, 0, 255), + ) + img = Image.open(io.BytesIO(png)).convert("RGBA") + position = render.map_data.vacuum_position + assert position is not None + image_position = (round(position.x * CONFIG.map_scale), round(position.y * CONFIG.map_scale)) + assert img.size == (8 * 4, 6 * 4) + assert img.getpixel(image_position) == (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. + """ + packet, trace = _calibrated_inputs(heading=0) + render = _render(packet, trace=trace) + png = draw_path_on_map( + render, + config=CONFIG, + position_color=(255, 211, 0, 255), + ) + img = Image.open(io.BytesIO(png)).convert("RGBA") + position = render.map_data.vacuum_position + assert position is not None + cx = round(position.x * CONFIG.map_scale) + cy = round(position.y * CONFIG.map_scale) + # Tick runs +x from the marker (4 * radius = 16 px at scale 4). + assert img.getpixel((cx + 8, cy)) == (255, 211, 0, 255) + # ...and not behind it (the marker is a small disc; sample well to the left) + assert img.getpixel((cx - 8, cy)) != (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).""" + packet, trace = _calibrated_inputs() + assert len(trace.points) < 20 # far too short for the full origin+resolution fit + + cal = solve_q10_calibration(packet, trace) + 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.""" + packet = _packet() + trace = Q10TracePacket(points=_floor_world_points(packet.layers, IDENTITY, 6)) + assert solve_q10_calibration(packet, trace) is None From 2db38da187b4407cd2562986761304550292bbf0 Mon Sep 17 00:00:00 2001 From: Vincent <2070309+tubededentifrice@users.noreply.github.com> Date: Sun, 19 Jul 2026 19:18:51 +0400 Subject: [PATCH 2/7] feat: compose Q10 map content from grouped traits --- roborock/cli.py | 89 ++++++++ roborock/devices/traits/b01/q10/__init__.py | 10 +- roborock/devices/traits/b01/q10/map.py | 216 ++++++++++++++------ tests/devices/traits/b01/q10/test_map.py | 211 +++++++++++++++++-- 4 files changed, 439 insertions(+), 87 deletions(-) diff --git a/roborock/cli.py b/roborock/cli.py index 9def26af..12aafc63 100644 --- a/roborock/cli.py +++ b/roborock/cli.py @@ -662,6 +662,59 @@ async def map_image(ctx, device_id: str, output_file: str): click.echo("No map image content available.") +@session.command() +@click.option("--device_id", required=True) +@click.option("--output-dir", default=None, help="If set, write one transparent PNG per layer here.") +@click.pass_context +@async_command +async def q10_map_layers(ctx, device_id: str, output_dir: str | None): + """List the Q10 map's separable layers (background/wall/floor/per-room). + + With --output-dir, also exports each layer as a transparent PNG that can be + stacked in a frontend (background, then floor, then walls, then each room). + """ + import os + + context: RoborockContext = ctx.obj + device_manager = await context.get_device_manager() + device = await device_manager.get_device(device_id) + if device.b01_q10_properties is None: + click.echo("Feature not supported by device") + return + properties = device.b01_q10_properties + await _await_q10_map_push(properties, lambda: properties.map.layers is not None) + layers = properties.map.layers + if layers is None: + click.echo("No map layers available.") + return + + summary = { + "size": {"width": layers.width, "height": layers.height}, + "class_counts": layers.class_counts, + "rooms": [ + {"id": r.id, "name": r.name, "pixel_count": r.pixel_count, "bbox": list(r.bbox)} for r in layers.rooms + ], + } + click.echo(dump_json(summary)) + + if output_dir: + os.makedirs(output_dir, exist_ok=True) + exports = { + "background": layers.render_class("background", (210, 210, 215, 255), scale=2), + "floor": layers.render_class("floor", (70, 170, 95, 200), scale=2), + "wall": layers.render_class("wall", (20, 20, 25, 255), scale=2), + } + for name, png in exports.items(): + with open(os.path.join(output_dir, f"layer_{name}.png"), "wb") as f: + f.write(png) + for room in layers.rooms: + png = layers.render_room(room.id, (90, 140, 220, 200), scale=2) + safe = "".join(c if c.isalnum() else "_" for c in room.name) or f"room{room.id}" + with open(os.path.join(output_dir, f"room_{room.id}_{safe}.png"), "wb") as f: + f.write(png) + click.echo(f"Wrote {3 + len(layers.rooms)} layer PNGs to {output_dir}") + + @session.command() @click.option("--device_id", required=True) @click.option("--include_path", is_flag=True, default=False, help="Include path data in the output.") @@ -721,6 +774,42 @@ async def q10_position(ctx, device_id: str, include_path: bool): click.echo(dump_json(summary)) +@session.command() +@click.option("--device_id", required=True) +@click.option("--output-file", required=True, help="Path to save the map image with the path drawn.") +@click.pass_context +@async_command +async def q10_map_with_path(ctx, device_id: str, output_file: str): + """Render the Q10 map with the current cleaning path + robot position drawn. + + Needs the robot to be actively cleaning (the path/calibration come from the + live trace). Fetches the map and the path, solves the world<->pixel + calibration, and writes the annotated PNG. + """ + context: RoborockContext = ctx.obj + device_manager = await context.get_device_manager() + device = await device_manager.get_device(device_id) + if device.b01_q10_properties is None: + click.echo("Feature not supported by device") + return + properties = device.b01_q10_properties + map_trait = properties.map + await _await_q10_map_push(properties, lambda: map_trait.image_content is not None) + got_path = await _await_q10_map_push(properties, lambda: bool(map_trait.path)) + if not got_path: + click.echo("No live path available (the robot only reports its path while cleaning).") + return + try: + image = map_trait.render_path_on_map() + except RoborockException as err: + click.echo(f"Could not render path on map: {err}") + return + with open(output_file, "wb") as f: + f.write(image) + cal = map_trait.calibration + click.echo(f"Saved map with {len(map_trait.path)}-point path to {output_file} (calibration: {cal})") + + @session.command() @click.option("--device_id", required=True) @click.pass_context diff --git a/roborock/devices/traits/b01/q10/__init__.py b/roborock/devices/traits/b01/q10/__init__.py index b85007f0..b5d14525 100644 --- a/roborock/devices/traits/b01/q10/__init__.py +++ b/roborock/devices/traits/b01/q10/__init__.py @@ -16,7 +16,7 @@ from .consumable import ConsumableTrait from .do_not_disturb import DoNotDisturbTrait from .dust_collection import DustCollectionTrait -from .map import MapContentTrait +from .map import MapContentTrait, MapDpsTrait from .network_info import NetworkInfoTrait from .remote import RemoteTrait from .status import StatusTrait @@ -32,6 +32,7 @@ "DoNotDisturbTrait", "DustCollectionTrait", "MapContentTrait", + "MapDpsTrait", "NetworkInfoTrait", "SoundVolumeTrait", "StatusTrait", @@ -79,6 +80,9 @@ class Q10PropertiesApi(Trait): map: MapContentTrait """Trait for fetching the current parsed map (image + rooms).""" + map_dps: MapDpsTrait + """Low-level DPS values used to compose map overlays.""" + clean_history: CleanHistoryTrait """Trait for fetching the device clean-record history (``dpCleanRecord``).""" @@ -96,7 +100,8 @@ def __init__(self, channel: B01Q10Channel) -> None: self.button_light = ButtonLightTrait(self.command) self.network_info = NetworkInfoTrait() self.consumable = ConsumableTrait() - self.map = MapContentTrait() + self.map_dps = MapDpsTrait() + self.map = MapContentTrait(self.map_dps) self.clean_history = CleanHistoryTrait(self.command) # Read-model traits updated from the device's DPS push stream. self._updatable_traits = [ @@ -108,6 +113,7 @@ def __init__(self, channel: B01Q10Channel) -> None: self.network_info, self.consumable, self.clean_history, + self.map_dps, ] self._subscribe_task: asyncio.Task[None] | None = None diff --git a/roborock/devices/traits/b01/q10/map.py b/roborock/devices/traits/b01/q10/map.py index c132def4..7e5f0100 100644 --- a/roborock/devices/traits/b01/q10/map.py +++ b/roborock/devices/traits/b01/q10/map.py @@ -1,110 +1,190 @@ -"""Map content trait for B01 Q10 devices. - -Unlike the v1 / Q7 maps, the Q10 has no synchronous "get map" command, so this -trait is purely push-driven and mirrors the Q10 ``StatusTrait`` contract: - -- The device pushes its current map/path as protocol-301 ``MAP_RESPONSE`` - messages (a ``dpRequestDps`` nudges it to do so). The protocol layer decodes - those into :class:`Q10MapPacket` / :class:`Q10TracePacket` objects and the - ``Q10PropertiesApi`` subscribe loop routes them to - :meth:`MapContentTrait.update_from_map_packet` / - :meth:`MapContentTrait.update_from_trace_packet`. -- Those methods render/cache the content and notify update listeners (register - via :meth:`add_update_listener`). -- ``image_content``, ``map_data``, ``rooms``, ``path`` and ``robot_position`` - are readable and reflect the most recently pushed map. - -Unlike the Q7, the Q10 map payload is unencrypted, so no map key is required. +"""Push-driven map traits for B01 Q10 devices. + +Map-related state arrives on two independent streams: + +* map and trace packets are decoded by the protocol layer; +* restricted zones and virtual walls arrive as ordinary DPS values. + +``MapDpsTrait`` owns the low-level DPS read model. ``MapContentTrait`` depends +on it and combines that state with the latest map/trace packets through the pure +functions in :mod:`roborock.map.b01_q10_render`. The high-level trait keeps only +one grouped source snapshot and one replace-whole rendered result; calibration, +path placement and overlay placement are not independently mutable trait state. """ import logging -from dataclasses import dataclass, field - -from vacuum_map_parser_base.map_data import MapData +from dataclasses import dataclass, field, replace from roborock.data import RoborockBase -from roborock.devices.traits.common import TraitUpdateListener +from roborock.data.b01_q10.b01_q10_code_mappings import B01_Q10_DP +from roborock.devices.traits.common import DpsDataConverter, TraitUpdateListener +from roborock.exceptions import RoborockException +from roborock.map.b01_grid_layers import GridCalibration from roborock.map.b01_q10_map_parser import ( - B01Q10MapParser, B01Q10MapParserConfig, Q10MapPacket, Q10Point, Q10Room, Q10TracePacket, ) +from roborock.map.b01_q10_overlays import Q10Zone, parse_virtual_wall_blob, parse_zone_blob +from roborock.map.b01_q10_render import Q10MapOverlays, Q10MapRender, draw_path_on_map, render_q10_map -_LOGGER = logging.getLogger(__name__) +from .common import UpdatableTrait -_TRUNCATE_LENGTH = 20 +_LOGGER = logging.getLogger(__name__) @dataclass -class MapContent(RoborockBase): - """Dataclass representing Q10 map content.""" +class MapDps(RoborockBase): + """Low-level map values delivered in the Q10 DPS stream.""" - image_content: bytes | None = None - """The rendered image of the map in PNG format.""" + restricted_zone_up: str | None = field(default=None, metadata={"dps": B01_Q10_DP.RESTRICTED_ZONE_UP}) + virtual_wall_up: str | None = field(default=None, metadata={"dps": B01_Q10_DP.VIRTUAL_WALL_UP}) - map_data: MapData | None = None - """Parsed map data (image metadata + room names).""" - rooms: list[Q10Room] = field(default_factory=list) - """Rooms (segments) reported by the device, with ids and names.""" +class MapDpsTrait(MapDps, UpdatableTrait): + """Converter-backed read model for map-related DPS values.""" - path: list[Q10Point] = field(default_factory=list) - """Full path of the current cleaning session (oldest point first). + _CONVERTER = DpsDataConverter.from_dataclass(MapDps) - The robot accumulates this server-side and serves the whole trajectory so - far in one packet, so it is complete even if we connect mid-session. Only - populated while a cleaning session is active.""" + def __init__(self) -> None: + MapDps.__init__(self) + UpdatableTrait.__init__(self, command=None, logger=_LOGGER) - robot_position: Q10Point | None = None - """Current robot position (the most recent path point), if known.""" - def __repr__(self) -> str: - img = self.image_content - if img and len(img) > _TRUNCATE_LENGTH: - img = img[: _TRUNCATE_LENGTH - 3] + b"..." - return f"MapContent(image_content={img!r}, rooms={self.rooms!r})" +@dataclass(frozen=True) +class Q10MapSource: + """The latest map-protocol inputs, replaced atomically on every push.""" + map_packet: Q10MapPacket | None = None + trace_packet: Q10TracePacket | None = None -class MapContentTrait(MapContent, TraitUpdateListener): - """Trait holding the most recently pushed parsed map content for Q10 devices. - The Q10 has no synchronous get-map request; the device pushes map and trace - packets, which the protocol layer decodes and the ``Q10PropertiesApi`` - subscribe loop feeds into :meth:`update_from_map_packet` / - :meth:`update_from_trace_packet`. Consumers read the cached fields and/or - register a callback with :meth:`add_update_listener` to be notified when new - map content arrives. +class MapContentTrait(TraitUpdateListener): + """High-level composed Q10 map view. + + Map and trace packet updates replace :attr:`_source`; DPS updates are owned + by the injected :class:`MapDpsTrait`. Rendering always produces one new + :class:`Q10MapRender`, keeping the externally visible fields consistent. """ def __init__( self, + map_dps: MapDpsTrait | None = None, *, map_parser_config: B01Q10MapParserConfig | None = None, ) -> None: - super().__init__() TraitUpdateListener.__init__(self, logger=_LOGGER) - self._map_parser = B01Q10MapParser(map_parser_config) + self._config = map_parser_config or B01Q10MapParserConfig() + self._map_dps = map_dps or MapDpsTrait() + self._source = Q10MapSource() + self._render: Q10MapRender | None = None + self._map_dps.add_update_listener(self._map_dps_updated) + + @property + def image_content(self) -> bytes | None: + """The rendered base map PNG, if a map has been pushed.""" + return self._render.image_content if self._render else None + + @property + def rooms(self) -> list[Q10Room]: + """Rooms reported by the device.""" + return self._render.rooms if self._render else [] + + @property + def calibration(self) -> GridCalibration | None: + """The calibration used by the current rendered result.""" + return self._render.calibration if self._render else None + + @property + def path(self) -> list[Q10Point]: + """Full path from the latest trace packet.""" + trace = self._source.trace_packet + return trace.points if trace else [] + + @property + def robot_position(self) -> Q10Point | None: + """Current robot position from the latest trace packet.""" + trace = self._source.trace_packet + return trace.robot_position if trace else None + + @property + def robot_heading(self) -> int | None: + """Current robot heading from the latest trace packet.""" + trace = self._source.trace_packet + return trace.heading if trace else None + + @property + def zones(self) -> list[Q10Zone]: + """Restricted zones decoded from the low-level DPS trait.""" + return parse_zone_blob(self._map_dps.restricted_zone_up) + + @property + def virtual_walls(self) -> list[Q10Zone]: + """Virtual walls decoded from the low-level DPS trait.""" + return parse_virtual_wall_blob(self._map_dps.virtual_wall_up) def update_from_map_packet(self, packet: Q10MapPacket) -> None: - """Render a pushed full-map packet into the cached image/rooms. - - Rendering failures are logged and skipped (listeners are not notified) so - a single bad push cannot tear down the subscribe loop. - """ - parsed = self._map_parser.parse_packet(packet) - if parsed.image_content is None: - _LOGGER.debug("Failed to render Q10 map image") + """Replace the current map packet and compose a consistent result.""" + source = replace(self._source, map_packet=packet) + render = self._compose(source) + if render is None: return - self.image_content = parsed.image_content - self.map_data = parsed.map_data - self.rooms = packet.rooms + self._source = source + self._render = render self._notify_update() def update_from_trace_packet(self, packet: Q10TracePacket) -> None: - """Cache the path/robot position from a pushed trace packet.""" - self.path = packet.points - self.robot_position = packet.robot_position + """Replace the complete current-session trace packet.""" + self._source = replace(self._source, trace_packet=packet) + if self._source.map_packet is not None: + self._rebuild() + self._notify_update() + + def render_path_on_map( + self, + *, + line_color: tuple[int, int, int, int] = (235, 64, 52, 255), + position_color: tuple[int, int, int, int] = (255, 211, 0, 255), + ) -> bytes: + """Return a PNG with the path, robot, zones and walls drawn.""" + if self._render is None: + raise RoborockException("No map available; no map has been pushed yet") + if self._render.calibration is None: + raise RoborockException( + "No calibration available; a cleaning path must be captured (pushed) during a clean" + ) + return draw_path_on_map( + self._render, + config=self._config, + line_color=line_color, + position_color=position_color, + ) + + def _map_dps_updated(self) -> None: + """Recompose placed overlays after the low-level DPS state changes.""" + if self._source.map_packet is not None: + self._rebuild() self._notify_update() + + def _compose(self, source: Q10MapSource) -> Q10MapRender | None: + """Compose a source snapshot, preserving the previous result on error.""" + if source.map_packet is None: + return None + try: + return render_q10_map( + source.map_packet, + source.trace_packet, + Q10MapOverlays(zones=tuple(self.zones), virtual_walls=tuple(self.virtual_walls)), + config=self._config, + ) + except RoborockException as ex: + _LOGGER.debug("Failed to render Q10 map packet: %s", ex) + return None + + def _rebuild(self) -> None: + """Replace the derived render from the current source snapshot.""" + render = self._compose(self._source) + if render is not None: + self._render = render diff --git a/tests/devices/traits/b01/q10/test_map.py b/tests/devices/traits/b01/q10/test_map.py index 65216496..e32a4f51 100644 --- a/tests/devices/traits/b01/q10/test_map.py +++ b/tests/devices/traits/b01/q10/test_map.py @@ -1,12 +1,17 @@ """Tests for the Q10 B01 map content trait. The Q10 map API is push-driven: the device publishes ``MAP_RESPONSE`` messages -and the trait updates its cached state from them via ``update_from_map_response`` -(there is no synchronous get-map request). +which the protocol layer decodes into typed map/trace packets; the trait updates +its cached state from them via ``update_from_map_packet`` / +``update_from_trace_packet`` (there is no synchronous get-map request). These +tests cover that state management; the pixel/geometry work it drives is tested in +``tests/map/test_b01_q10_render.py``. """ import asyncio +import base64 from collections.abc import AsyncGenerator +from dataclasses import replace from pathlib import Path from typing import cast from unittest.mock import Mock @@ -14,19 +19,52 @@ import pytest from roborock.cli import _await_q10_map_push, cli +from roborock.data.b01_q10.b01_q10_code_mappings import B01_Q10_DP from roborock.devices.traits.b01.q10 import Q10PropertiesApi, create -from roborock.devices.traits.b01.q10.map import MapContentTrait -from roborock.map.b01_q10_map_parser import Q10Point, parse_map_packet, parse_trace_packet +from roborock.devices.traits.b01.q10.map import MapContentTrait, MapDpsTrait +from roborock.exceptions import RoborockException +from roborock.map.b01_grid_layers import GridCalibration +from roborock.map.b01_q10_map_parser import ( + Q10HeaderCalibration, + Q10MapPacket, + Q10Point, + Q10TracePacket, + parse_map_packet, + parse_trace_packet, +) from roborock.protocols.b01_q10_protocol import Q10Message from .conftest import FakeB01Q10Channel FIXTURE = Path("tests/map/testdata/b01_q10_map.bin") -TRACE_FIXTURE = Path("tests/map/testdata/b01_q10_trace_multi.bin") +TRACE_SESSION_FIXTURE = Path("tests/map/testdata/b01_q10_trace_session.bin") + +# A header calibration whose pixel origin (0, 5) is usable (not a keepalive +# frame), so a short path can calibrate the fixture map. +_USABLE_HEADER = Q10HeaderCalibration(origin_x=0, origin_y=50, resolution=5, charger_x=0, charger_y=0, charger_phi=0) + + +def _trait_with_map() -> MapContentTrait: + """A trait with the fixture map already pushed into it.""" + trait = MapContentTrait() + trait.update_from_map_packet(parse_map_packet(FIXTURE.read_bytes())) + return trait + + +def _floor_world_points(packet: Q10MapPacket, cal: GridCalibration, count: int) -> list[Q10Point]: + """``count`` world points lying on the map's floor under ``cal``.""" + layers = packet.layers + 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_update_from_map_packet_populates_image_and_rooms() -> None: - """A parsed 01 01 map packet populates the image, rooms and map data.""" + """A pushed 01 01 map packet populates the image, rooms and map data.""" packet = parse_map_packet(FIXTURE.read_bytes()) trait = MapContentTrait() updates: list[None] = [] @@ -37,22 +75,23 @@ def test_update_from_map_packet_populates_image_and_rooms() -> None: assert trait.image_content is not None assert trait.image_content[:8] == b"\x89PNG\r\n\x1a\n" assert {room.id: room.name for room in trait.rooms} == {2: "Living Room", 3: "Bedroom"} - assert trait.map_data is not None assert len(updates) == 1 def test_update_from_trace_packet_populates_path_and_position() -> None: - """A parsed 02 01 trace packet populates the path and robot position.""" - packet = parse_trace_packet(TRACE_FIXTURE.read_bytes()) + """A pushed 02 01 trace packet populates the path, position and heading.""" + trace = parse_trace_packet(TRACE_SESSION_FIXTURE.read_bytes()) trait = MapContentTrait() updates: list[None] = [] trait.add_update_listener(lambda: updates.append(None)) - trait.update_from_trace_packet(packet) + trait.update_from_trace_packet(trace) - assert [(p.x, p.y) for p in trait.path] == [(100, 200), (150, 250), (-50, 300)] + assert len(trait.path) == 14 + assert (trait.path[0].x, trait.path[0].y) == (41, 64) assert trait.robot_position is not None - assert (trait.robot_position.x, trait.robot_position.y) == (-50, 300) + assert (trait.robot_position.x, trait.robot_position.y) == (276, -1) + assert trait.robot_heading == -34 assert len(updates) == 1 @@ -75,13 +114,13 @@ async def refresh(self) -> None: class _FakeQ10PropertiesWithTrace(_FakeQ10Properties): async def refresh(self) -> None: await super().refresh() - self.map.update_from_trace_packet(parse_trace_packet(TRACE_FIXTURE.read_bytes())) + self.map.update_from_trace_packet(parse_trace_packet(TRACE_SESSION_FIXTURE.read_bytes())) async def test_await_q10_map_push_waits_for_fresh_update() -> None: """A cached trace alone is not treated as a successful new map push.""" properties = _FakeQ10Properties() - properties.map.path = [Q10Point(1, 2)] + properties.map.update_from_trace_packet(Q10TracePacket(points=[Q10Point(1, 2)])) got_trace = await _await_q10_map_push( cast(Q10PropertiesApi, properties), lambda: bool(properties.map.path), timeout=0.01 @@ -99,12 +138,12 @@ async def test_await_q10_map_push_returns_true_after_update() -> None: ) assert got_trace is True - assert [(p.x, p.y) for p in properties.map.path] == [(100, 200), (150, 250), (-50, 300)] + assert len(properties.map.path) == 14 async def test_await_q10_map_push_can_fall_back_to_cached_map_on_timeout() -> None: properties = _FakeQ10Properties() - properties.map.image_content = b"cached-png" + properties.map.update_from_map_packet(parse_map_packet(FIXTURE.read_bytes())) got_map = await _await_q10_map_push( cast(Q10PropertiesApi, properties), @@ -171,7 +210,145 @@ async def test_subscribe_loop_routes_trace_push( """A trace pushed onto the stream is routed to the map trait by the loop.""" assert not q10_api.map.path - message_queue.put_nowait(parse_trace_packet(TRACE_FIXTURE.read_bytes())) + message_queue.put_nowait(parse_trace_packet(TRACE_SESSION_FIXTURE.read_bytes())) await _wait_for(lambda: bool(q10_api.map.path)) assert q10_api.map.robot_position is not None + + +# --- Calibration + rendering ------------------------------------------------- + + +def test_trace_without_map_has_no_calibration() -> None: + """No map -> no calibration, even with a path pushed.""" + trait = MapContentTrait() + trait.update_from_trace_packet(Q10TracePacket(points=[Q10Point(i, 0) for i in range(30)])) + assert trait.calibration is None + + +def test_trace_update_derives_calibration_from_map_and_short_path() -> None: + """A trace update derives calibration from the current map and trace.""" + trait = MapContentTrait() + packet = replace(parse_map_packet(FIXTURE.read_bytes()), header_calibration=_USABLE_HEADER) + trait.update_from_map_packet(packet) + true = GridCalibration(resolution=20.0, origin_x=0.0, origin_y=5.0, y_sign=1) + trait.update_from_trace_packet(Q10TracePacket(points=_floor_world_points(packet, true, 6))) + assert len(trait.path) < 20 # far too short for the full origin+resolution fit + + cal = trait.calibration + + assert cal is not None + assert (cal.origin_x, cal.origin_y) == (0.0, 5.0) # straight from the header + + +def test_short_trace_without_header_does_not_calibrate() -> None: + """Without a header origin a short path is still too sparse for the full fit.""" + packet = parse_map_packet(FIXTURE.read_bytes()) + trait = MapContentTrait() + trait.update_from_map_packet(packet) # the fixture header is a keepalive frame + true = GridCalibration(resolution=10.0, origin_x=0.0, origin_y=5.0, y_sign=1) + trait.update_from_trace_packet(Q10TracePacket(points=_floor_world_points(packet, true, 6))) + assert trait.calibration is None + + +def test_render_path_on_map_requires_map() -> None: + trait = MapContentTrait() + with pytest.raises(RoborockException, match="No map available"): + trait.render_path_on_map() + + +def test_render_path_on_map_uses_derived_content() -> None: + """The renderer draws the already-derived map content as a PNG.""" + trait = MapContentTrait() + packet = replace(parse_map_packet(FIXTURE.read_bytes()), header_calibration=_USABLE_HEADER) + trait.update_from_map_packet(packet) + true = GridCalibration(resolution=20.0, origin_x=0.0, origin_y=5.0, y_sign=1) + trait.update_from_trace_packet(Q10TracePacket(points=_floor_world_points(packet, true, 6))) + + png = trait.render_path_on_map() + + assert png[:8] == b"\x89PNG\r\n\x1a\n" + assert trait.calibration is not None + + +def test_render_path_on_map_without_path_cannot_calibrate() -> None: + """A map but no cleaning path -> no calibration -> a clear error.""" + trait = _trait_with_map() + with pytest.raises(RoborockException, match="No calibration available"): + trait.render_path_on_map() + + +# --- Overlays ---------------------------------------------------------------- + + +def test_load_overlays_places_zones_after_calibration() -> None: + """Decoded no-go / no-mop zones are placed on MapData once calibrated.""" + map_dps = MapDpsTrait() + trait = MapContentTrait(map_dps) + packet = replace(parse_map_packet(FIXTURE.read_bytes()), header_calibration=_USABLE_HEADER) + trait.update_from_map_packet(packet) + true = GridCalibration(resolution=20.0, origin_x=0.0, origin_y=5.0, y_sign=1) + trait.update_from_trace_packet(Q10TracePacket(points=_floor_world_points(packet, true, 6))) + assert trait.calibration is not None + before = trait.render_path_on_map() + + def rect(zone_type: int, corners: list[tuple[int, int]]) -> bytes: + out = bytes([zone_type, len(corners)]) + for x, y in corners: + out += int.to_bytes(x & 0xFFFF, 2, "big") + int.to_bytes(y & 0xFFFF, 2, "big") + return out.ljust(18, b"\x00") + + blob = bytes([1, 1]) + rect(0, [(0, 0), (40, 0), (40, 40), (0, 40)]) + map_dps.update_from_dps({B01_Q10_DP.RESTRICTED_ZONE_UP: base64.b64encode(blob).decode()}) + + assert len(trait.zones) == 1 + assert trait.render_path_on_map() != before + + +def test_load_overlays_partial_update_keeps_existing_zones() -> None: + """A status push without the zone DP (None) must not wipe loaded zones.""" + map_dps = MapDpsTrait() + trait = MapContentTrait(map_dps) + blob = ( + bytes([1, 1]) + + bytes([0, 4]) + + b"".join(int.to_bytes(v & 0xFFFF, 2, "big") for xy in [(0, 0), (4, 0), (4, 4), (0, 4)] for v in xy) + ) + map_dps.update_from_dps({B01_Q10_DP.RESTRICTED_ZONE_UP: base64.b64encode(blob).decode()}) + assert len(trait.zones) == 1 + # A later partial update carrying only the (empty) virtual-wall DP. + map_dps.update_from_dps({B01_Q10_DP.VIRTUAL_WALL_UP: "AA=="}) + assert len(trait.zones) == 1 # zones preserved + assert trait.virtual_walls == [] + + +def test_map_dps_trait_updates_high_level_map_content() -> None: + """The low-level DPS trait notifies the dependent high-level map trait.""" + map_dps = MapDpsTrait() + trait = MapContentTrait(map_dps) + blob = ( + bytes([1, 1]) + + bytes([0, 4]) + + b"".join(int.to_bytes(v & 0xFFFF, 2, "big") for xy in [(0, 0), (4, 0), (4, 4), (0, 4)] for v in xy) + ) + notified = [] + trait.add_update_listener(lambda: notified.append(True)) + + map_dps.update_from_dps({B01_Q10_DP.RESTRICTED_ZONE_UP: base64.b64encode(blob).decode()}) + + assert len(trait.zones) == 1 + assert notified # listeners learn the overlays changed + + +def test_map_dps_push_without_overlay_data_points_is_noop() -> None: + """A DPS push carrying neither overlay DP leaves both traits untouched.""" + map_dps = MapDpsTrait() + trait = MapContentTrait(map_dps) + notified = [] + trait.add_update_listener(lambda: notified.append(True)) + + map_dps.update_from_dps({B01_Q10_DP.BATTERY: 50}) + + assert trait.zones == [] + assert trait.virtual_walls == [] + assert not notified From 7f41f49643227dba8d85593f4ea9f3d2f8ace947 Mon Sep 17 00:00:00 2001 From: Vincent <2070309+tubededentifrice@users.noreply.github.com> Date: Sun, 19 Jul 2026 22:06:44 +0400 Subject: [PATCH 3/7] refactor: keep Q10 render intermediates internal --- roborock/map/b01_q10_render.py | 70 ++++++++++---------------------- tests/map/test_b01_q10_render.py | 49 +++++++++++++--------- 2 files changed, 51 insertions(+), 68 deletions(-) diff --git a/roborock/map/b01_q10_render.py b/roborock/map/b01_q10_render.py index 48d1ffa9..b5f58ea4 100644 --- a/roborock/map/b01_q10_render.py +++ b/roborock/map/b01_q10_render.py @@ -34,7 +34,6 @@ B01Q10MapParserConfig, Q10EraseZone, Q10MapPacket, - Q10Room, Q10TracePacket, erased_packet, ) @@ -69,13 +68,12 @@ class Q10MapOverlays: @dataclass class Q10MapRender: - """The fully composed result of rendering a Q10 map packet. + """The image and projected geometry produced by Q10 map composition. Built by :func:`render_q10_map` from one map packet, trace packet and DPS - overlay snapshot, so every derived field is consistent with one set of - source inputs. Analogous to - :class:`~roborock.map.map_parser.ParsedMapData`, but also carrying the - separable :attr:`layers` and derived :attr:`calibration`. + overlay snapshot. Source data and intermediate calibration/layer state stay + on their owning packet or inside the renderer rather than being repeated on + this result. """ image_content: bytes @@ -85,17 +83,6 @@ class Q10MapRender: """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, @@ -113,17 +100,15 @@ def render_q10_map( :class:`RoborockException` if map rendering fails. """ parser = B01Q10MapParser(config) - layers = packet.layers calibration = solve_q10_calibration(packet, trace) render_packet = packet if calibration is not None: - cells = _erased_cells(layers, packet.erase_zones, calibration) + cells = _erased_cells(packet.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). + # Blank the erase-zone cells before parsing the raster so 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: @@ -134,13 +119,7 @@ def render_q10_map( _place_trace(map_data, calibration, trace) _place_overlays(map_data, calibration, overlays) - return Q10MapRender( - image_content=parsed.image_content, - map_data=map_data, - layers=layers, - rooms=packet.rooms, - calibration=calibration, - ) + return Q10MapRender(image_content=parsed.image_content, map_data=map_data) def solve_q10_calibration( @@ -219,7 +198,9 @@ def _place_trace( robot_position = trace.robot_position 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, trace.heading) + # Store the heading in projected image coordinates so drawing does not + # need to retain the world-to-pixel calibration. + map_data.vacuum_position = Point(px, py, -calibration.y_sign * trace.heading) if pixels: map_data.charger = pixels[0] @@ -261,11 +242,10 @@ def draw_path_on_map( ) -> bytes: """Draw the projected ``MapData`` content onto the base map PNG. - ``render`` must carry its derived calibration. Returns a fresh PNG; the base - raster in :attr:`Q10MapRender.image_content` is left untouched. + Returns a fresh PNG; the base raster in + :attr:`Q10MapRender.image_content` is left untouched. """ - calibration = render.calibration - if calibration is None: + if render.map_data.path is None: raise RoborockException("No calibration available; a cleaning path must be captured during a clean") scale = config.map_scale @@ -314,21 +294,15 @@ def to_image(point: Point) -> tuple[float, float]: draw.ellipse([cx - radius, cy - radius, cx + radius, cy + radius], fill=position_color) robot_heading = robot_position.a 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. + # Heading was projected into image coordinates alongside the robot + # position, so no calibration state is needed during drawing. angle = math.radians(robot_heading) - dx = math.cos(angle) / calibration.resolution - dy = -calibration.y_sign * math.sin(angle) / calibration.resolution - norm = math.hypot(dx, dy) - if norm > 0: - tick = 4 * radius - draw.line( - [cx, cy, cx + dx / norm * tick, cy + dy / norm * tick], - fill=position_color, - width=max(1, scale // 2), - ) + tick = 4 * radius + draw.line( + [cx, cy, cx + math.cos(angle) * tick, cy + math.sin(angle) * 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 index 97a17049..fed967a6 100644 --- a/tests/map/test_b01_q10_render.py +++ b/tests/map/test_b01_q10_render.py @@ -9,8 +9,10 @@ from dataclasses import replace from pathlib import Path +import pytest from PIL import Image +from roborock.exceptions import RoborockException from roborock.map.b01_grid_layers import GridCalibration from roborock.map.b01_q10_map_parser import ( B01Q10MapParserConfig, @@ -31,6 +33,7 @@ _Q10_RESOLUTIONS, Q10MapOverlays, Q10MapRender, + _erased_cells, draw_path_on_map, render_q10_map, solve_q10_calibration, @@ -92,13 +95,10 @@ def _calibrated_inputs(*, heading: int = 0) -> tuple[Q10MapPacket, Q10TracePacke def test_render_base_map_without_calibration() -> None: - """Without a calibration only the base raster/layers/rooms are produced.""" + """Without a calibration only the base raster is 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 @@ -107,13 +107,14 @@ def test_render_places_path_and_position() -> None: """The map and trace derive calibration, path and position together.""" packet, trace = _calibrated_inputs(heading=45) render = _render(packet, trace=trace) - assert render.calibration is not None + calibration = solve_q10_calibration(packet, trace) + assert calibration is not None assert render.map_data.path is not None assert render.map_data.vacuum_position is not None assert trace.robot_position is not None - expected = render.calibration.world_to_pixel(trace.robot_position.x, trace.robot_position.y) + expected = calibration.world_to_pixel(trace.robot_position.x, trace.robot_position.y) assert (render.map_data.vacuum_position.x, render.map_data.vacuum_position.y) == expected - assert render.map_data.vacuum_position.a == 45 + assert render.map_data.vacuum_position.a == -45 def test_render_places_zones_and_charger() -> None: @@ -128,26 +129,27 @@ def test_render_places_zones_and_charger() -> None: assert len(render.map_data.no_go_areas or []) == 1 assert len(render.map_data.no_mopping_areas or []) == 1 assert len(render.map_data.walls or []) == 1 - assert render.calibration is not None + calibration = solve_q10_calibration(packet, trace) + assert calibration is not None assert render.map_data.charger is not None - expected = render.calibration.world_to_pixel(trace.points[0].x, trace.points[0].y) + expected = calibration.world_to_pixel(trace.points[0].x, trace.points[0].y) assert (render.map_data.charger.x, render.map_data.charger.y) == expected def test_render_applies_erase_zones() -> None: - """With a calibration, erase-zone cells are blanked from layers + image.""" + """With a calibration, erase-zone cells are blanked from the image.""" packet, trace = _calibrated_inputs() base = _render(packet, trace=trace) - before_floor = base.layers.class_counts.get("floor") - assert before_floor and before_floor > 0 - assert base.calibration is not None + calibration = solve_q10_calibration(packet, trace) + assert calibration is not None # A rectangle covering the whole grid in world coords erases every cell. corners = [(-1, -1), (8, -1), (8, 6), (-1, 6)] - erase_zone = Q10EraseZone(vertices=_world_vertices(base.calibration, corners)) + erase_zone = Q10EraseZone(vertices=_world_vertices(calibration, corners)) + cells = _erased_cells(packet.layers, [erase_zone], calibration) render = _render(replace(packet, erase_zones=[erase_zone]), trace=trace) - assert render.layers.class_counts.get("floor", 0) == 0 # all floor erased + assert len(cells) == packet.layers.width * packet.layers.height assert render.image_content != base.image_content # re-rendered @@ -155,16 +157,23 @@ def test_render_partial_erase() -> None: """An erase rectangle only blanks the cells it covers, leaving the rest.""" packet, trace = _calibrated_inputs() base = _render(packet, trace=trace) - before_floor = base.layers.class_counts.get("floor", 0) - assert base.calibration is not None + calibration = solve_q10_calibration(packet, trace) + assert calibration is not None # Cover only the top two grid rows. corners = [(-1, -1), (8, -1), (8, 2), (-1, 2)] - erase_zone = Q10EraseZone(vertices=_world_vertices(base.calibration, corners)) + erase_zone = Q10EraseZone(vertices=_world_vertices(calibration, corners)) + cells = _erased_cells(packet.layers, [erase_zone], calibration) render = _render(replace(packet, erase_zones=[erase_zone]), trace=trace) - after_floor = render.layers.class_counts.get("floor", 0) - assert 0 < after_floor < before_floor # some, not all, floor removed + assert 0 < len(cells) < packet.layers.width * packet.layers.height + assert render.image_content != base.image_content + + +def test_draw_path_on_map_requires_projected_path() -> None: + """Drawing fails clearly when the source streams could not calibrate.""" + with pytest.raises(RoborockException, match="No calibration available"): + draw_path_on_map(_render(), config=CONFIG) def test_draw_path_on_map_draws_position() -> None: From cdb17e285b3a01bb47a7aa7afcc542df258dca75 Mon Sep 17 00:00:00 2001 From: Vincent <2070309+tubededentifrice@users.noreply.github.com> Date: Sun, 19 Jul 2026 22:09:27 +0400 Subject: [PATCH 4/7] refactor: expose only composed Q10 map data --- roborock/cli.py | 61 ++---------------------- roborock/devices/traits/b01/q10/map.py | 18 ++----- tests/devices/traits/b01/q10/test_map.py | 27 +++++------ 3 files changed, 20 insertions(+), 86 deletions(-) diff --git a/roborock/cli.py b/roborock/cli.py index 12aafc63..7633510b 100644 --- a/roborock/cli.py +++ b/roborock/cli.py @@ -662,59 +662,6 @@ async def map_image(ctx, device_id: str, output_file: str): click.echo("No map image content available.") -@session.command() -@click.option("--device_id", required=True) -@click.option("--output-dir", default=None, help="If set, write one transparent PNG per layer here.") -@click.pass_context -@async_command -async def q10_map_layers(ctx, device_id: str, output_dir: str | None): - """List the Q10 map's separable layers (background/wall/floor/per-room). - - With --output-dir, also exports each layer as a transparent PNG that can be - stacked in a frontend (background, then floor, then walls, then each room). - """ - import os - - context: RoborockContext = ctx.obj - device_manager = await context.get_device_manager() - device = await device_manager.get_device(device_id) - if device.b01_q10_properties is None: - click.echo("Feature not supported by device") - return - properties = device.b01_q10_properties - await _await_q10_map_push(properties, lambda: properties.map.layers is not None) - layers = properties.map.layers - if layers is None: - click.echo("No map layers available.") - return - - summary = { - "size": {"width": layers.width, "height": layers.height}, - "class_counts": layers.class_counts, - "rooms": [ - {"id": r.id, "name": r.name, "pixel_count": r.pixel_count, "bbox": list(r.bbox)} for r in layers.rooms - ], - } - click.echo(dump_json(summary)) - - if output_dir: - os.makedirs(output_dir, exist_ok=True) - exports = { - "background": layers.render_class("background", (210, 210, 215, 255), scale=2), - "floor": layers.render_class("floor", (70, 170, 95, 200), scale=2), - "wall": layers.render_class("wall", (20, 20, 25, 255), scale=2), - } - for name, png in exports.items(): - with open(os.path.join(output_dir, f"layer_{name}.png"), "wb") as f: - f.write(png) - for room in layers.rooms: - png = layers.render_room(room.id, (90, 140, 220, 200), scale=2) - safe = "".join(c if c.isalnum() else "_" for c in room.name) or f"room{room.id}" - with open(os.path.join(output_dir, f"room_{room.id}_{safe}.png"), "wb") as f: - f.write(png) - click.echo(f"Wrote {3 + len(layers.rooms)} layer PNGs to {output_dir}") - - @session.command() @click.option("--device_id", required=True) @click.option("--include_path", is_flag=True, default=False, help="Include path data in the output.") @@ -782,9 +729,8 @@ async def q10_position(ctx, device_id: str, include_path: bool): async def q10_map_with_path(ctx, device_id: str, output_file: str): """Render the Q10 map with the current cleaning path + robot position drawn. - Needs the robot to be actively cleaning (the path/calibration come from the - live trace). Fetches the map and the path, solves the world<->pixel - calibration, and writes the annotated PNG. + Needs the robot to be actively cleaning so a live trace is available. + Fetches the map and path and writes the annotated PNG. """ context: RoborockContext = ctx.obj device_manager = await context.get_device_manager() @@ -806,8 +752,7 @@ async def q10_map_with_path(ctx, device_id: str, output_file: str): return with open(output_file, "wb") as f: f.write(image) - cal = map_trait.calibration - click.echo(f"Saved map with {len(map_trait.path)}-point path to {output_file} (calibration: {cal})") + click.echo(f"Saved map with {len(map_trait.path)}-point path to {output_file}") @session.command() diff --git a/roborock/devices/traits/b01/q10/map.py b/roborock/devices/traits/b01/q10/map.py index 7e5f0100..e5190c16 100644 --- a/roborock/devices/traits/b01/q10/map.py +++ b/roborock/devices/traits/b01/q10/map.py @@ -1,8 +1,9 @@ """Push-driven map traits for B01 Q10 devices. -Map-related state arrives on two independent streams: +Map-related state arrives on three independent streams: -* map and trace packets are decoded by the protocol layer; +* map packets are decoded from map-protocol responses; +* trace packets are decoded from trace-protocol responses; * restricted zones and virtual walls arrive as ordinary DPS values. ``MapDpsTrait`` owns the low-level DPS read model. ``MapContentTrait`` depends @@ -19,7 +20,6 @@ from roborock.data.b01_q10.b01_q10_code_mappings import B01_Q10_DP from roborock.devices.traits.common import DpsDataConverter, TraitUpdateListener from roborock.exceptions import RoborockException -from roborock.map.b01_grid_layers import GridCalibration from roborock.map.b01_q10_map_parser import ( B01Q10MapParserConfig, Q10MapPacket, @@ -90,12 +90,8 @@ def image_content(self) -> bytes | None: @property def rooms(self) -> list[Q10Room]: """Rooms reported by the device.""" - return self._render.rooms if self._render else [] - - @property - def calibration(self) -> GridCalibration | None: - """The calibration used by the current rendered result.""" - return self._render.calibration if self._render else None + packet = self._source.map_packet + return packet.rooms if packet else [] @property def path(self) -> list[Q10Point]: @@ -151,10 +147,6 @@ def render_path_on_map( """Return a PNG with the path, robot, zones and walls drawn.""" if self._render is None: raise RoborockException("No map available; no map has been pushed yet") - if self._render.calibration is None: - raise RoborockException( - "No calibration available; a cleaning path must be captured (pushed) during a clean" - ) return draw_path_on_map( self._render, config=self._config, diff --git a/tests/devices/traits/b01/q10/test_map.py b/tests/devices/traits/b01/q10/test_map.py index e32a4f51..844ee7d3 100644 --- a/tests/devices/traits/b01/q10/test_map.py +++ b/tests/devices/traits/b01/q10/test_map.py @@ -216,18 +216,19 @@ async def test_subscribe_loop_routes_trace_push( assert q10_api.map.robot_position is not None -# --- Calibration + rendering ------------------------------------------------- +# --- Source composition + rendering ------------------------------------------ -def test_trace_without_map_has_no_calibration() -> None: - """No map -> no calibration, even with a path pushed.""" +def test_trace_without_map_is_retained_without_rendering() -> None: + """A trace is retained even when no map is available to render yet.""" trait = MapContentTrait() trait.update_from_trace_packet(Q10TracePacket(points=[Q10Point(i, 0) for i in range(30)])) - assert trait.calibration is None + assert len(trait.path) == 30 + assert trait.image_content is None -def test_trace_update_derives_calibration_from_map_and_short_path() -> None: - """A trace update derives calibration from the current map and trace.""" +def test_trace_update_projects_short_path_using_header() -> None: + """A map header and short trace are sufficient to render a path.""" trait = MapContentTrait() packet = replace(parse_map_packet(FIXTURE.read_bytes()), header_calibration=_USABLE_HEADER) trait.update_from_map_packet(packet) @@ -235,20 +236,18 @@ def test_trace_update_derives_calibration_from_map_and_short_path() -> None: trait.update_from_trace_packet(Q10TracePacket(points=_floor_world_points(packet, true, 6))) assert len(trait.path) < 20 # far too short for the full origin+resolution fit - cal = trait.calibration + assert trait.render_path_on_map()[:8] == b"\x89PNG\r\n\x1a\n" - assert cal is not None - assert (cal.origin_x, cal.origin_y) == (0.0, 5.0) # straight from the header - -def test_short_trace_without_header_does_not_calibrate() -> None: - """Without a header origin a short path is still too sparse for the full fit.""" +def test_short_trace_without_header_cannot_be_projected() -> None: + """Without a header origin a short trace cannot be placed on the map.""" packet = parse_map_packet(FIXTURE.read_bytes()) trait = MapContentTrait() trait.update_from_map_packet(packet) # the fixture header is a keepalive frame true = GridCalibration(resolution=10.0, origin_x=0.0, origin_y=5.0, y_sign=1) trait.update_from_trace_packet(Q10TracePacket(points=_floor_world_points(packet, true, 6))) - assert trait.calibration is None + with pytest.raises(RoborockException, match="No calibration available"): + trait.render_path_on_map() def test_render_path_on_map_requires_map() -> None: @@ -268,7 +267,6 @@ def test_render_path_on_map_uses_derived_content() -> None: png = trait.render_path_on_map() assert png[:8] == b"\x89PNG\r\n\x1a\n" - assert trait.calibration is not None def test_render_path_on_map_without_path_cannot_calibrate() -> None: @@ -289,7 +287,6 @@ def test_load_overlays_places_zones_after_calibration() -> None: trait.update_from_map_packet(packet) true = GridCalibration(resolution=20.0, origin_x=0.0, origin_y=5.0, y_sign=1) trait.update_from_trace_packet(Q10TracePacket(points=_floor_world_points(packet, true, 6))) - assert trait.calibration is not None before = trait.render_path_on_map() def rect(zone_type: int, corners: list[tuple[int, int]]) -> bytes: From af1531e88eaef58754e7c0cfd0a46f603678b56a Mon Sep 17 00:00:00 2001 From: Vincent <2070309+tubededentifrice@users.noreply.github.com> Date: Sun, 19 Jul 2026 22:12:30 +0400 Subject: [PATCH 5/7] refactor: render Q10 maps through one image path --- roborock/map/b01_q10_render.py | 59 ++++++------------- tests/map/test_b01_q10_render.py | 99 ++++++++++---------------------- 2 files changed, 48 insertions(+), 110 deletions(-) diff --git a/roborock/map/b01_q10_render.py b/roborock/map/b01_q10_render.py index b5f58ea4..8c241c98 100644 --- a/roborock/map/b01_q10_render.py +++ b/roborock/map/b01_q10_render.py @@ -8,7 +8,7 @@ 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 +returned image 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. """ @@ -66,36 +66,18 @@ class Q10MapOverlays: virtual_walls: Sequence[Q10Zone] = () -@dataclass -class Q10MapRender: - """The image and projected geometry produced by Q10 map composition. - - Built by :func:`render_q10_map` from one map packet, trace packet and DPS - overlay snapshot. Source data and intermediate calibration/layer state stay - on their owning packet or inside the renderer rather than being repeated on - this result. - """ - - 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.""" - - def render_q10_map( packet: Q10MapPacket, trace: Q10TracePacket | None, overlays: Q10MapOverlays, *, config: B01Q10MapParserConfig, -) -> Q10MapRender: - """Compose the latest map, trace and DPS inputs into a render. +) -> bytes: + """Compose the latest map, trace and DPS inputs into one PNG image. Calibration is derived from ``packet`` (layers + header calibration) and ``trace`` (path points). Once calibrated, erase zones are blanked out of the - raster and trace/overlay data is projected into ``map_data`` pixel space. + raster and trace/overlay data is projected and drawn in pixel space. Without a usable trace only the base raster is rendered. Raises :class:`RoborockException` if map rendering fails. """ @@ -118,8 +100,9 @@ def render_q10_map( if calibration is not None and trace is not None: _place_trace(map_data, calibration, trace) _place_overlays(map_data, calibration, overlays) + return _draw_map_content(parsed.image_content, map_data, config=config) - return Q10MapRender(image_content=parsed.image_content, map_data=map_data) + return parsed.image_content def solve_q10_calibration( @@ -233,23 +216,17 @@ def to_area(zone: Q10Zone) -> Area | None: map_data.walls = walls or None -def draw_path_on_map( - render: Q10MapRender, +def _draw_map_content( + image_content: bytes, + map_data: MapData, *, config: B01Q10MapParserConfig, 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 projected ``MapData`` content onto the base map PNG. - - Returns a fresh PNG; the base raster in - :attr:`Q10MapRender.image_content` is left untouched. - """ - if render.map_data.path is None: - raise RoborockException("No calibration available; a cleaning path must be captured during a clean") - + """Draw projected map content onto a base PNG and return a fresh PNG.""" scale = config.map_scale - base = Image.open(io.BytesIO(render.image_content)).convert("RGBA") + base = Image.open(io.BytesIO(image_content)).convert("RGBA") def to_image(point: Point) -> tuple[float, float]: return (point.x * scale, point.y * scale) @@ -261,8 +238,8 @@ def to_image(point: Point) -> tuple[float, float]: # No-go (blue) and no-mop (magenta) zones beneath the path. for areas, fill, outline in ( - (render.map_data.no_go_areas or [], (0, 120, 255, 70), (0, 80, 200, 255)), - (render.map_data.no_mopping_areas or [], (255, 0, 200, 70), (200, 0, 160, 255)), + (map_data.no_go_areas or [], (0, 120, 255, 70), (0, 80, 200, 255)), + (map_data.no_mopping_areas or [], (255, 0, 200, 70), (200, 0, 160, 255)), ): for area in areas: polygon = [ @@ -274,20 +251,20 @@ def to_image(point: Point) -> tuple[float, float]: draw.polygon(polygon, fill=fill, outline=outline) # Virtual walls (line segments, not polygons) drawn over the zones. - for wall in render.map_data.walls or []: + for wall in map_data.walls or []: draw.line( [(wall.x0 * scale, wall.y0 * scale), (wall.x1 * scale, wall.y1 * scale)], fill=(255, 64, 64, 255), width=max(2, scale), ) - for path in render.map_data.path.path if render.map_data.path else []: + for path in map_data.path.path if map_data.path else []: if len(path) >= 2: draw.line([to_image(point) for point in path], fill=line_color, width=max(1, scale // 2)) - if render.map_data.charger is not None: - dx, dy = to_image(render.map_data.charger) + if map_data.charger is not None: + dx, dy = to_image(map_data.charger) draw.ellipse([dx - scale, dy - scale, dx + scale, dy + scale], outline=(40, 200, 40, 255), width=2) - robot_position = render.map_data.vacuum_position + robot_position = map_data.vacuum_position if robot_position is not None: cx, cy = to_image(robot_position) radius = scale diff --git a/tests/map/test_b01_q10_render.py b/tests/map/test_b01_q10_render.py index fed967a6..ff18ed65 100644 --- a/tests/map/test_b01_q10_render.py +++ b/tests/map/test_b01_q10_render.py @@ -9,10 +9,8 @@ from dataclasses import replace from pathlib import Path -import pytest from PIL import Image -from roborock.exceptions import RoborockException from roborock.map.b01_grid_layers import GridCalibration from roborock.map.b01_q10_map_parser import ( B01Q10MapParserConfig, @@ -32,9 +30,7 @@ from roborock.map.b01_q10_render import ( _Q10_RESOLUTIONS, Q10MapOverlays, - Q10MapRender, _erased_cells, - draw_path_on_map, render_q10_map, solve_q10_calibration, ) @@ -58,7 +54,7 @@ def _render( *, trace: Q10TracePacket | None = None, overlays: Q10MapOverlays | None = None, -) -> Q10MapRender: +) -> bytes: return render_q10_map( packet if packet is not None else _packet(), trace, @@ -96,44 +92,35 @@ def _calibrated_inputs(*, heading: int = 0) -> tuple[Q10MapPacket, Q10TracePacke def test_render_base_map_without_calibration() -> None: """Without a calibration only the base raster is produced.""" - render = _render() - assert render.image_content[:8] == b"\x89PNG\r\n\x1a\n" - assert render.map_data is not None - # Overlays are world-coordinate only, so nothing is placed yet. - assert render.map_data.path is None + image = _render() + assert image[:8] == b"\x89PNG\r\n\x1a\n" -def test_render_places_path_and_position() -> None: - """The map and trace derive calibration, path and position together.""" - packet, trace = _calibrated_inputs(heading=45) - render = _render(packet, trace=trace) +def test_render_draws_path_and_position() -> None: + """The map and trace derive calibration and draw the robot position.""" + packet, trace = _calibrated_inputs() + image = _render(packet, trace=trace) calibration = solve_q10_calibration(packet, trace) assert calibration is not None - assert render.map_data.path is not None - assert render.map_data.vacuum_position is not None assert trace.robot_position is not None - expected = calibration.world_to_pixel(trace.robot_position.x, trace.robot_position.y) - assert (render.map_data.vacuum_position.x, render.map_data.vacuum_position.y) == expected - assert render.map_data.vacuum_position.a == -45 + px, py = calibration.world_to_pixel(trace.robot_position.x, trace.robot_position.y) + image_position = (round(px * CONFIG.map_scale), round(py * CONFIG.map_scale)) + rendered = Image.open(io.BytesIO(image)).convert("RGBA") + assert rendered.size == (8 * 4, 6 * 4) + assert rendered.getpixel(image_position) == (255, 211, 0, 255) -def test_render_places_zones_and_charger() -> None: - """Decoded no-go / no-mop zones become pixel-space MapData areas + charger.""" +def test_render_draws_zones_and_virtual_walls() -> None: + """Decoded DPS overlays are included in the composed image.""" packet, trace = _calibrated_inputs() 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)]), ] walls = [Q10Zone(type=ZONE_TYPE_VIRTUAL_WALL, vertices=[(0, 0), (4, 0)])] - render = _render(packet, trace=trace, overlays=Q10MapOverlays(zones=zones, virtual_walls=walls)) - assert len(render.map_data.no_go_areas or []) == 1 - assert len(render.map_data.no_mopping_areas or []) == 1 - assert len(render.map_data.walls or []) == 1 - calibration = solve_q10_calibration(packet, trace) - assert calibration is not None - assert render.map_data.charger is not None - expected = calibration.world_to_pixel(trace.points[0].x, trace.points[0].y) - assert (render.map_data.charger.x, render.map_data.charger.y) == expected + base = _render(packet, trace=trace) + rendered = _render(packet, trace=trace, overlays=Q10MapOverlays(zones=zones, virtual_walls=walls)) + assert rendered != base def test_render_applies_erase_zones() -> None: @@ -150,7 +137,7 @@ def test_render_applies_erase_zones() -> None: render = _render(replace(packet, erase_zones=[erase_zone]), trace=trace) assert len(cells) == packet.layers.width * packet.layers.height - assert render.image_content != base.image_content # re-rendered + assert render != base def test_render_partial_erase() -> None: @@ -167,33 +154,10 @@ def test_render_partial_erase() -> None: render = _render(replace(packet, erase_zones=[erase_zone]), trace=trace) assert 0 < len(cells) < packet.layers.width * packet.layers.height - assert render.image_content != base.image_content - - -def test_draw_path_on_map_requires_projected_path() -> None: - """Drawing fails clearly when the source streams could not calibrate.""" - with pytest.raises(RoborockException, match="No calibration available"): - draw_path_on_map(_render(), config=CONFIG) - - -def test_draw_path_on_map_draws_position() -> None: - """The robot position is drawn at the mapped pixel.""" - packet, trace = _calibrated_inputs() - render = _render(packet, trace=trace) - png = draw_path_on_map( - render, - config=CONFIG, - position_color=(255, 211, 0, 255), - ) - img = Image.open(io.BytesIO(png)).convert("RGBA") - position = render.map_data.vacuum_position - assert position is not None - image_position = (round(position.x * CONFIG.map_scale), round(position.y * CONFIG.map_scale)) - assert img.size == (8 * 4, 6 * 4) - assert img.getpixel(image_position) == (255, 211, 0, 255) + assert render != base -def test_draw_path_on_map_draws_heading_indicator() -> None: +def test_render_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 @@ -201,21 +165,18 @@ def test_draw_path_on_map_draws_heading_indicator() -> None: the tick covers pixels at x > 12 along y == 12. """ packet, trace = _calibrated_inputs(heading=0) - render = _render(packet, trace=trace) - png = draw_path_on_map( - render, - config=CONFIG, - position_color=(255, 211, 0, 255), - ) - img = Image.open(io.BytesIO(png)).convert("RGBA") - position = render.map_data.vacuum_position - assert position is not None - cx = round(position.x * CONFIG.map_scale) - cy = round(position.y * CONFIG.map_scale) + image = _render(packet, trace=trace) + calibration = solve_q10_calibration(packet, trace) + assert calibration is not None + assert trace.robot_position is not None + px, py = calibration.world_to_pixel(trace.robot_position.x, trace.robot_position.y) + cx = round(px * CONFIG.map_scale) + cy = round(py * CONFIG.map_scale) + rendered = Image.open(io.BytesIO(image)).convert("RGBA") # Tick runs +x from the marker (4 * radius = 16 px at scale 4). - assert img.getpixel((cx + 8, cy)) == (255, 211, 0, 255) + assert rendered.getpixel((cx + 8, cy)) == (255, 211, 0, 255) # ...and not behind it (the marker is a small disc; sample well to the left) - assert img.getpixel((cx - 8, cy)) != (255, 211, 0, 255) + assert rendered.getpixel((cx - 8, cy)) != (255, 211, 0, 255) def test_solve_q10_calibration_uses_header_origin_with_short_path() -> None: From 2cd641cd2c42c670455bc015900e54401cb38546 Mon Sep 17 00:00:00 2001 From: Vincent <2070309+tubededentifrice@users.noreply.github.com> Date: Sun, 19 Jul 2026 22:14:53 +0400 Subject: [PATCH 6/7] refactor: cache one composed Q10 map image --- roborock/cli.py | 7 ++-- roborock/devices/traits/b01/q10/map.py | 34 ++++++-------------- tests/devices/traits/b01/q10/test_map.py | 41 ++++++------------------ 3 files changed, 21 insertions(+), 61 deletions(-) diff --git a/roborock/cli.py b/roborock/cli.py index 7633510b..9eeb7082 100644 --- a/roborock/cli.py +++ b/roborock/cli.py @@ -745,10 +745,9 @@ async def q10_map_with_path(ctx, device_id: str, output_file: str): if not got_path: click.echo("No live path available (the robot only reports its path while cleaning).") return - try: - image = map_trait.render_path_on_map() - except RoborockException as err: - click.echo(f"Could not render path on map: {err}") + image = map_trait.image_content + if image is None: + click.echo("No map image content available.") return with open(output_file, "wb") as f: f.write(image) diff --git a/roborock/devices/traits/b01/q10/map.py b/roborock/devices/traits/b01/q10/map.py index e5190c16..f47e02c4 100644 --- a/roborock/devices/traits/b01/q10/map.py +++ b/roborock/devices/traits/b01/q10/map.py @@ -9,7 +9,7 @@ ``MapDpsTrait`` owns the low-level DPS read model. ``MapContentTrait`` depends on it and combines that state with the latest map/trace packets through the pure functions in :mod:`roborock.map.b01_q10_render`. The high-level trait keeps only -one grouped source snapshot and one replace-whole rendered result; calibration, +one grouped source snapshot and one replace-whole rendered image; calibration, path placement and overlay placement are not independently mutable trait state. """ @@ -28,7 +28,7 @@ Q10TracePacket, ) from roborock.map.b01_q10_overlays import Q10Zone, parse_virtual_wall_blob, parse_zone_blob -from roborock.map.b01_q10_render import Q10MapOverlays, Q10MapRender, draw_path_on_map, render_q10_map +from roborock.map.b01_q10_render import Q10MapOverlays, render_q10_map from .common import UpdatableTrait @@ -66,7 +66,7 @@ class MapContentTrait(TraitUpdateListener): Map and trace packet updates replace :attr:`_source`; DPS updates are owned by the injected :class:`MapDpsTrait`. Rendering always produces one new - :class:`Q10MapRender`, keeping the externally visible fields consistent. + image, keeping the externally visible fields consistent. """ def __init__( @@ -79,13 +79,13 @@ def __init__( self._config = map_parser_config or B01Q10MapParserConfig() self._map_dps = map_dps or MapDpsTrait() self._source = Q10MapSource() - self._render: Q10MapRender | None = None + self._image_content: bytes | None = None self._map_dps.add_update_listener(self._map_dps_updated) @property def image_content(self) -> bytes | None: - """The rendered base map PNG, if a map has been pushed.""" - return self._render.image_content if self._render else None + """The composed map PNG, if a map has been pushed.""" + return self._image_content @property def rooms(self) -> list[Q10Room]: @@ -128,7 +128,7 @@ def update_from_map_packet(self, packet: Q10MapPacket) -> None: if render is None: return self._source = source - self._render = render + self._image_content = render self._notify_update() def update_from_trace_packet(self, packet: Q10TracePacket) -> None: @@ -138,29 +138,13 @@ def update_from_trace_packet(self, packet: Q10TracePacket) -> None: self._rebuild() self._notify_update() - def render_path_on_map( - self, - *, - line_color: tuple[int, int, int, int] = (235, 64, 52, 255), - position_color: tuple[int, int, int, int] = (255, 211, 0, 255), - ) -> bytes: - """Return a PNG with the path, robot, zones and walls drawn.""" - if self._render is None: - raise RoborockException("No map available; no map has been pushed yet") - return draw_path_on_map( - self._render, - config=self._config, - line_color=line_color, - position_color=position_color, - ) - def _map_dps_updated(self) -> None: """Recompose placed overlays after the low-level DPS state changes.""" if self._source.map_packet is not None: self._rebuild() self._notify_update() - def _compose(self, source: Q10MapSource) -> Q10MapRender | None: + def _compose(self, source: Q10MapSource) -> bytes | None: """Compose a source snapshot, preserving the previous result on error.""" if source.map_packet is None: return None @@ -179,4 +163,4 @@ def _rebuild(self) -> None: """Replace the derived render from the current source snapshot.""" render = self._compose(self._source) if render is not None: - self._render = render + self._image_content = render diff --git a/tests/devices/traits/b01/q10/test_map.py b/tests/devices/traits/b01/q10/test_map.py index 844ee7d3..ce5c9040 100644 --- a/tests/devices/traits/b01/q10/test_map.py +++ b/tests/devices/traits/b01/q10/test_map.py @@ -22,7 +22,6 @@ from roborock.data.b01_q10.b01_q10_code_mappings import B01_Q10_DP from roborock.devices.traits.b01.q10 import Q10PropertiesApi, create from roborock.devices.traits.b01.q10.map import MapContentTrait, MapDpsTrait -from roborock.exceptions import RoborockException from roborock.map.b01_grid_layers import GridCalibration from roborock.map.b01_q10_map_parser import ( Q10HeaderCalibration, @@ -232,11 +231,14 @@ def test_trace_update_projects_short_path_using_header() -> None: trait = MapContentTrait() packet = replace(parse_map_packet(FIXTURE.read_bytes()), header_calibration=_USABLE_HEADER) trait.update_from_map_packet(packet) + base = trait.image_content + assert base is not None true = GridCalibration(resolution=20.0, origin_x=0.0, origin_y=5.0, y_sign=1) trait.update_from_trace_packet(Q10TracePacket(points=_floor_world_points(packet, true, 6))) assert len(trait.path) < 20 # far too short for the full origin+resolution fit - assert trait.render_path_on_map()[:8] == b"\x89PNG\r\n\x1a\n" + assert trait.image_content is not None + assert trait.image_content != base def test_short_trace_without_header_cannot_be_projected() -> None: @@ -244,36 +246,10 @@ def test_short_trace_without_header_cannot_be_projected() -> None: packet = parse_map_packet(FIXTURE.read_bytes()) trait = MapContentTrait() trait.update_from_map_packet(packet) # the fixture header is a keepalive frame + base = trait.image_content true = GridCalibration(resolution=10.0, origin_x=0.0, origin_y=5.0, y_sign=1) trait.update_from_trace_packet(Q10TracePacket(points=_floor_world_points(packet, true, 6))) - with pytest.raises(RoborockException, match="No calibration available"): - trait.render_path_on_map() - - -def test_render_path_on_map_requires_map() -> None: - trait = MapContentTrait() - with pytest.raises(RoborockException, match="No map available"): - trait.render_path_on_map() - - -def test_render_path_on_map_uses_derived_content() -> None: - """The renderer draws the already-derived map content as a PNG.""" - trait = MapContentTrait() - packet = replace(parse_map_packet(FIXTURE.read_bytes()), header_calibration=_USABLE_HEADER) - trait.update_from_map_packet(packet) - true = GridCalibration(resolution=20.0, origin_x=0.0, origin_y=5.0, y_sign=1) - trait.update_from_trace_packet(Q10TracePacket(points=_floor_world_points(packet, true, 6))) - - png = trait.render_path_on_map() - - assert png[:8] == b"\x89PNG\r\n\x1a\n" - - -def test_render_path_on_map_without_path_cannot_calibrate() -> None: - """A map but no cleaning path -> no calibration -> a clear error.""" - trait = _trait_with_map() - with pytest.raises(RoborockException, match="No calibration available"): - trait.render_path_on_map() + assert trait.image_content == base # --- Overlays ---------------------------------------------------------------- @@ -287,7 +263,8 @@ def test_load_overlays_places_zones_after_calibration() -> None: trait.update_from_map_packet(packet) true = GridCalibration(resolution=20.0, origin_x=0.0, origin_y=5.0, y_sign=1) trait.update_from_trace_packet(Q10TracePacket(points=_floor_world_points(packet, true, 6))) - before = trait.render_path_on_map() + before = trait.image_content + assert before is not None def rect(zone_type: int, corners: list[tuple[int, int]]) -> bytes: out = bytes([zone_type, len(corners)]) @@ -299,7 +276,7 @@ def rect(zone_type: int, corners: list[tuple[int, int]]) -> bytes: map_dps.update_from_dps({B01_Q10_DP.RESTRICTED_ZONE_UP: base64.b64encode(blob).decode()}) assert len(trait.zones) == 1 - assert trait.render_path_on_map() != before + assert trait.image_content != before def test_load_overlays_partial_update_keeps_existing_zones() -> None: From 62adc2b62668aaf1f483132f1b092f836ce53532 Mon Sep 17 00:00:00 2001 From: Vincent <2070309+tubededentifrice@users.noreply.github.com> Date: Sun, 19 Jul 2026 22:54:04 +0400 Subject: [PATCH 7/7] refactor: simplify Q10 map trait composition --- roborock/cli.py | 33 -------- roborock/devices/traits/b01/q10/map.py | 98 +++++++++--------------- tests/devices/traits/b01/q10/test_map.py | 17 ++-- 3 files changed, 46 insertions(+), 102 deletions(-) diff --git a/roborock/cli.py b/roborock/cli.py index 9eeb7082..9def26af 100644 --- a/roborock/cli.py +++ b/roborock/cli.py @@ -721,39 +721,6 @@ async def q10_position(ctx, device_id: str, include_path: bool): click.echo(dump_json(summary)) -@session.command() -@click.option("--device_id", required=True) -@click.option("--output-file", required=True, help="Path to save the map image with the path drawn.") -@click.pass_context -@async_command -async def q10_map_with_path(ctx, device_id: str, output_file: str): - """Render the Q10 map with the current cleaning path + robot position drawn. - - Needs the robot to be actively cleaning so a live trace is available. - Fetches the map and path and writes the annotated PNG. - """ - context: RoborockContext = ctx.obj - device_manager = await context.get_device_manager() - device = await device_manager.get_device(device_id) - if device.b01_q10_properties is None: - click.echo("Feature not supported by device") - return - properties = device.b01_q10_properties - map_trait = properties.map - await _await_q10_map_push(properties, lambda: map_trait.image_content is not None) - got_path = await _await_q10_map_push(properties, lambda: bool(map_trait.path)) - if not got_path: - click.echo("No live path available (the robot only reports its path while cleaning).") - return - image = map_trait.image_content - if image is None: - click.echo("No map image content available.") - return - with open(output_file, "wb") as f: - f.write(image) - click.echo(f"Saved map with {len(map_trait.path)}-point path to {output_file}") - - @session.command() @click.option("--device_id", required=True) @click.pass_context diff --git a/roborock/devices/traits/b01/q10/map.py b/roborock/devices/traits/b01/q10/map.py index f47e02c4..1d3c3ba4 100644 --- a/roborock/devices/traits/b01/q10/map.py +++ b/roborock/devices/traits/b01/q10/map.py @@ -9,12 +9,12 @@ ``MapDpsTrait`` owns the low-level DPS read model. ``MapContentTrait`` depends on it and combines that state with the latest map/trace packets through the pure functions in :mod:`roborock.map.b01_q10_render`. The high-level trait keeps only -one grouped source snapshot and one replace-whole rendered image; calibration, -path placement and overlay placement are not independently mutable trait state. +the latest value from each source and one replace-whole rendered image; +calibration, path placement and overlay placement remain inside the renderer. """ import logging -from dataclasses import dataclass, field, replace +from dataclasses import dataclass, field from roborock.data import RoborockBase from roborock.data.b01_q10.b01_q10_code_mappings import B01_Q10_DP @@ -52,21 +52,22 @@ def __init__(self) -> None: MapDps.__init__(self) UpdatableTrait.__init__(self, command=None, logger=_LOGGER) + @property + def zones(self) -> list[Q10Zone]: + """Restricted zones decoded from the latest DPS value.""" + return parse_zone_blob(self.restricted_zone_up) -@dataclass(frozen=True) -class Q10MapSource: - """The latest map-protocol inputs, replaced atomically on every push.""" - - map_packet: Q10MapPacket | None = None - trace_packet: Q10TracePacket | None = None + @property + def virtual_walls(self) -> list[Q10Zone]: + """Virtual walls decoded from the latest DPS value.""" + return parse_virtual_wall_blob(self.virtual_wall_up) class MapContentTrait(TraitUpdateListener): """High-level composed Q10 map view. - Map and trace packet updates replace :attr:`_source`; DPS updates are owned - by the injected :class:`MapDpsTrait`. Rendering always produces one new - image, keeping the externally visible fields consistent. + The latest map and trace packets are combined with the injected + :class:`MapDpsTrait` whenever any of those three sources changes. """ def __init__( @@ -78,7 +79,8 @@ def __init__( TraitUpdateListener.__init__(self, logger=_LOGGER) self._config = map_parser_config or B01Q10MapParserConfig() self._map_dps = map_dps or MapDpsTrait() - self._source = Q10MapSource() + self._map_packet: Q10MapPacket | None = None + self._trace_packet: Q10TracePacket | None = None self._image_content: bytes | None = None self._map_dps.add_update_listener(self._map_dps_updated) @@ -90,77 +92,53 @@ def image_content(self) -> bytes | None: @property def rooms(self) -> list[Q10Room]: """Rooms reported by the device.""" - packet = self._source.map_packet - return packet.rooms if packet else [] + return self._map_packet.rooms if self._map_packet else [] @property def path(self) -> list[Q10Point]: """Full path from the latest trace packet.""" - trace = self._source.trace_packet - return trace.points if trace else [] + return self._trace_packet.points if self._trace_packet else [] @property def robot_position(self) -> Q10Point | None: """Current robot position from the latest trace packet.""" - trace = self._source.trace_packet - return trace.robot_position if trace else None + return self._trace_packet.robot_position if self._trace_packet else None @property def robot_heading(self) -> int | None: """Current robot heading from the latest trace packet.""" - trace = self._source.trace_packet - return trace.heading if trace else None - - @property - def zones(self) -> list[Q10Zone]: - """Restricted zones decoded from the low-level DPS trait.""" - return parse_zone_blob(self._map_dps.restricted_zone_up) - - @property - def virtual_walls(self) -> list[Q10Zone]: - """Virtual walls decoded from the low-level DPS trait.""" - return parse_virtual_wall_blob(self._map_dps.virtual_wall_up) + return self._trace_packet.heading if self._trace_packet else None def update_from_map_packet(self, packet: Q10MapPacket) -> None: - """Replace the current map packet and compose a consistent result.""" - source = replace(self._source, map_packet=packet) - render = self._compose(source) - if render is None: - return - self._source = source - self._image_content = render + """Store a map-protocol update and render the latest sources.""" + self._map_packet = packet + self._render() self._notify_update() def update_from_trace_packet(self, packet: Q10TracePacket) -> None: - """Replace the complete current-session trace packet.""" - self._source = replace(self._source, trace_packet=packet) - if self._source.map_packet is not None: - self._rebuild() + """Store a trace-protocol update and render the latest sources.""" + self._trace_packet = packet + self._render() self._notify_update() def _map_dps_updated(self) -> None: - """Recompose placed overlays after the low-level DPS state changes.""" - if self._source.map_packet is not None: - self._rebuild() + """Render after the low-level DPS source changes.""" + self._render() self._notify_update() - def _compose(self, source: Q10MapSource) -> bytes | None: - """Compose a source snapshot, preserving the previous result on error.""" - if source.map_packet is None: - return None + def _render(self) -> None: + """Render the latest map, trace and DPS sources, if a map is available.""" + if self._map_packet is None: + return try: - return render_q10_map( - source.map_packet, - source.trace_packet, - Q10MapOverlays(zones=tuple(self.zones), virtual_walls=tuple(self.virtual_walls)), + self._image_content = render_q10_map( + self._map_packet, + self._trace_packet, + Q10MapOverlays( + zones=tuple(self._map_dps.zones), + virtual_walls=tuple(self._map_dps.virtual_walls), + ), config=self._config, ) except RoborockException as ex: _LOGGER.debug("Failed to render Q10 map packet: %s", ex) - return None - - def _rebuild(self) -> None: - """Replace the derived render from the current source snapshot.""" - render = self._compose(self._source) - if render is not None: - self._image_content = render diff --git a/tests/devices/traits/b01/q10/test_map.py b/tests/devices/traits/b01/q10/test_map.py index ce5c9040..50622600 100644 --- a/tests/devices/traits/b01/q10/test_map.py +++ b/tests/devices/traits/b01/q10/test_map.py @@ -256,7 +256,7 @@ def test_short_trace_without_header_cannot_be_projected() -> None: def test_load_overlays_places_zones_after_calibration() -> None: - """Decoded no-go / no-mop zones are placed on MapData once calibrated.""" + """Decoded no-go / no-mop zones are drawn once the sources calibrate.""" map_dps = MapDpsTrait() trait = MapContentTrait(map_dps) packet = replace(parse_map_packet(FIXTURE.read_bytes()), header_calibration=_USABLE_HEADER) @@ -275,25 +275,24 @@ def rect(zone_type: int, corners: list[tuple[int, int]]) -> bytes: blob = bytes([1, 1]) + rect(0, [(0, 0), (40, 0), (40, 40), (0, 40)]) map_dps.update_from_dps({B01_Q10_DP.RESTRICTED_ZONE_UP: base64.b64encode(blob).decode()}) - assert len(trait.zones) == 1 + assert len(map_dps.zones) == 1 assert trait.image_content != before def test_load_overlays_partial_update_keeps_existing_zones() -> None: """A status push without the zone DP (None) must not wipe loaded zones.""" map_dps = MapDpsTrait() - trait = MapContentTrait(map_dps) blob = ( bytes([1, 1]) + bytes([0, 4]) + b"".join(int.to_bytes(v & 0xFFFF, 2, "big") for xy in [(0, 0), (4, 0), (4, 4), (0, 4)] for v in xy) ) map_dps.update_from_dps({B01_Q10_DP.RESTRICTED_ZONE_UP: base64.b64encode(blob).decode()}) - assert len(trait.zones) == 1 + assert len(map_dps.zones) == 1 # A later partial update carrying only the (empty) virtual-wall DP. map_dps.update_from_dps({B01_Q10_DP.VIRTUAL_WALL_UP: "AA=="}) - assert len(trait.zones) == 1 # zones preserved - assert trait.virtual_walls == [] + assert len(map_dps.zones) == 1 # zones preserved + assert map_dps.virtual_walls == [] def test_map_dps_trait_updates_high_level_map_content() -> None: @@ -310,7 +309,7 @@ def test_map_dps_trait_updates_high_level_map_content() -> None: map_dps.update_from_dps({B01_Q10_DP.RESTRICTED_ZONE_UP: base64.b64encode(blob).decode()}) - assert len(trait.zones) == 1 + assert len(map_dps.zones) == 1 assert notified # listeners learn the overlays changed @@ -323,6 +322,6 @@ def test_map_dps_push_without_overlay_data_points_is_noop() -> None: map_dps.update_from_dps({B01_Q10_DP.BATTERY: 50}) - assert trait.zones == [] - assert trait.virtual_walls == [] + assert map_dps.zones == [] + assert map_dps.virtual_walls == [] assert not notified