From b504f2403fabb3e418b0a96fa05e944b77402c5c Mon Sep 17 00:00:00 2001 From: Harry Coureau Date: Sat, 7 Nov 2026 19:02:09 +0000 Subject: [PATCH 1/3] feat: expose Q10 current cleaning room --- roborock/devices/traits/b01/q10/map.py | 20 ++- roborock/map/b01_q10_render.py | 129 +++++++++++++++--- .../b01/q10/__snapshots__/test_status.ambr | 3 + tests/devices/traits/b01/q10/test_map.py | 72 +++++++++- tests/map/test_b01_q10_render.py | 67 +++++++-- 5 files changed, 254 insertions(+), 37 deletions(-) diff --git a/roborock/devices/traits/b01/q10/map.py b/roborock/devices/traits/b01/q10/map.py index a33c2df22..8cdf51dc2 100644 --- a/roborock/devices/traits/b01/q10/map.py +++ b/roborock/devices/traits/b01/q10/map.py @@ -30,7 +30,7 @@ Q10TracePacket, ) from roborock.map.b01_q10_overlays import parse_virtual_wall_blob, parse_zone_blob -from roborock.map.b01_q10_render import Q10MapOverlays, render_q10_map +from roborock.map.b01_q10_render import Q10MapOverlays, render_q10_map, resolve_q10_current_room from .command import CommandTrait from .common import UpdatableTrait @@ -148,6 +148,17 @@ def robot_heading(self) -> int | None: """Current heading for orienting a robot marker on a caller-rendered map.""" return self._trace_packet.heading if self._trace_packet else None + @property + def current_room(self) -> Q10Room | None: + """Room inferred from the latest live robot position and segmented map. + + Returns ``None`` when position or calibration is unavailable, the pose + lies outside the map, or nearby room labels are ambiguous. + """ + if self._map_packet is None or self._trace_packet is None: + return None + return resolve_q10_current_room(self._map_packet, self._trace_packet) + def update_from_map_packet(self, packet: Q10MapPacket) -> None: """Store a map-protocol update and render the latest sources.""" self._map_packet = packet @@ -156,12 +167,15 @@ def update_from_map_packet(self, packet: Q10MapPacket) -> None: def update_from_trace_packet(self, packet: Q10TracePacket) -> None: """Store a trace-protocol update and render the latest sources.""" - self._trace_packet = packet + self._trace_packet = None if self._map_dps.robot_at_dock else packet self._render() self._notify_update() def _map_dps_updated(self) -> None: """Render after the low-level map DPS source changes.""" + if self._map_dps.robot_at_dock: + # A completed path no longer identifies the current pose at dock. + self._trace_packet = None if self._map_packet is None: return self._render() @@ -186,6 +200,7 @@ def _render(self) -> None: def as_dict(self, exclude: set[str] | None = None) -> dict[str, Any]: """Return the trait data as a dictionary, excluding large binary data.""" exclude_set = exclude or set() + current_room = self.current_room data = { "rooms": [room.as_dict() for room in self.rooms], "path": [point.as_dict() for point in self.path], @@ -193,6 +208,7 @@ def as_dict(self, exclude: set[str] | None = None) -> dict[str, Any]: {"x": position.x, "y": position.y} if (position := self.robot_position) is not None else None ), "robotHeading": self.robot_heading, + "currentRoom": current_room.as_dict() if current_room is not None else None, } for key in exclude_set: data.pop(key, None) diff --git a/roborock/map/b01_q10_render.py b/roborock/map/b01_q10_render.py index 1ee9ec006..2b3992e38 100644 --- a/roborock/map/b01_q10_render.py +++ b/roborock/map/b01_q10_render.py @@ -31,7 +31,6 @@ GridCalibration, GridLayers, solve_calibration, - solve_calibration_with_origin, ) from .b01_q10_map_parser import ( _BACKGROUND_VALUE, @@ -41,6 +40,8 @@ Q10EraseZone, Q10HistoricalTracePacket, Q10MapPacket, + Q10Point, + Q10Room, Q10TracePacket, erased_packet, ) @@ -62,13 +63,11 @@ # erase/restriction vectors use 5 mm units: one header unit therefore equals # two vector units. Trace points use a separate 2.5 mm coordinate scale. _Q10_VECTOR_UNITS_PER_HEADER_RESOLUTION_UNIT = 2.0 +# One centimetre header unit spans four 2.5 mm trace units. +_Q10_TRACE_UNITS_PER_HEADER_RESOLUTION_UNIT = 4.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 _Q10_DRAWABLE_TYPES = { Drawable.CHARGER, Drawable.NO_GO_AREAS, @@ -143,30 +142,120 @@ def solve_q10_calibration( ) -> 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. + When the map packet carries usable ss07 header metadata, its fixed origin + and resolution are authoritative and work from the first live pose. Older + or incomplete packets fall back to a full origin/resolution fit, which needs + a reasonably dense cleaning path. Returns ``None`` when neither is usable. """ - if trace is None: + if trace is None or not trace.points: return None + if calibration := _trace_calibration_from_header_metadata(packet, trace.points): + return calibration 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) + return _calibration_from_fit(packet.layers, points) -def _calibration_from_header( +def _trace_calibration_from_header_metadata( packet: Q10MapPacket, - points: list[tuple[float, float]], + points: Sequence[Q10Point], ) -> 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: + """Build the live-trace transform directly from validated header metadata.""" + header = packet.header_calibration + if header is None or header.resolution <= 0 or (origin := header.origin_pixels()) is None: + return None + resolution = header.resolution * _Q10_TRACE_UNITS_PER_HEADER_RESOLUTION_UNIT + if not min(_Q10_RESOLUTIONS) <= resolution <= max(_Q10_RESOLUTIONS): + return None + calibration = GridCalibration( + resolution=resolution, + origin_x=origin[0], + origin_y=origin[1], + y_sign=1, + ) + return calibration if _trace_projects_onto_floor(packet.layers, points, calibration) else None + + +def _trace_projects_onto_floor( + layers: GridLayers, + points: Sequence[Q10Point], + calibration: GridCalibration, +) -> bool: + """Validate header calibration against a bounded sample of trace points.""" + if not points: + return False + stride = max(1, len(points) // 256) + sample = points[::stride] + on_floor = 0 + for point in sample: + pixel_x, pixel_y = calibration.world_to_pixel(point.x, point.y) + column, row = math.floor(pixel_x), math.floor(pixel_y) + if not (0 <= column < layers.width and 0 <= row < layers.height): + continue + if layers.cell_class(layers.grid[row * layers.width + column]) == "floor": + on_floor += 1 + return on_floor >= len(sample) * 0.5 + + +def _room_at_position( + packet: Q10MapPacket, + position: Q10Point, + calibration: GridCalibration, + *, + search_radius: int = 2, +) -> Q10Room | None: + """Resolve a live position to a segmented room in the occupancy grid. + + The exact cell wins. A robot centre can briefly land on a wall or doorway + pixel, so a small expanding neighbourhood is used only when the exact cell + is not segmented. Ambiguous ties stay unknown rather than naming the wrong + room. + """ + pixel_x, pixel_y = calibration.world_to_pixel(position.x, position.y) + column, row = math.floor(pixel_x), math.floor(pixel_y) + if not (0 <= column < packet.width and 0 <= row < packet.height): + return None + + rooms_by_value = {room.pixel_value: room for room in packet.rooms} + + def room_at(column: int, row: int) -> Q10Room | None: + if not (0 <= column < packet.width and 0 <= row < packet.height): + return None + return rooms_by_value.get(packet.grid[row * packet.width + column]) + + if room := room_at(column, row): + return room + + for radius in range(1, max(0, search_radius) + 1): + room_ids = { + candidate.id + for nearby_row in range(row - radius, row + radius + 1) + for nearby_column in range(column - radius, column + radius + 1) + if (candidate := room_at(nearby_column, nearby_row)) is not None + } + if not room_ids: + continue + if len(room_ids) != 1: + return None + room_id = room_ids.pop() + return next((room for room in packet.rooms if room.id == room_id), None) + return None + + +def resolve_q10_current_room(packet: Q10MapPacket, trace: Q10TracePacket | None) -> Q10Room | None: + """Infer the robot's current room from its latest live pose and map grid. + + No authoritative active-segment field has been identified or observed on + ss07 firmware. The live trace's final point is the robot pose, and the full + map labels every segmented floor cell. Header metadata supplies the fixed + transform from trace coordinates to that grid; older/incomplete packets + fall back to the path-fit calibration used by rendering. + """ + if trace is None or (position := trace.robot_position) is None: return None - origin = header_calibration.origin_pixels() - if origin is None: # keepalive frame -- no usable origin + calibration = solve_q10_calibration(packet, trace) + if calibration is None: return None - return solve_calibration_with_origin(packet.layers, points, origin, resolutions=_Q10_RESOLUTIONS) + return _room_at_position(packet, position, calibration) def _calibration_from_header_metadata( diff --git a/tests/devices/traits/b01/q10/__snapshots__/test_status.ambr b/tests/devices/traits/b01/q10/__snapshots__/test_status.ambr index daac7f55b..3cc105e9b 100644 --- a/tests/devices/traits/b01/q10/__snapshots__/test_status.ambr +++ b/tests/devices/traits/b01/q10/__snapshots__/test_status.ambr @@ -24,6 +24,7 @@ 'dustSwitch': True, }), 'map': dict({ + 'currentRoom': None, 'path': list([ ]), 'robotHeading': None, @@ -404,6 +405,7 @@ 'dustSwitch': True, }), 'map': dict({ + 'currentRoom': None, 'path': list([ ]), 'robotHeading': None, @@ -480,6 +482,7 @@ 'dust_collection': dict({ }), 'map': dict({ + 'currentRoom': None, 'path': list([ ]), 'robotHeading': None, diff --git a/tests/devices/traits/b01/q10/test_map.py b/tests/devices/traits/b01/q10/test_map.py index 060553af5..707b989bb 100644 --- a/tests/devices/traits/b01/q10/test_map.py +++ b/tests/devices/traits/b01/q10/test_map.py @@ -9,6 +9,7 @@ import asyncio import base64 from collections.abc import AsyncGenerator, Generator +from dataclasses import replace from pathlib import Path from typing import cast from unittest.mock import Mock, patch @@ -23,6 +24,7 @@ from roborock.devices.traits.b01.q10.maps import MapsTrait from roborock.exceptions import RoborockException from roborock.map.b01_q10_map_parser import ( + Q10HeaderCalibration, Q10Point, Q10TracePacket, parse_map_packet, @@ -490,8 +492,8 @@ async def test_charging_status_renders_robot_at_dock(render_map: Mock) -> None: assert render_map.call_args.kwargs["robot_at_dock"] is True -def test_docked_state_hides_trace_only_from_rendering(render_map: Mock) -> None: - """A docked render omits the valid trace without deleting source data.""" +def test_docked_state_clears_live_trace(render_map: Mock) -> None: + """Docking clears the completed path and current robot pose.""" map_dps = MapDpsTrait() trait = _map_trait(map_dps) packet = parse_map_packet(FIXTURE.read_bytes()) @@ -504,13 +506,15 @@ def test_docked_state_hides_trace_only_from_rendering(render_map: Mock) -> None: map_dps.update_from_dps({B01_Q10_DP.STATUS: YXDeviceState.CHARGING.code}) - assert trait.path == trace.points + assert trait.path == [] + assert trait.robot_position is None + assert trait.current_room is None assert render_map.call_args.args[1] is None assert render_map.call_args.kwargs["robot_at_dock"] is True -def test_late_trace_is_retained_but_hidden_while_docked(render_map: Mock) -> None: - """A late trace stays available but is not part of a docked render.""" +def test_late_trace_is_ignored_while_docked(render_map: Mock) -> None: + """A late completed trace cannot repopulate the docked robot pose.""" map_dps = MapDpsTrait() map_dps.update_from_dps({B01_Q10_DP.STATUS: YXDeviceState.CHARGING.code}) trait = _map_trait(map_dps) @@ -520,7 +524,9 @@ def test_late_trace_is_retained_but_hidden_while_docked(render_map: Mock) -> Non trait.update_from_map_packet(parse_map_packet(FIXTURE.read_bytes())) trait.update_from_trace_packet(trace) - assert trait.path == trace.points + assert trait.path == [] + assert trait.robot_position is None + assert trait.current_room is None assert render_map.call_args.args[1] is None @@ -586,3 +592,57 @@ def test_map_content_trait_as_dict_camelizes_child_keys() -> None: } assert data["path"] == [{"x": 100, "y": 200}, {"x": 150, "y": 250}] assert data["robotPosition"] == {"x": 25875, "y": 26125} + + +@pytest.mark.parametrize("trace_first", [False, True]) +def test_current_room_tracks_latest_fixture_pose_and_map(trace_first: bool) -> None: + """Current room and robot position consume the same latest map/trace inputs.""" + map_dps = MapDpsTrait() + trait = _map_trait(map_dps) + packet = replace( + parse_map_packet(FIXTURE.read_bytes()), header_calibration=Q10HeaderCalibration(0, 50, 5, -1, -1, 0) + ) + trace = Q10TracePacket(points=[Q10Point(100, 60)]) + assert trait.current_room is None + if trace_first: + trait.update_from_trace_packet(trace) + assert trait.current_room is None + trait.update_from_map_packet(packet) + else: + trait.update_from_map_packet(packet) + assert trait.current_room is None + trait.update_from_trace_packet(trace) + assert trait.current_room == packet.rooms[1] + assert trait.robot_position == trace.points[-1].to_roborock() + assert trait.as_dict()["currentRoom"] == packet.rooms[1].as_dict() + assert "currentRoom" not in trait.as_dict(exclude={"currentRoom"}) + map_dps.update_from_dps({B01_Q10_DP.STATUS: YXDeviceState.IDLE.code}) + assert trait.current_room == packet.rooms[1] + trait.update_from_trace_packet(Q10TracePacket(points=[Q10Point(40, 60)])) + assert trait.current_room == packet.rooms[0] + trait.update_from_map_packet(replace(packet, rooms=[])) + assert trait.current_room is None + + +@pytest.mark.parametrize("status", [YXDeviceState.CHARGING, YXDeviceState.EMPTYING_THE_BIN]) +def test_docking_clears_room_and_ignores_late_fixture_trace(status: YXDeviceState) -> None: + map_dps = MapDpsTrait() + trait = _map_trait(map_dps) + packet = replace( + parse_map_packet(FIXTURE.read_bytes()), header_calibration=Q10HeaderCalibration(0, 50, 5, -1, -1, 0) + ) + trace = Q10TracePacket(points=[Q10Point(100, 60)]) + trait.update_from_map_packet(packet) + trait.update_from_trace_packet(trace) + assert trait.current_room == packet.rooms[1] + map_dps.update_from_dps({B01_Q10_DP.STATUS: status.code}) + assert trait.current_room is None + assert trait.robot_position is None + assert trait.path == [] + trait.update_from_trace_packet(trace) + assert trait.current_room is None + assert trait.path == [] + map_dps.update_from_dps({B01_Q10_DP.STATUS: YXDeviceState.CLEANING.code}) + assert trait.current_room is None + trait.update_from_trace_packet(trace) + assert trait.current_room == packet.rooms[1] diff --git a/tests/map/test_b01_q10_render.py b/tests/map/test_b01_q10_render.py index 131b9e86d..73cfd9709 100644 --- a/tests/map/test_b01_q10_render.py +++ b/tests/map/test_b01_q10_render.py @@ -9,6 +9,7 @@ from dataclasses import replace from pathlib import Path +import pytest from PIL import Image from vacuum_map_parser_base.config.size import Size, Sizes from vacuum_map_parser_base.map_data import MapData, Point @@ -33,7 +34,6 @@ Q10Zone, ) from roborock.map.b01_q10_render import ( - _Q10_RESOLUTIONS, Q10MapOverlays, _calibration_from_header_metadata, _erased_cells, @@ -41,6 +41,7 @@ _place_docked_robot, _vector_calibration, render_q10_map, + resolve_q10_current_room, solve_q10_calibration, ) @@ -252,7 +253,8 @@ def test_zero_degree_dock_places_robot_to_its_right() -> None: def test_render_applies_erase_zones() -> None: """With a calibration, erase-zone cells are blanked from the image.""" packet, trace = _calibrated_inputs() - base = _render(packet, trace=trace) + packet = replace(packet, header_calibration=replace(HEADER, charger_x=-1, charger_y=-1)) + base = _render(packet) trace_calibration = solve_q10_calibration(packet, trace) calibration = _vector_calibration(packet, trace_calibration) assert calibration == VECTOR_CALIBRATION @@ -261,7 +263,7 @@ def test_render_applies_erase_zones() -> None: corners = [(-1, -1), (8, -1), (8, 6), (-1, 6)] 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) + render = _render(replace(packet, erase_zones=[erase_zone])) assert len(cells) == packet.layers.width * packet.layers.height assert render != base @@ -294,7 +296,8 @@ def test_header_vector_calibration_is_independent_of_trace_orientation() -> None 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) + packet = replace(packet, header_calibration=replace(HEADER, charger_x=-1, charger_y=-1)) + base = _render(packet) trace_calibration = solve_q10_calibration(packet, trace) calibration = _vector_calibration(packet, trace_calibration) assert calibration == VECTOR_CALIBRATION @@ -303,7 +306,7 @@ def test_render_partial_erase() -> None: corners = [(-1, -1), (8, -1), (8, 2), (-1, 2)] 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) + render = _render(replace(packet, erase_zones=[erase_zone])) assert 0 < len(cells) < packet.layers.width * packet.layers.height assert render != base @@ -324,10 +327,7 @@ def test_solve_q10_calibration_uses_header_origin_with_short_path() -> None: 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 + assert cal == TRACE_CALIBRATION def test_solve_q10_calibration_short_path_without_header_returns_none() -> None: @@ -335,3 +335,52 @@ def test_solve_q10_calibration_short_path_without_header_returns_none() -> None: packet = _packet() trace = Q10TracePacket(points=_floor_world_points(packet.layers, IDENTITY, 6)) assert solve_q10_calibration(packet, trace) is None + + +@pytest.mark.parametrize( + "pixel, expected_id", [((2, 2), 2), ((5, 2), 3), ((0, 2), 2), ((3, 4), None), ((-1, 2), None), ((8, 2), None)] +) +def test_resolve_current_room_exact_nearby_ambiguous_and_outside( + pixel: tuple[int, int], + expected_id: int | None, +) -> None: + """Fixture segments resolve exactly or conservatively around doorways.""" + packet = replace(_packet(), header_calibration=HEADER) + points = [Q10Point(40, 60), Q10Point(100, 60)] + points.append(Q10Point(*(int(value) for value in TRACE_CALIBRATION.pixel_to_world(*pixel)))) + trace = Q10TracePacket(points=points) + room = resolve_q10_current_room(packet, trace) + assert (room.id if room is not None else None) == expected_id + + +def test_single_pose_uses_header_calibration_for_room_and_rendering() -> None: + packet = replace(_packet(), header_calibration=replace(HEADER, charger_x=-1, charger_y=-1)) + trace = Q10TracePacket(points=[Q10Point(100, 60)], heading=45) + assert solve_q10_calibration(packet, trace) == TRACE_CALIBRATION + assert resolve_q10_current_room(packet, trace) == packet.rooms[1] + rendered = Image.open(io.BytesIO(_render(packet, trace=trace))) + assert rendered.getpixel((5 * CONFIG.map_scale, 2 * CONFIG.map_scale)) == (255, 255, 255, 255) + + +@pytest.mark.parametrize( + "header", [None, replace(HEADER, resolution=100), replace(HEADER, origin_x=30000, origin_y=30000)] +) +def test_current_room_uses_dense_path_fallback_when_header_is_unusable(header: Q10HeaderCalibration | None) -> None: + packet = replace(_packet(), header_calibration=header) + floor_pixels = [(x, y) for y in (1, 2, 3) for x in (1, 2, 3, 4, 5, 6)] + pixels = floor_pixels * 2 + [(5, 2)] + trace = Q10TracePacket( + points=[Q10Point(*(int(v) for v in TRACE_CALIBRATION.pixel_to_world(x, y))) for x, y in pixels] + ) + calibration = solve_q10_calibration(packet, trace) + assert calibration is not None + assert resolve_q10_current_room(packet, trace) == packet.rooms[1] + + +def test_current_room_requires_pose_calibration_and_matching_metadata() -> None: + packet = _packet() + assert resolve_q10_current_room(packet, None) is None + assert resolve_q10_current_room(packet, Q10TracePacket()) is None + assert resolve_q10_current_room(packet, Q10TracePacket(points=[Q10Point(100, 60)])) is None + packet = replace(packet, header_calibration=HEADER, rooms=[]) + assert resolve_q10_current_room(packet, Q10TracePacket(points=[Q10Point(100, 60)])) is None From 3ee80aa5625a72e65ed0985c118bfc156c03ecec Mon Sep 17 00:00:00 2001 From: Harry Coureau Date: Sat, 7 Nov 2026 19:03:16 +0000 Subject: [PATCH 2/3] test: separate Q10 current room arrival scenarios --- tests/devices/traits/b01/q10/test_map.py | 28 +++++++++++++++--------- 1 file changed, 18 insertions(+), 10 deletions(-) diff --git a/tests/devices/traits/b01/q10/test_map.py b/tests/devices/traits/b01/q10/test_map.py index 707b989bb..9381fab16 100644 --- a/tests/devices/traits/b01/q10/test_map.py +++ b/tests/devices/traits/b01/q10/test_map.py @@ -594,8 +594,7 @@ def test_map_content_trait_as_dict_camelizes_child_keys() -> None: assert data["robotPosition"] == {"x": 25875, "y": 26125} -@pytest.mark.parametrize("trace_first", [False, True]) -def test_current_room_tracks_latest_fixture_pose_and_map(trace_first: bool) -> None: +def test_current_room_tracks_latest_fixture_pose_and_map() -> None: """Current room and robot position consume the same latest map/trace inputs.""" map_dps = MapDpsTrait() trait = _map_trait(map_dps) @@ -604,14 +603,9 @@ def test_current_room_tracks_latest_fixture_pose_and_map(trace_first: bool) -> N ) trace = Q10TracePacket(points=[Q10Point(100, 60)]) assert trait.current_room is None - if trace_first: - trait.update_from_trace_packet(trace) - assert trait.current_room is None - trait.update_from_map_packet(packet) - else: - trait.update_from_map_packet(packet) - assert trait.current_room is None - trait.update_from_trace_packet(trace) + trait.update_from_map_packet(packet) + assert trait.current_room is None + trait.update_from_trace_packet(trace) assert trait.current_room == packet.rooms[1] assert trait.robot_position == trace.points[-1].to_roborock() assert trait.as_dict()["currentRoom"] == packet.rooms[1].as_dict() @@ -624,6 +618,20 @@ def test_current_room_tracks_latest_fixture_pose_and_map(trace_first: bool) -> N assert trait.current_room is None +def test_map_arrival_resolves_current_room_from_existing_fixture_trace() -> None: + """A map arriving after the live trace resolves its latest robot pose.""" + trait = _map_trait() + packet = replace( + parse_map_packet(FIXTURE.read_bytes()), header_calibration=Q10HeaderCalibration(0, 50, 5, -1, -1, 0) + ) + trace = Q10TracePacket(points=[Q10Point(100, 60)]) + trait.update_from_trace_packet(trace) + assert trait.current_room is None + trait.update_from_map_packet(packet) + assert trait.current_room == packet.rooms[1] + assert trait.robot_position == trace.points[-1].to_roborock() + + @pytest.mark.parametrize("status", [YXDeviceState.CHARGING, YXDeviceState.EMPTYING_THE_BIN]) def test_docking_clears_room_and_ignores_late_fixture_trace(status: YXDeviceState) -> None: map_dps = MapDpsTrait() From 4d3baff916d4d995d4c4049cd0e7a954ccf1f8d0 Mon Sep 17 00:00:00 2001 From: Harry Coureau Date: Sun, 8 Nov 2026 18:59:52 +0000 Subject: [PATCH 3/3] fix: preserve Q10 room labels and sample header calibration --- roborock/diagnostics.py | 1 + roborock/map/b01_grid_layers.py | 52 ------------------------ roborock/map/b01_q10_map_parser.py | 2 + roborock/map/b01_q10_render.py | 10 +++-- tests/devices/traits/b01/q10/test_map.py | 6 ++- tests/map/test_b01_grid_layers.py | 25 ------------ tests/map/test_b01_q10_map_parser.py | 11 +++-- tests/map/test_b01_q10_render.py | 16 +++++++- tests/protocols/test_b01_q10_protocol.py | 2 +- tests/test_diagnostics.py | 9 +++- 10 files changed, 45 insertions(+), 89 deletions(-) diff --git a/roborock/diagnostics.py b/roborock/diagnostics.py index 0c761572e..26d1e89c7 100644 --- a/roborock/diagnostics.py +++ b/roborock/diagnostics.py @@ -106,6 +106,7 @@ def reset(self) -> None: "ipAddress", "ipAdress", "wifiName", + "rawName", "lat", "long", } diff --git a/roborock/map/b01_grid_layers.py b/roborock/map/b01_grid_layers.py index 494e48d75..86c826128 100644 --- a/roborock/map/b01_grid_layers.py +++ b/roborock/map/b01_grid_layers.py @@ -220,58 +220,6 @@ def solve_calibration( return best[1] -def solve_calibration_with_origin( - layers: GridLayers, - points: list[tuple[float, float]], - origin: tuple[float, float], - *, - resolutions: Iterable[float], - y_signs: Iterable[int] = (1, -1), - min_on_floor: float = 0.5, -) -> GridCalibration | None: - """Fit resolution + Y orientation around a *known* grid-pixel origin. - - Unlike :func:`solve_calibration`, the pixel origin ``(ox, oy)`` is fixed -- - e.g. read straight from the Q10 grid-frame header -- so this only sweeps the - candidate ``resolutions`` and ``y_signs`` and keeps the placement landing the - most ``points`` on floor. With the expensive 2-D offset slide gone, far fewer - points are needed to confirm the fit, so it works from a short path rather - than a dense clean. Returns ``None`` if no candidate lands a ``min_on_floor`` - fraction of points on floor (e.g. the origin or points are bogus). - """ - if not points: - return None - w, h = layers.width, layers.height - ox, oy = origin - cells = _calibration_cells(layers) - - best: tuple[float, GridCalibration] | None = None - for resolution in resolutions: - if resolution <= 0: - continue - for y_sign in y_signs: - on_floor = 0 - blocked = 0 - for x, y in points: - px = int(x / resolution + ox) - py = int(oy - y_sign * y / resolution) - if not (0 <= px < w and 0 <= py < h): - blocked += 1 - continue - cell_type = cells[py * w + px] - if cell_type == _CalibrationCell.FLOOR: - on_floor += 1 - elif cell_type == _CalibrationCell.BLOCKED: - blocked += 1 - score = on_floor - 1.5 * blocked - if best is None or score > best[0]: - best = (score, GridCalibration(float(resolution), float(ox), float(oy), y_sign)) - - if best is None or best[0] < len(points) * min_on_floor: - return None - return best[1] - - def decompose_grid( width: int, height: int, diff --git a/roborock/map/b01_q10_map_parser.py b/roborock/map/b01_q10_map_parser.py index c715007e7..13f920be6 100644 --- a/roborock/map/b01_q10_map_parser.py +++ b/roborock/map/b01_q10_map_parser.py @@ -119,6 +119,8 @@ class Q10Room(RoborockBase): @property def name(self) -> str: """User friendly room name (firmware ``rr_`` defaults are normalized).""" + if not self.raw_name.startswith("rr_"): + return self.raw_name return self.raw_name.removeprefix("rr_").replace("_", " ").strip().title() diff --git a/roborock/map/b01_q10_render.py b/roborock/map/b01_q10_render.py index e0ec5388d..e512002ae 100644 --- a/roborock/map/b01_q10_render.py +++ b/roborock/map/b01_q10_render.py @@ -28,6 +28,7 @@ from roborock.exceptions import RoborockException from .b01_grid_layers import ( + LAYER_FLOOR, GridCalibration, GridLayers, solve_calibration, @@ -43,6 +44,7 @@ Q10Point, Q10Room, Q10TracePacket, + classify_q10_cell, erased_packet, ) from .b01_q10_overlays import ZONE_TYPE_NO_GO, ZONE_TYPE_NO_MOP, Q10Zone @@ -186,11 +188,11 @@ def _trace_calibration_from_header_metadata( origin_y=origin[1], y_sign=1, ) - return calibration if _trace_projects_onto_floor(packet.layers, points, calibration) else None + return calibration if _trace_projects_onto_floor(packet, points, calibration) else None def _trace_projects_onto_floor( - layers: GridLayers, + packet: Q10MapPacket, points: Sequence[Q10Point], calibration: GridCalibration, ) -> bool: @@ -203,9 +205,9 @@ def _trace_projects_onto_floor( for point in sample: pixel_x, pixel_y = calibration.world_to_pixel(point.x, point.y) column, row = math.floor(pixel_x), math.floor(pixel_y) - if not (0 <= column < layers.width and 0 <= row < layers.height): + if not (0 <= column < packet.width and 0 <= row < packet.height): continue - if layers.cell_class(layers.grid[row * layers.width + column]) == "floor": + if classify_q10_cell(packet.grid[row * packet.width + column]) == LAYER_FLOOR: on_floor += 1 return on_floor >= len(sample) * 0.5 diff --git a/tests/devices/traits/b01/q10/test_map.py b/tests/devices/traits/b01/q10/test_map.py index 756bff96b..2f44e2f95 100644 --- a/tests/devices/traits/b01/q10/test_map.py +++ b/tests/devices/traits/b01/q10/test_map.py @@ -86,7 +86,7 @@ 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 {room.id: room.name for room in trait.rooms} == {2: "Living Room", 3: "bedroom"} assert len(updates) == 1 @@ -235,7 +235,7 @@ async def test_subscribe_loop_routes_map_push( message_queue.put_nowait(parse_map_packet(FIXTURE.read_bytes())) await _wait_for(lambda: q10_api.map.image_content is not None) - assert {room.id: room.name for room in q10_api.map.rooms} == {2: "Living Room", 3: "Bedroom"} + assert {room.id: room.name for room in q10_api.map.rooms} == {2: "Living Room", 3: "bedroom"} async def test_subscribe_loop_routes_trace_push( @@ -659,6 +659,8 @@ def test_docking_clears_room_and_ignores_late_fixture_trace(status: YXDeviceStat assert trait.current_room is None trait.update_from_trace_packet(trace) assert trait.current_room == packet.rooms[1] + + async def test_create_forwards_obstacle_drawables_to_rendering(fake_channel: FakeB01Q10Channel) -> None: """The public Q10 factory config changes the rendered image.""" enabled = create(fake_channel, map_parser_config=B01Q10MapParserConfig(drawables=[Drawable.OBSTACLES])) diff --git a/tests/map/test_b01_grid_layers.py b/tests/map/test_b01_grid_layers.py index 85ab14f70..9f543e91b 100644 --- a/tests/map/test_b01_grid_layers.py +++ b/tests/map/test_b01_grid_layers.py @@ -12,7 +12,6 @@ GridCalibration, decompose_grid, solve_calibration, - solve_calibration_with_origin, ) @@ -125,27 +124,3 @@ def test_solve_calibration_returns_none_when_unfittable() -> None: # Points so far apart no resolution keeps them on the 6x6 floor block. points = [(0.0, 0.0), (1000.0, 0.0), (0.0, 1000.0)] assert solve_calibration(layers, points, resolutions=[2.0]) is None - - -def test_solve_calibration_with_origin_fits_resolution_from_short_path() -> None: - """With a known origin only resolution is fit, so a tiny path suffices.""" - layers = _floor_block_layers() - true = GridCalibration(2.0, 3.0, 8.0, 1) - points = [true.pixel_to_world(px, py) for px, py in [(4, 7), (6, 5)]] # two points - cal = solve_calibration_with_origin(layers, points, (true.origin_x, true.origin_y), resolutions=[1.0, 2.0, 3.0]) - assert cal is not None - assert (cal.resolution, cal.origin_x, cal.origin_y, cal.y_sign) == (2.0, 3.0, 8.0, 1) - - -def test_solve_calibration_with_origin_returns_none_off_floor() -> None: - """A wrong origin that lands the path off floor is rejected, not forced.""" - layers = _floor_block_layers() - true = GridCalibration(2.0, 3.0, 8.0, 1) - points = [true.pixel_to_world(px, py) for px, py in [(4, 7), (6, 5)]] - # Origin shoved into the background corner: every point lands off floor. - assert solve_calibration_with_origin(layers, points, (0.0, 0.0), resolutions=[2.0]) is None - - -def test_solve_calibration_with_origin_returns_none_without_points() -> None: - layers = _floor_block_layers() - assert solve_calibration_with_origin(layers, [], (3.0, 8.0), resolutions=[2.0]) is None diff --git a/tests/map/test_b01_q10_map_parser.py b/tests/map/test_b01_q10_map_parser.py index 536b078fd..6d2c55376 100644 --- a/tests/map/test_b01_q10_map_parser.py +++ b/tests/map/test_b01_q10_map_parser.py @@ -153,7 +153,7 @@ def test_packet_layers_decompose_q10_fixture() -> None: """The Q10 synthetic fixture splits into floor + per-room layers.""" layers = parse_map_packet(_payload()).layers assert layers.class_counts.get(LAYER_FLOOR) == 26 - assert {room.id: room.name for room in layers.rooms} == {2: "Living Room", 3: "Bedroom"} + assert {room.id: room.name for room in layers.rooms} == {2: "Living Room", 3: "bedroom"} living = layers.render_room(2, (255, 0, 0, 255)) image = Image.open(io.BytesIO(living)) @@ -202,9 +202,9 @@ def test_parse_map_packet_dimensions_straddling_256() -> None: def test_room_name_normalization() -> None: - """Firmware ``rr_`` default names are normalized; custom names are titled.""" + """Firmware ``rr_`` defaults are normalized; custom names stay verbatim.""" assert Q10Room(id=2, raw_name="rr_living_room", pixel_value=8, pixel_count=9).name == "Living Room" - assert Q10Room(id=3, raw_name="bedroom", pixel_value=12, pixel_count=9).name == "Bedroom" + assert Q10Room(id=3, raw_name="bedroom", pixel_value=12, pixel_count=9).name == "bedroom" def test_room_pixel_count_matches_grid() -> None: @@ -627,3 +627,8 @@ def test_maximum_counted_obstacle_table_is_bounded() -> None: packet = parse_map_packet(FIXTURE.read_bytes() + _carpet_tail(8, 6, bytes(48)) + table) assert packet.obstacles == [Q10Obstacle(-10, 20)] * 255 assert packet.skip_cleaning_points == [] + + +@pytest.mark.parametrize("label", ["Salle d’eau", "Owner's bedroom", "lower_case", " Study ", ""]) +def test_custom_room_name_preserved(label: str) -> None: + assert Q10Room(id=3, raw_name=label, pixel_value=12, pixel_count=9).name == label diff --git a/tests/map/test_b01_q10_render.py b/tests/map/test_b01_q10_render.py index 3e27b35bc..04a9da1c6 100644 --- a/tests/map/test_b01_q10_render.py +++ b/tests/map/test_b01_q10_render.py @@ -149,7 +149,7 @@ def test_map_data_from_packet_populates_room_names_and_carpet_map() -> None: carpet[20] = 3 packet = replace(_packet(), carpet_mask=bytes(carpet)) map_data = _map_data_from_packet(packet, config=CONFIG) - assert map_data.additional_parameters["room_names"] == {2: "Living Room", 3: "Bedroom"} + assert map_data.additional_parameters["room_names"] == {2: "Living Room", 3: "bedroom"} assert map_data.carpet_map == {10, 20} @@ -386,6 +386,8 @@ def test_current_room_requires_pose_calibration_and_matching_metadata() -> None: assert resolve_q10_current_room(packet, Q10TracePacket(points=[Q10Point(100, 60)])) is None packet = replace(packet, header_calibration=HEADER, rooms=[]) assert resolve_q10_current_room(packet, Q10TracePacket(points=[Q10Point(100, 60)])) is None + + @pytest.mark.parametrize("origin", [(0, 500), (100, 600)]) def test_render_obstacle_at_header_anchored_position(origin: tuple[int, int]) -> None: """A marker is placed using 50 raw units per pixel, with a top-down image.""" @@ -430,3 +432,15 @@ def render(drawables: list[Drawable], source: Q10MapPacket = packet) -> bytes: assert render([Drawable.OBSTACLES]) != render([]) assert render([]) == render([], replace(packet, obstacles=[])) assert render([Drawable.OBSTACLES], replace(packet, obstacles=[])) == render([], replace(packet, obstacles=[])) + + +def test_resolve_current_room_valid_header_avoids_grid_decomposition(monkeypatch: pytest.MonkeyPatch) -> None: + """Reading the current room with a valid header only samples raw grid cells.""" + packet = replace(_packet(), header_calibration=HEADER) + trace = Q10TracePacket(points=[Q10Point(100, 60)]) + + def fail_decomposition(*args: object, **kwargs: object) -> None: + pytest.fail("Current-room lookup decomposed the full occupancy grid") + + monkeypatch.setattr("roborock.map.b01_q10_map_parser.decompose_grid", fail_decomposition) + assert resolve_q10_current_room(packet, trace) == packet.rooms[1] diff --git a/tests/protocols/test_b01_q10_protocol.py b/tests/protocols/test_b01_q10_protocol.py index 8fd745e09..ebd6b15bd 100644 --- a/tests/protocols/test_b01_q10_protocol.py +++ b/tests/protocols/test_b01_q10_protocol.py @@ -141,7 +141,7 @@ def test_decode_message_map_packet() -> None: message = _message(MAP_FIXTURE.read_bytes(), RoborockMessageProtocol.MAP_RESPONSE) decoded = decode_message(message) assert isinstance(decoded, Q10MapPacket) - assert {room.id: room.name for room in decoded.rooms} == {2: "Living Room", 3: "Bedroom"} + assert {room.id: room.name for room in decoded.rooms} == {2: "Living Room", 3: "bedroom"} @pytest.mark.parametrize( diff --git a/tests/test_diagnostics.py b/tests/test_diagnostics.py index 2fe8aff56..5de9a6950 100644 --- a/tests/test_diagnostics.py +++ b/tests/test_diagnostics.py @@ -2,7 +2,7 @@ import pytest -from roborock.diagnostics import Diagnostics, redact_device_uid, redact_topic_name +from roborock.diagnostics import REDACTED, Diagnostics, redact_device_data, redact_device_uid, redact_topic_name def test_empty_diagnostics(): @@ -99,3 +99,10 @@ def test_redact_device(duid: str, expected: str) -> None: redacted = redact_device_uid(duid) assert redacted == expected + + +def test_redact_q10_raw_room_name() -> None: + """Custom Q10 room labels are private in exported diagnostics.""" + assert redact_device_data( + {"map": {"rooms": [{"rawName": "Owner’s bedroom"}], "currentRoom": {"rawName": "Salle d’eau"}}} + ) == {"map": {"rooms": [{"rawName": REDACTED}], "currentRoom": {"rawName": REDACTED}}}