refactor(map): consolidate Q10 render config and grid constants

roborock/devices/device_manager.py

@@ -19,7 +19,6 @@
 from roborock.devices.device import DeviceReadyCallback, RoborockDevice
 from roborock.diagnostics import Diagnostics, redact_device_data
 from roborock.exceptions import RoborockException
-from roborock.map.b01_q10_map_parser import B01Q10MapParserConfig
 from roborock.map.map_parser import MapParserConfig
 from roborock.mqtt.roborock_session import create_lazy_mqtt_session
 from roborock.mqtt.session import MqttSession, SessionUnauthorizedHook
@@ -263,16 +262,7 @@ def device_creator(home_data: HomeData, device: HomeDataDevice, product: HomeDat
                 if "ss" in model_part:
                     b01_q10_channel = create_b01_q10_channel(mqtt_channel)
                     channel = b01_q10_channel
-                    trait = b01.q10.create(
-                        channel,
-                        map_parser_config=(
-                            B01Q10MapParserConfig(
-                                map_scale=map_parser_config.map_scale, drawables=map_parser_config.drawables
-                            )
-                            if map_parser_config
-                            else None
-                        ),
-                    )
+                    trait = b01.q10.create(channel, map_parser_config=map_parser_config)
                 elif "sc" in model_part:
                     # Q7 devices start with 'sc' in their model naming.
                     b01_q7_channel = create_b01_q7_channel(device, product, mqtt_channel)

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

@@ -8,7 +8,8 @@
 from roborock.data.containers import RoborockBase
 from roborock.devices.rpc.b01_q10_channel import B01Q10Channel
 from roborock.devices.traits import Trait
-from roborock.map.b01_q10_map_parser import B01Q10MapParserConfig, Q10MapPacket, Q10TracePacket
+from roborock.map.b01_q10_map_parser import Q10MapPacket, Q10TracePacket
+from roborock.map.map_parser import MapParserConfig
 from roborock.protocols.b01_q10_protocol import Q10DpsUpdate, Q10Message
 
 from .button_light import ButtonLightTrait
@@ -92,7 +93,12 @@ class Q10PropertiesApi(Trait):
     clean_history: CleanHistoryTrait
     """Trait for fetching the device clean-record history (``dpCleanRecord``)."""
 
-    def __init__(self, channel: B01Q10Channel, *, map_parser_config: B01Q10MapParserConfig | None = None) -> None:
+    def __init__(
+        self,
+        channel: B01Q10Channel,
+        *,
+        map_parser_config: MapParserConfig | None = None,
+    ) -> None:
         """Initialize the B01Props API."""
         self._channel = channel
         self.command = CommandTrait(channel)
@@ -179,6 +185,10 @@ def as_dict(self) -> dict[str, Any]:
         return result
 
 
-def create(channel: B01Q10Channel, *, map_parser_config: B01Q10MapParserConfig | None = None) -> Q10PropertiesApi:
+def create(
+    channel: B01Q10Channel,
+    *,
+    map_parser_config: MapParserConfig | None = None,
+) -> Q10PropertiesApi:
     """Create traits for B01 devices."""
     return Q10PropertiesApi(channel, map_parser_config=map_parser_config)

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

@@ -24,14 +24,14 @@
 from roborock.devices.traits.common import DpsDataConverter, TraitUpdateListener
 from roborock.exceptions import RoborockException
 from roborock.map.b01_q10_map_parser import (
-    B01Q10MapParserConfig,
     Q10MapPacket,
     Q10Point,
     Q10Room,
     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.map_parser import MapParserConfig
 from roborock.protocols.b01_q10_protocol import encode_restricted_zones, encode_room_name, encode_virtual_walls
 
 from .command import CommandTrait
@@ -101,10 +101,10 @@ def __init__(
         maps: MapsTrait,
         command: CommandTrait,
         *,
-        map_parser_config: B01Q10MapParserConfig | None = None,
+        map_parser_config: MapParserConfig | None = None,
     ) -> None:
         TraitUpdateListener.__init__(self, logger=_LOGGER)
-        self._config = map_parser_config or B01Q10MapParserConfig()
+        self._config = map_parser_config or MapParserConfig()
         self._map_dps = map_dps
         self._maps = maps
         self._command = command

roborock/map/b01_q10_map_parser.py

@@ -25,8 +25,6 @@
 import struct
 from dataclasses import dataclass, field, replace
 
-from vacuum_map_parser_base.config.drawable import Drawable
-
 from roborock.data.b01_q10.b01_q10_containers import Q10RoborockPoint
 from roborock.data.code_mappings import RoborockEnum
 from roborock.data.containers import RoborockBase
@@ -45,19 +43,21 @@
 # eyeballed): 243 is the background outside the home (~half the grid), 240 is
 # scanned floor not yet assigned to a room, other values >= 240 are walls, and
 # 0 < value < 240 are per-room floor cells (value == room_id * 4).
-_BACKGROUND_VALUE = 243
-_UNSEGMENTED_FLOOR_VALUE = 240
+Q10_BACKGROUND_VALUE = 243
+Q10_UNSEGMENTED_FLOOR_VALUE = 240
+# Grid cell values >= this are walls / borders rather than room segments.
+Q10_WALL_THRESHOLD = 240
 
 
 def classify_q10_cell(value: int) -> str:
     """Map a Q10 grid cell value to a canonical layer class."""
     if value == 0:
         return LAYER_UNKNOWN
-    if value == _BACKGROUND_VALUE:
+    if value == Q10_BACKGROUND_VALUE:
         return LAYER_BACKGROUND
-    if value == _UNSEGMENTED_FLOOR_VALUE:
+    if value == Q10_UNSEGMENTED_FLOOR_VALUE:
         return LAYER_FLOOR
-    if value >= _WALL_THRESHOLD:
+    if value >= Q10_WALL_THRESHOLD:
         return LAYER_WALL
     return LAYER_FLOOR
 
@@ -103,9 +103,6 @@ def classify_q10_cell(value: int) -> str:
 # although they use different coordinate frames as documented above.
 _HEADER_UNITS_PER_PIXEL = 10
 
-# Grid cell values >= this are walls / borders rather than room segments.
-_WALL_THRESHOLD = 240
-
 
 @dataclass
 class Q10Room(RoborockBase):
@@ -783,7 +780,7 @@ def _parse_clean_record_trace(
     )
 
 
-def erased_packet(packet: "Q10MapPacket", cells: set[int]) -> "Q10MapPacket":
+def erased_packet(packet: Q10MapPacket, cells: set[int]) -> Q10MapPacket:
     """Return a copy of ``packet`` with ``cells`` (grid indices) set to background.
 
     Used to apply the app's erase zones: cells inside an erase rectangle are blanked
@@ -794,15 +791,5 @@ def erased_packet(packet: "Q10MapPacket", cells: set[int]) -> "Q10MapPacket":
     grid = bytearray(packet.grid)
     for cell in cells:
         if 0 <= cell < len(grid):
-            grid[cell] = _BACKGROUND_VALUE
+            grid[cell] = Q10_BACKGROUND_VALUE
     return replace(packet, grid=bytes(grid))
-
-
-@dataclass
-class B01Q10MapParserConfig:
-    """Configuration for the Q10 map parser."""
-
-    map_scale: int = 4
-    """Scale factor for the rendered map image."""
-    drawables: list[Drawable] | None = None
-    """Enabled overlays, or ``None`` for the Q10 defaults."""

roborock/map/b01_q10_render.py

@@ -34,18 +34,17 @@
     solve_calibration_with_origin,
 )
 from .b01_q10_map_parser import (
-    _BACKGROUND_VALUE,
-    _UNSEGMENTED_FLOOR_VALUE,
-    _WALL_THRESHOLD,
-    B01Q10MapParserConfig,
+    Q10_BACKGROUND_VALUE,
+    Q10_UNSEGMENTED_FLOOR_VALUE,
+    Q10_WALL_THRESHOLD,
     Q10EraseZone,
     Q10HistoricalTracePacket,
     Q10MapPacket,
     Q10TracePacket,
     erased_packet,
 )
 from .b01_q10_overlays import ZONE_TYPE_NO_GO, ZONE_TYPE_NO_MOP, Q10Zone
-from .map_parser import DEFAULT_DRAWABLES, MapParserConfig, _create_image_generator
+from .map_parser import MapParserConfig, _create_image_generator
 from .room_colors import adjacency_aware_room_colors
 
 # Path-units-per-pixel candidates for calibration. A dense ss07 path lands a
@@ -78,7 +77,6 @@
     Drawable.VACUUM_POSITION,
     Drawable.VIRTUAL_WALLS,
 }
-_Q10_DRAWABLES = [drawable for drawable in DEFAULT_DRAWABLES if drawable in _Q10_DRAWABLE_TYPES]
 
 
 @dataclass(frozen=True)
@@ -94,7 +92,7 @@ def render_q10_map(
     trace: Q10TracePacket | Q10HistoricalTracePacket | None,
     overlays: Q10MapOverlays,
     *,
-    config: B01Q10MapParserConfig,
+    config: MapParserConfig,
     robot_at_dock: bool = False,
 ) -> bytes:
     """Compose the latest map, trace and DPS inputs into one PNG image.
@@ -119,19 +117,21 @@ def render_q10_map(
     if map_data.image is None:
         raise RoborockException("Failed to render Q10 map image")
 
-    has_drawables = _place_obstacles(map_data, packet)
-    if trace_calibration is not None and trace is not None:
-        charger_heading = packet.header_calibration.charger_phi if packet.header_calibration is not None else None
-        _place_trace(map_data, trace_calibration, trace, charger_heading=charger_heading)
-        has_drawables = True
-    has_drawables = _place_charger_from_header(map_data, packet) or has_drawables
-    if robot_at_dock:
-        has_drawables = _place_docked_robot(map_data) or has_drawables
-    if vector_calibration is not None:
-        _place_overlays(map_data, vector_calibration, overlays)
-        has_drawables = has_drawables or bool(map_data.no_go_areas or map_data.no_mopping_areas or map_data.walls)
-    if has_drawables:
-        return _draw_map_content(map_data, config=config)
+    drawables = [drawable for drawable in config.drawables if drawable in _Q10_DRAWABLE_TYPES]
+    if drawables:
+        has_drawables = _place_obstacles(map_data, packet)
+        if trace_calibration is not None and trace is not None:
+            charger_heading = packet.header_calibration.charger_phi if packet.header_calibration is not None else None
+            _place_trace(map_data, trace_calibration, trace, charger_heading=charger_heading)
+            has_drawables = True
+        has_drawables = _place_charger_from_header(map_data, packet) or has_drawables
+        if robot_at_dock:
+            has_drawables = _place_docked_robot(map_data) or has_drawables
+        if vector_calibration is not None:
+            _place_overlays(map_data, vector_calibration, overlays)
+            has_drawables = has_drawables or bool(map_data.no_go_areas or map_data.no_mopping_areas or map_data.walls)
+        if has_drawables:
+            return _draw_map_content(map_data, config=config, drawables=drawables)
 
     buffer = io.BytesIO()
     map_data.image.data.save(buffer, format="PNG")
@@ -341,17 +341,13 @@ def to_area(zone: Q10Zone) -> Area | None:
 def _draw_map_content(
     map_data: MapData,
     *,
-    config: B01Q10MapParserConfig,
+    config: MapParserConfig,
+    drawables: list[Drawable],
 ) -> bytes:
     """Draw Q10 content with the shared V1 image generator."""
     if map_data.image is None:
         raise RoborockException("Failed to render Q10 map image")
-    drawables = (
-        _Q10_DRAWABLES
-        if config.drawables is None
-        else [drawable for drawable in config.drawables if drawable in _Q10_DRAWABLE_TYPES]
-    )
-    generator = _create_image_generator(MapParserConfig(map_scale=config.map_scale), drawables=drawables)
+    generator = _create_image_generator(config, drawables=drawables)
     generator.draw_map(map_data)
     buffer = io.BytesIO()
     map_data.image.data.save(buffer, format="PNG")
@@ -361,10 +357,10 @@ def _draw_map_content(
 def _map_data_from_packet(
     packet: Q10MapPacket,
     *,
-    config: B01Q10MapParserConfig,
+    config: MapParserConfig,
 ) -> MapData:
     """Build MapData with a rendered base image, without PNG serialization."""
-    image = _render_base_grid(packet, scale=config.map_scale)
+    image = _render_base_grid(packet, config=config)
 
     map_data = MapData()
     map_data.image = ImageData(
@@ -382,29 +378,28 @@ def _map_data_from_packet(
         # overlays vertically.
         img_transformation=lambda p: Point(p.x, packet.height - p.y - 1, p.a),
     )
-    room_names = {room.id: room.name for room in packet.rooms}
-    if room_names:
-        map_data.additional_parameters["room_names"] = room_names
-
-    # Carpet cells are flat grid indices (y*width + x) in the same top-down
-    # pixel space as the rendered raster, so they line up with the image.
-    if packet.carpet_mask is not None:
-        map_data.carpet_map = {i for i, value in enumerate(packet.carpet_mask) if value}
-
     return map_data
 
 
-def _render_base_grid(packet: Q10MapPacket, *, scale: int) -> Image.Image:
+def _render_base_grid(packet: Q10MapPacket, *, config: MapParserConfig) -> Image.Image:
     """Render the Q10 grid with the V1 map palette."""
-    palette = _build_palette(packet.grid, packet.width)
+    palette = _build_palette(
+        packet.grid,
+        packet.width,
+        show_walls=config.show_walls,
+        show_rooms=config.show_rooms,
+    )
     flat_palette = [channel for color in palette for channel in color]
     p_img = Image.frombytes("P", (packet.width, packet.height), packet.grid)
     p_img.putpalette(flat_palette, rawmode="RGBA")
     img = p_img.convert("RGBA")
     # The ss07 grid is stored top-down (row 0 = top of the home), so it is
     # rendered as-is -- unlike the V1/Q7 convention, no vertical flip.
-    if scale > 1:
-        img = img.resize((packet.width * scale, packet.height * scale), resample=Image.Resampling.NEAREST)
+    if config.map_scale > 1:
+        img = img.resize(
+            (packet.width * config.map_scale, packet.height * config.map_scale),
+            resample=Image.Resampling.NEAREST,
+        )
     return img
 
 
@@ -413,27 +408,34 @@ def _opaque(color: tuple[int, ...]) -> tuple[int, int, int, int]:
     return (color[0], color[1], color[2], color[3] if len(color) == 4 else 255)
 
 
-def _build_palette(grid: bytes, width: int) -> list[tuple[int, int, int, int]]:
+def _build_palette(
+    grid: bytes,
+    width: int,
+    *,
+    show_walls: bool = True,
+    show_rooms: bool = True,
+) -> list[tuple[int, int, int, int]]:
     """Map Q10 cells onto the same colors used by the V1 map renderer."""
 
     def room_id(value: int) -> int | None:
-        return max(1, value // 4) if 0 < value < _WALL_THRESHOLD else None
+        return max(1, value // 4) if 0 < value < Q10_WALL_THRESHOLD else None
 
     colors = ColorsPalette()
-    room_colors = adjacency_aware_room_colors(
-        grid,
-        width,
-        colors,
-        room_id,
-    )
     outside = (0, 0, 0, 0)
     palette = [outside] * 256
-    for value in {value for value in grid if 0 < value < _WALL_THRESHOLD}:
-        palette[value] = _opaque(room_colors[max(1, value // 4)])
-    wall = _opaque(colors.get_color(SupportedColor.GREY_WALL))
-    for value in range(_WALL_THRESHOLD, 256):
-        palette[value] = wall
-    palette[_UNSEGMENTED_FLOOR_VALUE] = _opaque(colors.get_color(SupportedColor.MAP_INSIDE))
-    palette[_BACKGROUND_VALUE] = outside
-    palette[0] = outside
+    if show_rooms:
+        room_colors = adjacency_aware_room_colors(
+            grid,
+            width,
+            colors,
+            room_id,
+        )
+        for value in {value for value in grid if 0 < value < Q10_WALL_THRESHOLD}:
+            palette[value] = _opaque(room_colors[max(1, value // 4)])
+        palette[Q10_UNSEGMENTED_FLOOR_VALUE] = _opaque(colors.get_color(SupportedColor.MAP_INSIDE))
+    if show_walls:
+        wall = _opaque(colors.get_color(SupportedColor.GREY_WALL))
+        for value in range(Q10_WALL_THRESHOLD, 256):
+            if value not in (Q10_UNSEGMENTED_FLOOR_VALUE, Q10_BACKGROUND_VALUE):
+                palette[value] = wall
     return palette