feat(pointcloud): complete semantic annotation loop

shuishen

23 hours ago 8f75db57f3055b2a3575c86ee3e6acf1dd56ce47

 tests/test_serve_workbench_console.py

@@ -430,6 +430,74 @@
        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_trained_model_inference_is_discovered_as_semantic_case_only(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            inference = root / "shared" / "outputs" / "05-3d-pointcloud" / "model-inference-runs" / "semantic-inference-test"
            automatic = root / "shared" / "outputs" / "05-3d-pointcloud" / "auto-annotation-runs" / "automatic-test"
            for directory in (inference, automatic):
                directory.mkdir(parents=True)
                for name in ("predicted-semantic-preview.ply", "predicted-semantic-classified.las", "prediction-summary.json", "class-counts.csv"):
                    (directory / name).write_bytes(b"artifact")
                (directory / "run_metadata.json").write_text(json.dumps({
                    "capability": "05-3d-pointcloud", "created_at": "2026-08-31T00:00:00+00:00",
                    "input": {"mesh_file": "new-scene.ply"}, "model": {"path": "training-runs/model/model.pt"},
                    "prediction": {"class_counts": {"5": 10}},
                    "artifacts": {"preview": "predicted-semantic-preview.ply", "classified_las": "predicted-semantic-classified.las", "summary": "prediction-summary.json", "class_counts": "class-counts.csv"},
                }), encoding="utf-8")
            runs = MODULE.pointcloud_runs(root)
            self.assertEqual([item["id"] for item in runs], ["semantic-inference-test"])
            self.assertEqual(runs[0]["kind"], "trained-model-inference")

    def test_trained_model_semantic_case_has_a_limited_removal_plan(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            run_id = "semantic-inference-test"
            output = root / "shared" / "outputs" / "05-3d-pointcloud" / "model-inference-runs" / run_id
            raw = root / "shared" / "data" / "raw" / "05-3d-pointcloud" / "model-inference-runs" / run_id
            processed = root / "shared" / "data" / "processed" / "05-3d-pointcloud" / "model-inference-runs" / run_id
            for directory in (output, raw, processed):
                directory.mkdir(parents=True)
            for name in ("predicted-semantic-preview.ply", "predicted-semantic-classified.las", "prediction-summary.json", "class-counts.csv"):
                (output / name).write_bytes(b"artifact")
            (output / "run_metadata.json").write_text(json.dumps({
                "capability": "05-3d-pointcloud", "input": {"path": "new-scene.las"},
                "model": {"path": "training-runs/model/model.pt"}, "prediction": {"class_counts": {"5": 10}},
                "artifacts": {"preview": "predicted-semantic-preview.ply", "classified_las": "predicted-semantic-classified.las", "summary": "prediction-summary.json", "class_counts": "class-counts.csv"},
            }), encoding="utf-8")
            external = root / "baseData" / "new-scene.las"; external.parent.mkdir(); external.write_bytes(b"keep")
            plan = MODULE.run_deletion_plan(root, "05-3d-pointcloud", run_id)
            self.assertTrue(plan["removable"])
            self.assertEqual(plan["outputDirectories"], [f"shared/outputs/05-3d-pointcloud/model-inference-runs/{run_id}"])
            self.assertEqual(plan["rawDirectories"], [f"shared/data/raw/05-3d-pointcloud/model-inference-runs/{run_id}"])
            self.assertEqual(plan["processedDirectories"], [f"shared/data/processed/05-3d-pointcloud/model-inference-runs/{run_id}"])
            MODULE.delete_console_run(root, "05-3d-pointcloud", run_id)
            self.assertFalse(output.exists()); self.assertFalse(raw.exists()); self.assertFalse(processed.exists())
            self.assertEqual(external.read_bytes(), b"keep")

    def test_trained_model_human_review_ply_is_discoverable_for_human_review(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            run_id = "semantic-inference-review"
            artifact = root / "shared" / "outputs" / "05-3d-pointcloud" / "model-inference-runs" / run_id
            artifact.mkdir(parents=True)
            classified = artifact / "predicted-semantic-classified.las"
            classified.write_bytes(b"classified-with-observed-rgb")
            review = artifact / "human-review-source.ply"
            review.write_text(
                "ply\nformat ascii 1.0\nelement vertex 42\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", "input": {"path": "new-scene.las"},
                "prediction": {"input_points": 42}, "artifacts": {"classified_las": classified.name, "human_review_source": review.name},
                "human_review_source": {"file": review.name, "sha256": MODULE.file_sha256(review), "point_count": 42},
            }), encoding="utf-8")
            sources = MODULE.model_inference_annotation_sources(root)
            self.assertEqual([item["id"] for item in sources], [f"{run_id}:{review.name}"])
            self.assertEqual(sources[0]["pointCount"], 42)
            self.assertTrue(sources[0]["sourceHasRgb"])

    def test_multiview_annotation_source_requires_complete_ordered_contract(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
@@ -476,6 +544,31 @@
            dataset.unlink()
            self.assertEqual(MODULE.pointcloud_annotation_sources(root), [])

    def test_pointcloud_concentrated_training_accepts_one_revision_per_rgb_source(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            annotation_root = root / "shared" / "outputs" / "05-3d-pointcloud" / "annotations"
            records = []
            for index in range(2):
                annotation_id = f"annotation-source-{index}"
                location = annotation_root / annotation_id
                location.mkdir(parents=True)
                source_id = f"source-{index}"
                (location / "annotation.json").write_text(json.dumps({"schema_version": 1, "id": annotation_id, "source_id": source_id}), encoding="utf-8")
                records.append({"id": source_id, "artifactRoot": f"shared/outputs/05-3d-pointcloud/runs/source-{index}", "sourceHasRgb": True})
            handler = object.__new__(MODULE.WorkbenchConsoleHandler)
            handler.directory = str(root)
            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_annotation_sources", return_value=records), 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({"annotationIds": ["annotation-source-0", "annotation-source-1"], "device": "cpu"})
            self.assertEqual(job["trainer"], "rgb_xyz_baseline")
            self.assertEqual(job["sourceCount"], 2)
            self.assertEqual(job["annotationIds"], ["annotation-source-0", "annotation-source-1"])
            self.assertEqual(len(thread.call_args.kwargs["args"][2]), 2)
            with self.assertRaisesRegex(MODULE.ApiError, "each annotation revision only once"):
                handler.create_pointcloud_training_run({"annotationIds": ["annotation-source-0", "annotation-source-0"], "device": "cpu"})

    def test_pointcloud_request_requires_allowlisted_files(self) -> None:
        handler = self.make_handler()
        with self.assertRaisesRegex(MODULE.ApiError, "PLY, PCD, XYZ, LAS, or LAZ"):
@@ -510,6 +603,26 @@
                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_annotation_taxonomy_color_update_preserves_stable_class_metadata(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            handler = object.__new__(MODULE.WorkbenchConsoleHandler)
            handler.directory = str(root)
            before = next(item for item in MODULE.annotation_classes(root) if item["code"] == 15)
            annotation = root / "shared" / "outputs" / "05-3d-pointcloud" / "annotations" / "annotation-used"
            annotation.mkdir(parents=True)
            record = {"labels": [[0, 15]], "class_schema": {"15": before}}
            path = annotation / "annotation.json"
            path.write_text(json.dumps(record), encoding="utf-8")

            updated = handler.update_pointcloud_annotation_class_color(15, {"color": [12, 34, 56]})

            self.assertEqual(updated["color"], [12, 34, 56])
            self.assertEqual({key: updated[key] for key in ("code", "key", "label", "builtIn")}, {key: before[key] for key in ("code", "key", "label", "builtIn")})
            self.assertEqual(json.loads(path.read_text(encoding="utf-8")), record)
            with self.assertRaisesRegex(MODULE.ApiError, "RGB"):
                handler.update_pointcloud_annotation_class_color(15, {"color": [256, 0, 0]})

    def test_preview_only_annotation_source_is_discovered_without_geometry_artifacts(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
@@ -552,6 +665,52 @@
            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_textured_mesh_header_requires_declared_uv_texture_tiles(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            mesh = Path(temp_dir) / "BlockBA.ply"
            mesh.write_bytes(
                b"ply\nformat binary_little_endian 1.0\ncomment TextureFile BlockBA_0_0.jpg\n"
                b"element vertex 3\nproperty float x\nproperty float y\nproperty float z\n"
                b"element face 1\nproperty list uchar uint vertex_indices\n"
                b"property list uchar float texcoord\nproperty int texnumber\nend_header\n"
            )
            self.assertEqual(MODULE.textured_mesh_texture_names(mesh), ["BlockBA_0_0.jpg"])
            mesh.write_bytes(b"ply\nformat binary_little_endian 1.0\nelement vertex 3\nend_header\n")
            with self.assertRaisesRegex(MODULE.ApiError, "TextureFile"):
                MODULE.textured_mesh_texture_names(mesh)

    def test_textured_model_inference_requires_exact_declared_texture_tiles(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            model = root / "shared" / "outputs" / "05-3d-pointcloud" / "training-runs" / "semantic-good"
            model.mkdir(parents=True)
            (model / "model.pt").write_bytes(b"weights")
            (model / "metrics.json").write_text(json.dumps({
                "capability": "05-3d-pointcloud", "classification": "B", "classes": {
                    "5": {"key": "vegetation", "label": "Vegetation", "color": [1, 2, 3]},
                    "16": {"key": "power_line", "label": "Power line", "color": [4, 5, 6]},
                },
            }), encoding="utf-8")
            mesh_upload = root / "shared" / "data" / "raw" / "05-3d-pointcloud" / "uploads" / ("a" * 32)
            texture_upload = root / "shared" / "data" / "raw" / "05-3d-pointcloud" / "uploads" / ("b" * 32)
            mesh_upload.mkdir(parents=True); texture_upload.mkdir(parents=True)
            (mesh_upload / "pointcloud.ply").write_bytes(
                b"ply\nformat binary_little_endian 1.0\ncomment TextureFile expected.jpg\n"
                b"element vertex 3\nproperty float x\nproperty float y\nproperty float z\n"
                b"element face 1\nproperty list uchar uint vertex_indices\n"
                b"property list uchar float texcoord\nproperty int texnumber\nend_header\n"
            )
            (mesh_upload / "pointcloud.json").write_text(json.dumps({"role": "pointcloud", "name": "mesh.ply"}), encoding="utf-8")
            (texture_upload / "texture.jpg").write_bytes(b"jpeg")
            (texture_upload / "texture.json").write_text(json.dumps({"role": "texture", "name": "unexpected.jpg"}), encoding="utf-8")
            handler = object.__new__(MODULE.WorkbenchConsoleHandler)
            handler.directory = str(root)
            with self.assertRaisesRegex(MODULE.ApiError, "must exactly match"):
                handler.create_textured_mesh_model_inference_run({
                    "modelId": "semantic-good", "pointCloud": {"uploadId": "a" * 32},
                    "textures": [{"uploadId": "b" * 32}],
                })

    def test_annotation_source_removal_deletes_only_its_complete_local_chain(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
@@ -643,6 +802,53 @@
            self.assertFalse(revision.exists())
            self.assertEqual(source_sentinel.read_bytes(), b"source-must-remain")

    def test_pointcloud_task_registry_marks_unfinished_work_interrupted_after_restart(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            registry = MODULE.pointcloud_task_registry(root)
            registry.write("training", {"id": "task-1", "runId": "semantic-model-1", "status": "running", "stage": "training", "createdAt": "2026-09-01T00:00:00+00:00"})
            self.assertEqual(registry.reconcile_startup(), 1)
            task = registry.get("task-1")
            self.assertEqual(task["status"], "interrupted")
            self.assertIn("restarted", task["error"])

    def test_pointcloud_artifact_health_reports_missing_and_valid_artifacts(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            valid = root / "shared" / "outputs" / "05-3d-pointcloud" / "training-runs" / "valid-model"
            invalid = valid.parent / "broken-model"
            valid.mkdir(parents=True); invalid.mkdir(parents=True)
            (valid / "metrics.json").write_text(json.dumps({"capability": "05-3d-pointcloud"}), encoding="utf-8")
            (valid / "model.pt").write_bytes(b"weights")
            (valid / "predicted-semantic-preview.ply").write_bytes(b"preview")
            (invalid / "metrics.json").write_text("{bad", encoding="utf-8")
            health = MODULE.pointcloud_artifact_health(root)
            records = {(item["kind"], item["id"]): item for item in health["records"]}
            self.assertEqual(records[("training_model", "valid-model")]["status"], "healthy")
            self.assertEqual(records[("training_model", "broken-model")]["status"], "invalid")
            self.assertIn("model.pt", records[("training_model", "broken-model")]["missing"])

    def test_pointcloud_auto_annotation_merge_preview_counts_review_and_human_overrides(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            run_id, source_id = "auto-test", "source:preview.ply"
            artifact = root / "shared" / "outputs" / "05-3d-pointcloud" / "auto-annotation-runs" / run_id
            artifact.mkdir(parents=True)
            (artifact / "automatic-annotation-candidates.json").write_text(json.dumps({"schema_version": 1, "source_id": source_id, "source_sha256": "source-hash", "labels": [[0, 5, 0.99], [1, 15, 0.99], [2, 16, 0.99]]}), encoding="utf-8")
            (artifact / "review-corrections.json").write_text(json.dumps({"schema_version": 1, "run_id": run_id, "source_id": source_id, "source_sha256": "source-hash", "corrections": [[0, 0], [1, 16], [3, 5]]}), encoding="utf-8")
            base = root / "shared" / "outputs" / "05-3d-pointcloud" / "annotations" / "human-1"
            base.mkdir(parents=True)
            (base / "annotation.json").write_text(json.dumps({"source_id": source_id, "labels": [[1, 15], [2, 5]]}), encoding="utf-8")
            source = {"id": source_id, "sha256": "source-hash", "pointCount": 4}
            classes = [{"code": 5}, {"code": 15}, {"code": 16}]
            with mock.patch.object(MODULE, "pointcloud_annotation_source", return_value=source), mock.patch.object(MODULE, "annotation_classes", return_value=classes):
                preview = MODULE.pointcloud_auto_annotation_merge_preview(root, run_id, source_id, "human-1")
            self.assertEqual(preview["highConfidenceCandidateCount"], 3)
            self.assertEqual(preview["rejectedCandidateCount"], 1)
            self.assertEqual(preview["reclassifiedCandidateCount"], 2)
            self.assertEqual(preview["baseHumanOverrideCount"], 2)
            self.assertEqual(preview["classCounts"], {"5": 2, "15": 1})

    def test_risk_rule_validation_run_is_discovered(self) -> None:
        runs = MODULE.risk_rule_runs(ROOT)
        validation = next(item for item in runs if item["id"] == "validation-normal-20260820")