refactor(q7): address maintainer review follow-ups

Author: arduano

.pre-commit-config.yaml

@@ -38,10 +38,11 @@ repos:
         rev: v0.13.2
         hooks:
             -   id: ruff-format
-                exclude: ^roborock/map/proto/.*_pb2\.py$
+                args:
+                    - --force-exclude
             -   id: ruff
-                exclude: ^roborock/map/proto/.*_pb2\.py$
                 args:
+                    - --force-exclude
                     - --fix
     -   repo: https://github.com/pre-commit/mirrors-mypy
         rev: v1.7.1

pyproject.toml

@@ -98,6 +98,7 @@ major_tags= ["refactor"]
 lint.ignore = ["F403", "E741"]
 lint.select=["E", "F", "UP", "I"]
 line-length = 120
+extend-exclude = ["roborock/map/proto/*_pb2.py"]
 
 [tool.ruff.lint.per-file-ignores]
 "*/__init__.py" = ["F401"]

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

@@ -57,6 +57,9 @@ def __init__(self, channel: MqttChannel, *, device: HomeDataDevice, product: Hom
         self._device = device
         self._product = product
 
+        if not device.sn or not product.model:
+            raise ValueError("B01 Q7 map content requires device serial number and product model metadata")
+
         self.clean_summary = CleanSummaryTrait(channel)
         self.map = MapTrait(channel)
         self.map_content = MapContentTrait(

roborock/devices/traits/b01/q7/map_content.py

@@ -55,8 +55,8 @@ def __init__(
         self,
         map_trait: MapTrait,
         *,
-        serial: str | None,
-        model: str | None,
+        serial: str,
+        model: str,
         map_parser_config: B01MapParserConfig | None = None,
     ) -> None:
         super().__init__()
@@ -79,11 +79,6 @@ def parse_map_content(self, response: bytes) -> MapContent:
         This mirrors the v1 trait behavior so cached map payload bytes can be
         reparsed without going back to the device.
         """
-        if not self._serial or not self._model:
-            raise RoborockException(
-                "B01 map parsing requires device serial number and model metadata, but they were missing"
-            )
-
         try:
             parsed_data = self._map_parser.parse(
                 response,

roborock/map/b01_map_parser.py

@@ -18,11 +18,11 @@
 import io
 import zlib
 from dataclasses import dataclass, field
-from typing import Any
+from typing import Protocol, cast
 
 from Crypto.Cipher import AES
 from Crypto.Util.Padding import pad, unpad
-from google.protobuf.message import DecodeError
+from google.protobuf.message import DecodeError, Message
 from PIL import Image
 from vacuum_map_parser_base.config.image_config import ImageConfig
 from vacuum_map_parser_base.map_data import ImageData, MapData
@@ -36,6 +36,72 @@
 _MAP_FILE_FORMAT = "PNG"
 
 
+class _ProtoMessage(Protocol):
+    def HasField(self, field_name: str) -> bool: ...
+
+
+class _ScPointMessage(_ProtoMessage, Protocol):
+    x: float
+    y: float
+
+
+class _ScMapBoundaryInfoMessage(_ProtoMessage, Protocol):
+    mapMd5: str
+    vMinX: int
+    vMaxX: int
+    vMinY: int
+    vMaxY: int
+
+
+class _ScMapExtInfoMessage(_ProtoMessage, Protocol):
+    taskBeginDate: int
+    mapUploadDate: int
+    mapValid: int
+    radian: int
+    force: int
+    cleanPath: int
+    boudaryInfo: _ScMapBoundaryInfoMessage
+    mapVersion: int
+    mapValueType: int
+
+
+class _ScMapHeadMessage(_ProtoMessage, Protocol):
+    mapHeadId: int
+    sizeX: int
+    sizeY: int
+    minX: float
+    minY: float
+    maxX: float
+    maxY: float
+    resolution: float
+
+
+class _ScRoomDataMessage(_ProtoMessage, Protocol):
+    roomId: int
+    roomName: str
+    roomTypeId: int
+    meterialId: int
+    cleanState: int
+    roomClean: int
+    roomCleanIndex: int
+    roomNamePost: _ScPointMessage
+    colorId: int
+    floor_direction: int
+    global_seq: int
+
+
+class _ScMapDataContainerMessage(_ProtoMessage, Protocol):
+    mapData: bytes
+
+
+class _ScMapMessage(_ProtoMessage, Protocol):
+    mapType: int
+    mapExtInfo: _ScMapExtInfoMessage
+    mapHead: _ScMapHeadMessage
+    mapData: _ScMapDataContainerMessage
+    roomDataInfo: list[_ScRoomDataMessage]
+
+
 @dataclass(frozen=True)
 class _ScPoint:
     x: float | None = None
@@ -185,7 +251,7 @@ def _decode_b01_map_payload(raw_payload: bytes, *, serial: str, model: str) -> b
         raise RoborockException("Failed to decode B01 map payload") from err
 
 
-def _parse_proto(blob: bytes, message: Any, *, context: str) -> None:
+def _parse_proto(blob: bytes, message: Message, *, context: str) -> None:
     try:
         message.ParseFromString(blob)
     except DecodeError as err:
@@ -199,14 +265,14 @@ def _decode_map_data_bytes(value: bytes) -> bytes:
         return value
 
 
-def _parse_sc_point(parsed: Any) -> _ScPoint:
+def _parse_sc_point(parsed: _ScPointMessage) -> _ScPoint:
     return _ScPoint(
         x=parsed.x if parsed.HasField("x") else None,
         y=parsed.y if parsed.HasField("y") else None,
     )
 
 
-def _parse_sc_map_boundary_info(parsed: Any) -> _ScMapBoundaryInfo:
+def _parse_sc_map_boundary_info(parsed: _ScMapBoundaryInfoMessage) -> _ScMapBoundaryInfo:
     return _ScMapBoundaryInfo(
         map_md5=parsed.mapMd5 if parsed.HasField("mapMd5") else None,
         v_min_x=parsed.vMinX if parsed.HasField("vMinX") else None,
@@ -216,7 +282,7 @@ def _parse_sc_map_boundary_info(parsed: Any) -> _ScMapBoundaryInfo:
     )
 
 
-def _parse_sc_map_ext_info(parsed: Any) -> _ScMapExtInfo:
+def _parse_sc_map_ext_info(parsed: _ScMapExtInfoMessage) -> _ScMapExtInfo:
     return _ScMapExtInfo(
         task_begin_date=parsed.taskBeginDate if parsed.HasField("taskBeginDate") else None,
         map_upload_date=parsed.mapUploadDate if parsed.HasField("mapUploadDate") else None,
@@ -230,7 +296,7 @@ def _parse_sc_map_ext_info(parsed: Any) -> _ScMapExtInfo:
     )
 
 
-def _parse_sc_map_head(parsed: Any) -> _ScMapHead:
+def _parse_sc_map_head(parsed: _ScMapHeadMessage) -> _ScMapHead:
     return _ScMapHead(
         map_head_id=parsed.mapHeadId if parsed.HasField("mapHeadId") else None,
         size_x=parsed.sizeX if parsed.HasField("sizeX") else None,
@@ -243,7 +309,7 @@ def _parse_sc_map_head(parsed: Any) -> _ScMapHead:
     )
 
 
-def _parse_sc_room_data(parsed: Any) -> _ScRoomData:
+def _parse_sc_room_data(parsed: _ScRoomDataMessage) -> _ScRoomData:
     return _ScRoomData(
         room_id=parsed.roomId if parsed.HasField("roomId") else None,
         room_name=parsed.roomName if parsed.HasField("roomName") else None,
@@ -261,8 +327,8 @@ def _parse_sc_room_data(parsed: Any) -> _ScRoomData:
 
 def _parse_scmap_payload(payload: bytes) -> _ScMapPayload:
     """Parse inflated SCMap bytes into typed map metadata."""
-    parsed: Any = getattr(b01_scmap_pb2, "RobotMap")()
-    _parse_proto(payload, parsed, context="B01 SCMap")
+    parsed = cast(_ScMapMessage, getattr(b01_scmap_pb2, "RobotMap")())
+    _parse_proto(payload, cast(Message, parsed), context="B01 SCMap")
 
     map_data = None
     if parsed.HasField("mapData"):

roborock/map/proto/b01_scmap.proto

@@ -1,8 +1,4 @@
-// Source of truth for the B01/Q7 SCMap schema.
-//
-// Regenerate the checked-in Python module after edits with:
-//   python -m grpc_tools.protoc -I. --python_out=. roborock/map/proto/b01_scmap.proto
-//
+// Checked-in B01/Q7 SCMap schema for the generated runtime protobuf module.
 // The generated file `b01_scmap_pb2.py` is checked in for runtime use and should
 // not be edited by hand.
 syntax = "proto2";

tests/devices/traits/b01/q7/test_map_content.py

@@ -57,25 +57,23 @@ def test_q7_map_content_preserves_specific_roborock_errors(q7_api: Q7PropertiesA
             q7_api.map_content.parse_map_content(b"raw")
 
 
-def test_q7_map_content_missing_metadata_fails_lazily(fake_channel: FakeChannel):
+def test_q7_map_content_requires_metadata_at_init(fake_channel: FakeChannel):
     from roborock.data import HomeDataDevice, HomeDataProduct, RoborockCategory
 
-    q7_api = Q7PropertiesApi(
-        cast(MqttChannel, fake_channel),
-        device=HomeDataDevice(
-            duid="abc123",
-            name="Q7",
-            local_key="key123key123key1",
-            product_id="product-id-q7",
-            sn=None,
-        ),
-        product=HomeDataProduct(
-            id="product-id-q7",
-            name="Roborock Q7",
-            model="roborock.vacuum.sc05",
-            category=RoborockCategory.VACUUM,
-        ),
-    )
-
-    with pytest.raises(RoborockException, match="requires device serial number and model metadata"):
-        q7_api.map_content.parse_map_content(b"raw")
+    with pytest.raises(ValueError, match="requires device serial number and product model metadata"):
+        Q7PropertiesApi(
+            cast(MqttChannel, fake_channel),
+            device=HomeDataDevice(
+                duid="abc123",
+                name="Q7",
+                local_key="key123key123key1",
+                product_id="product-id-q7",
+                sn=None,
+            ),
+            product=HomeDataProduct(
+                id="product-id-q7",
+                name="Roborock Q7",
+                model="roborock.vacuum.sc05",
+                category=RoborockCategory.VACUUM,
+            ),
+        )