ci(measure): mesh_bodies_vs_ifc and saved_meshes_vs_ifc mesh across the CPUs, share their meshes and prune surface distance exactly, output unchanged (#1237)

tools/ci/measure-reference-models.sh

@@ -258,7 +258,7 @@ finished() { # dir name started
 # longest ones do not start last (the BIM4LCA office's mesh_bodies_vs_ifc took
 # 477 s of its 1,953 run one after another, run 41589642386).
 model_tasks() { # name rvt ref
-  local name="$1" rvt="$2" ref="$3" s
+  local name="$1" rvt="$2" ref="$3" s meshes=
   echo "1 base"
   # ports_vs_ifc reads the findings the schema step's validation wrote, so it
   # runs after it, in the same task.
@@ -270,8 +270,10 @@ model_tasks() { # name rvt ref
       wanted "$s" || continue
       [ "$s" = ports_vs_ifc ] && wanted schema && continue
       case "$s" in
-        mesh_bodies_vs_ifc) echo "10 ifc:$s" ;;
-        saved_meshes_vs_ifc) echo "8 ifc:$s" ;;
+        # These two mesh the same bodies of the same two files, so they run
+        # one after the other as one task and the second reads what the
+        # first meshed.
+        mesh_bodies_vs_ifc | saved_meshes_vs_ifc) meshes=1 ;;
         layer_sets_vs_ifc | wall_heights_vs_ifc | door_bodies_vs_ifc) echo "2 ifc:$s" ;;
         *) echo "1 ifc:$s" ;;
       esac
@@ -282,6 +284,7 @@ model_tasks() { # name rvt ref
       echo "1 ifc:$s"
     done
   fi
+  [ -n "$meshes" ] && echo "18 meshes"
   [ -f "$MODELS/${rvt%.rvt}.revit-sets.ifc" ] && wanted revit_property_sets_vs_ifc && echo "2 sets"
   return 0
 }
@@ -326,6 +329,14 @@ model_task() { # name rvt ref task
       s=${task#ifc:}
       score "$dir/$s.txt" "$SCORERS/$s.py" "$dir/model.ifc" "$MODELS/$ref"
       ;;
+    meshes)
+      for s in saved_meshes_vs_ifc mesh_bodies_vs_ifc; do
+        wanted "$s" || continue
+        RVT_MESH_CACHE="$dir/mesh-cache" score "$dir/$s.txt" "$SCORERS/$s.py" "$dir/model.ifc" "$MODELS/$ref"
+      done
+      echo "$name: mesh cache $(du -sh "$dir/mesh-cache" 2>/dev/null | cut -f1), removed"
+      rm -rf "$dir/mesh-cache"
+      ;;
     sets)
       # RE-210: where Revit's export with its own property sets is beside the
       # model (<model>.revit-sets.ifc, the R2025 models), an export with
@@ -350,6 +361,7 @@ model_task() { # name rvt ref task
 model_done() { # name
   local dir ms
   dir=$(model_dir "$1")
+  rm -rf "$dir/mesh-cache"
   rm -f "$dir/model.glb" "$dir/validation.jsonl"
   [ -n "${KEEP_IFC:-}" ] || rm -f "$dir/model.ifc"
   ms=$(awk '{ms += $2} END {print ms + 0}' "$dir/timings.log" 2>/dev/null || echo 0)

tools/re/ifc_common.py

@@ -1,7 +1,15 @@
 """Helpers the tools/re scorers share, one copy of each so they cannot drift."""
 
+import concurrent.futures
+import concurrent.futures.process
+import hashlib
+import multiprocessing
+import os
 import re
+import sys
+import time
 
+import ifcopenshell.geom
 import ifcopenshell.util.placement
 import numpy as np
 
@@ -36,3 +44,134 @@ def report_mesh_error(error):
     if message not in MESH_ERRORS:
         MESH_ERRORS.add(message)
         print(f"IfcOpenShell cannot mesh: {message}")
+
+
+_MESHING = {}
+# Instance id -> the first line of IfcOpenShell's reason, for each part
+# `mesh_parts` could not mesh (a part kept in the cache keeps its reason too).
+# Nothing prints them: a scorer's output does not change when a part fails.
+MESH_FAILURES = {}
+# Fewer parts than this are meshed in this process: a forked worker parses
+# again the parts of the file it meshes, which a handful do not repay.
+POOL_FROM = 16
+
+
+def _in_worker():
+    _MESHING["parent"] = os.getppid()
+
+
+def _mesh_one(part_id):
+    """(vertices, faces) of the part, or the first line of why IfcOpenShell would not mesh it."""
+    # An OOM-killed parent leaves its workers with the queued parts: stop at the next one.
+    if "parent" in _MESHING and os.getppid() != _MESHING["parent"]:
+        os._exit(1)
+    try:
+        shape = ifcopenshell.geom.create_shape(_MESHING["settings"], _MESHING["model"].by_id(part_id))
+    except RuntimeError as error:
+        return (str(error).strip().splitlines() or [""])[0] or type(error).__name__
+    return (
+        np.array(shape.geometry.verts, dtype=float).reshape(-1, 3),
+        np.array(shape.geometry.faces, dtype=int).reshape(-1, 3),
+    )
+
+
+def _settings_key(settings):
+    items = []
+    for name in sorted(settings.setting_names()):
+        try:
+            items.append(f"{name}={settings.get(name)!r}")
+        except RuntimeError:
+            items.append(f"{name}=unset")
+    return ";".join(items)
+
+
+def _mesh_cache(path, settings):
+    root = os.environ.get("RVT_MESH_CACHE")
+    if not root or not path:
+        return None
+    stat = os.stat(path)
+    key = f"{os.path.realpath(path)} {stat.st_size} {stat.st_mtime_ns} {_settings_key(settings)}"
+    return os.path.join(root, hashlib.sha256(key.encode()).hexdigest()[:16])
+
+
+def _read_meshes(path):
+    """What `_write_meshes` kept: instance id -> (vertices, faces), or the reason IfcOpenShell gave for a part it could not mesh."""
+    out = {}
+    with np.load(path, allow_pickle=False) as kept:
+        ids, reasons, nverts, nfaces = kept["ids"], kept["reasons"], kept["nverts"], kept["nfaces"]
+        verts, faces = kept["verts"], kept["faces"]
+    vstart = np.concatenate([[0], np.cumsum(nverts)])
+    fstart = np.concatenate([[0], np.cumsum(nfaces)])
+    for k, part_id in enumerate(ids):
+        out[int(part_id)] = str(reasons[k]) or (verts[vstart[k] : vstart[k + 1]], faces[fstart[k] : fstart[k + 1]])
+    return out
+
+
+def _write_meshes(path, meshes):
+    ids = list(meshes)
+    parts = [(np.empty((0, 3)), np.empty((0, 3), dtype=int)) if isinstance(meshes[i], str) else meshes[i] for i in ids]
+    scratch = f"{path}.tmp"
+    with open(scratch, "wb") as f:
+        np.savez(
+            f,
+            ids=np.array(ids, dtype=np.int64),
+            reasons=np.array([meshes[i] if isinstance(meshes[i], str) else "" for i in ids]),
+            nverts=np.array([len(v) for v, _ in parts], dtype=np.int64),
+            nfaces=np.array([len(t) for _, t in parts], dtype=np.int64),
+            verts=np.concatenate([v for v, _ in parts]),
+            faces=np.concatenate([t for _, t in parts]),
+        )
+    os.replace(scratch, path)
+
+
+def mesh_parts(model, settings, parts, path=None):
+    """Instance id -> (vertices (N, 3), faces (M, 3)) for each of `parts` that
+    IfcOpenShell meshes, one `create_shape` per part spread over the CPUs. The
+    triangles are those `create_shape` gives, in the same order (the
+    iterator's are the same, but it took twice the CPU time per element in
+    a measurement on the BIM4LCA office, and no less wall time with 12 threads
+    than with 4); a part it cannot mesh is left out, with its reason in
+    `MESH_FAILURES`.
+
+    With `RVT_MESH_CACHE` set to a directory and the model's `path`, what is
+    meshed is kept there for the next scorer to read, which Measure sets for
+    the scorers it runs one after another on one pair of files and deletes
+    afterwards. Each call keeps its own shard, named by process and time, and
+    the cache is keyed by the file's path, size and modification time and by
+    every setting."""
+    ids = list(dict.fromkeys(part.id() for part in parts))
+    cache = _mesh_cache(path, settings)
+    known = {}
+    if cache and os.path.isdir(cache):
+        for name in sorted(os.listdir(cache)):
+            if name.endswith(".npz"):
+                known.update(_read_meshes(os.path.join(cache, name)))
+    todo = [i for i in ids if i not in known]
+    cpus = len(os.sched_getaffinity(0)) if hasattr(os, "sched_getaffinity") else os.cpu_count() or 1
+    workers = min(int(os.environ.get("RVT_MESH_WORKERS") or cpus), len(todo))
+    if len(todo) < POOL_FROM:
+        workers = 1
+    _MESHING.update(model=model, settings=settings)
+    meshes = None
+    if workers > 1:
+        # A forked worker flushes the buffered output it inherits when it ends.
+        sys.stdout.flush()
+        sys.stderr.flush()
+        try:
+            with concurrent.futures.ProcessPoolExecutor(
+                workers, mp_context=multiprocessing.get_context("fork"), initializer=_in_worker
+            ) as pool:
+                meshes = list(pool.map(_mesh_one, todo, chunksize=1))
+        except concurrent.futures.process.BrokenProcessPool:
+            print("mesh_parts: a worker died, so the parts are meshed one after another", file=sys.stderr)
+            meshes = None
+    if meshes is None:
+        meshes = [_mesh_one(i) for i in todo]
+    _MESHING.clear()
+    fresh = dict(zip(todo, meshes))
+    if cache and fresh:
+        os.makedirs(cache, exist_ok=True)
+        _write_meshes(os.path.join(cache, f"{os.getpid()}-{time.time_ns()}.npz"), fresh)
+    known.update(fresh)
+    MESH_FAILURES.update({i: known[i] for i in ids if isinstance(known[i], str)})
+    return {i: known[i] for i in ids if not isinstance(known[i], str)}