Attaches custom IFC metadata to IFC4 files as property sets and relationships

ifc_metadata_attach.py

import ifcopenshell
import uuid
import csv
from typing import List, Dict, Any


ALLOWED_AUTH_TYPES = {"none", "token", "mtls"}


def new_guid() -> str:
    return ifcopenshell.guid.compress(uuid.uuid1().hex)


def get_or_create_property_set(ifc_file, ifc_entity, pset_name: str):
    for rel in getattr(ifc_entity, "IsDefinedBy", []):
        if rel.is_a("IfcRelDefinesByProperties"):
            pset = rel.RelatingPropertyDefinition
            if pset.is_a("IfcPropertySet") and pset.Name == pset_name:
                return pset

    pset = ifc_file.create_entity(
        "IfcPropertySet",
        GlobalId=new_guid(),
        Name=pset_name,
        HasProperties=[],
    )

    ifc_file.create_entity(
        "IfcRelDefinesByProperties",
        GlobalId=new_guid(),
        RelatedObjects=[ifc_entity],
        RelatingPropertyDefinition=pset,
    )

    return pset


def create_property(ifc_file, name: str, value: Any):
    if value is None:
        nominal = None
    elif isinstance(value, bool):
        nominal = ifc_file.create_entity("IfcBoolean", value)
    elif isinstance(value, int):
        nominal = ifc_file.create_entity("IfcInteger", value)
    elif isinstance(value, float):
        nominal = ifc_file.create_entity("IfcReal", value)
    else:
        nominal = ifc_file.create_entity("IfcLabel", str(value))

    return ifc_file.create_entity(
        "IfcPropertySingleValue",
        Name=name,
        NominalValue=nominal,
        Unit=None,
    )


def load_servers_from_csv(csv_path: str) -> List[Dict[str, Any]]:
    servers = []

    with open(csv_path, newline="", encoding="utf-8") as f:
        reader = csv.DictReader(f)

        required = {
            "protocol",
            "host",
            "port",
            "path",
            "metric_namespace",
            "auth_type",
        }

        if not required.issubset(reader.fieldnames):
            missing = required - set(reader.fieldnames or [])
            raise ValueError(f"Missing required CSV columns: {missing}")

        for row in reader:
            auth_type = row["auth_type"].strip().lower()
            if auth_type not in ALLOWED_AUTH_TYPES:
                raise ValueError(f"Invalid auth_type: {auth_type}")

            servers.append(
                {
                    "protocol": row["protocol"].strip(),
                    "host": row["host"].strip(),
                    "port": int(row["port"]),
                    "path": row["path"].strip(),
                    "metric_namespace": row["metric_namespace"].strip(),
                    "auth_type": auth_type,
                }
            )

    return servers


def attach_timeseries_servers(
    ifc_file,
    space,
    server_templates: List[Dict[str, Any]],
):
    pset = get_or_create_property_set(
        ifc_file,
        space,
        "Pset_TimeseriesEndpoints",
    )

    pset.HasProperties.clear()

    servers = []
    for tmpl in server_templates:
        servers.append(
            {
                **tmpl,
                "metric_namespace": tmpl["metric_namespace"].replace(
                    "{SPACE_ID}", space.GlobalId
                ),
            }
        )

    properties = [
        create_property(ifc_file, "TS_ServerCount", len(servers))
    ]

    for idx, server in enumerate(servers, start=1):
        prefix = f"TS_{idx}"

        properties.extend([
            create_property(ifc_file, f"{prefix}_Protocol", server["protocol"]),
            create_property(ifc_file, f"{prefix}_Host", server["host"]),
            create_property(ifc_file, f"{prefix}_Port", server["port"]),
            create_property(ifc_file, f"{prefix}_Path", server["path"]),
            create_property(
                ifc_file,
                f"{prefix}_MetricNamespace",
                server["metric_namespace"],
            ),
            create_property(
                ifc_file,
                f"{prefix}_AuthType",
                server["auth_type"],
            ),
        ])

    pset.HasProperties.extend(properties)


def process_ifc(
    input_ifc: str,
    output_ifc: str,
    servers_csv: str,
):
    ifc_file = ifcopenshell.open(input_ifc)

    server_templates = load_servers_from_csv(servers_csv)

    spaces = ifc_file.by_type("IfcSpace")

    for space in spaces:
        attach_timeseries_servers(
            ifc_file,
            space,
            server_templates,
        )

    ifc_file.write(output_ifc)


if __name__ == "__main__":
    process_ifc(
        input_ifc="input.ifc",
        output_ifc="output.ifc",
        servers_csv="timeseries_servers.csv",
    )

The timeseries_servers.csv file should be formatted like the example, below:

protocol,host,port,path,metric_namespace,auth_type
https,prometheus.bms.local,443,/api/v1/query,spaces.{SPACE_ID},token
mqtt,mqtt.telemetry.local,1883,building/telemetry,spaces/{SPACE_ID},none