Move decompress_7z_base64_data to rust

.devcontainer.json

@@ -44,7 +44,8 @@
   },
   // Add the IDs of extensions you want installed when the container is created.
   "features": {
-    "ghcr.io/devcontainers/features/docker-in-docker:2": {}
+    "ghcr.io/devcontainers/features/docker-in-docker:2": {},
+    "ghcr.io/devcontainers/features/rust:1": {}
   },
   "image": "mcr.microsoft.com/devcontainers/base:debian",
   "name": "Deebot client",

.gitignore

@@ -21,4 +21,9 @@ nosetests.xml
 .*_cache
 
 test.py
-.env
+.env
+
+/target
+
+# rust so
+*.cpython*.so

Cargo.toml

@@ -0,0 +1,14 @@
+[package]
+name = "deebot_client"
+version = "0.1.0"
+edition = "2021"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+[lib]
+name = "deebot_client"
+crate-type = ["cdylib"]
+
+[dependencies]
+base64 = "0.22.1"
+pyo3 = "0.23.3"
+rust-lzma = "0.6.0"

deebot_client/commands/json/map.py

@@ -18,7 +18,7 @@
 from deebot_client.events.map import CachedMapInfoEvent
 from deebot_client.logging_filter import get_logger
 from deebot_client.message import HandlingResult, HandlingState, MessageBodyDataDict
-from deebot_client.util import decompress_7z_base64_data
+from deebot_client.rs import decompress_7z_base64_data
 
 from .common import JsonCommandWithMessageHandling
 
@@ -275,7 +275,7 @@ def _handle_body_data_dict(
             # This command is used by new and old bots
             if data.get("compress", 0) == 1:
                 # Newer bot's return coordinates as base64 decoded string
-                coordinates = decompress_7z_base64_data(data["value"]).decode()
+                coordinates = decompress_7z_base64_data(data["value"])
             else:
                 # Older bot's return coordinates direct as comma/semicolon separated list
                 coordinates = data["value"]
@@ -305,7 +305,7 @@ def _get_subset_ids(
     ) -> list[int] | None:
         """Return subset ids."""
         # subset is based64 7z compressed
-        subsets = json.loads(decompress_7z_base64_data(data["subsets"]).decode())
+        subsets = json.loads(decompress_7z_base64_data(data["subsets"]))
 
         match data["type"]:
             case MapSetType.ROOMS:

deebot_client/map.py

@@ -35,10 +35,12 @@
 from .exceptions import MapError
 from .logging_filter import get_logger
 from .models import Room
+from .rs import (
+    decompress_7z_base64_data,
+)
 from .util import (
     OnChangedDict,
     OnChangedList,
-    decompress_7z_base64_data,
 )
 
 if TYPE_CHECKING:
@@ -393,7 +395,7 @@ async def on_map_subset(event: MapSubsetEvent) -> None:
 
     def _update_trace_points(self, data: str) -> None:
         _LOGGER.debug("[_update_trace_points] Begin")
-        trace_points = decompress_7z_base64_data(data)
+        trace_points = decompress_7z_base64_data(data).encode()
 
         for i in range(0, len(trace_points), 5):
             position_x, position_y = struct.unpack("<hh", trace_points[i : i + 4])
@@ -620,7 +622,7 @@ def image(self) -> Image.Image:
 
     def update_points(self, base64_data: str) -> None:
         """Add map piece points."""
-        decoded = decompress_7z_base64_data(base64_data)
+        decoded = decompress_7z_base64_data(base64_data).encode()
         old_crc32 = self._crc32
         self._crc32 = zlib.crc32(decoded)
 

deebot_client/rs.pyi

@@ -0,0 +1,2 @@
+def decompress_7z_base64_data(input: str) -> str:
+    """Decompress base64 decoded 7z compressed string."""

deebot_client/util/__init__.py

@@ -4,11 +4,9 @@
 
 from abc import ABC
 import asyncio
-import base64
 from contextlib import suppress
 from enum import Enum
 import hashlib
-import lzma
 from typing import TYPE_CHECKING, Any, TypeVar
 
 from deebot_client.logging_filter import get_logger
@@ -37,22 +35,6 @@ def verify_required_class_variables_exists(
                 raise ValueError(msg)
 
 
-def decompress_7z_base64_data(data: str) -> bytes:
-    """Decompress base64 decoded 7z compressed string."""
-    final_array = bytearray()
-
-    # Decode Base64
-    decoded = base64.b64decode(data)
-
-    for i, idx in enumerate(decoded):
-        if i == 8:
-            final_array.extend(b"\x00\x00\x00\x00")
-        final_array.append(idx)
-
-    dec = lzma.LZMADecompressor(lzma.FORMAT_AUTO, None, None)
-    return dec.decompress(final_array)
-
-
 def create_task(
     tasks: set[asyncio.Future[Any]], target: Coroutine[Any, Any, _T]
 ) -> asyncio.Task[_T]: