Trigger viomi map upload before downloading the map

custom_components/xiaomi_cloud_map_extractor/connector/vacuums/vacuum_viomi.py

@@ -1,9 +1,22 @@
+import asyncio
+import logging
 from typing import Self
 
 from vacuum_map_parser_viomi.map_data_parser import ViomiMapDataParser
 from .base.model import VacuumConfig, VacuumApi
 from .base.vacuum_v2 import BaseXiaomiCloudVacuumV2
 
+_LOGGER = logging.getLogger(__name__)
+
+# Viomi vacuums do not keep a map in the cloud. They upload the current map only
+# when asked to via the `set_uploadmap` miio command; the uploaded map then
+# becomes available as cloud object "1". Without this request the cloud object
+# does not exist and the map download fails with "Object Not Found".
+_UPLOAD_MAP_METHOD = "set_uploadmap"
+_UPLOAD_MAP_PARAMS = [1]
+_VIOMI_MAP_NAME = "1"
+_MAP_UPLOAD_DELAY = 3
+
 
 class ViomiCloudVacuum(BaseXiaomiCloudVacuumV2):
     _viomi_map_data_parser: ViomiMapDataParser
@@ -29,3 +42,13 @@ def map_archive_extension(self: Self) -> str:
     @property
     def map_data_parser(self: Self) -> ViomiMapDataParser:
         return self._viomi_map_data_parser
+
+    async def get_map_name(self: Self) -> str | None:
+        _LOGGER.debug("Requesting viomi map upload via cloud RPC")
+        response = await self._connector.get_other_info(
+            self._device_id, _UPLOAD_MAP_METHOD, _UPLOAD_MAP_PARAMS)
+        if response is None or response.get("code") != 0:
+            _LOGGER.warning("Viomi map upload request was not acknowledged: %s", response)
+            return None
+        await asyncio.sleep(_MAP_UPLOAD_DELAY)
+        return _VIOMI_MAP_NAME

custom_components/xiaomi_cloud_map_extractor/connector/xiaomi_cloud/connector.py

@@ -714,7 +714,7 @@ async def get_device_details(self: Self, device_id: str, server: Optional[str] =
         matching_device = filter(lambda device: device.device_id == device_id, devices)
         return next(matching_device, None)
 
-    async def get_other_info(self: Self, device_id: str, method: str, parameters: dict) -> Any:
+    async def get_other_info(self: Self, device_id: str, method: str, parameters: dict | list) -> Any:
         url = self.get_api_url() + "/v2/home/rpc/" + device_id
         params = {
             "data": json.dumps({"method": method, "params": parameters}, separators=(",", ":"))