IFC4x3 Schema Mapping for Civil Infrastructure Python Pipelines

# ifcopenshell>=0.8.0, pandas>=2.0, shapely>=2.0, pyproj>=3.6
pip install ifcopenshell pandas shapely pyproj

IfcProject
  └─ IfcSite
       ├─ IfcAlignment          (road/rail centreline)
       │    └─ IfcAlignmentSegment (per-segment geometry)
       ├─ IfcBridge             (bridge as facility)
       │    └─ IfcBridgePart    (abutment, deck, pier, …)
       └─ IfcRailway
            └─ IfcFacilityPart

# ifcopenshell>=0.8.0
import ifcopenshell

def load_and_validate_ifc4x3(filepath: str) -> ifcopenshell.file:
    """Load an IFC file and assert IFC4X3 schema compliance."""
    # Fast pre-check: scan raw text before full parse
    with open(filepath, "r", encoding="utf-8", errors="replace") as fh:
        header_text = "".join(fh.readline() for _ in range(50))
    if "IFC4X3" not in header_text:
        raise ValueError(
            f"{filepath} does not declare IFC4X3 in FILE_SCHEMA. "
            "Check model.schema before proceeding."
        )

    model = ifcopenshell.open(filepath)
    schema = model.schema
    if not schema.startswith("IFC4X3"):
        raise ValueError(f"Expected IFC4X3, resolved schema: {schema}")
    return model

# ifcopenshell>=0.8.0
from typing import Generator

CIVIL_TYPES = ("IfcAlignment", "IfcBridge", "IfcRailway", "IfcMarineFacility")

def iter_civil_entities(model: ifcopenshell.file) -> Generator:
    """Yield civil facility entities from the IFC4x3 spatial graph."""
    for type_name in CIVIL_TYPES:
        for entity in model.by_type(type_name):
            yield entity

def iter_alignment_segments(
    alignment: "ifcopenshell.entity_instance",
) -> Generator:
    """Yield IfcAlignmentSegment children of a given IfcAlignment."""
    for rel in alignment.IsDecomposedBy:
        for child in rel.RelatedObjects:
            if child.is_a("IfcAlignmentSegment"):
                yield child
            # Recurse if nested decomposition exists
            for grandchild in child.IsDecomposedBy:
                for obj in grandchild.RelatedObjects:
                    yield obj

# ifcopenshell>=0.8.0, pandas>=2.0
import pandas as pd
from typing import Any, Dict

def extract_psets(entity) -> Dict[str, Any]:
    """Flatten all IfcPropertySet and IfcElementQuantity data to a dict."""
    props: Dict[str, Any] = {
        "ifc_guid": entity.GlobalId,
        "ifc_type": entity.is_a(),
        "name": getattr(entity, "Name", None) or "Unnamed",
    }

    for rel in getattr(entity, "IsDefinedBy", []):
        if not rel.is_a("IfcRelDefinesByProperties"):
            continue
        definition = rel.RelatingPropertyDefinition

        if definition.is_a("IfcPropertySet"):
            pset_name = definition.Name or "UnnamedPset"
            for prop in definition.HasProperties:
                key = f"{pset_name}.{prop.Name}"
                if prop.is_a("IfcPropertySingleValue"):
                    val = prop.NominalValue
                    props[key] = (
                        val.wrappedValue if hasattr(val, "wrappedValue") else str(val)
                    ) if val is not None else None
                elif prop.is_a("IfcPropertyEnumeratedValue"):
                    props[key] = [
                        v.wrappedValue if hasattr(v, "wrappedValue") else str(v)
                        for v in prop.EnumerationValues
                    ]
                elif prop.is_a("IfcPropertyTableValue"):
                    # Preserve key→value structure; do NOT collapse to scalar
                    props[key] = [
                        {
                            "k": (
                                dk.wrappedValue
                                if hasattr(dk, "wrappedValue") else str(dk)
                            ),
                            "v": (
                                dv.wrappedValue
                                if hasattr(dv, "wrappedValue") else str(dv)
                            ),
                        }
                        for dk, dv in zip(
                            prop.DefiningValues or [],
                            prop.DefinedValues or [],
                        )
                    ]

        elif definition.is_a("IfcElementQuantity"):
            qset_name = definition.Name or "UnnamedQset"
            for qty in definition.Quantities:
                key = f"{qset_name}.{qty.Name}"
                for attr in ("LengthValue", "AreaValue", "VolumeValue", "WeightValue", "CountValue"):
                    if hasattr(qty, attr) and getattr(qty, attr) is not None:
                        props[key] = getattr(qty, attr)
                        break

    return props

# pyproj>=3.6
import pyproj
from typing import Optional, Tuple
import math

def resolve_map_conversion(
    model: ifcopenshell.file,
) -> Optional[Tuple[pyproj.CRS, dict]]:
    """
    Return (target_CRS, conversion_params) if IfcMapConversion is present.
    conversion_params keys: eastings, northings, orthogonal_height,
    x_axis_abscissa, x_axis_ordinate, scale.
    """
    for ctx in model.by_type("IfcGeometricRepresentationContext"):
        if getattr(ctx, "ContextType", None) != "Model":
            continue
        for op in getattr(ctx, "HasCoordinateOperation", []):
            if not op.is_a("IfcMapConversion"):
                continue
            target = op.TargetCRS
            if target is None:
                continue
            epsg_name = getattr(target, "Name", "") or ""
            try:
                crs = pyproj.CRS.from_string(epsg_name)
            except Exception:
                continue
            params = {
                "eastings": op.Eastings or 0.0,
                "northings": op.Northings or 0.0,
                "orthogonal_height": op.OrthogonalHeight or 0.0,
                "x_axis_abscissa": op.XAxisAbscissa or 1.0,
                "x_axis_ordinate": op.XAxisOrdinate or 0.0,
                "scale": op.Scale or 1.0,
            }
            return crs, params
    return None

def apply_map_conversion(
    local_x: float,
    local_y: float,
    params: dict,
) -> Tuple[float, float]:
    """Rotate and translate a local (x,y) into the mapped CRS frame."""
    angle = math.atan2(params["x_axis_ordinate"], params["x_axis_abscissa"])
    scale = params["scale"]
    rx = local_x * math.cos(angle) - local_y * math.sin(angle)
    ry = local_x * math.sin(angle) + local_y * math.cos(angle)
    return rx * scale + params["eastings"], ry * scale + params["northings"]

# ifcopenshell>=0.8.0, pandas>=2.0, pyproj>=3.6
import json
import warnings

def run_ifc4x3_pipeline(filepath: str, output_parquet: str) -> pd.DataFrame:
    """Execute the full IFC4x3 schema mapping pipeline."""
    model = load_and_validate_ifc4x3(filepath)

    crs_result = resolve_map_conversion(model)
    if crs_result:
        target_crs, conv_params = crs_result
        transformer = pyproj.Transformer.from_crs(
            target_crs, pyproj.CRS.from_epsg(4326), always_xy=True
        )
        print(f"CRS resolved: {target_crs.to_epsg()}")
    else:
        transformer = None
        conv_params = None
        warnings.warn(
            "No IfcProjectedCRS found. Coordinates retained in local frame. "
            "Tag output with crs_source=local_fallback."
        )

    records = []
    for entity in iter_civil_entities(model):
        record = extract_psets(entity)
        record["crs_source"] = (
            f"EPSG:{target_crs.to_epsg()}" if transformer else "local_fallback"
        )
        records.append(record)

    df = pd.DataFrame(records)
    df.to_parquet(output_parquet, index=False)
    print(f"Exported {len(df)} civil entities → {output_parquet}")
    return df

import re

VENDOR_PREFIX_RE = re.compile(r"^(?:Civil_|Structural_|MEP_)")

def normalize_pset_name(raw_name: str) -> str:
    return VENDOR_PREFIX_RE.sub("", raw_name)

# ifcopenshell>=0.8.0, pandas>=2.0
import pytest

def validate_pipeline_output(
    model: ifcopenshell.file, df: pd.DataFrame
) -> None:
    """Raise AssertionError on any structural mapping failure."""
    assert model.schema.startswith("IFC4X3"), (
        f"Wrong schema: {model.schema}"
    )

    for alignment in model.by_type("IfcAlignment"):
        segments = list(iter_alignment_segments(alignment))
        assert segments, (
            f"IfcAlignment {alignment.GlobalId} has no segments — "
            "authoring error or unsupported representation"
        )

    assert df["ifc_guid"].is_unique, (
        "Duplicate GUIDs in output — check multi-file merge logic"
    )

    alignment_rows = df[df["ifc_type"] == "IfcAlignment"]
    if not alignment_rows.empty:
        missing = alignment_rows[
            alignment_rows.get("Pset_AlignmentCommon.StartStation", pd.Series(dtype=object)).isna()
        ]
        if not missing.empty:
            import warnings
            warnings.warn(
                f"{len(missing)} alignment(s) missing StartStation — "
                "check authoring tool Pset export settings"
            )

# pandas>=2.0
import pyarrow as pa
import pyarrow.parquet as pq

CHUNK_SIZE = 5_000

def write_chunked_parquet(
    entity_iter,
    output_path: str,
) -> int:
    """Write entity records to Parquet in chunks; return total row count."""
    writer = None
    total = 0
    chunk: list = []

    for entity in entity_iter:
        chunk.append(extract_psets(entity))
        if len(chunk) >= CHUNK_SIZE:
            table = pa.Table.from_pylist(chunk)
            if writer is None:
                writer = pq.ParquetWriter(output_path, table.schema)
            writer.write_table(table)
            total += len(chunk)
            chunk = []

    if chunk:
        table = pa.Table.from_pylist(chunk)
        if writer is None:
            writer = pq.ParquetWriter(output_path, table.schema)
        writer.write_table(table)
        total += len(chunk)

    if writer:
        writer.close()
    return total