native.py

"""The native pieces of the pipeline, wherever it runs.

On the Space: point-cloud-utils, xatlas, and QuadriFlow through ctypes with
Instant Meshes behind it. In the browser the same Python runs under Pyodide,
and the worker registers `remesh_native_js`: QuadriFlow, xatlas and a
closest-point search compiled to WebAssembly (browser/build.sh).
"""

import numpy as np

try:
    import remesh_native_js as _js
    from pyodide.ffi import to_js
except ImportError:
    _js = None

if _js is None:
    import point_cloud_utils as pcu
    import pynanoinstantmeshes as pim
    import xatlas

    import quadriflow


def _array(proxy, dtype):
    return np.asarray(proxy.to_py()).astype(dtype, copy=False)


def closest_points_on_mesh(points, v, tris):
    """For each point: distance to the mesh, the nearest triangle, and where on it (barycentric)."""
    points = np.ascontiguousarray(points, dtype=np.float64)
    v = np.ascontiguousarray(v, dtype=np.float64)
    tris = np.ascontiguousarray(tris, dtype=np.int32)
    if _js is None:
        return pcu.closest_points_on_mesh(points, v, tris)
    out = _js.closest(to_js(points.ravel()), to_js(v.ravel()), to_js(tris.ravel()))
    return _array(out.distance, np.float64), _array(out.face, np.int32), _array(out.bc, np.float64).reshape(-1, 3)


def parametrize(v, tris):
    """UVs from xatlas: (the input vertex of each new vertex, faces, uvs)."""
    v = np.ascontiguousarray(v, dtype=np.float32)
    tris = np.ascontiguousarray(tris, dtype=np.uint32)
    if _js is None:
        return xatlas.parametrize(v, tris)
    out = _js.unwrap(to_js(v.ravel()), to_js(tris.ravel()))
    vmap = _array(out.vmap, np.uint32)
    if len(vmap) == 0:
        raise RuntimeError("The new mesh could not be unwrapped.")
    return vmap, _array(out.indices, np.uint32).reshape(-1, 3), _array(out.uvs, np.float32).reshape(-1, 2)


def quadriflow_remesh(v, f, quads, seed=0):
    """One QuadriFlow run: (vertices, quads). Raises RuntimeError when the solver gives up."""
    if _js is None:
        # In a child process, so a crash or a stall costs one attempt.
        return quadriflow.remesh(v, f, quads, seed=seed)
    v = np.ascontiguousarray(v, dtype=np.float64)
    f = np.ascontiguousarray(f, dtype=np.int32)
    out = _js.quadriflow(to_js(v.ravel()), to_js(f.ravel()), int(quads), int(seed))
    if getattr(out, "fatal", False):
        # The solver crashed and no fresh instance is ready: nothing more can
        # run here. Not a RuntimeError, so no other seed or face count is tried.
        raise MemoryError(str(out.error))
    if out.error:
        raise RuntimeError(str(out.error))
    return _array(out.vertices, np.float64).reshape(-1, 3), _array(out.faces, np.int32).reshape(-1, 4).astype(np.int64)


def instant_meshes(v, f, quads):
    """Instant Meshes, the fallback when QuadriFlow gives up: (vertices, quads)."""
    if _js is not None:
        # Not built for the browser: the page sends such a model to the server instead.
        raise RuntimeError("QuadriFlow could not remesh this model.")
    # The binding's count is about a quarter of the quads it produces.
    out_v, out_f = pim.remesh(v, f, max(64, int(quads / 4)), rosy=4, posy=4, deterministic=True)
    return np.asarray(out_v, dtype=np.float64), np.asarray(out_f, dtype=np.int64)