BuildingSmart IFCRail changes to BIMTester (#1309)

* Add a way to validate the IFC file in BIMTester

* Fix an issue in Report where only the first error was shown

* Fix the usage of Schema in BIMTester

* Add Aggreation steps to BIMTester

src/ifcbimtester/bimtester/features/steps/aggregation/en.py

@@ -0,0 +1,191 @@
+from behave import step, given, when, then, use_step_matcher
+from bimtester.ifc import IfcStore
+from bimtester.table import TableModel
+from bimtester import util
+
+use_step_matcher("parse")
+
+
+@step('There must be exactly {number} "{ifc_class}" element')
+@step('There must be exactly {number} "{ifc_class}" elements')
+def step_impl(context, number, ifc_class):
+    num = len(IfcStore.file.by_type(ifc_class))
+    assert num == int(
+        number
+    ), "Could not find {} elements of {}. \
+        Found {} element(s).".format(
+        number, ifc_class, num
+    )
+
+
+@given('a set of (key,value) called ("{key_name}","{value_name}")')
+def step_impl(context, key_name, value_name):
+    model = getattr(context, "model", None)
+    if not model:
+        context.model = TableModel(key_name, value_name)
+    for row in context.table:
+        context.model.add_row(row[key_name], row[value_name])
+
+
+@given('a set of (key,value) called ("{key_name}","{value_name}") taken from the file "{path_file}"')
+def step_impl(context, key_name, value_name, path_file):
+    import csv
+    import os
+
+    model = getattr(context, "model", None)
+    if not model:
+        context.model = TableModel(key_name, value_name)
+    if context.config.userdata.get("path"):
+        path_file = os.path.join(context.config.userdata.get("path"), path_file)
+    if not os.path.exists(path_file):
+        assert False, "File {} not found".format(path_file)
+    with open(path_file, "r", encoding="utf-8-sig") as csvfile:
+        reader = csv.DictReader(csvfile)
+        for row in reader:
+            context.model.add_row(row[key_name], row[value_name])
+
+
+@then('there is an element of type key with an attribute "{attribute_name}" equals to value')
+def step_impl(context, attribute_name):
+    if not context.model.key_name and not context.model.value_name:
+        assert False, "Missing (key,value)"
+    errors = []
+    rows = context.model.rows
+    for key, values in rows.items():
+        for value in values:
+            found = check_equals(attribute_name, value, IfcStore.file.by_type(key))
+            if not found:
+                errors.append(f"The row ({key}, {value}) was not found.")
+    assert not errors, "Errors occured:\n{}".format("\n".join(errors))
+
+
+@then('there is an element of type key with an attribute "{attribute_name}" starting with value')
+def step_impl(context, attribute_name):
+    # import re
+    if not context.model.key_name and not context.model.value_name:
+        assert False, "Missing (key,value)"
+    errors = []
+    rows = context.model.rows
+    for key, values in rows.items():
+        for value in values:
+            found = check_starts_with(attribute_name, value, IfcStore.file.by_type(key))
+            if not found:
+                errors.append(f"The row ({key}, {value}) was not found.")
+    assert not errors, "Errors occured:\n{}".format("\n".join(errors))
+
+
+@then('there must be exactly a number of "{ifc_class}" equals to the number of distinct value')
+def step_impl(context, ifc_class):
+    try:
+        context.execute_steps(
+            """
+            then There must be exactly {number} "{ifc_class}" elements
+            """.format(
+                ifc_class=ifc_class, number=context.model.get_count_distinct_values()
+            )
+        )
+    except AssertionError as error:
+        str_error = str(error)
+        assert False, str_error[: str_error.find("Traceback")]
+    assert True
+
+
+@then('there is a relationship "{ifc_class}" between the two elements of each row')
+def step_impl(context, ifc_class):
+    if not context.model.key_name and not context.model.value_name:
+        assert False, "Missing (key,value)"
+    errors = []
+    elements = IfcStore.file.by_type(ifc_class)
+
+    # first, loop on the dataset to check if there are all in the relationships
+    rows = context.model.rows
+    for key, values in rows.items():
+        found = False
+        # value is actually a list of values
+        for value in values:
+            for element in elements:
+                if (
+                    any(x.Name == key for x in getattr(element, context.model.key_name))
+                    and getattr(element, context.model.value_name).Name == value
+                ):
+                    found = True
+            if not found:
+                errors.append(f"The row ({key}, {value}) does not have the relationship.")
+
+    # second, loop on the elements to check if there are not too many relationships
+    for element in elements:
+        if hasattr(element, context.model.key_name) and hasattr(element, context.model.value_name):
+            # list here
+            keys = getattr(element, context.model.key_name)
+            value = getattr(element, context.model.value_name).Name
+            for key in keys:
+                if not key.Name in rows or not value in rows[key.Name]:
+                    errors.append(f"The element ({element}) has a relation that cannot be found in the dataset: {key}")
+    assert not errors, "Errors occured:\n{}".format("\n".join(errors))
+
+
+use_step_matcher("re")
+
+
+@step('all IfcGroup must be linked to a type "(?P<linked_ifc_classes>.*)"')
+def step_impl(context, linked_ifc_classes):
+    groups = IfcStore.file.by_type("IfcGroup")
+    errors = []
+    for group in groups:
+        if not hasattr(group, "IsGroupedBy"):
+            errors.append(f'The element "{group.Name}" has no "IsGroupedBy" attribute.')
+        else:
+            for grouped_by in getattr(group, "IsGroupedBy"):
+                if not hasattr(grouped_by, "RelatedObjects"):
+                    errors.append(f'The element "{grouped_by.Name}" has no "RelatedObjects" attribute.')
+                else:
+                    for related_object in getattr(grouped_by, "RelatedObjects"):
+                        found = False
+                        for linked_ifc_class in linked_ifc_classes.split(","):
+                            if related_object.is_a(linked_ifc_class):
+                                found = True
+                        if not found:
+                            errors.append(
+                                f'The element "{related_object.Name}" does not have the right associated type.'
+                            )
+    assert not errors, "Errors occured:\n{}".format("\n".join(errors))
+
+
+@then('there is an element of type "(?P<ifc_types>.*)" with an attribute "(?P<attribute_name>.*)" for each key')
+def step_impl(context, ifc_types, attribute_name):
+    check_if_element_exists_by_types_with_attribute_name(ifc_types, attribute_name, context.model.rows.keys())
+
+
+@then('there is an element of type "(?P<ifc_types>.*)" with an attribute "(?P<attribute_name>.*)" for each value')
+def step_impl(context, ifc_types, attribute_name):
+    values = set(item for sublist in context.model.rows.values() for item in sublist)
+    check_if_element_exists_by_types_with_attribute_name(ifc_types, attribute_name, values)
+
+
+def check_if_element_exists_by_types_with_attribute_name(ifc_types, attribute_name, attribute_values):
+    errors = []
+    # retrieve all elements of that type
+    elements = util.by_types(IfcStore.file, ifc_types)
+    # loop
+    for attribute_value in attribute_values:
+        found = False
+        for element in elements:
+            if hasattr(element, attribute_name) and getattr(element, attribute_name) == attribute_value:
+                found = True
+        if not found:
+            errors.append(f'An element with attribute "{attribute_name}" equals to "{attribute_value}" was not found.')
+    assert not errors, "Errors occured:\n{}".format("\n".join(errors))
+
+
+def check_starts_with(attribute_name, value, elements):
+    """Make sure at least an element of the list has a attribute starting with the value"""
+    if any(hasattr(x, attribute_name) and getattr(x, attribute_name).startswith(value) for x in elements):
+        return True
+    return False
+
+
+def check_equals(attribute_name, value, elements):
+    """Make sure at least an element of the list has a attribute equals to the value"""
+    if any(hasattr(x, attribute_name) and getattr(x, attribute_name) == value for x in elements):
+        return True
+    return False

src/ifcbimtester/bimtester/features/steps/all.py

@@ -1,15 +1,23 @@
 use_step_matcher("parse")
 from bimtester.features.steps.classification import en
+
 use_step_matcher("parse")
 from bimtester.features.steps.element_classes import en
+
 use_step_matcher("parse")
 from bimtester.features.steps.geocoding import en
+
 use_step_matcher("parse")
 from bimtester.features.steps.geolocation import en
+
 use_step_matcher("parse")
 from bimtester.features.steps.geometric_detail import en
+
 use_step_matcher("parse")
 from bimtester.features.steps.model_federation import en
+
 use_step_matcher("parse")
 from bimtester.features.steps.project_setup import de, en, fr, it, nl
+
 use_step_matcher("parse")
+from bimtester.features.steps.aggregation import en

src/ifcbimtester/bimtester/features/steps/project_setup/en.py

@@ -10,6 +10,12 @@ def step_impl(context, schema):
     assert real_schema == schema, _("We expected a schema of {} but instead got {}").format(schema, real_schema)
 
 
+@step("The IFC file must be valid")
+def step_impl(context):
+    errors = util.validate(IfcStore.file)
+    assert not errors, "Errors occured:\n{}".format("\n".join(errors))
+
+
 @step('The IFC file "{file}" is exempt from being provided')
 def step_impl(context, file):
     pass

src/ifcbimtester/bimtester/reports.py

@@ -93,9 +93,7 @@ class ReportGenerator:
             step["result"] = {}
             step["result"]["status"] = "skipped"
             step["result"]["duration"] = 0
-            step["result"][
-                "error_message"
-            ] = "This requirement has been skipped due to a previous failing step."
+            step["result"]["error_message"] = "This requirement has been skipped due to a previous failing step."
         elif step["result"]["status"] == "undefined":
             step["result"] = {}
             step["result"]["status"] = "undefined"
@@ -108,14 +106,12 @@ class ReportGenerator:
             "is_success": step["result"]["status"] == "passed",
             "is_unspecified": step["result"]["status"] == "undefined",
             "is_skipped": step["result"]["status"] == "skipped",
-            "error_message": None
-            if step["result"]["status"] == "passed"
-            else step["result"]["error_message"],
+            "error_message": None if step["result"]["status"] == "passed" else step["result"]["error_message"],
         }
 
-        # TODO: there is probably a better way of doing this
-        if isinstance(data["error_message"], list):
-            data["error_message"] = data["error_message"][1]
+        # Remove the first "Assertion Failed" message
+        if isinstance(data["error_message"], list) and data["error_message"]:
+            data["error_message"].pop(0)
         return data
 
     def get_template_strings(self):

src/ifcbimtester/bimtester/run.py

@@ -3,10 +3,11 @@ import sys
 import shutil
 import tempfile
 import ifcopenshell
+
 try:
     import ifcopenshell.express
 except:
-    pass # They are using an old version of IfcOpenShell. Gracefully degrade for now.
+    pass  # They are using an old version of IfcOpenShell. Gracefully degrade for now.
 import behave.formatter.pretty  # Needed for pyinstaller to package it
 from bimtester.ifc import IfcStore
 from distutils.dir_util import copy_tree
@@ -14,8 +15,14 @@ from behave.__main__ import main as behave_main
 
 
 class TestRunner:
-    def __init__(self, ifc_path, ifc=None):
+    def __init__(self, ifc_path, schema_path=None, ifc=None):
         IfcStore.path = ifc_path
+
+        # can't load the IFC file if the schema is not loaded before
+        if schema_path:
+            schema = ifcopenshell.express.parse(schema_path)
+            ifcopenshell.register_schema(schema)
+
         IfcStore.file = ifc if ifc else ifcopenshell.open(ifc_path)
 
         try:
@@ -27,10 +34,6 @@ class TestRunner:
         self.locale_path = os.path.join(self.base_path, "locale")
 
     def run(self, args):
-        if args["schema_file"]:
-            schema = ifcopenshell.express.parse(args["schema_file"])
-            ifcopenshell.register_schema(args["schema_name"])
-
         tmpdir = tempfile.mkdtemp()
         features_path = os.path.join(tmpdir, "features")
         steps_path = os.path.join(features_path, "steps")

src/ifcbimtester/bimtester/table.py

@@ -0,0 +1,19 @@
+from collections import defaultdict
+
+
+class TableModel:
+    """This class represents a table of data"""
+
+    def __init__(self, key_name=None, value_name=None):
+        self.rows = defaultdict(list)
+        self.key_name = key_name
+        self.value_name = value_name
+
+    def add_row(self, related, relating):
+        self.rows[related].append(relating)
+
+    def get_count(self):
+        return len(self.rows)
+
+    def get_count_distinct_values(self):
+        return len(set(item for sublist in self.rows.values() for item in sublist))

src/ifcbimtester/bimtester/util.py

@@ -1,8 +1,29 @@
 import ifcopenshell
 import ifcopenshell.util.element
+import ifcopenshell.validate
+import json
 from bimtester.lang import _
 
 
+def by_types(ifc, ifc_types):
+    elements = []
+    for ifc_type in ifc_types.split(","):
+        elements += ifc.by_type(ifc_type.strip())
+    return elements
+
+
+def validate(ifc):
+    errors = []
+    logger = ifcopenshell.validate.json_logger()
+    try:
+        ifcopenshell.validate.validate(ifc, logger)
+        if logger.statements:
+            for statement in logger.statements:
+                errors.append(f"{json.dumps(statement, default=str)}")
+    except RuntimeError as error:
+        assert False, str(error)
+
+
 def assert_guid(ifc, guid):
     try:
         return ifc.by_guid(guid)

src/ifcbimtester/cli.py

@@ -22,7 +22,7 @@ parser.add_argument("--lang", type=str, help="Specify a language e.g. en/de/fr/i
 args = vars(parser.parse_args())
 
 if args["action"] == "run":
-    report_json = bimtester.run.TestRunner(args["ifc"]).run(args)
+    report_json = bimtester.run.TestRunner(args["ifc"], args["schema_file"]).run(args)
     if args["report"]:
         bimtester.reports.ReportGenerator().generate(report_json, args["report"])
 elif args["action"] == "purge":