From 385be2eca72eb3833efa4be0a0088b34e764788a Mon Sep 17 00:00:00 2001
From: shuishen <1109946754@qq.com>
Date: Mon, 31 Aug 2026 09:04:45 +0800
Subject: [PATCH] feat(pointcloud): complete annotation and result lifecycle workflows
---
tests/test_serve_workbench_console.py | 261 ++++++++++++++++++++++++++++++++++++++++++++++++++++
1 files changed, 260 insertions(+), 1 deletions(-)
diff --git a/tests/test_serve_workbench_console.py b/tests/test_serve_workbench_console.py
index 0bec32a..9d394aa 100644
--- a/tests/test_serve_workbench_console.py
+++ b/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)
--
Gitblit v1.9.3