mps-cli-py: complete binary (.mpb) persistency implementation

mps-cli-py/src/mpscli/model/SNodeRef.py

@@ -1,14 +1,11 @@
-
-
 class SNodeRef:
-
-    def __init__(self, model_uuid, node_uuid):
+    def __init__(self, model_uuid, node_uuid, resolve_info=None):
         self.model_uuid = model_uuid
         self.node_uuid = node_uuid
+        self.resolve_info = resolve_info
 
     def resolve(self, repo):
         model = repo.get_model_by_uuid(self.model_uuid)
         if model is None:
             return None
         return model.get_node_by_uuid(self.node_uuid)
-

mps-cli-py/src/mpscli/model/builder/NodeIdEncodingUtils.py

@@ -1,24 +1,39 @@
-
 class NodeIdEncodingUtils:
-
+    INDEX_CHARS = "0123456789abcdefghijklmnopqrstuvwxyz$_ABCDEFGHIJKLMNOPQRSTUVWXYZ"
+    MIN_CHAR = ord("$")
+
     def __init__(self):
-        self.myIndexChars = "0123456789abcdefghijklmnopqrstuvwxyz$_ABCDEFGHIJKLMNOPQRSTUVWXYZ"
-        self.MIN_CHAR = ord('$')
-        self.myCharToValue = {}
-        for i in range(0, len(self.myIndexChars)):
-            charValue = ord(self.myIndexChars[i])
-            self.myCharToValue[charValue - self.MIN_CHAR] = i
-
-    # transforms the node id string saved in MPS files and converts it to the proper node id as long 
-    def decode_string(self, uid_string):
-        res = 0
-        c = uid_string[0]
-        value = self.myCharToValue[ord(c) - self.MIN_CHAR]
-        res = value
-        for idx in range(1, len(uid_string)):
+        self._char_to_value = {}
+        for i, ch in enumerate(self.INDEX_CHARS):
+            self._char_to_value[ord(ch) - self.MIN_CHAR] = i
+
+    def decode(self, uid_string: str) -> str:
+        if not uid_string:
+            return ""
+
+        res = self._char_to_value[ord(uid_string[0]) - self.MIN_CHAR]
+
+        for ch in uid_string[1:]:
             res = res << 6
-            c = uid_string[idx]
-            value = self.myIndexChars.index(c)
-            res = res | value
-        return res
-
+            res |= self.INDEX_CHARS.index(ch)
+
+        return str(res)
+
+    def encode(self, decimal_string: str) -> str:
+        try:
+            uid_number = int(decimal_string)
+        except ValueError:
+            return "Invalid decimal input"
+
+        if uid_number == 0:
+            return self.INDEX_CHARS[0]
+
+        result = []
+
+        while uid_number > 0:
+            remainder = uid_number % 64
+            result.append(self.INDEX_CHARS[remainder])
+            uid_number //= 64
+
+        result.reverse()
+        return "".join(result)

mps-cli-py/src/mpscli/model/builder/SModelBuilderBinaryPersistency.py

@@ -0,0 +1,84 @@
+# mpscli/model/builder/SModelBuilderBinaryPersistency.py
+
+from mpscli.model.builder.binary.constants import *
+from mpscli.model.builder.binary.utils import (
+    read_uuid,
+    read_string,
+    advance_until_after,
+)
+from mpscli.model.builder.binary.reader import BinaryReader
+from mpscli.model.builder.SModelBuilderBase import SModelBuilderBase
+from mpscli.model.builder.binary.registry import load_registry
+from mpscli.model.builder.binary.imports import load_imports
+from mpscli.model.builder.binary.used_languages import load_used_languages
+from mpscli.model.builder.binary.module_refs import load_module_ref_list
+from mpscli.model.builder.binary.nodes import read_children
+from mpscli.model.SModel import SModel
+
+
+class SModelBuilderBinaryPersistency(SModelBuilderBase):
+
+    def __init__(self):
+        super().__init__()
+        self.model_refs = []
+
+    def build(self, path_to_model: str):
+        with open(path_to_model, "rb") as f:
+            reader = BinaryReader(f.read())
+
+        header = reader.read_u32()
+        if header != HEADER_START:
+            raise ValueError(f"Invalid header: 0x{header:X}")
+
+        stream_id = reader.read_u32()
+        if stream_id != STREAM_ID:
+            raise ValueError(f"Unsupported stream id: 0x{stream_id:X}")
+
+        model = self.read_model_header(reader)
+
+        self.index_2_imported_model_uuid["0"] = model.uuid
+
+        advance_until_after(reader, REGISTRY_START)
+        load_registry(reader, self)
+        advance_until_after(reader, REGISTRY_END)
+
+        load_used_languages(reader)
+
+        load_module_ref_list(reader)
+
+        load_module_ref_list(reader)
+
+        load_imports(reader, self)
+
+        advance_until_after(reader, MODEL_START)
+        read_children(reader, self, model, None)
+
+        return model
+
+    def read_model_header(self, reader):
+        ref_kind = reader.read_u8()
+
+        if ref_kind == NULL:
+            raise ValueError("Unexpected NULL model reference")
+
+        if ref_kind == MODELREF_INDEX:
+            index = reader.read_u32()
+            return self.model_refs[index]
+
+        model_id_kind = reader.read_u8()
+
+        if model_id_kind != MODELID_REGULAR:
+            raise ValueError(f"Unsupported model id type: 0x{model_id_kind:X}")
+
+        uuid = read_uuid(reader)
+        model_id = f"r:{uuid}"
+
+        model_name = read_string(reader)
+
+        advance_until_after(reader, HEADER_END)
+
+        model = SModel(model_name, model_id, False)
+
+        self.model_refs.append(model)
+
+        return model

mps-cli-py/src/mpscli/model/builder/SSolutionBuilder.py

@@ -3,7 +3,7 @@
 from mpscli.model.SSolution import SSolution
 from mpscli.model.builder.SModelBuilderDefaultPersistency import SModelBuilderDefaultPersistency
 from mpscli.model.builder.SModelBuilderFilePerRootPersistency import SModelBuilderFilePerRootPersistency
-
+from mpscli.model.builder.SModelBuilderBinaryPersistency import SModelBuilderBinaryPersistency      
 
 class SSolutionBuilder:
 
@@ -20,6 +20,8 @@ def build_solution(self, path_to_msd_file):
         for path_to_model in path_to_models_dir.iterdir():
             if path_to_model.is_dir():
                 builder = SModelBuilderFilePerRootPersistency()
+            elif path_to_model.suffix == ".mpb":
+                builder = SModelBuilderBinaryPersistency()
             else:
                 builder = SModelBuilderDefaultPersistency()
             model = builder.build(path_to_model)

mps-cli-py/src/mpscli/model/builder/binary/constants.py

@@ -0,0 +1,29 @@
+# mpscli/model/builder/binary/constants.py
+
+HEADER_START = 0x91ABABA9
+HEADER_END = 0xABABABAB
+STREAM_ID_V2 = 0x00000400
+STREAM_ID = STREAM_ID_V2
+
+REGISTRY_START = 0x5A5A5A5A
+REGISTRY_END = 0xA5A5A5A5
+MODEL_START = 0xBABABABA
+
+NULL = 0x70
+MODELREF_INDEX = 9
+MODELID_REGULAR = 0x28
+
+STRING_INDEX = 1
+
+REF_THIS_MODEL = 17
+REF_OTHER_MODEL = 18
+
+NODEID_STRING = 0x17
+NODEID_LONG = 0x18
+
+MODULEREF_NAMEONLY = 0x18
+MODULEREF_INDEX = 0x19
+MODULEREF_MODULEID = 0x17
+
+MODULEID_REGULAR = 0x48
+MODULEID_FOREIGN = 0x47

mps-cli-py/src/mpscli/model/builder/binary/imports.py

@@ -0,0 +1,33 @@
+# mpscli/model/builder/binary/imports.py
+
+from mpscli.model.builder.binary.constants import *
+from mpscli.model.builder.binary.utils import (
+    read_uuid,
+    read_string,
+    read_module_reference,
+)
+
+
+def load_imports(reader, builder):
+    imports_count = reader.read_u32()
+
+    for i in range(imports_count):
+        ref_kind = reader.read_u8()
+
+        if ref_kind == MODELREF_INDEX:
+            raise ValueError("Unexpected MODELREF_INDEX in imports")
+
+        model_id_kind = reader.read_u8()
+
+        if model_id_kind == MODELID_REGULAR:
+            uuid = read_uuid(reader)
+            model_id = f"r:{uuid}"
+        elif model_id_kind == NULL:
+            model_id = ""
+        else:
+            raise ValueError(f"Unsupported model id kind: 0x{model_id_kind:X}")
+
+        model_name = read_string(reader)
+        read_module_reference(reader)
+
+        builder.index_2_imported_model_uuid[str(i + 1)] = model_id

mps-cli-py/src/mpscli/model/builder/binary/module_refs.py

@@ -0,0 +1,7 @@
+from mpscli.model.builder.binary.utils import read_module_reference
+
+
+def load_module_ref_list(reader):
+    count = reader.read_u16()
+    for _ in range(count):
+        read_module_reference(reader)

mps-cli-py/src/mpscli/model/builder/binary/node_id_utils.py

@@ -0,0 +1,43 @@
+class NodeIdEncodingUtils:
+    INDEX_CHARS = "0123456789abcdefghijklmnopqrstuvwxyz$_ABCDEFGHIJKLMNOPQRSTUVWXYZ"
+    MIN_CHAR = ord("$")
+
+    def __init__(self):
+        self.char_to_value = {}
+        for i, ch in enumerate(self.INDEX_CHARS):
+            key = ord(ch) - self.MIN_CHAR
+            self.char_to_value[key] = i
+
+    def encode(self, uid_string: str) -> str:
+        try:
+            num = int(uid_string)
+        except ValueError:
+            return uid_string
+
+        if num == 0:
+            return self.INDEX_CHARS[0]
+
+        result = []
+
+        while num > 0:
+            pos = num % 64
+            result.append(self.INDEX_CHARS[pos])
+            num //= 64
+
+        result.reverse()
+        return "".join(result)
+
+    def decode(self, uid_string: str) -> str:
+        if not uid_string:
+            return ""
+
+        res = 0
+
+        for ch in uid_string:
+            res = res << 6
+            value = self.INDEX_CHARS.find(ch)
+            if value == -1:
+                raise ValueError(f"Invalid encoded node id character: {ch}")
+            res |= value
+
+        return str(res)