[WIP] Added ijai models basic support, fixed initialization of Palette

and ImageConfig objects

Author: Alexander "Tarh

custom_components/xiaomi_cloud_map_extractor/__init__.py

@@ -2,10 +2,11 @@
 import logging
 from datetime import timedelta
 from enum import StrEnum
+from dataclasses import fields
 
 from vacuum_map_parser_base.config.color import ColorsPalette
 from vacuum_map_parser_base.config.drawable import Drawable
-from vacuum_map_parser_base.config.image_config import ImageConfig
+from vacuum_map_parser_base.config.image_config import ImageConfig, TrimConfig
 from vacuum_map_parser_base.config.size import Sizes
 from vacuum_map_parser_base.config.text import Text
 
@@ -27,6 +28,7 @@
 from .vacuum_platforms.vacuum_viomi import ViomiCloudVacuum
 from .vacuum_platforms.vacuum_ijai import IjaiCloudVacuum
 from .vacuum_platforms.vacuum_unsupported import UnsupportedCloudVacuum
+from .initializer import from_dict
 from .const import *
 
 
@@ -122,24 +124,26 @@
     })
 
 
+
 async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
     await async_setup_reload_service(hass, DOMAIN, PLATFORMS)
-
+    _LOGGER.debug(f"config={config}")
     host = config[CONF_HOST]
     token = config[CONF_TOKEN]
     username = config[CONF_USERNAME]
     password = config[CONF_PASSWORD]
     country = config[CONF_COUNTRY]
     name = config[CONF_NAME]
     should_poll = config[CONF_AUTO_UPDATE]
-    image_config = config[CONF_MAP_TRANSFORM]
+    image_config = from_dict(ImageConfig, config[CONF_MAP_TRANSFORM])
     colors = config[CONF_COLORS]
     room_colors = config[CONF_ROOM_COLORS]
     for room, color in room_colors.items():
         colors[f"{COLOR_ROOM_PREFIX}{room}"] = color
     drawables = config[CONF_DRAW]
-    sizes = config[CONF_SIZES]
-    texts = config[CONF_TEXTS]
+    sizes = Sizes(config[CONF_SIZES])
+    texts = from_dict(list, config[CONF_TEXTS])
+    
     if DRAWABLE_ALL in drawables:
         drawables = CONF_AVAILABLE_DRAWABLES[1:]
     attributes = config[CONF_ATTRIBUTES]
@@ -149,7 +153,7 @@ async def async_setup_platform(hass, config, async_add_entities, discovery_info=
     force_api = config[CONF_FORCE_API]
     entity_id = generate_entity_id(ENTITY_ID_FORMAT, name, hass=hass)
     async_add_entities([VacuumCamera(entity_id, host, token, username, password, country, name, should_poll,
-                                     image_config, colors, drawables, sizes, texts, attributes, store_map_raw,
+                                     image_config, ColorsPalette(colors, room_colors), drawables, sizes, texts, attributes, store_map_raw,
                                      store_map_image, store_map_path, force_api)])
 
 
@@ -230,6 +234,7 @@ def should_poll(self) -> bool:
 
     @staticmethod
     def extract_attributes(map_data: MapData, attributes_to_return: list[str], country) -> dict[str, any]:
+        _LOGGER.debug(f"extract_attributes{map_data}, {attributes_to_return}, country")
         attributes = {}
         rooms = []
         if map_data.rooms is not None:
@@ -299,7 +304,7 @@ def _login(self):
 
     def _initialize_device(self):
         _LOGGER.debug("Retrieving device info, country: %s", self._country)
-        country, user_id, device_id, model, mac = self._connector.get_device_details(self._vacuum.token, self._country)
+        country, user_id, device_id, model, mac = self._connector.get_device_details(self._token, self._country)
         if model is not None:
             self._country = country
             _LOGGER.debug("Retrieved device model: %s", model)
@@ -345,26 +350,29 @@ def _create_device(self, user_id: str, device_id: str, model: str, mac: str) ->
         self._used_api = self._detect_api(model)
         store_map_path = self._store_map_path if self._store_map_raw else None
         vacuum_config = VacuumConfig(
-            self._connector,
-            self._country,
-            user_id,
-            device_id,
-            self._host,
-            self._token,
-            model,
-            self._colors,
-            self._drawables,
-            self._image_config,
-            self._sizes,
-            self._texts,
-            store_map_path
+            connector = self._connector,
+            country = self._country,
+            user_id = user_id,
+            device_id = device_id,
+            host = self._host,
+            token = self._token,
+            model = model,
+            _mac = mac,
+            palette = self._colors,
+            drawables = self._drawables,
+            image_config = self._image_config,
+            sizes = self._sizes,
+            texts = self._texts,
+            store_map_path = store_map_path
         )
         if self._used_api == CONF_AVAILABLE_API_XIAOMI:
             return RoborockCloudVacuum(vacuum_config)
         if self._used_api == CONF_AVAILABLE_API_VIOMI:
             return ViomiCloudVacuum(vacuum_config)
         if self._used_api == CONF_AVAILABLE_API_IJAI:
-            return IjaiCloudVacuum(self._connector, self._country, user_id, device_id, model, mac)
+            _LOGGER.debug(f"palette={vacuum_config.palette}")
+            _LOGGER.debug(f"image_config={vacuum_config.image_config}")
+            return IjaiCloudVacuum(vacuum_config)
         if self._used_api == CONF_AVAILABLE_API_ROIDMI:
             return RoidmiCloudVacuum(vacuum_config)
         if self._used_api == CONF_AVAILABLE_API_DREAME:

custom_components/xiaomi_cloud_map_extractor/camera.py

@@ -43,7 +43,7 @@
 CONF_Y = "y"
 
 CONF_AVAILABLE_APIS = [CONF_AVAILABLE_API_XIAOMI, CONF_AVAILABLE_API_VIOMI, CONF_AVAILABLE_API_ROIDMI,
-                       CONF_AVAILABLE_API_DREAME]
+                       CONF_AVAILABLE_API_DREAME, CONF_AVAILABLE_API_IJAI]
 
 CONF_AVAILABLE_SIZES = [CONF_SIZE_VACUUM_RADIUS, CONF_SIZE_PATH_WIDTH, CONF_SIZE_IGNORED_OBSTACLE_RADIUS,
                         CONF_SIZE_IGNORED_OBSTACLE_WITH_PHOTO_RADIUS, CONF_SIZE_MOP_PATH_WIDTH,

custom_components/xiaomi_cloud_map_extractor/const.py

@@ -0,0 +1,8 @@
+from dataclasses import fields
+
+def from_dict(cls, d):
+    try:
+        fieldtypes = {f.name:f.type for f in fields(cls)}
+        return cls(**{f:from_dict(fieldtypes[f],d[f]) for f in d})
+    except:
+        return d

custom_components/xiaomi_cloud_map_extractor/initializer.py

@@ -11,7 +11,6 @@
 
 from .xiaomi_cloud_connector import XiaomiCloudConnector
 
-
 @dataclass
 class VacuumConfig:
     connector: XiaomiCloudConnector
@@ -21,6 +20,7 @@ class VacuumConfig:
     host: str
     token: str
     model: str
+    _mac: str
     palette: ColorsPalette
     drawables: list[Drawable]
     image_config: ImageConfig

custom_components/xiaomi_cloud_map_extractor/vacuum_platforms/vacuum_base.py

@@ -1,39 +1,59 @@
-import zlib
 
-from custom_components.xiaomi_cloud_map_extractor.common.map_data import MapData
-from custom_components.xiaomi_cloud_map_extractor.common.vacuum_v2 import XiaomiCloudVacuumV2
-from custom_components.xiaomi_cloud_map_extractor.common.xiaomi_cloud_connector import XiaomiCloudConnector
-from custom_components.xiaomi_cloud_map_extractor.types import Colors, Drawables, ImageConfig, Sizes, Texts
-from custom_components.xiaomi_cloud_map_extractor.ijai.map_data_parser import MapDataParserIjai
-from custom_components.xiaomi_cloud_map_extractor.ijai.aes_decryptor import *
+from vacuum_map_parser_ijai.map_data_parser import IjaiMapDataParser
+from miio.miot_device import MiotDevice
 
-class IjaiCloudVacuum(XiaomiCloudVacuumV2):
-
-    def __init__(self, connector: XiaomiCloudConnector, country: str, user_id: str, device_id: str, model: str, mac:str):
-        super().__init__(connector, country, user_id, device_id, model, mac)
+from .vacuum_v2 import XiaomiCloudVacuumV2
+from .vacuum_base import VacuumConfig
+import logging
+_LOGGER = logging.getLogger(__name__)
 
-    def get_map_url(self, map_name: str) -> str | None:
-        url = self._connector.get_api_url(self._country) + '/v2/home/get_interim_file_url/pro'
-
-    def decode_map(self,
-                   raw_map: bytes,
-                   colors: Colors,
-                   drawables: Drawables,
-                   texts: Texts,
-                   sizes: Sizes,
-                   image_config: ImageConfig) -> MapData:
+class IjaiCloudVacuum(XiaomiCloudVacuumV2):
+    WIFI_STR_LEN = 18
+    WIFI_STR_POS = 11
+    def __init__(self, vacuum_config: VacuumConfig):
+        super().__init__(vacuum_config)
+        self._token = vacuum_config.token
+        self._host = vacuum_config.host
+        self._mac = vacuum_config._mac
+        self._wifi_info_sn = None
 
-        unzipped = zlib.decompress(raw_map)
-
-        return MapDataParserIjai.parse(unzipped, colors, drawables, texts, sizes, image_config)
-
-    def get_map_archive_extension(self) -> str:
+        self._ijai_map_data_parser = IjaiMapDataParser(
+            vacuum_config.palette,
+            vacuum_config.sizes,
+            vacuum_config.drawables,
+            vacuum_config.image_config,
+            vacuum_config.texts
+        )
+    @property
+    def map_archive_extension(self) -> str:
         return "zlib"
 
-    def decrypt_map(self, data:bytes, wifi_info_sn:str, user_id:str, device_id:str, model:str, mac:str):
-        return unGzipCommon(data=data, \
-                            wifi_info_sn=wifi_info_sn, \
-                            owner_id=str(user_id), \
-                            device_id=device_id, \
-                            model=model, \
-                            device_mac=mac);
+    @property
+    def map_data_parser(self) -> IjaiMapDataParser:
+        return self._ijai_map_data_parser
+
+    def get_map_url(self, map_name: str) -> str | None:
+        url = self._connector.get_api_url(self._country) + '/v2/home/get_interim_file_url_pro'
+        params = {
+            "data": f'{{"obj_name":"{self._user_id}/{self._device_id}/{map_name}"}}'
+        }
+        api_response = self._connector.execute_api_call_encrypted(url, params)
+        if api_response is None or ("result" not in api_response) or (api_response["result"] is None) or ("url" not in api_response["result"]):
+            self._LOGGER.debug(f"API returned {api_response['code']}" + "(" + api_response["message"] + ")")
+            return None
+        return api_response["result"]["url"]
+
+    def decode_and_parse(self, raw_map: bytes):
+        if self._wifi_info_sn is None or self._wifi_info_sn is "":
+            device = MiotDevice(self._host, self._token)
+            props = device.get_property_by(7, 45)[0]["value"].split(',')
+            self._wifi_info_sn = props[self.WIFI_STR_POS].replace('"','')[:self.WIFI_STR_LEN]
+            _LOGGER.debug(f"wifi_sn = {self._wifi_info_sn}")
+
+        decoded_map = self.map_data_parser.unpack_map(raw_map,
+            wifi_sn=self._wifi_info_sn,
+            owner_id = str(self._user_id),
+            device_id = str(self._device_id),
+            model = self.model,
+            device_mac = self._mac)
+        return self.map_data_parser.parse(decoded_map)