feat(pointcloud): complete annotation and result lifecycle workflows

shuishen

11 hours ago 385be2eca72eb3833efa4be0a0088b34e764788a

 tests/test_serve_workbench_console.py

@@ -7,6 +7,8 @@
import tempfile
import unittest
import json
import struct
import zipfile
from unittest import mock
from http import HTTPStatus
from urllib.parse import quote
@@ -21,7 +23,64 @@
SPEC.loader.exec_module(MODULE)


def write_test_npz(path: Path, point_count: int) -> None:
    """Create just enough valid NPY headers for the console's dependency-free validator."""
    def npy_header(shape: tuple[int, int]) -> bytes:
        text = repr({"descr": "<f8", "fortran_order": False, "shape": shape}).encode("ascii")
        padding = (-((10 + len(text) + 1) % 16)) % 16
        header = text + (b" " * padding) + b"\n"
        return b"\x93NUMPY\x01\x00" + struct.pack("<H", len(header)) + header

    with zipfile.ZipFile(path, "w", compression=zipfile.ZIP_DEFLATED) as archive:
        archive.writestr("xyz.npy", npy_header((point_count, 3)))
        archive.writestr("las_rgb.npy", npy_header((point_count, 3)))


class WorkbenchConsoleHandlerTests(unittest.TestCase):
    def test_run_deletion_plan_rejects_unknown_capability_and_path_traversal(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            with self.assertRaises(MODULE.ApiError):
                MODULE.run_deletion_plan(root, "not-a-capability", "run-1")
            with self.assertRaises(MODULE.ApiError):
                MODULE.run_deletion_plan(root, "02-semantic-mapping", "../run-1")

    def test_run_deletion_plan_protects_non_console_baseline(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            artifact = root / "shared" / "outputs" / "02-semantic-mapping" / "validation-baseline"
            artifact.mkdir(parents=True)
            (artifact / "run_metadata.json").write_text(json.dumps({"capability": "02-semantic-mapping", "images": []}), encoding="utf-8")
            plan = MODULE.run_deletion_plan(root, "02-semantic-mapping", "validation-baseline")
            self.assertFalse(plan["removable"])
            self.assertEqual(plan["outputDirectories"], [])

    def test_console_run_deletion_removes_only_discovered_owned_directories(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            run_id = "semantic-test-run"
            output = root / "shared" / "outputs" / "02-semantic-mapping" / "runs" / run_id
            raw = root / "shared" / "data" / "raw" / "02-semantic-mapping" / "runs" / run_id
            processed = root / "shared" / "data" / "processed" / "02-semantic-mapping" / run_id
            sibling = root / "shared" / "outputs" / "02-semantic-mapping" / "runs" / "keep-me"
            external = root / "baseData" / "keep.jpg"
            for directory in (output, raw, processed, sibling, external.parent):
                directory.mkdir(parents=True, exist_ok=True)
            (output / "run_metadata.json").write_text(json.dumps({"capability": "02-semantic-mapping", "images": [], "created_at": "2026-08-28T00:00:00Z"}), encoding="utf-8")
            external.write_bytes(b"external")
            plan = MODULE.run_deletion_plan(root, "02-semantic-mapping", run_id)
            self.assertTrue(plan["removable"])
            self.assertEqual(plan["outputDirectories"], [f"shared/outputs/02-semantic-mapping/runs/{run_id}"])
            self.assertEqual(plan["rawDirectories"], [f"shared/data/raw/02-semantic-mapping/runs/{run_id}"])
            self.assertEqual(plan["processedDirectories"], [f"shared/data/processed/02-semantic-mapping/{run_id}"])
            removed = MODULE.delete_console_run(root, "02-semantic-mapping", run_id)
            self.assertEqual(removed["runId"], run_id)
            self.assertFalse(output.exists())
            self.assertFalse(raw.exists())
            self.assertFalse(processed.exists())
            self.assertTrue(sibling.is_dir())
            self.assertEqual(external.read_bytes(), b"external")

    def make_handler(self) -> MODULE.WorkbenchConsoleHandler:
        handler = object.__new__(MODULE.WorkbenchConsoleHandler)
        handler.directory = str(ROOT)
@@ -74,6 +133,9 @@
    def test_large_tiff_upload_limits_allow_one_gibibyte_files(self) -> None:
        self.assertEqual(MODULE.MAX_FILE_BYTES, 1024 * 1024 * 1024)
        self.assertEqual(MODULE.MAX_REQUEST_BYTES, 3072 * 1024 * 1024)

    def test_photo_reconstruction_timeout_supports_long_cpu_runs(self) -> None:
        self.assertEqual(MODULE.PHOTO_RECONSTRUCTION_TIMEOUT, 86_400)

    def test_binary_change_upload_preserves_original_bytes(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
@@ -147,7 +209,7 @@
            with self.assertRaisesRegex(MODULE.ApiError, "没有点记录"):
                MODULE.validate_pointcloud_model_input(empty_las)

    def test_photo_reconstruction_request_requires_three_to_thirty_photos(self) -> None:
    def test_photo_reconstruction_request_requires_three_to_one_thousand_photos(self) -> None:
        handler = self.make_handler()
        with self.assertRaisesRegex(MODULE.ApiError, "at least three"):
            handler.create_photo_reconstruction_run({"photos": []})
@@ -155,6 +217,20 @@
            handler.create_photo_reconstruction_run({"photos": [{}] * (MODULE.MAX_PHOTO_RECONSTRUCTION_IMAGES_PER_RUN + 1)})
        with self.assertRaisesRegex(MODULE.ApiError, "true or false"):
            handler.create_photo_reconstruction_run({"photos": [{}, {}, {}], "usePositionPriors": "yes"})

    def test_photo_reconstruction_job_reads_progress_without_exposing_path(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            progress_path = Path(temp_dir) / "progress.json"
            progress_path.write_text(json.dumps({"percent": 68, "stage": "dense_fusion", "message": "running", "inputImages": 18, "estimate": True}), encoding="utf-8")
            with MODULE.PHOTO_RECONSTRUCTION_JOBS_LOCK:
                MODULE.PHOTO_RECONSTRUCTION_JOBS["progress-test"] = {"id": "progress-test", "runId": "run", "status": "running", "stage": "dense_mvs", "inputImages": 18, "progressPath": str(progress_path)}
            try:
                job = MODULE.photo_reconstruction_job("progress-test")
            finally:
                with MODULE.PHOTO_RECONSTRUCTION_JOBS_LOCK:
                    MODULE.PHOTO_RECONSTRUCTION_JOBS.pop("progress-test", None)
            self.assertEqual(job["progress"]["percent"], 68)
            self.assertNotIn("progressPath", job)

    def test_binary_anomaly_upload_decodes_chinese_file_name(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
@@ -238,6 +314,33 @@
            with mock.patch.object(MODULE.subprocess, "run", return_value=subprocess.CompletedProcess([], 0, '{"cuda": false, "torch": "2.11.0+cu128"}\n', "")):
                with self.assertRaisesRegex(MODULE.ApiError, "CUDA was requested"):
                    MODULE.pointcloud_execution_environment(root, "cuda")

    def test_explicit_cpu_execution_never_probes_cuda(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            cpu_python = root / ".venvs" / MODULE.OBJECT_DETECTION_CPU_ENVIRONMENT / "Scripts" / "python.exe"
            cpu_python.parent.mkdir(parents=True)
            cpu_python.write_bytes(b"fixed-cpu-interpreter")
            with mock.patch.object(MODULE.subprocess, "run") as probe:
                execution = MODULE.object_detection_execution_environment(root, "cpu")
            probe.assert_not_called()
            self.assertEqual(execution["requestedDevice"], "cpu")
            self.assertEqual(execution["device"], "cpu")
            self.assertFalse(execution["fallbackUsed"])

    def test_auto_cpu_fallback_records_the_probe_reason(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            cpu_python = root / ".venvs" / MODULE.CHANGE_DETECTION_CPU_ENVIRONMENT / "Scripts" / "python.exe"
            gpu_python = root / ".venvs" / MODULE.CHANGE_DETECTION_GPU_ENVIRONMENT / "Scripts" / "python.exe"
            cpu_python.parent.mkdir(parents=True)
            gpu_python.parent.mkdir(parents=True)
            cpu_python.write_bytes(b"fixed-cpu-interpreter")
            gpu_python.write_bytes(b"fixed-gpu-interpreter")
            with mock.patch.object(MODULE.subprocess, "run", return_value=subprocess.CompletedProcess([], 0, '{"cuda": false, "torch": "2.11.0+cu128"}\n', "")):
                execution = MODULE.change_detection_execution_environment(root, "auto")
            self.assertTrue(execution["fallbackUsed"])
            self.assertIn("cannot use CUDA", execution["fallbackReason"])

    def test_object_detection_execution_uses_fixed_gpu_environment_when_probe_succeeds(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
@@ -327,6 +430,52 @@
        validation = next(item for item in runs if item["id"] == "validation-normal-20260821-v2")
        self.assertTrue(validation["artifactRoot"].startswith("shared/outputs/05-3d-pointcloud/"))

    def test_multiview_annotation_source_requires_complete_ordered_contract(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            artifact = root / "shared" / "outputs" / "05-3d-pointcloud" / "multiview-feature-valid"
            artifact.mkdir(parents=True)
            cloud = artifact / "multiview-annotation-source.ply"
            cloud.write_text(
                "ply\nformat ascii 1.0\nelement vertex 2\nproperty float x\nproperty float y\nproperty float z\nend_header\n0 0 0\n1 1 1\n",
                encoding="ascii",
            )
            dataset = artifact / "multiview-point-features.npz"
            write_test_npz(dataset, 2)
            (artifact / "run_metadata.json").write_text(json.dumps({
                "capability": "05-3d-pointcloud",
                "artifacts": {"annotation_source": cloud.name, "feature_dataset": dataset.name},
                "annotation_source": {
                    "schema_version": 1,
                    "kind": "multiview_photo_feature_fusion",
                    "point_cloud": cloud.name,
                    "feature_dataset": dataset.name,
                    "point_count": 2,
                    "point_cloud_sha256": MODULE.file_sha256(cloud),
                    "feature_dataset_sha256": MODULE.file_sha256(dataset),
                },
            }), encoding="utf-8")

            sources = MODULE.pointcloud_annotation_sources(root)
            self.assertEqual(len(sources), 1)
            self.assertEqual(sources[0]["pointCount"], 2)
            self.assertIn("多视角照片特征融合", sources[0]["sourceKind"])

            handler = object.__new__(MODULE.WorkbenchConsoleHandler)
            handler.directory = str(root)
            annotation = handler.create_pointcloud_annotation({"sourceId": sources[0]["id"], "labels": [[1, 15]]})
            self.assertEqual(annotation["labelCount"], 1)
            execution = {"python": "fixed-python", "device": "cpu", "environment": "05-3d-pointcloud", "torchVersion": "2.13.0+cpu", "requestedDevice": "cpu", "fallbackUsed": False, "fallbackReason": None}
            with mock.patch.object(MODULE, "pointcloud_execution_environment", return_value=execution), mock.patch.object(MODULE.threading, "Thread") as thread:
                thread.return_value.start.return_value = None
                job = handler.create_pointcloud_training_run({"annotationId": annotation["id"], "device": "cpu"})
            self.assertEqual(job["trainer"], "multiview_local_attention_baseline")
            self.assertTrue(job["artifactRoot"] if "artifactRoot" in job else True)
            self.assertEqual(thread.call_args.kwargs["args"][-1], "multiview_local_attention_baseline")

            dataset.unlink()
            self.assertEqual(MODULE.pointcloud_annotation_sources(root), [])

    def test_pointcloud_request_requires_allowlisted_files(self) -> None:
        handler = self.make_handler()
        with self.assertRaisesRegex(MODULE.ApiError, "PLY, PCD, XYZ, LAS, or LAZ"):
@@ -334,6 +483,116 @@
        with self.assertRaisesRegex(MODULE.ApiError, "Unsupported file type"):
            handler.create_pointcloud_run({"pointClouds": [{"name": "unsafe.exe", "content": "eA=="}]})

    def test_annotation_taxonomy_adds_custom_las_class_and_preserves_used_code(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            handler = object.__new__(MODULE.WorkbenchConsoleHandler)
            handler.directory = str(root)
            defaults = MODULE.annotation_classes(root)
            self.assertEqual({item["code"] for item in defaults}, {1, 2, 5, 6, 15, 16})
            created = handler.create_pointcloud_annotation_class({"key": "transformer", "label": "变压器", "color": [30, 144, 255]})
            self.assertEqual(created["code"], 17)
            self.assertTrue(MODULE.annotation_classes_path(root).is_file())
            revision = root / "shared" / "outputs" / "05-3d-pointcloud" / "annotations" / "annotation-used"
            revision.mkdir(parents=True)
            (revision / "annotation.json").write_text(json.dumps({"schema_version": 1, "labels": [[0, 17]]}), encoding="utf-8")
            with self.assertRaisesRegex(MODULE.ApiError, "used by a saved annotation"):
                handler.delete_pointcloud_annotation_class(17)
            (revision / "annotation.json").unlink()
            self.assertEqual(handler.delete_pointcloud_annotation_class(17), 17)
            self.assertNotIn(17, {item["code"] for item in MODULE.annotation_classes(root)})

    def test_annotation_taxonomy_rejects_invalid_custom_metadata(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            handler = object.__new__(MODULE.WorkbenchConsoleHandler)
            handler.directory = temp_dir
            with self.assertRaisesRegex(MODULE.ApiError, "lowercase English"):
                handler.create_pointcloud_annotation_class({"key": "Transformer", "label": "变压器", "color": [30, 144, 255]})
            with self.assertRaisesRegex(MODULE.ApiError, "RGB"):
                handler.create_pointcloud_annotation_class({"key": "transformer", "label": "变压器", "color": [999, 1, 1]})

    def test_preview_only_annotation_source_is_discovered_without_geometry_artifacts(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            run_id = "annotation-source-preview-only"
            artifact = root / "shared" / "outputs" / "05-3d-pointcloud" / "runs" / run_id
            artifact.mkdir(parents=True)
            preview = artifact / "upload.annotation-source.ply"
            preview.write_text(
                "ply\nformat binary_little_endian 1.0\nelement vertex 2\nproperty float x\nproperty float y\nproperty float z\nproperty uchar red\nproperty uchar green\nproperty uchar blue\nend_header\n",
                encoding="ascii",
            )
            (artifact / "run_metadata.json").write_text(json.dumps({
                "capability": "05-3d-pointcloud", "annotation_source_job": True,
                "annotation_source": {"schema_version": 1, "file": preview.name, "point_count": 2, "sha256": MODULE.file_sha256(preview)},
                "input": {"file": "upload.ply", "has_rgb": False},
            }), encoding="utf-8")
            sources = MODULE.pointcloud_annotation_sources(root)
            self.assertEqual(len(sources), 1)
            self.assertEqual(sources[0]["id"], f"{run_id}:{preview.name}")
            self.assertFalse(sources[0]["sourceHasRgb"])
            self.assertEqual(MODULE.pointcloud_annotation_source_deletion_plan(root, sources[0]["id"])["outputDirectories"], 1)

    def test_texture_baked_annotation_source_is_discovered_at_capability_root(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            run_id = "texture-baked-preview"
            artifact = root / "shared" / "outputs" / "05-3d-pointcloud" / run_id
            artifact.mkdir(parents=True)
            preview = artifact / "texture-baked-annotation-source.ply"
            preview.write_text(
                "ply\nformat binary_little_endian 1.0\nelement vertex 1\nproperty float x\nproperty float y\nproperty float z\nend_header\n",
                encoding="ascii",
            )
            (artifact / "run_metadata.json").write_text(json.dumps({
                "capability": "05-3d-pointcloud", "annotation_source_job": True,
                "annotation_source": {"schema_version": 1, "file": preview.name, "point_count": 1, "sha256": MODULE.file_sha256(preview)},
                "input": {"file": "Model_0.zip", "has_rgb": True},
            }), encoding="utf-8")
            sources = MODULE.pointcloud_annotation_sources(root)
            self.assertEqual([item["id"] for item in sources], [f"{run_id}:{preview.name}"])
            self.assertTrue(sources[0]["sourceHasRgb"])

    def test_annotation_source_removal_deletes_only_its_complete_local_chain(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            run_id = "annotation-source-test"
            artifact = root / "shared" / "outputs" / "05-3d-pointcloud" / "runs" / run_id
            artifact.mkdir(parents=True)
            annotation_cloud = artifact / "block.semantic-annotation-source.ply"
            annotation_cloud.write_bytes(b"annotation-preview")
            (artifact / "preview.png").write_bytes(b"preview")
            (artifact / "vector.geojson").write_text('{"type":"FeatureCollection","features":[]}', encoding="utf-8")
            (artifact / "run_metadata.json").write_text(json.dumps({"capability": "05-3d-pointcloud", "created_at": "2026-08-28T00:00:00+00:00", "annotation_source_job": True, "point_clouds": [{"file": "block.las", "preview_file": "preview.png", "vector_file": "vector.geojson", "semantic_annotation_source_point_cloud": annotation_cloud.name, "semantic_preview_points": 1}]}), encoding="utf-8")
            source_id = f"{run_id}:{annotation_cloud.name}"
            revision = root / "shared" / "outputs" / "05-3d-pointcloud" / "annotations" / "annotation-test"
            revision.mkdir(parents=True)
            annotation_path = revision / "annotation.json"
            annotation_path.write_text(json.dumps({"schema_version": 1, "id": "annotation-test", "source_id": source_id, "labels": [[0, 5]]}), encoding="utf-8")
            training = root / "shared" / "outputs" / "05-3d-pointcloud" / "training-runs" / "model-test"
            training.mkdir(parents=True)
            model = training / "model.pt"
            model.write_bytes(b"weights")
            (training / "metrics.json").write_text(json.dumps({"annotation": str(annotation_path)}), encoding="utf-8")
            inference = root / "shared" / "outputs" / "05-3d-pointcloud" / "model-inference-runs" / "inference-test"
            inference.mkdir(parents=True)
            (inference / "run_metadata.json").write_text(json.dumps({"model": {"path": str(model)}}), encoding="utf-8")
            raw = root / "shared" / "data" / "raw" / "05-3d-pointcloud" / "annotation-source-runs" / run_id
            processed = root / "shared" / "data" / "processed" / "05-3d-pointcloud" / "annotation-source-runs" / run_id
            raw.mkdir(parents=True); processed.mkdir(parents=True)
            (raw / "block.las").write_bytes(b"raw"); (processed / "block.las").write_bytes(b"processed")
            external = root / "baseData" / "block.las"; external.parent.mkdir(); external.write_bytes(b"must-remain")

            handler = object.__new__(MODULE.WorkbenchConsoleHandler)
            handler.directory = str(root)
            plan = MODULE.pointcloud_annotation_source_deletion_plan(root, source_id)
            self.assertEqual((plan["rawDirectories"], plan["processedDirectories"], plan["annotationRevisions"], plan["trainingRuns"], plan["inferenceRuns"]), (1, 1, 1, 1, 1))
            result = handler.delete_pointcloud_annotation_source(source_id)
            self.assertEqual(result["removed"]["trainingRuns"], 1)
            self.assertFalse(artifact.exists()); self.assertFalse(raw.exists()); self.assertFalse(processed.exists())
            self.assertFalse(revision.exists()); self.assertFalse(training.exists()); self.assertFalse(inference.exists())
            self.assertEqual(external.read_bytes(), b"must-remain")

    def test_semantic_model_discovery_requires_complete_training_artifacts(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)