"""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", "IfcWallStandardCase", "IfcDoor", "IfcWindow", "IfcSlab", "IfcColumn", "IfcBeam", "IfcStair", "IfcRoof", "IfcSpace", "IfcFurnishingElement", "IfcBuildingElementProxy", ] for ifc_type in element_types: for ifc_element in ifc.by_type(ifc_type): 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