diff --git a/api/mcp_server.py b/api/mcp_server.py index fa9127cf2..3c7968f6c 100644 --- a/api/mcp_server.py +++ b/api/mcp_server.py @@ -19,6 +19,7 @@ """ import asyncio +import json import mimetypes import httpx from mcp.server import Server @@ -83,6 +84,37 @@ async def list_tools() -> list[Tool]: "required": ["job_id"], }, ), + Tool( + name="modly_inspect_mesh", + description=( + "Diagnose a mesh without modifying it: watertight, manifold, holes, " + "self-intersections, parts, dimensions, volume, UVs/normals, printability." + ), + inputSchema={ + "type": "object", + "properties": { + "path": { + "type": "string", + "description": "Workspace-relative path to the mesh (e.g. 'Default/mesh.glb').", + }, + }, + "required": ["path"], + }, + ), + Tool( + name="modly_suggest_fixes", + description="Inspect a mesh and list the operations (with params) that would fix its problems.", + inputSchema={ + "type": "object", + "properties": { + "path": { + "type": "string", + "description": "Workspace-relative path to the mesh (e.g. 'Default/mesh.glb').", + }, + }, + "required": ["path"], + }, + ), Tool( name="modly_decimate_mesh", description="Reduce the polygon count of a mesh using quadric edge collapse decimation.", @@ -217,6 +249,12 @@ async def _dispatch(client: httpx.AsyncClient, name: str, args: dict) -> str: parts.append(f"Error: {s['error']}") return " | ".join(parts) + elif name in ("modly_inspect_mesh", "modly_suggest_fixes"): + op = name.removeprefix("modly_") + r = await client.post(f"{API_BASE}/optimize/op/{op}", json={"path": args["path"]}) + r.raise_for_status() + return json.dumps(r.json(), indent=2) + elif name == "modly_decimate_mesh": r = await client.post( f"{API_BASE}/optimize/mesh", diff --git a/api/routers/agent.py b/api/routers/agent.py index 1fb389eb6..6a563b9b0 100644 --- a/api/routers/agent.py +++ b/api/routers/agent.py @@ -28,6 +28,8 @@ - **list_models** — List all downloaded 3D generation models ready to use. - **unload_models** — Unload all 3D generation models from GPU VRAM to free memory. - **get_mesh_info** — Get info about the current mesh in the 3D viewer (path, triangle count). +- **inspect_mesh(path)** — Diagnose a mesh without modifying it: watertight, manifold, holes, self-intersections, parts, dimensions, volume, UVs/normals, and whether it is 3D-printable. +- **suggest_fixes(path)** — Inspect a mesh and list the operations (with params) that would fix its problems. - **decimate_mesh(path, target_faces)** — Reduce the polygon count of a mesh. - **smooth_mesh(path, iterations)** — Apply Laplacian smoothing to a mesh. - **get_generation_status(job_id)** — Poll the status of an ongoing 3D generation job. @@ -39,6 +41,7 @@ - Always use tools to act on the scene — never just describe what you would do. - If you need the current mesh path, call get_mesh_info first. +- To answer questions about a mesh's quality or printability, call inspect_mesh — never guess. Before repairing a mesh, call suggest_fixes to pick the right operations. - If you need to run a workflow but don't know the ID, call list_workflows first. - To create a workflow, ONLY use extension ids listed under "Available extensions" in the context. Never invent an id. Chain steps so each step's input type matches the previous step's output type. - For a workflow's input, `input_type` MUST be exactly one of: `image`, `text`, or `mesh`. These map to the Image, Text, and Load 3D Mesh nodes. Never invent another input. Pick the one matching the first step's expected input. @@ -72,6 +75,40 @@ "parameters": {"type": "object", "properties": {}}, }, }, + { + "type": "function", + "function": { + "name": "inspect_mesh", + "description": "Diagnose a mesh without modifying it: watertight, manifold, holes, self-intersections, disconnected parts, dimensions, volume, UVs/normals, and whether it is 3D-printable.", + "parameters": { + "type": "object", + "properties": { + "path": { + "type": "string", + "description": "Workspace-relative path to the mesh file. Use get_mesh_info to obtain it.", + }, + }, + "required": ["path"], + }, + }, + }, + { + "type": "function", + "function": { + "name": "suggest_fixes", + "description": "Inspect a mesh and list the operations (with parameters) that would fix its problems. Does not modify the mesh.", + "parameters": { + "type": "object", + "properties": { + "path": { + "type": "string", + "description": "Workspace-relative path to the mesh file. Use get_mesh_info to obtain it.", + }, + }, + "required": ["path"], + }, + }, + }, { "type": "function", "function": { @@ -259,6 +296,42 @@ def _build_workflow_graph(name: str, description: str, input_type: str, steps: l return {"name": name, "description": description, "nodes": all_nodes, "edges": edges} +def _format_inspection(report: dict) -> str: + """Render an inspect_mesh report as compact text for the model.""" + def yes_no(value) -> str: + return "unknown" if value is None else ("yes" if value else "no") + + dims = " × ".join(f"{d:.4g}" for d in report.get("dimensions", [])) + volume = report.get("volume") + self_hits = report.get("self_intersecting_face_count") + lines = [ + f"Printable: {yes_no(report.get('printable'))}", + f"Faces: {report.get('face_count', '?'):,}, vertices: {report.get('vertex_count', '?'):,}, parts: {report.get('component_count', '?')}", + f"Watertight: {yes_no(report.get('is_watertight'))}, manifold: {yes_no(report.get('is_manifold'))}, consistent winding: {yes_no(report.get('is_winding_consistent'))}", + f"Holes: {report.get('hole_count', 0)} ({report.get('boundary_edge_count', 0)} open edges), non-manifold edges: {report.get('non_manifold_edge_count', 0)}, degenerate faces: {report.get('degenerate_face_count', 0)}", + f"Self-intersecting faces: {'not checked' if self_hits is None else self_hits}", + f"Dimensions: {dims}, surface area: {report.get('surface_area', 0):.4g}, volume: {'n/a (not closed)' if volume is None else f'{volume:.4g}'}", + f"UVs: {yes_no(report.get('has_uvs'))}, normals: {yes_no(report.get('has_normals'))}", + ] + issues = report.get("issues") or [] + if issues: + lines.append("Issues: " + "; ".join(issues)) + return "\n".join(lines) + + +def _format_fixes(data: dict) -> str: + """Render a suggest_fixes result as compact text for the model.""" + fixes = data.get("fixes") or [] + notes = data.get("notes") or [] + if not fixes and not notes: + return "No fixes needed — the mesh has no detected problems." + lines = ["Suggested operations:"] if fixes else ["No operation can fix the detected problems."] + for fix in fixes: + lines.append(f"- {fix['op']} {json.dumps(fix.get('params', {}))}: {fix.get('reason', '')}") + lines.extend(f"Note: {note}" for note in notes) + return "\n".join(lines) + + async def execute_tool(name: str, arguments: dict, context: dict) -> tuple[str, dict | None]: """Execute a tool and return (result_text, action_payload). action_payload carries data the frontend needs to react (e.g. new mesh URL). @@ -289,6 +362,22 @@ async def execute_tool(name: str, arguments: dict, context: dict) -> tuple[str, info += f" ({mesh_triangles:,} triangles)" return info, None + elif name == "inspect_mesh": + r = await client.post( + f"{MODLY_API}/optimize/op/inspect_mesh", + json={"path": arguments["path"]}, + ) + r.raise_for_status() + return _format_inspection(r.json()), None + + elif name == "suggest_fixes": + r = await client.post( + f"{MODLY_API}/optimize/op/suggest_fixes", + json={"path": arguments["path"]}, + ) + r.raise_for_status() + return _format_fixes(r.json()), None + elif name == "decimate_mesh": r = await client.post( f"{MODLY_API}/optimize/mesh", diff --git a/api/routers/optimize.py b/api/routers/optimize.py index c325ff206..bcfb9f6b0 100644 --- a/api/routers/optimize.py +++ b/api/routers/optimize.py @@ -101,6 +101,10 @@ def _run_operation( def _operation_response(result: MeshOpResult) -> dict[str, object]: + if result.file_path is None: + # Analysis ops (inspect_mesh, suggest_fixes) return only their report. + return dict(result.details) + output_path = result.file_path.resolve() try: relative_path = output_path.relative_to(registry.WORKSPACE_DIR.resolve()).as_posix() diff --git a/api/services/mesh_ops/builtin.py b/api/services/mesh_ops/builtin.py index 900bd1933..75574fe41 100644 --- a/api/services/mesh_ops/builtin.py +++ b/api/services/mesh_ops/builtin.py @@ -1,5 +1,6 @@ """Built-in mesh operation definitions.""" +from .inspection import inspect_mesh, suggest_fixes from .operations import decimate_mesh, repair_mesh, smooth_mesh from .types import MeshOp @@ -106,6 +107,19 @@ ) +INSPECT_PARAMS = ( + { + "id": "check_self_intersections", + "label": "Check Self-Intersections", + "type": "boolean", + "default": True, + "tooltip": ( + "Detect intersecting faces. The slowest check on very dense meshes." + ), + }, +) + + BUILTIN_MESH_OPS = ( MeshOp( id="repair", @@ -128,4 +142,18 @@ fn=smooth_mesh, category="optimization", ), + MeshOp( + id="inspect_mesh", + label="Inspect Mesh", + params_schema=INSPECT_PARAMS, + fn=inspect_mesh, + category="analysis", + ), + MeshOp( + id="suggest_fixes", + label="Suggest Fixes", + params_schema=INSPECT_PARAMS, + fn=suggest_fixes, + category="analysis", + ), ) diff --git a/api/services/mesh_ops/inspection.py b/api/services/mesh_ops/inspection.py new file mode 100644 index 000000000..a8dfc3f14 --- /dev/null +++ b/api/services/mesh_ops/inspection.py @@ -0,0 +1,358 @@ +"""Read-only mesh diagnostics: inspection report and derived fix suggestions.""" + +import json +import struct +from pathlib import Path +from typing import Any, Mapping, Optional + +from .types import ( + MeshOpContext, + MeshOpExecutionError, + MeshOpResult, + MeshOpUnavailableError, +) + + +# Above this triangle count, suggest a decimation pass (slicers and the viewer +# both struggle well before the generators' multi-million-face outputs). +HIGH_FACE_COUNT = 500_000 +DECIMATE_TARGET_FACES = 200_000 +# Mirrors the repair op's max_hole_size bounds. +MAX_HOLE_SIZE_BOUNDS = (10, 10000) + + +def _trimesh(): + try: + import trimesh + except ImportError as exc: + raise MeshOpUnavailableError( + "inspect_mesh: trimesh is not available on this system" + ) from exc + return trimesh + + +def _scene_meshes(loaded, trimesh_module) -> list[tuple]: + """Return (geometry, world transform or None) for every non-empty triangle mesh.""" + if not isinstance(loaded, trimesh_module.Scene): + candidates = [(loaded, None)] + else: + candidates = [] + for node in loaded.graph.nodes_geometry: + transform, geometry_name = loaded.graph[node] + candidates.append((loaded.geometry.get(geometry_name), transform)) + return [ + (geometry, transform) + for geometry, transform in candidates + if isinstance(geometry, trimesh_module.Trimesh) and len(geometry.faces) + ] + + +def _has_uvs(geometry, trimesh_module) -> bool: + visual = geometry.visual + if not isinstance(visual, trimesh_module.visual.TextureVisuals): + return False + return visual.uv is not None and len(visual.uv) > 0 + + +def _gltf_has_normals(document: Mapping[str, Any]) -> bool: + primitives = [ + primitive + for mesh in document.get("meshes", []) + for primitive in mesh.get("primitives", []) + # Only triangle primitives (mode 4, the default) are inspected. + if primitive.get("mode", 4) == 4 + ] + return bool(primitives) and all( + "NORMAL" in primitive.get("attributes", {}) for primitive in primitives + ) + + +def _file_has_normals(input_path: Path) -> Optional[bool]: + """Whether the file stores vertex normals; None when the format is unknown. + + trimesh computes normals on demand, so its meshes always have them; the + file itself is the only reliable source. + """ + suffix = input_path.suffix.lower() + try: + if suffix == ".glb": + with open(input_path, "rb") as handle: + header = handle.read(20) + if len(header) < 20 or header[:4] != b"glTF": + return None + chunk_length = struct.unpack_from(" dict[str, Any]: + import numpy as np + + edges, counts = np.unique(merged.edges_sorted, axis=0, return_counts=True) + boundary = edges[counts == 1] + loops = ( + trimesh_module.graph.connected_components(boundary, min_len=1) + if len(boundary) + else [] + ) + return { + "boundary_edge_count": int(len(boundary)), + "non_manifold_edge_count": int(np.count_nonzero(counts > 2)), + "hole_count": int(len(loops)), + # A simple boundary loop has as many edges as vertices. + "largest_hole_edges": int(max((len(loop) for loop in loops), default=0)), + } + + +def _self_intersecting_faces(merged, context: MeshOpContext) -> Optional[int]: + try: + import pymeshlab + except ImportError: + context.log("Self-intersection check skipped: pymeshlab is unavailable") + return None + + mesh_set = pymeshlab.MeshSet() + mesh_set.add_mesh( + pymeshlab.Mesh(vertex_matrix=merged.vertices, face_matrix=merged.faces) + ) + try: + mesh_set.compute_selection_by_self_intersections_per_face() + except Exception as exc: + context.log(f"Self-intersection check skipped: {exc}") + return None + return int(mesh_set.current_mesh().selected_face_number()) + + +def _printable(report: Mapping[str, Any]) -> Optional[bool]: + """True/False when the verdict is certain; None when it hinges on an unchecked test.""" + if not ( + report["is_watertight"] + and report["is_winding_consistent"] + and report["is_manifold"] + ): + return False + if report["self_intersecting_face_count"] is None: + return None + return report["self_intersecting_face_count"] == 0 + + +def _issues(report: Mapping[str, Any]) -> list[str]: + issues = [] + if report["hole_count"]: + issues.append( + f"{report['hole_count']} hole(s) " + f"({report['boundary_edge_count']} open boundary edges)" + ) + if report["non_manifold_edge_count"]: + issues.append( + f"{report['non_manifold_edge_count']} non-manifold edge(s)" + ) + if not report["is_winding_consistent"]: + issues.append("inconsistent face winding (normals flipped on some faces)") + if report["self_intersecting_face_count"]: + issues.append( + f"{report['self_intersecting_face_count']} self-intersecting face(s)" + ) + if report["self_intersecting_face_count"] is None: + issues.append("self-intersections were not checked") + if report["degenerate_face_count"]: + issues.append(f"{report['degenerate_face_count']} degenerate face(s)") + if report["component_count"] > 1: + issues.append(f"{report['component_count']} disconnected parts") + return issues + + +def build_report( + input_path: Path, + context: MeshOpContext, + check_self_intersections: bool = True, +) -> dict[str, Any]: + """Diagnose a mesh file without modifying it; every value is JSON-safe.""" + trimesh = _trimesh() + + context.progress(10, "Loading mesh…") + loaded = trimesh.load(str(input_path), process=False) + parts = _scene_meshes(loaded, trimesh) + if not parts: + raise MeshOpExecutionError( + f"inspect_mesh: {input_path.name} contains no triangle mesh" + ) + + context.progress(30, "Analysing topology…") + merged = _merged_geometry(parts, trimesh) + topology = _edge_topology(merged, trimesh) + is_watertight = bool(merged.is_watertight) + is_winding_consistent = bool(merged.is_winding_consistent) + + self_intersections = None + if check_self_intersections: + context.progress(60, "Checking self-intersections…") + self_intersections = _self_intersecting_faces(merged, context) + + context.progress(90, "Measuring…") + bounds = merged.bounds + report: dict[str, Any] = { + "vertex_count": int(sum(len(geometry.vertices) for geometry, _ in parts)), + "face_count": int(sum(len(geometry.faces) for geometry, _ in parts)), + "component_count": int(merged.body_count), + "is_watertight": is_watertight, + "is_winding_consistent": is_winding_consistent, + "is_manifold": topology["non_manifold_edge_count"] == 0, + **topology, + "degenerate_face_count": int( + len(merged.faces) - int(merged.nondegenerate_faces().sum()) + ), + "self_intersecting_face_count": self_intersections, + "bounds": { + "min": [float(value) for value in bounds[0]], + "max": [float(value) for value in bounds[1]], + }, + "dimensions": [float(value) for value in merged.extents], + "surface_area": float(merged.area), + # The signed volume of an open or mis-wound mesh is meaningless. + "volume": ( + float(abs(merged.volume)) + if is_watertight and is_winding_consistent + else None + ), + "has_uvs": any(_has_uvs(geometry, trimesh) for geometry, _ in parts), + "has_normals": _file_has_normals(input_path), + } + report["printable"] = _printable(report) + report["issues"] = _issues(report) + context.log( + f"Inspected {report['face_count']} faces: " + + ("printable" if report["printable"] else "; ".join(report["issues"])) + ) + context.progress(100, "Done") + return report + + +def fixes_for_report(report: Mapping[str, Any]) -> tuple[list[dict], list[str]]: + """Derive registry op calls from a report, plus notes for unfixable issues.""" + fixes: list[dict] = [] + notes: list[str] = [] + + repair_reasons = [] + if report["hole_count"]: + repair_reasons.append(f"fill {report['hole_count']} hole(s)") + if report["non_manifold_edge_count"]: + repair_reasons.append("fix non-manifold edges") + if report["degenerate_face_count"]: + repair_reasons.append("remove degenerate faces") + if repair_reasons: + low, high = MAX_HOLE_SIZE_BOUNDS + fixes.append( + { + "op": "repair", + "params": { + "remove_duplicates": True, + "remove_degenerate": bool(report["degenerate_face_count"]), + "fix_non_manifold": bool(report["non_manifold_edge_count"]), + "fill_holes": bool(report["hole_count"]), + "max_hole_size": max( + low, min(high, report["largest_hole_edges"] or low) + ), + }, + "reason": "Repair to " + ", ".join(repair_reasons) + ".", + } + ) + if report["largest_hole_edges"] > high: + notes.append( + f"The largest hole has {report['largest_hole_edges']} boundary " + f"edges, above the repair limit of {high}; it may stay open." + ) + + if report["face_count"] > HIGH_FACE_COUNT: + fixes.append( + { + "op": "decimate", + "params": {"target_faces": DECIMATE_TARGET_FACES}, + "reason": ( + f"{report['face_count']} faces is heavy; reduce to " + f"{DECIMATE_TARGET_FACES}." + ), + } + ) + + if not report["is_winding_consistent"]: + notes.append("Inconsistent winding has no automatic fix yet.") + if report["self_intersecting_face_count"]: + notes.append("Self-intersections have no automatic fix yet.") + if report["component_count"] > 1: + notes.append( + f"{report['component_count']} disconnected parts; remove stray " + "fragments manually if they are not intended." + ) + return fixes, notes + + +def inspect_mesh( + input_path: Path, + params: Mapping[str, Any], + context: MeshOpContext, +) -> MeshOpResult: + report = build_report( + input_path, + context, + check_self_intersections=bool(params.get("check_self_intersections", True)), + ) + return MeshOpResult(details=report) + + +def suggest_fixes( + input_path: Path, + params: Mapping[str, Any], + context: MeshOpContext, +) -> MeshOpResult: + report = build_report( + input_path, + context, + check_self_intersections=bool(params.get("check_self_intersections", True)), + ) + fixes, notes = fixes_for_report(report) + return MeshOpResult(details={"report": report, "fixes": fixes, "notes": notes}) diff --git a/api/services/mesh_ops/processor.py b/api/services/mesh_ops/processor.py index 9af04353c..f31c5376d 100644 --- a/api/services/mesh_ops/processor.py +++ b/api/services/mesh_ops/processor.py @@ -56,6 +56,10 @@ def run_processor(operation_id: str, processor_id: str) -> None: data.get("params") or {}, context, ) + if result.file_path is None: + raise ValueError( + f"{processor_id}: {operation_id} produces no mesh file" + ) _emit( { "type": "done", diff --git a/api/services/mesh_ops/types.py b/api/services/mesh_ops/types.py index 5951bfe12..1a1929db6 100644 --- a/api/services/mesh_ops/types.py +++ b/api/services/mesh_ops/types.py @@ -44,9 +44,12 @@ def log(self, message: str) -> None: @dataclass(frozen=True) class MeshOpResult: - """The file produced by an operation and optional JSON-safe measurements.""" + """The file produced by an operation and optional JSON-safe measurements. - file_path: Path + Read-only (analysis) operations produce no file and leave file_path None. + """ + + file_path: Optional[Path] = None details: Mapping[str, Any] = field(default_factory=dict) diff --git a/api/tests/test_agent_router.py b/api/tests/test_agent_router.py index ecd2ed7c1..4de9cc07f 100644 --- a/api/tests/test_agent_router.py +++ b/api/tests/test_agent_router.py @@ -49,5 +49,74 @@ def handler(request: httpx.Request) -> httpx.Response: self.assertIsNone(payload) +class InspectMeshToolTests(unittest.TestCase): + REPORT = { + "printable": False, + "face_count": 11, + "vertex_count": 8, + "component_count": 1, + "is_watertight": False, + "is_manifold": True, + "is_winding_consistent": True, + "hole_count": 1, + "boundary_edge_count": 3, + "non_manifold_edge_count": 0, + "degenerate_face_count": 0, + "self_intersecting_face_count": None, + "dimensions": [1.0, 2.0, 0.5], + "surface_area": 5.5, + "volume": None, + "has_uvs": False, + "has_normals": True, + "issues": ["1 hole(s) (3 open boundary edges)"], + } + + def _run(self, name: str, response: dict) -> tuple[str, list]: + requests = [] + + def handler(request: httpx.Request) -> httpx.Response: + requests.append(request) + return httpx.Response(200, json=response) + + factory = _MockClientFactory(handler) + with mock.patch.object(agent.httpx, "AsyncClient", factory): + text, payload = asyncio.run( + agent.execute_tool(name, {"path": "Default/mesh.glb"}, {}) + ) + self.assertIsNone(payload) + return text, requests + + def test_inspect_mesh_calls_the_registry_op_and_summarises(self) -> None: + text, requests = self._run("inspect_mesh", self.REPORT) + + self.assertEqual(requests[0].url.path, "/optimize/op/inspect_mesh") + self.assertIn("Printable: no", text) + self.assertIn("Watertight: no", text) + self.assertIn("Self-intersecting faces: not checked", text) + self.assertIn("volume: n/a", text) + self.assertIn("Issues: 1 hole(s)", text) + + def test_suggest_fixes_lists_ops_and_notes(self) -> None: + text, requests = self._run( + "suggest_fixes", + { + "report": self.REPORT, + "fixes": [ + {"op": "repair", "params": {"fill_holes": True}, "reason": "Repair to fill 1 hole(s)."} + ], + "notes": ["2 disconnected parts"], + }, + ) + + self.assertEqual(requests[0].url.path, "/optimize/op/suggest_fixes") + self.assertIn('- repair {"fill_holes": true}', text) + self.assertIn("Note: 2 disconnected parts", text) + + def test_suggest_fixes_on_a_clean_mesh(self) -> None: + text, _ = self._run("suggest_fixes", {"report": {}, "fixes": [], "notes": []}) + + self.assertIn("No fixes needed", text) + + if __name__ == "__main__": unittest.main() diff --git a/api/tests/test_mesh_ops_inspection.py b/api/tests/test_mesh_ops_inspection.py new file mode 100644 index 000000000..f1ff4fc4c --- /dev/null +++ b/api/tests/test_mesh_ops_inspection.py @@ -0,0 +1,249 @@ +import hashlib +import importlib.util +import tempfile +import unittest +from pathlib import Path + +import numpy as np +import trimesh + +from services.mesh_ops import MeshOpContext, MeshOpExecutionError, mesh_ops_registry +from services.mesh_ops.inspection import ( + DECIMATE_TARGET_FACES, + HIGH_FACE_COUNT, + fixes_for_report, +) + + +HAS_PYMESHLAB = importlib.util.find_spec("pymeshlab") is not None + + +def _box(offset=(0.0, 0.0, 0.0)) -> trimesh.Trimesh: + mesh = trimesh.creation.box() + mesh.apply_translation(offset) + return mesh + + +class InspectMeshTests(unittest.TestCase): + def setUp(self) -> None: + self._directory = tempfile.TemporaryDirectory() + self.workspace = Path(self._directory.name) + self.context = MeshOpContext( + workspace_dir=self.workspace, + temp_dir=self.workspace, + ) + + def tearDown(self) -> None: + self._directory.cleanup() + + def _export(self, geometry, name="mesh.glb", **kwargs) -> Path: + path = self.workspace / name + geometry.export(str(path), **kwargs) + return path + + def _inspect(self, path: Path, **params) -> dict: + result = mesh_ops_registry.run("inspect_mesh", path, params, self.context) + self.assertIsNone(result.file_path) + return dict(result.details) + + def test_closed_box_is_printable(self) -> None: + path = self._export(_box(), include_normals=True) + + report = self._inspect(path) + + self.assertTrue(report["is_watertight"]) + self.assertTrue(report["is_winding_consistent"]) + self.assertTrue(report["is_manifold"]) + self.assertEqual(report["face_count"], 12) + self.assertEqual(report["component_count"], 1) + self.assertEqual(report["hole_count"], 0) + self.assertAlmostEqual(report["volume"], 1.0, places=6) + self.assertAlmostEqual(report["surface_area"], 6.0, places=6) + np.testing.assert_allclose(report["dimensions"], [1.0, 1.0, 1.0]) + self.assertTrue(report["has_normals"]) + self.assertFalse(report["has_uvs"]) + if HAS_PYMESHLAB: + self.assertEqual(report["self_intersecting_face_count"], 0) + self.assertTrue(report["printable"]) + self.assertEqual(report["issues"], []) + + def test_open_box_reports_its_hole_and_no_volume(self) -> None: + mesh = _box() + mesh.update_faces(np.arange(1, len(mesh.faces))) + path = self._export(mesh) + + report = self._inspect(path, check_self_intersections=False) + + self.assertFalse(report["is_watertight"]) + self.assertFalse(report["printable"]) + self.assertEqual(report["hole_count"], 1) + self.assertEqual(report["boundary_edge_count"], 3) + self.assertEqual(report["largest_hole_edges"], 3) + self.assertIsNone(report["volume"]) + self.assertIsNone(report["self_intersecting_face_count"]) + self.assertTrue(any("hole" in issue for issue in report["issues"])) + self.assertFalse(report["has_normals"]) + + def test_unchecked_self_intersections_leave_printability_unknown(self) -> None: + path = self._export(_box()) + + report = self._inspect(path, check_self_intersections=False) + + self.assertTrue(report["is_watertight"]) + self.assertIsNone(report["printable"]) + self.assertIn("self-intersections were not checked", report["issues"]) + + def test_normals_are_read_from_the_file_format(self) -> None: + cases = { + "normals.obj": (dict(include_normals=True), True), + "plain.obj": (dict(include_normals=False), False), + "normals.ply": (dict(vertex_normal=True), True), + "mesh.stl": ({}, False), + } + for name, (kwargs, expected) in cases.items(): + with self.subTest(name=name): + path = self._export(_box(), name=name, **kwargs) + report = self._inspect(path, check_self_intersections=False) + self.assertIs(report["has_normals"], expected) + + def test_uv_seams_do_not_count_as_holes(self) -> None: + mesh = _box() + mesh.unmerge_vertices() + uv = np.random.default_rng(0).random((len(mesh.vertices), 2)) + mesh.visual = trimesh.visual.TextureVisuals(uv=uv) + path = self._export(mesh) + + report = self._inspect(path, check_self_intersections=False) + + self.assertTrue(report["has_uvs"]) + self.assertEqual(report["vertex_count"], 36) + self.assertTrue(report["is_watertight"]) + self.assertEqual(report["hole_count"], 0) + + def test_scene_parts_are_counted_in_world_space(self) -> None: + scene = trimesh.Scene() + scene.add_geometry(_box(), node_name="a") + scene.add_geometry( + _box(), + node_name="b", + transform=trimesh.transformations.translation_matrix([3.0, 0.0, 0.0]), + ) + path = self._export(scene) + + report = self._inspect(path, check_self_intersections=False) + + self.assertEqual(report["component_count"], 2) + self.assertEqual(report["face_count"], 24) + np.testing.assert_allclose(report["dimensions"], [4.0, 1.0, 1.0]) + self.assertAlmostEqual(report["volume"], 2.0, places=6) + + @unittest.skipUnless(HAS_PYMESHLAB, "pymeshlab is not installed") + def test_overlapping_parts_are_self_intersecting(self) -> None: + mesh = trimesh.util.concatenate([_box(), _box((0.5, 0.3, 0.2))]) + path = self._export(mesh) + + report = self._inspect(path) + + self.assertGreater(report["self_intersecting_face_count"], 0) + self.assertFalse(report["printable"]) + + def test_inspection_never_modifies_the_input(self) -> None: + mesh = _box() + mesh.update_faces(np.arange(1, len(mesh.faces))) + path = self._export(mesh) + before = hashlib.sha256(path.read_bytes()).hexdigest() + mtime = path.stat().st_mtime_ns + + self._inspect(path) + mesh_ops_registry.run("suggest_fixes", path, {}, self.context) + + self.assertEqual(hashlib.sha256(path.read_bytes()).hexdigest(), before) + self.assertEqual(path.stat().st_mtime_ns, mtime) + self.assertEqual([p.name for p in self.workspace.iterdir()], ["mesh.glb"]) + + def test_file_without_triangles_is_rejected(self) -> None: + path = self._export(trimesh.PointCloud(np.zeros((3, 3))), name="cloud.ply") + + with self.assertRaises(MeshOpExecutionError): + self._inspect(path) + + def test_suggest_fixes_proposes_a_repair_for_holes(self) -> None: + mesh = _box() + mesh.update_faces(np.arange(1, len(mesh.faces))) + path = self._export(mesh) + + result = mesh_ops_registry.run( + "suggest_fixes", + path, + {"check_self_intersections": False}, + self.context, + ) + + self.assertIsNone(result.file_path) + self.assertEqual(result.details["report"]["hole_count"], 1) + fixes = result.details["fixes"] + self.assertEqual([fix["op"] for fix in fixes], ["repair"]) + self.assertTrue(fixes[0]["params"]["fill_holes"]) + self.assertEqual(fixes[0]["params"]["max_hole_size"], 10) + + +class FixesForReportTests(unittest.TestCase): + CLEAN = { + "face_count": 12, + "component_count": 1, + "is_winding_consistent": True, + "hole_count": 0, + "largest_hole_edges": 0, + "non_manifold_edge_count": 0, + "degenerate_face_count": 0, + "self_intersecting_face_count": 0, + } + + def test_clean_mesh_needs_nothing(self) -> None: + self.assertEqual(fixes_for_report(self.CLEAN), ([], [])) + + def test_suggested_params_are_valid_for_the_registry(self) -> None: + report = { + **self.CLEAN, + "face_count": HIGH_FACE_COUNT + 1, + "hole_count": 2, + "largest_hole_edges": 50_000, + "non_manifold_edge_count": 4, + "degenerate_face_count": 1, + } + + fixes, notes = fixes_for_report(report) + + self.assertEqual([fix["op"] for fix in fixes], ["repair", "decimate"]) + self.assertEqual(fixes[0]["params"]["max_hole_size"], 10000) + self.assertEqual(fixes[1]["params"], {"target_faces": DECIMATE_TARGET_FACES}) + self.assertTrue(any("repair limit" in note for note in notes)) + repair = mesh_ops_registry.get("repair") + declared = {schema["id"] for schema in repair.params_schema} + self.assertLessEqual(set(fixes[0]["params"]), declared) + + def test_problems_without_an_op_become_notes(self) -> None: + report = { + **self.CLEAN, + "is_winding_consistent": False, + "self_intersecting_face_count": 7, + "component_count": 3, + } + + fixes, notes = fixes_for_report(report) + + self.assertEqual(fixes, []) + self.assertEqual(len(notes), 3) + + +class InspectionRegistryTests(unittest.TestCase): + def test_analysis_ops_are_registered_as_read_only(self) -> None: + described = {op["id"]: op for op in mesh_ops_registry.describe()} + + for op_id in ("inspect_mesh", "suggest_fixes"): + self.assertEqual(described[op_id]["category"], "analysis") + self.assertFalse(described[op_id]["destructive"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/api/tests/test_mesh_ops_registry.py b/api/tests/test_mesh_ops_registry.py index 31a777a20..8656f3d51 100644 --- a/api/tests/test_mesh_ops_registry.py +++ b/api/tests/test_mesh_ops_registry.py @@ -19,10 +19,10 @@ def test_builtin_metadata_is_serializable_and_complete(self) -> None: self.assertEqual( [description["id"] for description in descriptions], - ["repair", "decimate", "smooth"], + ["repair", "decimate", "smooth", "inspect_mesh", "suggest_fixes"], ) for description in descriptions: - self.assertIn(description["category"], {"repair", "optimization"}) + self.assertIn(description["category"], {"repair", "optimization", "analysis"}) self.assertFalse(description["destructive"]) self.assertTrue(description["undoable"]) self.assertIsInstance(description["params_schema"], list) diff --git a/api/tests/test_optimize_mesh_ops.py b/api/tests/test_optimize_mesh_ops.py index 6c702ddb6..c2c9d071c 100644 --- a/api/tests/test_optimize_mesh_ops.py +++ b/api/tests/test_optimize_mesh_ops.py @@ -105,6 +105,25 @@ def test_legacy_routes_delegate_with_their_existing_clamps_and_names(self) -> No self.assertEqual(optimize_response["face_count"], 42) self.assertEqual(smooth_response["url"], "/workspace/model_smooth20.glb") + def test_analysis_operation_returns_only_its_report(self) -> None: + class AnalysisRegistry: + def run(self, operation_id, input_path, params, context): + return MeshOpResult(details={"printable": False, "hole_count": 1}) + + with tempfile.TemporaryDirectory() as directory: + workspace = Path(directory) + (workspace / "input.glb").touch() + with ( + patch.object(optimize.registry, "WORKSPACE_DIR", workspace), + patch.object(optimize, "mesh_ops_registry", AnalysisRegistry()), + ): + response = optimize.run_mesh_operation( + "inspect_mesh", + optimize.MeshOpRequest(path="input.glb"), + ) + + self.assertEqual(response, {"printable": False, "hole_count": 1}) + def test_unknown_generic_operation_is_a_404(self) -> None: class MissingRegistry: def run(self, operation_id, input_path, params, context):