mirror of
https://github.com/IfcOpenShell/IfcOpenShell.git
synced 2026-08-09 17:31:45 +00:00
Experimental support for storage of IFC in MySQL and SQLite
This commit is contained in:
Dion Moult
@@ -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()
|
||||
Reference in New Issue
Block a user