roborock.devices.traits.v1.map_content

  1"""Trait for fetching the map content from Roborock devices."""
  2
  3import asyncio
  4import logging
  5from dataclasses import dataclass
  6
  7from vacuum_map_parser_base.map_data import MapData
  8
  9from roborock.data import RoborockBase
 10from roborock.devices.traits.v1 import common
 11from roborock.exceptions import RoborockParsingException
 12from roborock.map.map_parser import MapParser, MapParserConfig
 13from roborock.roborock_typing import RoborockCommand
 14
 15_LOGGER = logging.getLogger(__name__)
 16
 17_TRUNCATE_LENGTH = 20
 18
 19
 20@dataclass
 21class MapContent(RoborockBase):
 22    """Dataclass representing map content."""
 23
 24    image_content: bytes | None = None
 25    """The rendered image of the map in PNG format."""
 26
 27    map_data: MapData | None = None
 28    """The parsed map data which contains metadata for points on the map."""
 29
 30    raw_api_response: bytes | None = None
 31    """The raw bytes of the map data from the API for caching for future use.
 32
 33    This should be treated as an opaque blob used only internally by this library
 34    to re-parse the map data when needed.
 35    """
 36
 37    def __repr__(self) -> str:
 38        """Return a string representation of the MapContent."""
 39        img = self.image_content
 40        if self.image_content and len(self.image_content) > _TRUNCATE_LENGTH:
 41            img = self.image_content[: _TRUNCATE_LENGTH - 3] + b"..."
 42        return f"MapContent(image_content={img!r}, map_data={self.map_data!r})"
 43
 44
 45class MapContentConverter(common.V1TraitDataConverter):
 46    """Convert map response data to MapContent."""
 47
 48    def __init__(self, map_parser: MapParser) -> None:
 49        """Initialize MapContentConverter."""
 50        self._map_parser = map_parser
 51
 52    def convert(self, response: common.V1ResponseData) -> MapContent:
 53        """Parse the response from the device into a MapContent instance."""
 54        if not isinstance(response, bytes):
 55            raise ValueError(f"Unexpected MapContentTrait response format: {type(response)}")
 56        return self.parse_map_content(response)
 57
 58    def parse_map_content(self, response: bytes) -> MapContent:
 59        """Parse the map content from raw bytes.
 60
 61        This method is exposed so that cached map data can be parsed without
 62        needing to go through the RPC channel.
 63
 64        Args:
 65            response: The raw bytes of the map data from the API.
 66
 67        Returns:
 68            MapContent: The parsed map content.
 69
 70        Raises:
 71            RoborockException: If the map data cannot be parsed.
 72        """
 73        parsed_data = self._map_parser.parse(response)
 74        if parsed_data is None:
 75            raise ValueError("Failed to parse map data")
 76
 77        return MapContent(
 78            image_content=parsed_data.image_content,
 79            map_data=parsed_data.map_data,
 80            raw_api_response=response,
 81        )
 82
 83    async def async_parse_map_content(self, response: bytes) -> MapContent:
 84        """Parse the map content from raw bytes asynchronously.
 85
 86        Args:
 87            response: The raw bytes of the map data from the API.
 88
 89        Returns:
 90            MapContent: The parsed map content.
 91
 92        Raises:
 93            RoborockException: If the map data cannot be parsed.
 94        """
 95        return await asyncio.to_thread(self.parse_map_content, response)
 96
 97
 98@common.map_rpc_channel
 99class MapContentTrait(MapContent, common.V1TraitMixin):
100    """Trait for fetching the map content."""
101
102    command = RoborockCommand.GET_MAP_V1
103    converter: MapContentConverter
104
105    def __init__(self, map_parser_config: MapParserConfig | None = None) -> None:
106        """Initialize MapContentTrait."""
107        super().__init__()
108        self.converter = MapContentConverter(MapParser(map_parser_config or MapParserConfig()))
109
110    async def refresh(self) -> None:
111        """Refresh the contents of this trait without blocking the event loop."""
112        response = await self.rpc_channel.send_command(self.command)
113        if self.raw_api_response is not None and self.raw_api_response == response:
114            return
115        if not isinstance(response, bytes):
116            raise ValueError(f"Unexpected MapContentTrait response format: {type(response)}")
117        try:
118            new_data = await self.converter.async_parse_map_content(response)
119        except (TypeError, ValueError) as err:
120            raise RoborockParsingException(
121                trait_name=type(self).__name__,
122                command=self.command,
123                payload=response,
124                inner_error=err,
125            ) from err
126        common.merge_trait_values(self, new_data)