Rudimentary ifcopenshell-compatible file/instance interface

Author: aothms

main.py

@@ -1,3 +1,7 @@
+import builtins
+from dataclasses import dataclass
+import itertools
+import numbers
 import time
 import sys
 import os
@@ -6,6 +10,8 @@
 import re
 
 from collections import defaultdict
+import types
+import typing
 
 from lark import Lark, Transformer, Tree, Token
 from lark.exceptions import UnexpectedToken, UnexpectedCharacters
@@ -236,6 +242,18 @@ def enumeration(self, s):
     STAR = str
 
 
+@dataclass
+class entity_instance:
+    id : int
+    type : str
+    attributes: tuple
+    lines: tuple
+    def __getitem__(self, k):
+        # compatibility with dict
+        return getattr(self, k)
+    def __repr__(self):
+        return f'#{self.id}={self.type}({",".join(map(str, self.attributes))})'
+
 def create_step_entity(entity_tree):
     entity = {}
     t = T(visit_tokens=True).transform(entity_tree)
@@ -252,31 +270,38 @@ def traverse(fn, x):
 
     lines = list(traverse(get_line_number, entity_tree))
 
-    id_tree = t.children[0].children[0]
-
     entity_id = t.children[0].children[0]
     entity_type = t.children[0].children[1].children[0]
 
     attributes_tree = t.children[0].children[1].children[1]
     attributes = list(attributes_tree)
 
-    return {
-        "id": entity_id,
-        "type": entity_type,
-        "attributes": attributes,
-        "lines": (min(lines), max(lines)),
-    }
+    return entity_instance(
+        entity_id,
+        entity_type,
+        attributes,
+        (min(lines), max(lines)),
+    )
 
 
-def process_tree(filecontent, file_tree, with_progress):
+def process_tree(filecontent, file_tree, with_progress, with_header=False):
     ents = defaultdict(list)
+    header, data = file_tree.children
+
+    def make_header_ent(ast):
+        kw, param_list = ast.children
+        kw = kw.children[0].value
+        return kw, T(visit_tokens=True).transform(param_list)
+
+    if with_header:
+        header = dict(map(make_header_ent, header.children[0].children))
 
-    n = len(file_tree.children[1].children)
+    n = len(data.children)
     if n:
         percentages = [i * 100.0 / n for i in range(n + 1)]
         num_dots = [int(b) - int(a) for a, b in zip(percentages, percentages[1:])]
 
-    for idx, entity_tree in enumerate(file_tree.children[1].children):
+    for idx, entity_tree in enumerate(data.children):
         if with_progress:
             sys.stdout.write(num_dots[idx] * ".")
             sys.stdout.flush()
@@ -286,13 +311,16 @@ def process_tree(filecontent, file_tree, with_progress):
             raise DuplicateNameError(filecontent, ent["id"], ent["lines"])
         ents[id_].append(ent)
 
-    return ents
+    if with_header:
+        return header, ents
+    else:
+        return ents
 
 
-def parse(*, filename=None, filecontent=None, with_progress=False, with_tree=True):
+def parse(*, filename=None, filecontent=None, with_progress=False, with_tree=True, with_header=False):
     if filename:
         assert not filecontent
-        filecontent = open(filename, encoding=None).read()
+        filecontent = builtins.open(filename, encoding=None).read()
 
     # Match and remove the comments
     p = r"/\*[\s\S]*?\*/"
@@ -340,7 +368,7 @@ def replace_fn(match):
         raise SyntaxError(filecontent, e)
 
     if with_tree:
-        return process_tree(filecontent, ast, with_progress)
+        return process_tree(filecontent, ast, with_progress, with_header)
     else:
         # process_tree() would take care of duplicate identifiers,
         # but we need to do it ourselves now using our rudimentary
@@ -352,6 +380,78 @@ def replace_fn(match):
             seen.add(iden)
 
 
+class file:
+    """
+    A somewhat compatible interface (but very limited) to ifcopenshell.file
+    """
+    def __init__(self, parse_outcomes):
+        self.header_, self.data_ = parse_outcomes
+        
+    @property
+    def schema_identifier(self) -> str:
+        return self.header_['FILE_SCHEMA'][0][0]
+        
+    @property
+    def schema(self) -> str:
+        """General IFC schema version: IFC2X3, IFC4, IFC4X3."""
+        prefixes = ("IFC", "X", "_ADD", "_TC")
+        reg = "".join(f"(?P<{s}>{s}\\d+)?" for s in prefixes)
+        match = re.match(reg, self.schema_identifier)
+        version_tuple = tuple(
+            map(
+                lambda pp: int(pp[1][len(pp[0]) :]) if pp[1] else None,
+                ((p, match.group(p)) for p in prefixes),
+            )
+        )
+        return "".join("".join(map(str, t)) if t[1] else "" for t in zip(prefixes, version_tuple[0:2]))
+
+    @property
+    def schema_version(self) -> tuple[int, int, int, int]:
+        """Numeric representation of the full IFC schema version.
+
+        E.g. IFC4X3_ADD2 is represented as (4, 3, 2, 0).
+        """
+        schema = self.wrapped_data.schema
+        version = []
+        for prefix in ("IFC", "X", "_ADD", "_TC"):
+            number = re.search(prefix + r"(\d)", schema)
+            version.append(int(number.group(1)) if number else 0)
+        return tuple(version)
+    
+    @property
+    def header(self):
+        return types.SimpleNamespace(**{k.lower(): v for k, v in self.header_.items()})
+
+    def __getitem__(self, key: numbers.Integral) -> entity_instance:
+        return self.by_id(key)
+        
+    def by_id(self, id: int) -> entity_instance:
+        """Return an IFC entity instance filtered by IFC ID.
+
+        :param id: STEP numerical identifier
+        :type id: int
+
+        :raises RuntimeError: If `id` is not found or multiple definitions exist for `id`.
+
+        :rtype: entity_instance
+        """
+        ns = self.data_.get(id, [])
+        if len(ns) == 0:
+            raise RuntimeError(f"Instance with id {id} not found")
+        elif len(ns) > 1:
+            raise RuntimeError(f"Duplicate definition for id {id}")
+        return entity_instance(ns, ns[0])
+
+    def by_type(self, type: str) -> list[entity_instance]:
+        """Return IFC objects filtered by IFC Type and wrapped with the entity_instance class.
+        :rtype: list[entity_instance]
+        """
+        type_lc = type.lower()
+        return list(filter(lambda ent: ent.type.lower() == type_lc, itertools.chain.from_iterable(self.data_.values())))
+
+def open(fn) -> file:
+    return file(parse(filename=fn, with_tree=True, with_header=True))
+
 if __name__ == "__main__":
     args = [x for x in sys.argv[1:] if not x.startswith("-")]
     flags = [x for x in sys.argv[1:] if x.startswith("-")]