Request for Xiaomi S20+ ( xiaomi.vacuum.b108gl )

[anton-knoc]

Checklist

• I have updated the integration to the latest version available
• I have checked if the vacuum/platform is already requested
• I have sent raw map file to piotr.machowski.dev [at] gmail.com (Retrieving map; please provide your GitHub username in the email)

What vacuum model do you want to be supported?

xiaomi.vacuum.b108gl

What is its name?

Xiaomi S20+ (S20 plus)

Available APIs

• xiaomi
• viomi
• roidmi
• dreame

Errors shown in the HA logs (if applicable)

2024-10-29 18:41:27.400 ERROR (MainThread) [homeassistant.helpers.entity] Update for camera.xiaomi_cloud_map_extractor fails
Traceback (most recent call last):
  File "/usr/src/homeassistant/homeassistant/helpers/entity.py", line 942, in async_update_ha_state
    await self.async_device_update()
  File "/usr/src/homeassistant/homeassistant/helpers/entity.py", line 1302, in async_device_update
    await hass.async_add_executor_job(self.update)
  File "/usr/local/lib/python3.12/concurrent/futures/thread.py", line 58, in run
    result = self.fn(*self.args, **self.kwargs)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/config/custom_components/xiaomi_cloud_map_extractor/camera.py", line 278, in update
    self._handle_map_data(map_name)
  File "/config/custom_components/xiaomi_cloud_map_extractor/camera.py", line 335, in _handle_map_data
    map_data, map_stored = self._device.get_map(map_name, self._colors, self._drawables, self._texts,
                           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/config/custom_components/xiaomi_cloud_map_extractor/common/vacuum.py", line 27, in get_map
    response = self.get_raw_map_data(map_name)
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/config/custom_components/xiaomi_cloud_map_extractor/common/vacuum.py", line 45, in get_raw_map_data
    map_url = self.get_map_url(map_name)
              ^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/config/custom_components/xiaomi_cloud_map_extractor/common/vacuum_v2.py", line 18, in get_map_url
    if api_response is None or "result" not in api_response or "url" not in api_response["result"]:
                                                               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
TypeError: argument of type 'NoneType' is not iterable

Other info

Please add this vacuum, if i can help - let me know please
Thank you!

[Nicooow]

Judging by the model number (xiaomi.vacuum.b108), I think it's the same integration as #460 (xiaomi.vacuum.b106)

[tam481]

Adding my +1
Thank you

[dankarization]

+1 waiting for this model

[0rangutan]

+1 here too - just got this popular model and would be awesome to have this working, thanks

[mdudek]

Hi @PiotrMachowski , I have reverse engineered MI Home app plugin for S20+ and here is the python code that decrypts downloaded map. It should help with integration of S20+ to your map extractor.

from Crypto.Cipher import AES
from Crypto.Hash import MD5
from Crypto.Util.Padding import pad, unpad
import base64
import json
import zlib
import hashlib

def inflate(byte_array: bytes):
    # Inflate using zlib
    inflated_string = zlib.decompress(byte_array).decode('utf-8')    
    return inflated_string

def loadMapFromFile(file_path: str):
    with open(file_path, 'rb') as f:
        rawMapContent = f.read()
        jsoMapContent = json.loads(rawMapContent)
        return base64_decode(jsoMapContent["data"].encode('latin1'))
    

def encrypt(source: bytes, key: bytes, iv: bytes):
    """
    Encrypts a string using AES encryption in CBC mode.
    """
    cipher = AES.new(key, AES.MODE_CBC, iv)
    encrypted = cipher.encrypt(pad(source, AES.block_size))
    return encrypted.hex().upper()

def decrypt(encrypted_bytes: bytes, key: bytes, iv: bytes):
    """
    Decrypts a string using AES decryption in CBC mode.
    """
    try:
        cipher = AES.new(key, AES.MODE_CBC, iv)
        decrypted = cipher.decrypt(encrypted_bytes)
        decryptedUnpadded = unpad(decrypted, AES.block_size, 'pkcs7')
        return decryptedUnpadded
    except Exception as e:
        return ""


def md5_hash(data: bytes):
    """
    Returns the MD5 hash of the given data.
    """
    return hashlib.md5(data).hexdigest()

def base64Encoding(input):
  dataBase64 = base64.b64encode(input)
  dataBase64P = dataBase64.decode("UTF-8")
  return dataBase64P

def base64_decode(input: bytes):
    """
    Decodes a Base64 string to hexadecimal.
    """
    decoded_bytes = base64.decodebytes(input)
    return decoded_bytes.hex()

def decryptMap(encryptedMapContent: bytes, modelKey: str, did: str):

    originalWork = modelKey + did

    iv = b"ABCDEF1234123412" # iv as a byte array

    encKey = encrypt(originalWork.encode('latin1'), modelKey.encode('latin1'), iv)
    encKey2 = bytes.fromhex(encKey)
    md5Key = md5_hash(encKey2)
    decryptKey = bytes.fromhex(md5Key)

    encryptedBytes = bytes.fromhex(encryptedMapContent)
    decryptedBase64Bytes = decrypt(encryptedBytes, decryptKey, iv)
    inflatedString = inflate(decryptedBase64Bytes)
    
    ## Write decrypted map to file
    #with open("0.decrypted.map.json", "w") as decryptedFile:
    #    # Writing data to a file
    #    decryptedFile.write(inflatedString)

    return inflatedString

def transformMapData(map_data):
    if map_data is None:
        return None

    map_data = json.loads(map_data)

    map_id = map_data.get("map_id")
    map_version = map_data.get("map_type")
    map_height = map_data.get("height")
    map_width = map_data.get("width")
    origin_x = map_data.get("origin_x")
    origin_y = map_data.get("origin_y")
    have_charge_pile = map_data.get("have_pile")
    charge_pile_x = map_data.get("pile_x")
    charge_pile_y = map_data.get("pile_y")
    charge_pile_yaw = map_data.get("pile_yaw")
    map_resolution = map_data.get("resolution")

    # Inflate the base64-encoded map data
    byte_array = zlib.decompress(base64.b64decode(map_data.get("map_data")))

    # Convert to a uint8 array (bytearray in Python)
    data = bytearray(byte_array)
    fb_walls = map_data.get("fb_walls")
    fb_area = map_data.get("fb_regions")
    zone = map_data.get("part_regions")
    rooms = map_data.get("room_attrs")
    room_colors = None

    if isinstance(map_data.get("map_room_info"), list):
        room_colors = {}
        for item in map_data["map_room_info"]:
            room_colors[item["grid_id"]] = item["color"]

    path = map_data.get("paths")

    # Charge details
    charge = {
        "haveChargePile": have_charge_pile,
        "chargePileX": charge_pile_x,
        "chargePileY": charge_pile_y,
        "chargePileYaw": charge_pile_yaw,
    }

    # Origin details
    origin = {
        "x": round(origin_x / 1000, 2) if origin_x is not None else None,
        "y": round(origin_y / 1000, 2) if origin_y is not None else None,
    }

    # Accuracy
    accuracy = round(map_resolution / 1000, 2) if map_resolution is not None else None

    # Header
    header = {
        "mapId": map_id,
        "mapVersion": map_version,
        "mapWidth": map_width,
        "mapHeight": map_height,
        "charge": charge,
        "origin": origin,
        "accuracy": accuracy,
        "roomColors": room_colors,
    }

    # Extra
    extra = {
        "fbWalls": fb_walls,
        "fbArea": fb_area,
        "zone": zone,
        "rooms": rooms,
        "path": path,
    }

    # Final map structure
    map_result = {
        "header": header,
        "data": data,
        "extra": extra,
    }

    return map_result


def main(): 
    modelKey = "mi.vacuum.b108gl"
    did = "1068470163"

    mapContent = loadMapFromFile("0.encrypted.map")
    decryptedMapContent = decryptMap(mapContent, modelKey, did)

    transformedMapData = transformMapData(decryptedMapContent)

    print ('Map content:', transformedMapData)

if __name__ == "__main__":
    main()
    
    ``` 

[PiotrMachowski]

@mdudek that is great, thank you for your work! <3 I will check it out when I will have some free time