refactor(map): standardize protocol commands and improve state consistency

- Standardize RoborockCommand enum with missing map edit constants

- Implement additive edit batching in TranslationLayer to prevent state overwrite

- Enhance VirtualState with comprehensive state capture (walls, zones, mop zones)

- Add refresh_base_map to VirtualState for post-sync synchronization

- Fix MapVerifier to accurately validate virtual wall endpoints and directions

- Add map version safety check to CLI to detect state drift

- Improve map_edit_status CLI output with granular edit details

Entire-Checkpoint: f7bfcf7b1097

Author: Coldaine

roborock/cli.py

@@ -182,9 +182,24 @@ def __init__(self):
         self._virtual_states: dict[str, Any] = {}
 
     def get_virtual_state(self, device_id: str, map_data: Any = None) -> Any:
-        """Get or create a VirtualState for a device."""
+        """Get or create a VirtualState for a device with version safety."""
         from roborock.map import CoordinateTransformer, VirtualState
 
+        if device_id in self._virtual_states:
+            state = self._virtual_states[device_id]
+            # If map_data is provided, check for state drift
+            if map_data is not None and state._base_map is not None:
+                # Simple version check: room count or timestamp
+                curr_rooms = set(map_data.rooms.keys()) if map_data.rooms else set()
+                base_rooms = set(state._base_map.rooms.keys()) if state._base_map.rooms else set()
+                
+                if curr_rooms != base_rooms:
+                    click.echo(f"WARNING: Device map for {device_id} has changed. Local edits may be invalid.")
+                    if click.confirm("Clear pending edits and refresh state?"):
+                        del self._virtual_states[device_id]
+                    else:
+                        click.echo("Continuing with potentially stale state (DANGEROUS)")
+            
         if device_id not in self._virtual_states:
             if map_data is None:
                 return None
@@ -1373,17 +1388,18 @@ def _generate_preview(map_data, virtual_state, transformer, output_path: str = "
         draw = ImageDraw.Draw(img)
 
         # Draw each pending edit in red
+        from roborock.map.editor import EditType
         for edit in virtual_state.pending_edits:
-            if edit.edit_type.value == "split_room":
+            if edit.edit_type == EditType.SPLIT_ROOM:
                 # Draw split line in red
                 p1 = transformer.robot_to_image(Point(edit.x1, edit.y1))
                 p2 = transformer.robot_to_image(Point(edit.x2, edit.y2))
                 draw.line([(int(p1.x), int(p1.y)), (int(p2.x), int(p2.y))], fill=(255, 0, 0), width=3)
-            elif edit.edit_type.value in ["virtual_wall", "no_go_zone"]:
+            elif edit.edit_type in [EditType.VIRTUAL_WALL, EditType.NO_GO_ZONE]:
                 # Draw virtual walls/no-go zones in red
                 p1 = transformer.robot_to_image(Point(edit.x1, edit.y1))
                 p2 = transformer.robot_to_image(Point(edit.x2, edit.y2))
-                if edit.edit_type.value == "virtual_wall":
+                if edit.edit_type == EditType.VIRTUAL_WALL:
                     draw.line([(int(p1.x), int(p1.y)), (int(p2.x), int(p2.y))], fill=(255, 0, 0), width=3)
                 else:
                     # No-go zone - draw rectangle
@@ -1865,6 +1881,14 @@ async def add_no_go_zone(ctx, device_id: str, x1: int, y1: int, x2: int, y2: int
 @async_command
 async def map_edit_status(ctx, device_id: str):
     """Show pending edits in the virtual state."""
+    from roborock.map.editor import (
+        MergeRoomsEdit,
+        NoGoZoneEdit,
+        RenameRoomEdit,
+        SplitRoomEdit,
+        VirtualWallEdit,
+    )
+    
     context: RoborockContext = ctx.obj
     virtual_state = context.get_virtual_state(device_id)
 
@@ -1874,7 +1898,19 @@ async def map_edit_status(ctx, device_id: str):
 
     click.echo(f"Pending edits for device {device_id}:")
     for i, edit in enumerate(virtual_state.pending_edits):
-        click.echo(f"  {i+1}. {edit.edit_type.name} (Status: {edit.status.name})")
+        details = ""
+        if isinstance(edit, VirtualWallEdit):
+            details = f"({edit.x1:.0f}, {edit.y1:.0f}) -> ({edit.x2:.0f}, {edit.y2:.0f})"
+        elif isinstance(edit, NoGoZoneEdit):
+            details = f"[{edit.x1:.0f}, {edit.y1:.0f}, {edit.x2:.0f}, {edit.y2:.0f}]"
+        elif isinstance(edit, SplitRoomEdit):
+            details = f"Room {edit.segment_id} @ ({edit.x1:.0f}, {edit.y1:.0f}) -> ({edit.x2:.0f}, {edit.y2:.0f})"
+        elif isinstance(edit, MergeRoomsEdit):
+            details = f"Rooms: {edit.segment_ids}"
+        elif isinstance(edit, RenameRoomEdit):
+            details = f"Room {edit.segment_id}: '{edit.old_name}' -> '{edit.new_name}'"
+            
+        click.echo(f"  {i+1}. {edit.edit_type.name:15} {details} (Status: {edit.status.name})")
 
 
 @session.command()

roborock/map/editor.py

@@ -395,14 +395,39 @@ def __init__(self, map_data: MapData | None = None, transformer: CoordinateTrans
         self._transformer = transformer
         self._edit_stack: list[EditObject] = []
         self._redo_stack: list[EditObject] = []
+        
+        # Capture original state for full physical rollback support
         self._original_room_names: dict[int, str] = {}
-
-        # Capture original room names for rollback
-        if map_data and map_data.rooms:
-            for room_id, room in map_data.rooms.items():
-                if hasattr(room, 'name'):
-                    self._original_room_names[room_id] = room.name
-
+        self._original_walls: list[tuple[float, float, float, float]] = []
+        self._original_no_go_zones: list[tuple[float, float, float, float]] = []
+        self._original_mop_zones: list[tuple[float, float, float, float]] = []
+
+        if map_data:
+            # Capture rooms
+            if map_data.rooms:
+                for room_id, room in map_data.rooms.items():
+                    if hasattr(room, 'name') and room.name:
+                        self._original_room_names[room_id] = room.name
+
+            # Capture walls (if exposed by parser)
+            if hasattr(map_data, 'walls') and map_data.walls:
+                for wall in map_data.walls:
+                    self._original_walls.append((wall.x1, wall.y1, wall.x2, wall.y2))
+
+            # Capture zones (if exposed by parser)
+            if hasattr(map_data, 'no_go_areas') and map_data.no_go_areas:
+                for zone in map_data.no_go_areas:
+                    # Note: we simplify to 4-coord bbox for compatibility
+                    xs = [p.x for p in zone.points]
+                    ys = [p.y for p in zone.points]
+                    self._original_no_go_zones.append((min(xs), min(ys), max(xs), max(ys)))
+
+            # Capture mop zones
+            if hasattr(map_data, 'no_mop_areas') and map_data.no_mop_areas:
+                for zone in map_data.no_mop_areas:
+                    xs = [p.x for p in zone.points]
+                    ys = [p.y for p in zone.points]
+                    self._original_mop_zones.append((min(xs), min(ys), max(xs), max(ys)))
     @property
     def has_pending_edits(self) -> bool:
         """Return True if there are pending edits."""
@@ -492,6 +517,21 @@ def redo(self) -> EditObject | None:
         _LOGGER.debug(f"Redone edit: {edit.edit_id}")
         return edit
 
+    def refresh_base_map(self, map_data: MapData) -> None:
+        """Refresh the base map and coordinate transformer.
+
+        This is used after a physical sync to ensure the virtual state
+        matches the current device state.
+
+        Args:
+            map_data: The fresh map data from device.
+        """
+        from .geometry import CoordinateTransformer
+        
+        self._base_map = map_data
+        self._transformer = CoordinateTransformer.from_map_data(map_data)
+        _LOGGER.debug("Refreshed base map in VirtualState")
+
     def clear(self) -> None:
         """Clear all edits and reset to base state."""
         self._edit_stack.clear()
@@ -541,6 +581,8 @@ def to_dict(self) -> dict[str, Any]:
         return {
             "edits": [e.to_dict() for e in self._edit_stack],
             "original_room_names": self._original_room_names,
+            "original_walls": self._original_walls,
+            "original_no_go_zones": self._original_no_go_zones,
         }
 
     def __len__(self) -> int:

roborock/map/translation.py

@@ -95,12 +95,12 @@ def translate_virtual_wall(self, edit: VirtualWallEdit) -> tuple[str, Any]:
         """
         # V1 uses the full map update approach
         wall_data = [int(edit.x1), int(edit.y1), int(edit.x2), int(edit.y2)]
-        return "set_virtual_wall", [wall_data]
+        return RoborockCommand.SET_VIRTUAL_WALL, [wall_data]
 
     def translate_no_go_zone(self, edit: NoGoZoneEdit) -> tuple[str, Any]:
         """Translate to SET_NO_GO_ZONES command."""
         zone_data = [int(edit.x1), int(edit.y1), int(edit.x2), int(edit.y2)]
-        return "set_no_go_zones", [zone_data]
+        return RoborockCommand.SET_NO_GO_ZONES, [zone_data]
 
     def translate_split_room(self, edit: SplitRoomEdit) -> tuple[str, Any]:
         """Translate to SPLIT_SEGMENT command.
@@ -264,12 +264,110 @@ async def execute_edits(
         additive_edits = virtual_state.get_additive_edits()
         if additive_edits and topology_success:
             _LOGGER.info(f"Stage 3: Executing {len(additive_edits)} additive edits")
-            for edit in additive_edits:
-                result = await self._execute_edit(edit, map_flag)
-                results.append(result)
+            
+            if self._protocol == "v1":
+                # V1 requires batching additive edits as they overwrite the entire state
+                batch_results = await self._execute_v1_additive_batch(virtual_state, additive_edits, map_flag)
+                results.extend(batch_results)
+            else:
+                # B01 might support individual updates (verify)
+                for edit in additive_edits:
+                    result = await self._execute_edit(edit, map_flag)
+                    results.append(result)
 
         return results
 
+    async def _execute_v1_additive_batch(
+        self,
+        virtual_state: VirtualState,
+        edits: list[EditObject],
+        map_flag: int | None,
+    ) -> list[TranslationResult]:
+        """Execute additive edits as a batch for V1 protocol.
+        
+        V1 commands like set_virtual_wall overwrite the entire list,
+        so we must combine new edits with existing state.
+        """
+        results: list[TranslationResult] = []
+        pending_walls = [e for e in edits if isinstance(e, VirtualWallEdit) and e.status == EditStatus.APPLIED]
+        pending_zones = [e for e in edits if isinstance(e, NoGoZoneEdit) and e.status == EditStatus.APPLIED]
+        pending_mop_zones = [e for e in edits if e.edit_type == EditType.MOP_FORBIDDEN_ZONE and e.status == EditStatus.APPLIED]
+        
+        # 1. Handle Virtual Walls
+        if pending_walls:
+            # Combine original walls with new ones
+            all_walls = list(virtual_state._original_walls)
+            for wall in pending_walls:
+                all_walls.append((wall.x1, wall.y1, wall.x2, wall.y2))
+            
+            # Format: [map_flag, [[x1, y1, x2, y2], ...]]
+            params = [list(w) for w in all_walls]
+            if map_flag is not None:
+                params = [map_flag, params]
+            
+            try:
+                _LOGGER.debug(f"Sending V1 wall batch: {params}")
+                await self._command.send(RoborockCommand.SET_VIRTUAL_WALL, params)
+                for edit in pending_walls:
+                    edit.status = EditStatus.SYNCED
+                    results.append(TranslationResult(success=True, edit=edit))
+            except Exception as e:
+                _LOGGER.error(f"Failed to sync wall batch: {e}")
+                for edit in pending_walls:
+                    edit.status = EditStatus.FAILED
+                    results.append(TranslationResult(success=False, edit=edit, error=str(e)))
+
+        # 2. Handle No-Go Zones
+        if pending_zones:
+            # Combine original zones with new ones
+            all_zones = list(virtual_state._original_no_go_zones)
+            for zone in pending_zones:
+                all_zones.append((zone.x1, zone.y1, zone.x2, zone.y2))
+            
+            # Format: [map_flag, [[x1, y1, x2, y2], ...]]
+            params = [list(z) for z in all_zones]
+            if map_flag is not None:
+                params = [map_flag, params]
+                
+            try:
+                _LOGGER.debug(f"Sending V1 zone batch: {params}")
+                await self._command.send(RoborockCommand.SET_NO_GO_ZONES, params)
+                for edit in pending_zones:
+                    edit.status = EditStatus.SYNCED
+                    results.append(TranslationResult(success=True, edit=edit))
+            except Exception as e:
+                _LOGGER.error(f"Failed to sync zone batch: {e}")
+                for edit in pending_zones:
+                    edit.status = EditStatus.FAILED
+                    results.append(TranslationResult(success=False, edit=edit, error=str(e)))
+
+        # 3. Handle Mop Forbidden Zones
+        if pending_mop_zones:
+            # Combine original mop zones with new ones
+            all_mop_zones = list(virtual_state._original_mop_zones)
+            for zone in pending_mop_zones:
+                # Mop zones are typically same 4-coord format
+                if hasattr(zone, 'x1'):
+                    all_mop_zones.append((zone.x1, zone.y1, zone.x2, zone.y2))
+            
+            params = [list(z) for z in all_mop_zones]
+            if map_flag is not None:
+                params = [map_flag, params]
+                
+            try:
+                _LOGGER.debug(f"Sending V1 mop zone batch: {params}")
+                await self._command.send(RoborockCommand.SET_MOP_FORBIDDEN_ZONE, params)
+                for edit in pending_mop_zones:
+                    edit.status = EditStatus.SYNCED
+                    results.append(TranslationResult(success=True, edit=edit))
+            except Exception as e:
+                _LOGGER.error(f"Failed to sync mop zone batch: {e}")
+                for edit in pending_mop_zones:
+                    edit.status = EditStatus.FAILED
+                    results.append(TranslationResult(success=False, edit=edit, error=str(e)))
+                    
+        return results
+
     async def _execute_edit(
         self,
         edit: EditObject,
@@ -500,5 +598,5 @@ async def _repopulate_room_ids(self, virtual_state: VirtualState) -> None:
 
         # 5. Update the base map for the VirtualState so subsequent matches are correct
         # This is important if we're doing multiple rounds or just to keep state consistent.
-        # virtual_state._base_map = new_map (Internal access needed or a public setter)
-        # For now, we've updated the pending edits which is the primary goal for execution.
+        virtual_state.refresh_base_map(new_map)
+        _LOGGER.info("Refreshed VirtualState base map after remapping")

roborock/map/verifier.py

@@ -200,14 +200,20 @@ def _verify_virtual_wall(
         # Check for matching wall (with tolerance)
         tolerance = 100  # 10cm in mm
         for wall in map_data.walls:
-            wall_point1 = Point(wall.x, wall.y)
-            # Wall may be stored as single point with orientation, or two points
-            if self._points_match(Point(edit.x1, edit.y1), wall_point1, tolerance):
+            # Wall has x1, y1, x2, y2 from parser
+            w_p1 = Point(wall.x1, wall.y1)
+            w_p2 = Point(wall.x2, wall.y2)
+            e_p1 = Point(edit.x1, edit.y1)
+            e_p2 = Point(edit.x2, edit.y2)
+            
+            # Check both directions
+            if (self._points_match(e_p1, w_p1, tolerance) and self._points_match(e_p2, w_p2, tolerance)) or \
+               (self._points_match(e_p1, w_p2, tolerance) and self._points_match(e_p2, w_p1, tolerance)):
                 return VerificationResult(
                     verified=True,
                     edit_type="VIRTUAL_WALL",
                     expected=expected,
-                    actual={"x": wall.x, "y": wall.y},
+                    actual={"x1": wall.x1, "y1": wall.y1, "x2": wall.x2, "y2": wall.y2},
                 )
 
         return VerificationResult(

roborock/roborock_typing.py

@@ -208,9 +208,11 @@ class RoborockCommand(str, Enum):
     SET_LAB_STATUS = "set_lab_status"
     SET_LED_STATUS = "set_led_status"
     SET_MAP_BEAUTIFICATION_STATUS = "set_map_beautification_status"
+    SET_MOP_FORBIDDEN_ZONE = "set_mop_forbidden_zone"
     SET_MOP_MODE = "set_mop_mode"
     SET_MOP_MOTOR_STATUS = "set_mop_motor_status"
     SET_MOP_TEMPLATE_ID = "set_mop_template_id"
+    SET_NO_GO_ZONES = "set_no_go_zones"
     SET_OFFLINE_MAP_STATUS = "set_offline_map_status"
     SET_SCENES_SEGMENTS = "set_scenes_segments"
     SET_SCENES_ZONES = "set_scenes_zones"
@@ -221,6 +223,7 @@ class RoborockCommand(str, Enum):
     SET_TIMER = "set_timer"
     SET_TIMEZONE = "set_timezone"
     SET_VALLEY_ELECTRICITY_TIMER = "set_valley_electricity_timer"
+    SET_VIRTUAL_WALL = "set_virtual_wall"
     SET_VOICE_CHAT_VOLUME = "set_voice_chat_volume"
     SET_WASH_DEBUG_PARAMS = "set_wash_debug_params"
     SET_WASH_TOWEL_MODE = "set_wash_towel_mode"