feat: expose Q10 current cleaning room

roborock/devices/traits/b01/q10/map.py

@@ -31,7 +31,7 @@
     Q10TracePacket,
 )
 from roborock.map.b01_q10_overlays import Q10RestrictedZone, Q10VirtualWall, 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 roborock.protocols.b01_q10_protocol import encode_restricted_zones, encode_room_name, encode_virtual_walls
 
 from .command import CommandTrait
@@ -212,6 +212,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
@@ -220,12 +231,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()
@@ -250,13 +264,15 @@ 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],
             "robotPosition": (
                 {"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)

roborock/diagnostics.py

@@ -106,6 +106,7 @@ def reset(self) -> None:
     "ipAddress",
     "ipAdress",
     "wifiName",
+    "rawName",
     "lat",
     "long",
 }

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,

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()
 
 

roborock/map/b01_q10_render.py

@@ -28,10 +28,10 @@
 from roborock.exceptions import RoborockException
 
 from .b01_grid_layers import (
+    LAYER_FLOOR,
     GridCalibration,
     GridLayers,
     solve_calibration,
-    solve_calibration_with_origin,
 )
 from .b01_q10_map_parser import (
     _BACKGROUND_VALUE,
@@ -41,7 +41,10 @@
     Q10EraseZone,
     Q10HistoricalTracePacket,
     Q10MapPacket,
+    Q10Point,
+    Q10Room,
     Q10TracePacket,
+    classify_q10_cell,
     erased_packet,
 )
 from .b01_q10_overlays import ZONE_TYPE_NO_GO, ZONE_TYPE_NO_MOP, Q10Zone
@@ -62,13 +65,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,
@@ -157,30 +158,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, points, calibration) else None
+
+
+def _trace_projects_onto_floor(
+    packet: Q10MapPacket,
+    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 < packet.width and 0 <= row < packet.height):
+            continue
+        if classify_q10_cell(packet.grid[row * packet.width + column]) == LAYER_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(

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,