feat: Implement depth surface strip cleanup and organize tool CLI

- Added a proposal for depth surface strip cleanup, detailing the rationale and changes to the depth plane and cutout nodes.
- Created specifications for the new cleanup requirements, including scenarios for strip removal and boundary edge detection.
- Developed tasks for implementing shared strip detection logic and integrating new interfaces for depth nodes.
- Established a new CLI structure for organizing node groups and model commands, replacing old entry points with a more intuitive command structure.
- Introduced a prototype for panorama surface generation, including requirements for distance reconstruction and geometry node surfaces.
- Added comprehensive tests for depth strip cleanup functionality, ensuring correct behavior for various mesh topologies and edge cases.
This commit is contained in:
2026-09-10 00:53:43 +08:00
parent 1622afa7a2
commit 27ca25224a
42 changed files with 1970 additions and 70 deletions

View File

@@ -3,6 +3,97 @@ import importlib.util
import sys
import unittest
from unittest.mock import patch
from dataclasses import replace
from pathlib import Path
import pytest
@pytest.mark.parametrize("command", ["prepare", "sync"])
def test_model_commands_prepare_all_before_upload(command):
module = load_sync_models()
events = []
with (
patch("sys.argv", ["model", command]),
patch.object(module, "prepare_model", side_effect=lambda model: events.append(model)),
patch.object(module, "upload_models", side_effect=lambda: events.append("upload")),
):
module.main()
assert events == [*module.MODEL_FILES, *(["upload"] if command == "sync" else [])]
def test_prepare_failure_prevents_upload():
module = load_sync_models()
with (
patch("sys.argv", ["model", "sync"]),
patch.object(module, "prepare_model", side_effect=RuntimeError("invalid model")),
patch.object(module, "upload_models") as upload,
):
with pytest.raises(RuntimeError, match="invalid model"):
module.main()
upload.assert_not_called()
def test_valid_model_cache_skips_all_preparation():
module = load_sync_models()
with (
patch.object(module, "verify_file", return_value=True),
patch.object(module, "build_model") as build,
patch.object(module, "download_file") as download,
patch.object(module.subprocess, "run") as run,
):
for model in module.MODEL_FILES:
module.prepare_model(model)
build.assert_not_called()
download.assert_not_called()
run.assert_not_called()
@pytest.mark.parametrize("valid", [True, False])
def test_pisa_export_validates_all_outputs_before_replacing(tmp_path, valid):
module = load_sync_models()
digest = module.hashlib.sha256(b"new").hexdigest()
models = tuple(module.ModelFile("PISA_SR", "pisa-sr", name, 3, digest) for name in ("a.onnx", "b.onnx"))
def export(command, **kwargs):
destination = Path(command[-1])
(destination / "a.onnx").write_bytes(b"new")
(destination / "b.onnx").write_bytes(b"new" if valid else b"bad")
with patch.object(module, "MODELS_DIR", tmp_path), patch.object(module, "MODEL_FILES", models), patch.object(module.subprocess, "run", side_effect=export):
models[1].destination.parent.mkdir()
models[1].destination.write_bytes(b"old")
if valid:
module.prepare_model(models[0])
assert all(model.destination.read_bytes() == b"new" for model in models)
else:
with pytest.raises(RuntimeError, match="SHA-256"):
module.prepare_model(models[0])
assert not models[0].destination.exists()
assert models[1].destination.read_bytes() == b"old"
assert not list(tmp_path.glob(".pisa-sr-*"))
def test_local_export_failure_cleans_temporary_output(tmp_path):
module = load_sync_models()
model = replace(next(model for model in module.MODEL_FILES if model.build), folder="test")
def export(exporter, source, destination):
destination.write_bytes(b"partial")
raise RuntimeError("export failed")
with (
patch.object(module, "MODELS_DIR", tmp_path),
patch.object(module, "PROJECT_ROOT", tmp_path),
patch.object(module, "prepare_source", return_value=tmp_path / "weights.pth"),
patch("tools.models.exports.export_model", side_effect=export),
):
model.destination.parent.mkdir()
model.destination.write_bytes(b"original")
with pytest.raises(RuntimeError, match="export failed"):
module.build_model(model)
assert model.destination.read_bytes() == b"original"
assert not model.destination.with_name(model.filename + ".part").exists()
def load_sync_models():