feat: Improve the structure of the model to more closely match am L5X file.

Author: hutcheb

acd/generated/comps/rx_generic.py

@@ -159,6 +159,17 @@ def cip_data_type(self):
             self._io.seek(_pos)
             return getattr(self, '_m_cip_data_type', None)
 
+        @property
+        def radix(self):
+            if hasattr(self, '_m_radix'):
+                return self._m_radix
+
+            _pos = self._io.pos()
+            self._io.seek(32)
+            self._m_radix = self._io.read_u2le()
+            self._io.seek(_pos)
+            return getattr(self, '_m_radix', None)
+
         @property
         def data_type(self):
             if hasattr(self, '_m_data_type'):
@@ -200,6 +211,17 @@ def valid(self):
             self._m_valid = True
             return getattr(self, '_m_valid', None)
 
+        @property
+        def external_access(self):
+            if hasattr(self, '_m_external_access'):
+                return self._m_external_access
+
+            _pos = self._io.pos()
+            self._io.seek(34)
+            self._m_external_access = self._io.read_u2le()
+            self._io.seek(_pos)
+            return getattr(self, '_m_external_access', None)
+
         @property
         def dimension_1(self):
             if hasattr(self, '_m_dimension_1'):

acd/l5x/elements.py

@@ -1,7 +1,8 @@
 import os
 import shutil
 import struct
-from dataclasses import dataclass
+from dataclasses import dataclass, field
+from enum import Enum
 from os import PathLike
 from pathlib import Path
 from sqlite3 import Cursor
@@ -38,7 +39,7 @@ def to_xml(self):
                 if isinstance(attribute_value, L5xElement):
                     child_list.append(attribute_value.to_xml())
                 elif isinstance(attribute_value, list):
-                    if attribute == "tags":
+                    if attribute == "tags" or attribute == "data_types" or attribute == "members":
                         new_child_list: List[str] = []
                         for element in attribute_value:
                             if isinstance(element, L5xElement):
@@ -48,16 +49,30 @@ def to_xml(self):
                         child_list.append(f'<{attribute.title().replace("_", "")}>{" ".join(new_child_list)}</{attribute.title().replace("_", "")}>')
 
                 else:
+                    if attribute == "cls":
+                        attribute = "class"
                     attribute_list.append(f'{attribute.title().replace("_", "")}="{attribute_value}"')
 
         _export_name = self.__class__.__name__.title().replace("_", "")
         return f'<{_export_name} {" ".join(attribute_list)}>{" ".join(child_list)}</{_export_name}>'
 
 
+@dataclass
+class Member(L5xElement):
+    name: str
+    data_type: str
+    dimension: int
+    radix: str
+    hidden: bool
+    external_access: str
+
+
 @dataclass
 class DataType(L5xElement):
     name: str
-    children: List[str]
+    family: str
+    cls: str
+    members: List[Member]
 
 
 @dataclass
@@ -131,6 +146,85 @@ def __post_init__(self):
         self._name = "RSLogix5000Content"
 
 
+def radix_enum(i: int) -> str:
+    if i == 0:
+        return "NullType"
+    if i == 1:
+        return "General"
+    if i == 2:
+        return "Binary"
+    if i == 3:
+        return "Octal"
+    if i == 4:
+        return "Decimal"
+    if i == 5:
+        return "Hex"
+    if i == 6:
+        return "Exponential"
+    if i == 7:
+        return "Float"
+    if i == 8:
+        return "ASCII"
+    if i == 9:
+        return "Unicode"
+    if i == 10:
+        return "Date/Time"
+    if i == 11:
+        return "Date/Time (ns)"
+    if i == 12:
+        return "UseTypeStyle"
+    return "General"
+
+
+def external_access_enum(i: int) -> str:
+    if i == 0:
+        return "Read/Write"
+    if i == 1:
+        return "Read Only"
+    if i == 2:
+        return "None"
+    return "Read/Write"
+
+@dataclass
+class MemberBuilder(L5xElementBuilder):
+    record: List[int] = field(default_factory=[])
+
+    def build(self) -> Member:
+        self._cur.execute(
+            "SELECT comp_name, object_id, parent_id, record FROM comps WHERE object_id=" + str(
+                self._object_id))
+        results = self._cur.fetchall()
+
+        name = results[0][0]
+        r = RxGeneric.from_bytes(results[0][3])
+        try:
+            r = RxGeneric.from_bytes(results[0][3])
+        except Exception as e:
+            return Member(name, name, "", 0, "Decimal", False, "Read/Write")
+
+        extended_records: Dict[int, List[int]] = {}
+        for extended_record in r.extended_records:
+            extended_records[extended_record.attribute_id] = extended_record.value
+        extended_records[r.last_extended_record.attribute_id] = r.last_extended_record.value
+
+        cip_data_typoe = struct.unpack_from("<I", self.record, 0x78)[0]
+        dimension = struct.unpack_from("<I", self.record, 0x5C)[0]
+        radix = radix_enum(struct.unpack_from("<I", self.record, 0x54)[0])
+        data_type_id = struct.unpack_from("<I", self.record, 0x58)[0]
+        hidden = bool(struct.unpack_from("<I", self.record, 0x70)[0])
+        external_access = external_access_enum(struct.unpack_from("<I", self.record, 0x74)[0])
+
+
+        self._cur.execute(
+            "SELECT comp_name, object_id, parent_id, record FROM comps WHERE object_id=" + str(
+                data_type_id))
+        data_type_results = self._cur.fetchall()
+        data_type = data_type_results[0][0]
+
+
+        return Member(name, name, data_type, dimension, radix, hidden, external_access)
+
+
 @dataclass
 class DataTypeBuilder(L5xElementBuilder):
 
@@ -140,25 +234,53 @@ def build(self) -> DataType:
                 self._object_id))
         results = self._cur.fetchall()
 
-        record = results[0][3]
         name = results[0][0]
 
+        try:
+            r = RxGeneric.from_bytes(results[0][3])
+        except Exception as e:
+            return DataType(name, name, "NoFamily", "User", [])
+
+        extended_records: Dict[int, List[int]] = {}
+        for extended_record in r.extended_records:
+            extended_records[extended_record.attribute_id] = extended_record.value
+        extended_records[r.last_extended_record.attribute_id] = r.last_extended_record.value
+
+        string_family_int = struct.unpack("<I", extended_records[0x6C])[0]
+        string_family = "StringFamily" if string_family_int == 1 else "NoFamily"
+
+        built_in = struct.unpack("<I", extended_records[0x67])[0]
+        module_defined = struct.unpack("<I", extended_records[0x69])[0]
+
+        class_type = "User"
+        if module_defined > 0:
+            class_type = "IO"
+        if built_in > 0:
+            class_type = "ProductDefined"
+        if len(extended_records[0x64]) == 0x04:
+            member_count = struct.unpack("<I", extended_records[0x64])[0]
+        else:
+            member_count = 0
+
         self._cur.execute(
             "SELECT comp_name, object_id, parent_id, record FROM comps WHERE parent_id=" + str(
                 self._object_id))
         member_results = self._cur.fetchall()
+        children: List[Member] = []
         if len(member_results) == 1:
             member_collection_id = member_results[0][1]
 
             self._cur.execute(
-                "SELECT comp_name, object_id, parent_id, record FROM comps WHERE parent_id=" + str(
-                    member_collection_id))
+                f"SELECT comp_name, object_id, parent_id, seq_number, record FROM comps WHERE parent_id={member_collection_id} ORDER BY seq_number")
             children_results = self._cur.fetchall()
-            children = []
-            for child in children_results:
-                children.append(child[0])
-            return DataType(name, name, children)
-        return DataType(name, name, [])
+
+            if member_count != len(children_results):
+                raise Exception("Member and children list arent the same length")
+
+            for idx, child in enumerate(children_results):
+                children.append(MemberBuilder(self._cur, child[1], extended_records[0x6E + idx]).build())
+
+        return DataType(name, name, string_family, class_type, children)
 
 
 @dataclass
@@ -236,13 +358,16 @@ def build(self) -> Tag:
         name_length = struct.unpack("<H", extended_records[0x01][0:2])[0]
         name = bytes(extended_records[0x01][2:name_length+2]).decode('utf-8')
 
+        radix = radix_enum(r.main_record.radix)
+        external_access = external_access_enum(r.main_record.external_access)
+
         if r.main_record.dimension_1 != 0:
             data_type = data_type + "[" + str(r.main_record.dimension_1) + "]"
         if r.main_record.dimension_2 != 0:
             data_type = data_type + "[" + str(r.main_record.dimension_2) + "]"
         if r.main_record.dimension_3 != 0:
             data_type = data_type + "[" + str(r.main_record.dimension_3) + "]"
-        return Tag(name, name, "Base",  data_type, "Decimal", "Read/Write", r.main_record.data_table_instance, comment_results)
+        return Tag(name, name, "Base",  data_type, radix, external_access, r.main_record.data_table_instance, comment_results)
 
 
 @dataclass

resources/templates/Comps/RxGeneric.ksy

@@ -64,6 +64,12 @@ types:
       data_type:
         pos: 0x1C
         type: u4
+      radix:
+        pos: 0x20
+        type: u2
+      external_access:
+        pos: 0x22
+        type: u2
       data_table_instance:
         pos: 0x24
         type: u4

setup.py

@@ -48,7 +48,7 @@ def run(self):
 
 setup(
     name="acd-tools",
-    version="0.2a4",
+    version="0.2a5",
     description="Rockwell ACD File Tools",
     classifiers=[
         "Development Status :: 3 - Alpha",

test/test_api.py

@@ -25,7 +25,7 @@ def test_dump_to_files():
     DumpCompsRecords(export._cur, 0).dump(0)
 
 
-def manual_test_to_xml():
+def test_to_xml():
     importer = ImportProjectFromFile(Path(os.path.join("..", "resources", "CuteLogix.ACD")))
     project: RSLogix5000Content = importer.import_project()
     ss = project.to_xml()

test/test_database.py

@@ -45,9 +45,9 @@ def test_parse_rungs_dat(controller):
 
 def test_parse_datatypes_dat(controller):
     data_type = controller.data_types[-1].name
-    child =  controller.data_types[-1].children[-1]
+    child =  controller.data_types[-1].members[-1]
     assert data_type == 'STRING20'
-    assert child == 'DATA'
+    assert child.name == 'DATA'
 
 
 def test_parse_tags_dat(controller):