From ed990c3e834bc80edca81e0acc3c130de7685949 Mon Sep 17 00:00:00 2001 From: Dion Moult Date: Fri, 9 Jun 2023 22:16:34 +1000 Subject: [PATCH] Experimental support for storage of IFC in MySQL and SQLite --- src/blenderbim/scripts/ifc2sql.py | 262 ++++++++++++++++++++++++++++++ 1 file changed, 262 insertions(+) create mode 100644 src/blenderbim/scripts/ifc2sql.py diff --git a/src/blenderbim/scripts/ifc2sql.py b/src/blenderbim/scripts/ifc2sql.py new file mode 100644 index 0000000000..135f037b24 --- /dev/null +++ b/src/blenderbim/scripts/ifc2sql.py @@ -0,0 +1,262 @@ +import os +import json +import itertools +import ifcopenshell +import ifcopenshell.util.schema +import ifcopenshell.util.attribute + +import sqlite3 +import mysql.connector + + +class Ifc2Sql: + def execute(self): + self.sql_type = "sqlite" + # self.sql_type = "mysql" + + self.is_strict = False + self.should_get_psets = True + self.should_expand = True + self.should_get_geometry_blob = True + self.should_skip_geometry_data = True + + filepath = "/home/dion/test2.ifc" + self.file = ifcopenshell.open(filepath) + self.schema = ifcopenshell.ifcopenshell_wrapper.schema_by_name(self.file.schema) + + if self.sql_type == "sqlite": + db_file = "/home/dion/test.db" + os.remove(db_file) + self.db = sqlite3.connect(db_file) + elif self.sql_type == "mysql": + self.db = mysql.connector.connect( + host="localhost", + user="root", + password="root", + database="test" + ) + + self.create_id_map() + if self.should_get_psets: + self.create_pset_table() + + for ifc_class in self.file.wrapped_data.types(): + declaration = self.schema.declaration_by_name(ifc_class) + + if self.should_skip_geometry_data: + if ifcopenshell.util.schema.is_a(declaration, "IfcRepresentation") or ifcopenshell.util.schema.is_a( + declaration, "IfcRepresentationItem" + ): + continue + + if self.sql_type == "sqlite": + self.create_sqlite_table(ifc_class, declaration) + elif self.sql_type == "mysql": + self.create_mysql_table(ifc_class, declaration) + self.insert_data(ifc_class) + + self.db.close() + + def create_id_map(self): + if self.sql_type == "sqlite": + statement = ( + "CREATE TABLE IF NOT EXISTS id_map (ifc_id integer PRIMARY KEY NOT NULL UNIQUE, ifc_class text);" + ) + elif self.sql_type == "mysql": + statement = """ + CREATE TABLE `id_map` ( + `ifc_id` int(10) unsigned NOT NULL, + `ifc_class` varchar(255) NOT NULL, + PRIMARY KEY (`ifc_id`) + ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb3 COLLATE=utf8mb3_general_ci; + """ + c = self.db.cursor() + c.execute(statement) + + def create_pset_table(self): + statement = """ + CREATE TABLE IF NOT EXISTS psets ( + ifc_id integer NOT NULL, + pset_name text, + name text, + value text + ); + """ + c = self.db.cursor() + c.execute(statement) + + def create_sqlite_table(self, ifc_class, declaration): + statement = f"CREATE TABLE IF NOT EXISTS {ifc_class} (" + + if self.should_expand: + statement += "ifc_id INTEGER NOT NULL," + else: + statement += "ifc_id INTEGER PRIMARY KEY NOT NULL UNIQUE," + + total_attributes = declaration.attribute_count() + derived = declaration.derived() + for i in range(0, total_attributes): + attribute = declaration.attribute_by_index(i) + primitive = ifcopenshell.util.attribute.get_primitive_type(attribute) + print(attribute.name(), primitive) + if primitive in ("string", "enum"): + data_type = "TEXT" + elif primitive in ("entity", "integer", "boolean"): + data_type = "INTEGER" + elif primitive == "float": + data_type = "REAL" + elif self.should_expand and self.is_entity_list(primitive): + data_type = "INTEGER" + elif isinstance(primitive, tuple): + data_type = "JSON" + else: + print(attribute, primitive) # Not implemented? + if not self.is_strict or derived[i]: + optional = "" + else: + optional = "" if attribute.optional() else " NOT NULL" + comma = "" if i == total_attributes - 1 else "," + statement += f" {attribute.name()} {data_type}{optional}{comma}" + statement += ");" + print(statement) + c = self.db.cursor() + c.execute(statement) + + def create_mysql_table(self, ifc_class, declaration): + declaration = self.schema.declaration_by_name(ifc_class) + statement = f"CREATE TABLE IF NOT EXISTS {ifc_class} (" + statement += "`ifc_id` int(10) unsigned NOT NULL," + derived = declaration.derived() + for attribute in declaration.all_attributes(): + primitive = ifcopenshell.util.attribute.get_primitive_type(attribute) + if primitive in ("string", "enum"): + if "IfcText" in str(attribute.type_of_attribute()): + data_type = "text" + else: + data_type = "varchar(255)" + elif primitive == "entity": + data_type = "int(10) unsigned" + elif primitive == "boolean": + data_type = "tinyint(1)" + elif primitive == "integer": + data_type = "int(10)" + if "Positive" in str(attribute.type_of_attribute()): + data_type += " unsigned" + elif primitive == "float": + data_type = "decimal(10,0)" + elif self.should_expand and self.is_entity_list(primitive): + data_type = "int(10) unsigned" + elif isinstance(primitive, tuple): + data_type = "JSON" + else: + print(attribute, primitive) # Not implemented? + if not self.is_strict or derived[i]: + optional = "DEFAULT NULL" + else: + optional = "DEFAULT NULL" if attribute.optional() else "NOT NULL" + statement += f" `{attribute.name()}` {data_type} {optional}," + statement += " PRIMARY KEY (`ifc_id`)" + statement += ") ENGINE=InnoDB DEFAULT CHARSET=utf8mb3 COLLATE=utf8mb3_general_ci;" + print(statement) + c = self.db.cursor() + c.execute(statement) + + def insert_data(self, ifc_class): + elements = self.file.by_type(ifc_class, include_subtypes=False) + + rows = [] + id_map_rows = [] + pset_rows = [] + + for element in elements: + nested_indices = [] + values = [element.id()] + for i, attribute in enumerate(element): + if isinstance(attribute, ifcopenshell.entity_instance): + if attribute.id(): + values.append(attribute.id()) + else: + values.append(json.dumps({"type": attribute.is_a(), "value": attribute.wrappedValue})) + elif ( + self.should_expand + and attribute + and isinstance(attribute, tuple) + and isinstance(attribute[0], ifcopenshell.entity_instance) + ): + nested_indices.append(i + 1) + attribute = self.serialise_value(element, attribute) + values.append(attribute) + elif isinstance(attribute, tuple): + attribute = self.serialise_value(element, attribute) + values.append(json.dumps(attribute)) + else: + values.append(attribute) + + if self.should_expand: + print(values) + rows.extend(self.get_permutations(values, nested_indices)) + else: + rows.append(values) + + id_map_rows.append([element.id(), ifc_class]) + + if self.should_get_psets: + psets = ifcopenshell.util.element.get_psets(element) + if psets: + print(psets) + for pset_name, pset_data in psets.items(): + for prop_name, value in pset_data.items(): + if prop_name == "id": + continue + pset_rows.append([element.id(), pset_name, prop_name, value]) + c = self.db.cursor() + + print(rows) + + if self.sql_type == "sqlite": + c.executemany(f"INSERT INTO {ifc_class} VALUES ({','.join(['?']*len(rows[0]))});", rows) + c.executemany("INSERT INTO id_map VALUES (?, ?);", id_map_rows) + if pset_rows: + c.executemany("INSERT INTO psets VALUES (?, ?, ?, ?);", pset_rows) + elif self.sql_type == "mysql": + c.executemany(f"INSERT INTO {ifc_class} VALUES ({','.join(['%s']*len(rows[0]))});", rows) + c.executemany("INSERT INTO id_map VALUES (%s, %s);", id_map_rows) + if pset_rows: + c.executemany("INSERT INTO psets VALUES (%s, %s, %s, %s);", pset_rows) + + self.db.commit() + + def serialise_value(self, element, value): + return element.walk( + lambda v: isinstance(v, ifcopenshell.entity_instance), + lambda v: v.id() if v.id() else {"type": v.is_a(), "value": v.wrappedValue}, + value, + ) + + def get_permutations(self, lst, indexes): + nested_lists = [lst[i] for i in indexes] + + # Generate the Cartesian product of the nested lists + products = list(itertools.product(*nested_lists)) + + # Place the elements of each product back in their original positions + final_lists = [] + for product in products: + temp_list = lst[:] + for i, index in enumerate(indexes): + temp_list[index] = product[i] + final_lists.append(temp_list) + + return final_lists + + def is_entity_list(self, primitive): + if not isinstance(primitive, tuple): + return False + elif primitive[1] == "entity": + return True + elif isinstance(primitive[1], tuple): + return self.is_entity_list(primitive[1]) + return False + + +Ifc2Sql().execute()