packages feed

moonlight-triangulation-1.0.1.0: bindings/python/src/moonlight_triangulation/__init__.py

from __future__ import annotations

import ctypes
import os
import weakref
from collections.abc import Callable, Sequence
from pathlib import Path
from typing import Final


class _Obstruction(ctypes.Structure):
    _fields_ = [
        ("code", ctypes.c_uint32),
        ("coordinate_error", ctypes.c_uint32),
        ("input_index", ctypes.c_uint64),
        ("first_index", ctypes.c_uint64),
        ("second_index", ctypes.c_uint64),
        ("first_value", ctypes.c_double),
        ("second_value", ctypes.c_double),
        ("point_x", ctypes.c_double),
        ("point_y", ctypes.c_double),
        ("message", ctypes.c_char * 256),
    ]


class MoonlightError(RuntimeError):
    def __init__(self, status: int, obstruction: _Obstruction) -> None:
        self.status = status
        self.code = obstruction.code
        self.coordinate_error = obstruction.coordinate_error
        self.input_index = obstruction.input_index
        self.first_index = obstruction.first_index
        self.second_index = obstruction.second_index
        self.first_value = obstruction.first_value
        self.second_value = obstruction.second_value
        self.point = (obstruction.point_x, obstruction.point_y)
        message = bytes(obstruction.message).split(b"\0", 1)[0].decode("utf-8", errors="replace")
        super().__init__(message or f"Moonlight ABI failure {status}:{self.code}")


class _NativeApi:
    _OK: Final = 0

    def __init__(self, library_path: Path) -> None:
        library = ctypes.CDLL(str(library_path))
        mesh = ctypes.c_void_p
        mesh_output = ctypes.POINTER(mesh)
        obstruction = ctypes.POINTER(_Obstruction)
        count_output = ctypes.POINTER(ctypes.c_size_t)
        _configure(library, "ml_abi_version", ())
        _configure(library, "ml_runtime_initialize", ())
        _configure(library, "ml_delaunay_f64", (ctypes.POINTER(ctypes.c_double), ctypes.c_size_t, mesh_output, obstruction))
        _configure(library, "ml_mesh_insert_many_f64", (mesh, ctypes.POINTER(ctypes.c_double), ctypes.c_size_t, mesh_output, obstruction))
        _configure(library, "ml_mesh_union", (mesh, mesh, mesh_output, obstruction))
        _configure(library, "ml_mesh_intersection", (mesh, mesh, mesh_output, obstruction))
        _configure(library, "ml_mesh_difference", (mesh, mesh, mesh_output, obstruction))
        _configure(library, "ml_mesh_symmetric_difference", (mesh, mesh, mesh_output, obstruction))
        _configure(library, "ml_mesh_vertex_count", (mesh, count_output, obstruction))
        _configure(library, "ml_mesh_triangle_count", (mesh, count_output, obstruction))
        _configure(library, "ml_mesh_copy_vertices_f64", (mesh, ctypes.POINTER(ctypes.c_double), ctypes.c_size_t, count_output, obstruction))
        _configure(library, "ml_mesh_copy_triangles_u32", (mesh, ctypes.POINTER(ctypes.c_uint32), ctypes.c_size_t, count_output, obstruction))
        _configure(library, "ml_mesh_free", (mesh,), None)

        status = int(library.ml_runtime_initialize())
        if status != self._OK:
            raise RuntimeError(f"Moonlight runtime initialization failed with status {status}")
        abi_version = int(library.ml_abi_version())
        if abi_version != 1:
            raise RuntimeError(f"unsupported Moonlight ABI version {abi_version}")
        self.library = library

    def check(self, status: int, obstruction: _Obstruction) -> None:
        if status != self._OK:
            raise MoonlightError(status, obstruction)


Point = tuple[float, float]
Triangle = tuple[int, int, int]
_BinaryNativeOperation = Callable[[ctypes.c_void_p, ctypes.c_void_p, object, object], int]


class Moonlight:
    def __init__(self, library_path: str | os.PathLike[str] | None = None) -> None:
        configured_path = library_path or os.environ.get("MOONLIGHT_TRIANGULATION_LIBRARY")
        if configured_path is None:
            raise ValueError("set MOONLIGHT_TRIANGULATION_LIBRARY or pass library_path")
        self._native = _NativeApi(Path(configured_path))

    def delaunay(self, points: Sequence[Point]) -> Mesh:
        coordinates, pointer = _coordinate_buffer(points)
        output = ctypes.c_void_p()
        obstruction = _Obstruction()
        status = int(
            self._native.library.ml_delaunay_f64(
                pointer,
                len(points),
                ctypes.byref(output),
                ctypes.byref(obstruction),
            )
        )
        self._native.check(status, obstruction)
        return Mesh(self._native, _required_handle(output))


class Mesh:
    __slots__ = ("_native", "_handle", "_finalizer", "__weakref__")

    def __init__(self, native: _NativeApi, handle: ctypes.c_void_p) -> None:
        self._native = native
        self._handle = handle
        self._finalizer = weakref.finalize(self, native.library.ml_mesh_free, handle)

    def close(self) -> None:
        self._finalizer()
        self._handle = ctypes.c_void_p()

    def __enter__(self) -> Mesh:
        return self

    def __exit__(self, _type: object, _value: object, _traceback: object) -> None:
        self.close()

    @property
    def vertex_count(self) -> int:
        return self._count(self._native.library.ml_mesh_vertex_count)

    @property
    def triangle_count(self) -> int:
        return self._count(self._native.library.ml_mesh_triangle_count)

    @property
    def vertices(self) -> tuple[Point, ...]:
        count = self.vertex_count
        output = (ctypes.c_double * (count * 2))()
        written = ctypes.c_size_t()
        obstruction = _Obstruction()
        status = int(
            self._native.library.ml_mesh_copy_vertices_f64(
                self._live_handle(),
                output,
                count,
                ctypes.byref(written),
                ctypes.byref(obstruction),
            )
        )
        self._native.check(status, obstruction)
        return tuple((float(output[index * 2]), float(output[index * 2 + 1])) for index in range(written.value))

    @property
    def triangles(self) -> tuple[Triangle, ...]:
        count = self.triangle_count
        output = (ctypes.c_uint32 * (count * 3))()
        written = ctypes.c_size_t()
        obstruction = _Obstruction()
        status = int(
            self._native.library.ml_mesh_copy_triangles_u32(
                self._live_handle(),
                output,
                count,
                ctypes.byref(written),
                ctypes.byref(obstruction),
            )
        )
        self._native.check(status, obstruction)
        return tuple(
            (int(output[index * 3]), int(output[index * 3 + 1]), int(output[index * 3 + 2]))
            for index in range(written.value)
        )

    def insert_many(self, points: Sequence[Point]) -> Mesh:
        coordinates, pointer = _coordinate_buffer(points)
        output = ctypes.c_void_p()
        obstruction = _Obstruction()
        status = int(
            self._native.library.ml_mesh_insert_many_f64(
                self._live_handle(),
                pointer,
                len(points),
                ctypes.byref(output),
                ctypes.byref(obstruction),
            )
        )
        self._native.check(status, obstruction)
        return Mesh(self._native, _required_handle(output))

    def union(self, other: Mesh) -> Mesh:
        return self._binary(other, self._native.library.ml_mesh_union)

    def intersection(self, other: Mesh) -> Mesh:
        return self._binary(other, self._native.library.ml_mesh_intersection)

    def difference(self, other: Mesh) -> Mesh:
        return self._binary(other, self._native.library.ml_mesh_difference)

    def symmetric_difference(self, other: Mesh) -> Mesh:
        return self._binary(other, self._native.library.ml_mesh_symmetric_difference)

    def _binary(self, other: Mesh, operation: _BinaryNativeOperation) -> Mesh:
        if self._native is not other._native:
            raise ValueError("both meshes must belong to the same Moonlight runtime")
        output = ctypes.c_void_p()
        obstruction = _Obstruction()
        status = int(
            operation(
                self._live_handle(),
                other._live_handle(),
                ctypes.byref(output),
                ctypes.byref(obstruction),
            )
        )
        self._native.check(status, obstruction)
        return Mesh(self._native, _required_handle(output))

    def _count(self, operation: Callable[[ctypes.c_void_p, object, object], int]) -> int:
        output = ctypes.c_size_t()
        obstruction = _Obstruction()
        status = int(operation(self._live_handle(), ctypes.byref(output), ctypes.byref(obstruction)))
        self._native.check(status, obstruction)
        return int(output.value)

    def _live_handle(self) -> ctypes.c_void_p:
        if not self._finalizer.alive:
            raise RuntimeError("mesh is closed")
        return self._handle


def _coordinate_buffer(points: Sequence[Point]) -> tuple[object, object]:
    values = tuple(component for x, y in points for component in (x, y))
    if not values:
        return (), None
    coordinates = (ctypes.c_double * len(values))(*values)
    return coordinates, coordinates


def _required_handle(handle: ctypes.c_void_p) -> ctypes.c_void_p:
    if not handle.value:
        raise RuntimeError("Moonlight returned success without a mesh handle")
    return handle


def _configure(library: ctypes.CDLL, name: str, parameters: Sequence[object], result: object = ctypes.c_uint32) -> None:
    function = getattr(library, name)
    setattr(function, "argtypes", list(parameters))
    setattr(function, "restype", result)


__all__ = ["Mesh", "Moonlight", "MoonlightError", "Point", "Triangle"]