eb2c662d0e
Some checks are pending
CI / backend-lint-and-test (push) Waiting to run
- Remove IfcWallStandardCase from parser (covered by IfcWall as parent type) - Add seen_global_ids set to prevent duplicate element insertion - Use scalars().first() instead of scalar_one_or_none() for resilient lookup - Re-parse required: clears and rebuilds element data without duplicates
133 lines
4.2 KiB
Python
133 lines
4.2 KiB
Python
"""Service for parsing IFC files and storing elements in the database."""
|
|
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
import ifcopenshell
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.element import Element, Project, Property
|
|
|
|
|
|
async def parse_ifc_file(
|
|
filename: str,
|
|
content: bytes,
|
|
db: AsyncSession,
|
|
) -> Project:
|
|
"""
|
|
Parse an IFC file and persist its structure to the database.
|
|
|
|
Extracts building elements (walls, doors, slabs, windows, etc.)
|
|
along with their property sets into the relational model.
|
|
"""
|
|
# Write to temp file (ifcopenshell needs a file path)
|
|
with tempfile.NamedTemporaryFile(suffix=".ifc", delete=False) as tmp:
|
|
tmp.write(content)
|
|
tmp_path = Path(tmp.name)
|
|
|
|
try:
|
|
ifc = ifcopenshell.open(str(tmp_path))
|
|
finally:
|
|
tmp_path.unlink()
|
|
|
|
# Create project record
|
|
project = Project(
|
|
name=_extract_project_name(ifc, filename),
|
|
filename=filename,
|
|
description=_extract_project_description(ifc),
|
|
ifc_schema=ifc.schema,
|
|
)
|
|
db.add(project)
|
|
await db.flush()
|
|
|
|
# Extract elements
|
|
element_types = [
|
|
"IfcWall", "IfcDoor", "IfcWindow", "IfcSlab",
|
|
"IfcColumn", "IfcBeam", "IfcStair", "IfcRoof",
|
|
"IfcSpace", "IfcFurnishingElement", "IfcBuildingElementProxy",
|
|
]
|
|
|
|
seen_global_ids = set()
|
|
|
|
for ifc_type in element_types:
|
|
for ifc_element in ifc.by_type(ifc_type):
|
|
if ifc_element.GlobalId in seen_global_ids:
|
|
continue
|
|
seen_global_ids.add(ifc_element.GlobalId)
|
|
storey = _get_storey(ifc_element)
|
|
|
|
element = Element(
|
|
project_id=project.id,
|
|
global_id=ifc_element.GlobalId,
|
|
ifc_type=ifc_element.is_a(),
|
|
name=ifc_element.Name or "",
|
|
description=ifc_element.Description or "",
|
|
storey=storey,
|
|
)
|
|
db.add(element)
|
|
await db.flush()
|
|
|
|
# Extract property sets
|
|
for pset_name, props in _get_properties(ifc_element).items():
|
|
for prop_name, prop_value in props.items():
|
|
prop = Property(
|
|
element_id=element.id,
|
|
pset_name=pset_name,
|
|
name=prop_name,
|
|
value=str(prop_value) if prop_value is not None else "",
|
|
)
|
|
db.add(prop)
|
|
|
|
await db.commit()
|
|
await db.refresh(project)
|
|
return project
|
|
|
|
def _extract_project_name(ifc, fallback_filename: str) -> str:
|
|
"""Try to get the project name from the IFC file."""
|
|
try:
|
|
ifc_project = ifc.by_type("IfcProject")[0]
|
|
if ifc_project.Name:
|
|
return ifc_project.Name
|
|
except (IndexError, AttributeError):
|
|
pass
|
|
return Path(fallback_filename).stem
|
|
|
|
|
|
def _extract_project_description(ifc) -> str:
|
|
"""Try to get the project description from the IFC file."""
|
|
try:
|
|
ifc_project = ifc.by_type("IfcProject")[0]
|
|
return ifc_project.Description or ""
|
|
except (IndexError, AttributeError):
|
|
return ""
|
|
|
|
|
|
def _get_storey(element) -> str:
|
|
"""Determine which building storey an element belongs to."""
|
|
try:
|
|
for rel in element.ContainedInStructure:
|
|
if rel.RelatingStructure.is_a("IfcBuildingStorey"):
|
|
return rel.RelatingStructure.Name or ""
|
|
except AttributeError:
|
|
pass
|
|
return ""
|
|
|
|
|
|
def _get_properties(element) -> dict[str, dict[str, any]]:
|
|
"""Extract all property sets from an IFC element."""
|
|
result = {}
|
|
try:
|
|
for definition in element.IsDefinedBy:
|
|
if definition.is_a("IfcRelDefinesByProperties"):
|
|
pset = definition.RelatingPropertyDefinition
|
|
if pset.is_a("IfcPropertySet"):
|
|
props = {}
|
|
for prop in pset.HasProperties:
|
|
if prop.is_a("IfcPropertySingleValue"):
|
|
value = prop.NominalValue.wrappedValue if prop.NominalValue else None
|
|
props[prop.Name] = value
|
|
result[pset.Name] = props
|
|
except AttributeError:
|
|
pass
|
|
return result
|
|
|