code/ezflow_v3/app/premesh.py

"""Detail-preserving premesher for the v2 app.

The data pipeline's `make_watertight` rebuilds every body as a smooth voxel
*envelope* (great for CFD robustness, but it erases mirrors/wheels/sharp edges).
For app uploads we usually want to KEEP the real geometry:

  prepare_surface()  - if the upload is already watertight, keep its true surface
                       (only light cleanup + optional decimation); fall back to a
                       higher-resolution voxel envelope only for non-watertight
                       "triangle soup". Normalizes the body to L = 1.
  build_volume()     - external-flow tet mesh refined near the body, with optional
                       curvature-aware sizing so detailed regions get a finer mesh.

RESOLUTION presets trade geometric detail against node count / speed.

NOTE: the surrogate models were trained on the *enveloped* (smoothed) geometries,
so a high-detail mesh mainly improves the geometry/visualisation and lets the model
run at higher resolution; fully exploiting fine features would require retraining on
high-fidelity geometry.
"""
from __future__ import annotations
import os, math
os.environ.setdefault("KMP_DUPLICATE_LIB_OK", "TRUE")
os.environ.setdefault("OMP_NUM_THREADS", "1")
from pathlib import Path
import numpy as np
from datasets.gen.shapenet.watertight import make_watertight, _load_mesh
from datasets.gen.shapenet.mesh_stl import _add_box_surfaces

# preset -> meshing knobs. h_* in units of body length L_ref = 1.
PRESETS = {
    "coarse": dict(h_min=0.12,  h_max=1.2, dist_min=0.4, dist_max=4.0, curv=0,
                   surf_faces=8000,   vox_res=64,  preserve=False),   # = legacy envelope
    "medium": dict(h_min=0.045, h_max=0.9, dist_min=0.2, dist_max=3.0, curv=20,
                   surf_faces=80000,  vox_res=128, preserve=True),
    "fine":   dict(h_min=0.022, h_max=0.7, dist_min=0.1, dist_max=2.5, curv=30,
                   surf_faces=150000, vox_res=192, preserve=True),
    # much denser near-wall band. Slow to build and heavy on memory -> the app auto-runs inference
    # on CPU when the node count would overflow GPU VRAM. h_min is in body lengths (0.012*L),
    # so on a 5 m car this is ~60 mm cells; a true 16 mm would need h_min~0.003 and explode the count.
    "very fine": dict(h_min=0.012, h_max=0.6, dist_min=0.08, dist_max=2.2, curv=40,
                      surf_faces=250000, vox_res=256, preserve=True),
}
DEFAULT_PRESET = "medium"


def quick_extents(in_stl):
    """Raw bounding-box extents (x, y, z) in the file's native units, no repair/meshing.

    Used to advise a reference length before simulation: the model normalizes geometry to unit
    size, so the real longest dimension has to come from the file. process=False keeps it fast
    on heavy CAD (no vertex merging)."""
    import trimesh
    m = trimesh.load(str(in_stl), process=False, force="mesh")
    e = m.extents
    return [float(e[0]), float(e[1]), float(e[2])]


def inspect_geometry(in_stl) -> dict:
    """Cheap watertightness / topology report for an STL (one load, no meshing).

    Returns a dict the UI can show BEFORE meshing so the user knows what they uploaded:
    watertight?, face/vertex counts, Euler number + genus (for closed bodies), number of
    open (boundary) edges, and whether the face winding is consistent.
    """
    import trimesh
    try:
        mesh = _load_mesh(Path(in_stl))
    except Exception as e:
        return {"ok": False, "error": f"{type(e).__name__}: {e}"}
    if mesh.is_empty or len(mesh.faces) == 0:
        return {"ok": False, "error": "empty / unreadable mesh"}
    wt = bool(mesh.is_watertight)
    try:
        euler = int(mesh.euler_number)
    except Exception:
        euler = None
    try:                                          # edges used by exactly one face = open boundary
        open_edges = int(len(trimesh.grouping.group_rows(mesh.edges_sorted, require_count=1)))
    except Exception:
        open_edges = None
    try:
        winding = bool(mesh.is_winding_consistent)
    except Exception:
        winding = None
    genus = ((2 - euler) // 2) if (wt and euler is not None) else None
    return {"ok": True, "watertight": wt, "n_faces": int(len(mesh.faces)),
            "n_vertices": int(len(mesh.vertices)), "euler": euler, "genus": genus,
            "open_edges": open_edges, "winding_consistent": winding}


def _repair_mesh(mesh):
    """Light, geometry-preserving repair: merge coincident verts, drop degenerate/duplicate
    faces, make winding/normals consistent, and fill small holes. Returns (mesh, notes)."""
    import trimesh
    notes = []
    n0 = len(mesh.faces); wt0 = bool(mesh.is_watertight)
    mesh.merge_vertices()
    mesh.update_faces(mesh.nondegenerate_faces())
    mesh.update_faces(mesh.unique_faces())
    mesh.remove_unreferenced_vertices()
    try:
        trimesh.repair.fix_winding(mesh)
        trimesh.repair.fix_inversion(mesh)
        trimesh.repair.fix_normals(mesh)
    except Exception:
        pass
    if not mesh.is_watertight:
        try:
            trimesh.repair.fill_holes(mesh)       # triangle-fan fill of small boundary loops
        except Exception:
            pass
    if len(mesh.faces) != n0:
        notes.append(f"cleaned {n0}->{len(mesh.faces)} faces")
    if not wt0 and bool(mesh.is_watertight):
        notes.append("filled holes -> watertight")
    return mesh, notes


def prepare_surface(in_stl, out_stl, preset=DEFAULT_PRESET, target_L=1.0,
                    force_envelope=False, repair="auto") -> dict:
    cfg = PRESETS[preset]
    mesh = _load_mesh(Path(in_stl))
    if mesh.is_empty or len(mesh.faces) == 0:
        raise ValueError(f"empty mesh: {in_stl}")
    mesh.apply_translation(-mesh.centroid)
    mesh.apply_scale(target_L / float(mesh.extents.max()))      # normalize L=1, centered
    # remedy for non-watertight uploads: try a light repair first (keeps real geometry);
    # if it becomes watertight+genus0 we take the detail-preserving path below, otherwise
    # we still fall through to the robust voxel envelope.
    repair_notes = []
    if repair == "auto" and not force_envelope and not bool(mesh.is_watertight):
        mesh, repair_notes = _repair_mesh(mesh)
    watertight = bool(mesh.is_watertight)

    # direct (true-geometry) path only for clean, genus-0 closed surfaces; complex/
    # high-genus bodies (DrivAer underbody/wheels) go straight to the robust envelope.
    genus0 = False
    try:
        genus0 = int(mesh.euler_number) == 2
    except Exception:
        genus0 = False
    if cfg["preserve"] and watertight and genus0 and not force_envelope:
        # KEEP true geometry: light cleanup + decimate only if very heavy (quadric
        # decimation preserves features far better than a voxel envelope).
        mesh.update_faces(mesh.nondegenerate_faces())
        mesh.remove_unreferenced_vertices()
        mesh.fix_normals()
        if len(mesh.faces) > cfg["surf_faces"]:
            try:
                mesh = mesh.simplify_quadric_decimation(face_count=cfg["surf_faces"])
                mesh.fix_normals()
            except Exception:
                pass
        try:
            a_ref = float(abs(mesh.projected(normal=[1.0, 0.0, 0.0]).area))
        except Exception:
            a_ref = float(mesh.extents[1] * mesh.extents[2])
        if not np.isfinite(a_ref) or a_ref <= 1e-6:
            a_ref = float(mesh.extents[1] * mesh.extents[2])
        Path(out_stl).parent.mkdir(parents=True, exist_ok=True)
        mesh.export(str(out_stl))
        method = f"direct (true geometry, {len(mesh.faces)} faces)"
        if repair_notes:
            method += " [repaired: " + "; ".join(repair_notes) + "]"
        return {"L_ref": float(mesh.extents.max()), "A_ref": a_ref,
                "extents": [float(x) for x in mesh.extents], "n_faces": int(len(mesh.faces)),
                "watertight": True, "method": method}

    # voxel envelope (smooth genus-0 solid). Keep the face count low (~8k) so gmsh's
    # surface parametrization stays robust; vox_res controls how faithfully the blob
    # samples the body (higher = closer to the true silhouette, still a smooth shell).
    meta = make_watertight(Path(in_stl), Path(out_stl), target_L=target_L,
                           vox_res=cfg["vox_res"], max_faces=8000)
    tag = "" if watertight else " [non-watertight input]"
    meta["method"] = f"voxel envelope (res {cfg['vox_res']}){tag}"
    return meta


def build_volume(stl_path, out_msh, preset=DEFAULT_PRESET) -> dict:
    """External-flow tet mesh around the body surface.

    Primary path reparametrizes the surface (classifySurfaces + createGeometry): nice smooth
    patches, but gmsh CANNOT reparametrize a high-genus surface, so it fails on cars/aircraft
    envelopes (wheels/wings/props punch handles -> "Wrong topology of boundary mesh"). On that
    failure we fall back to a reparametrization-FREE discrete mesh: the merged STL triangles are
    used directly as the volume boundary, which meshes at ANY genus (just no smooth patches)."""
    cfg = PRESETS[preset]
    import gmsh

    def _build(discrete):
        # interruptible=False -> skip gmsh's signal.signal(SIGINT) install, which raises
        # "signal only works in main thread" when meshing runs on a background worker thread.
        gmsh.initialize(interruptible=False)
        gmsh.option.setNumber("General.Terminal", 0)
        gmsh.option.setNumber("Geometry.Tolerance", 1e-4)
        gmsh.option.setNumber("Mesh.ToleranceInitialDelaunay", 1e-6)
        gmsh.option.setNumber("Mesh.Algorithm3D", 1)
        gmsh.model.add("v2app")
        try:
            gmsh.merge(str(stl_path))
            if not discrete:                       # reparametrize the STL into smooth CAD patches
                gmsh.model.mesh.classifySurfaces(math.radians(40), True, True, math.radians(180))
                gmsh.model.mesh.createGeometry()
            body_surfs = [e[1] for e in gmsh.model.getEntities(2)]
            if not body_surfs:
                raise RuntimeError("no body surface recovered from STL")
            body_loop = gmsh.model.geo.addSurfaceLoop(body_surfs)
            box_surfs, cls = _add_box_surfaces(gmsh)
            box_loop = gmsh.model.geo.addSurfaceLoop(box_surfs)
            vol = gmsh.model.geo.addVolume([box_loop, body_loop])
            gmsh.model.geo.synchronize()
            gmsh.model.addPhysicalGroup(3, [vol], name="fluid")
            gmsh.model.addPhysicalGroup(2, cls["inlet"], name="inlet")
            gmsh.model.addPhysicalGroup(2, cls["outlet"], name="outlet")
            gmsh.model.addPhysicalGroup(2, cls["farfield"], name="farfield")
            gmsh.model.addPhysicalGroup(2, body_surfs, name="wall")
            # size field: fine near body, coarse far. Sampling 100 (600 is pathologically slow).
            gmsh.model.mesh.field.add("Distance", 1)
            gmsh.model.mesh.field.setNumbers(1, "SurfacesList", body_surfs)
            gmsh.model.mesh.field.setNumber(1, "Sampling", 100)
            gmsh.model.mesh.field.add("Threshold", 2)
            gmsh.model.mesh.field.setNumber(2, "InField", 1)
            gmsh.model.mesh.field.setNumber(2, "SizeMin", cfg["h_min"])
            gmsh.model.mesh.field.setNumber(2, "SizeMax", cfg["h_max"])
            gmsh.model.mesh.field.setNumber(2, "DistMin", cfg["dist_min"])
            gmsh.model.mesh.field.setNumber(2, "DistMax", cfg["dist_max"])
            gmsh.model.mesh.field.setAsBackgroundMesh(2)
            gmsh.option.setNumber("Mesh.MeshSizeExtendFromBoundary", 0)
            gmsh.option.setNumber("Mesh.MeshSizeFromPoints", 0)
            gmsh.option.setNumber("Mesh.MeshSizeMin", cfg["h_min"])
            # curvature refinement only makes sense on the reparametrized patches
            gmsh.option.setNumber("Mesh.MeshSizeFromCurvature", 0 if discrete else cfg["curv"])
            gmsh.model.mesh.generate(3)
            try:
                gmsh.model.mesh.optimize("Netgen")
            except Exception:
                pass
            out = Path(out_msh); out.parent.mkdir(parents=True, exist_ok=True)
            gmsh.write(str(out))
            n_nodes = gmsh.model.mesh.getNodes()[0].size
            return {"n_body_surfs": len(body_surfs), "n_nodes_approx": int(n_nodes),
                    "mesher": "discrete" if discrete else "reparam"}
        finally:
            gmsh.finalize()

    try:
        return _build(discrete=False)
    except Exception:
        return _build(discrete=True)               # robust fallback: meshes any genus (F1, cars, ...)