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
---
scripts/serve_workbench_console.py | 1262 +++++++++++++++++++++++++++++++++++++++++++++++++++++----
1 files changed, 1,170 insertions(+), 92 deletions(-)
diff --git a/scripts/serve_workbench_console.py b/scripts/serve_workbench_console.py
index 6a97f99..a7fd5e8 100644
--- a/scripts/serve_workbench_console.py
+++ b/scripts/serve_workbench_console.py
@@ -3,15 +3,18 @@
from __future__ import annotations
import argparse
+import ast
import base64
import binascii
import hashlib
import json
+import math
import os
import re
import shutil
import subprocess
import threading
+import zipfile
from datetime import UTC, datetime
from http import HTTPStatus
from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
@@ -31,8 +34,8 @@
MAX_SEGMENTATION_IMAGES_PER_RUN = 6
MAX_MEASUREMENT_RASTERS_PER_RUN = 4
MAX_POINTCLOUDS_PER_RUN = 2
-MAX_PHOTO_RECONSTRUCTION_IMAGES_PER_RUN = 30
-PHOTO_RECONSTRUCTION_TIMEOUT = 7200
+MAX_PHOTO_RECONSTRUCTION_IMAGES_PER_RUN = 1_000
+PHOTO_RECONSTRUCTION_TIMEOUT = 86_400
RISK_RULE_REQUIRED_FILES = {"observations", "zones", "rules"}
MAX_ANOMALY_IMAGES_PER_ROLE = 6
CHANGE_THRESHOLD_DEFAULT = 0.5
@@ -74,14 +77,27 @@
POINTCLOUD_TRAINING_JOBS_LOCK = threading.Lock()
POINTCLOUD_INFERENCE_JOBS: dict[str, dict[str, Any]] = {}
POINTCLOUD_INFERENCE_JOBS_LOCK = threading.Lock()
-MAX_ANNOTATION_LABELS = 400_000
-POINTCLOUD_CLASS_CODES = {1, 2, 5, 6, 15, 16}
+POINTCLOUD_ANNOTATION_SOURCE_JOBS: dict[str, dict[str, Any]] = {}
+POINTCLOUD_ANNOTATION_SOURCE_JOBS_LOCK = threading.Lock()
+POINTCLOUD_DETAIL_REQUEST_LOCK = threading.Lock()
+MAX_POINTCLOUD_DETAIL_RADIUS = 10_000_000.0
+DEFAULT_POINTCLOUD_ANNOTATION_CLASSES = (
+ {"code": 1, "key": "other_unknown", "label": "其他/未知", "color": [128, 128, 128], "builtIn": True},
+ {"code": 2, "key": "ground", "label": "地面", "color": [151, 111, 51], "builtIn": True},
+ {"code": 5, "key": "vegetation", "label": "植被", "color": [59, 163, 87], "builtIn": True},
+ {"code": 6, "key": "building_structure", "label": "建筑物", "color": [224, 115, 55], "builtIn": True},
+ {"code": 15, "key": "pole_tower", "label": "杆塔", "color": [149, 89, 210], "builtIn": True},
+ {"code": 16, "key": "power_line", "label": "电线", "color": [231, 196, 61], "builtIn": True},
+)
+POINTCLOUD_CLASS_CODES = {item["code"] for item in DEFAULT_POINTCLOUD_ANNOTATION_CLASSES}
+ANNOTATION_CLASS_KEY = re.compile(r"^[a-z][a-z0-9_]{0,47}$")
POINTCLOUD_CPU_ENVIRONMENT = "05-3d-pointcloud"
POINTCLOUD_GPU_ENVIRONMENT = "05-3d-pointcloud-gpu"
OBJECT_DETECTION_CPU_ENVIRONMENT = "01-object-detection"
OBJECT_DETECTION_GPU_ENVIRONMENT = "01-object-detection-cuda"
CHANGE_DETECTION_CPU_ENVIRONMENT = "00-change-detection"
CHANGE_DETECTION_GPU_ENVIRONMENT = "00-change-detection-cuda"
+COMPUTE_DEVICES = {"auto", "cpu", "cuda"}
class ApiError(ValueError):
@@ -117,7 +133,9 @@
def relative_path(root: Path, path: Path) -> str:
- return path.relative_to(root).as_posix()
+ # Windows may present a temporary root with its 8.3 spelling while an
+ # input directory has already been resolved to its long spelling.
+ return path.resolve().relative_to(root.resolve()).as_posix()
def load_json(path: Path) -> dict[str, Any]:
@@ -136,18 +154,19 @@
return digest.hexdigest()
-def pointcloud_execution_environment(root: Path, requested_device: str = "auto") -> dict[str, str]:
- """Select only a fixed point-cloud interpreter after a short CUDA probe.
+def select_execution_environment(root: Path, requested_device: str, cpu_environment: str, gpu_environment: str, label: str) -> dict[str, Any]:
+ """Choose an allowlisted CPU/CUDA interpreter and retain the selection evidence."""
+ if requested_device not in COMPUTE_DEVICES:
+ raise ApiError(f"{label} device must be auto, cpu, or cuda.")
+ cpu_python = root / ".venvs" / cpu_environment / "Scripts" / "python.exe"
+ gpu_python = root / ".venvs" / gpu_environment / "Scripts" / "python.exe"
+ if requested_device == "cpu":
+ if not cpu_python.is_file():
+ raise ApiError(f"{label} CPU virtual environment is unavailable. Run the capability setup first.")
+ return {"python": str(cpu_python), "device": "cpu", "environment": cpu_environment, "torchVersion": "unknown", "requestedDevice": "cpu", "fallbackUsed": False, "fallbackReason": None}
- The console never accepts a browser-supplied Python path. A failed or
- unavailable GPU environment is an expected condition for ``auto`` and
- falls back to the retained CPU environment.
- """
- if requested_device not in {"auto", "cpu", "cuda"}:
- raise ApiError("Point-cloud device must be auto, cpu, or cuda.")
- cpu_python = root / ".venvs" / POINTCLOUD_CPU_ENVIRONMENT / "Scripts" / "python.exe"
- gpu_python = root / ".venvs" / POINTCLOUD_GPU_ENVIRONMENT / "Scripts" / "python.exe"
- if requested_device != "cpu" and gpu_python.is_file():
+ probe_reason = "Fixed CUDA virtual environment is unavailable."
+ if gpu_python.is_file():
try:
probe = subprocess.run(
[str(gpu_python), "-c", "import json, torch; print(json.dumps({'cuda': bool(torch.cuda.is_available()), 'torch': torch.__version__}))"],
@@ -157,64 +176,48 @@
timeout=20,
check=False,
)
- payload = json.loads(probe.stdout.strip().splitlines()[-1]) if probe.returncode == 0 and probe.stdout.strip() else {}
- if payload.get("cuda") is True and isinstance(payload.get("torch"), str):
- return {"python": str(gpu_python), "device": "cuda", "environment": POINTCLOUD_GPU_ENVIRONMENT, "torchVersion": payload["torch"]}
+ if probe.returncode == 0 and probe.stdout.strip():
+ payload = json.loads(probe.stdout.strip().splitlines()[-1])
+ if payload.get("cuda") is True and isinstance(payload.get("torch"), str):
+ return {"python": str(gpu_python), "device": "cuda", "environment": gpu_environment, "torchVersion": payload["torch"], "requestedDevice": requested_device, "fallbackUsed": False, "fallbackReason": None}
+ probe_reason = "CUDA probe reports that PyTorch cannot use CUDA."
+ else:
+ probe_reason = "CUDA probe process did not complete successfully."
except (OSError, subprocess.SubprocessError, json.JSONDecodeError, IndexError):
- pass
+ probe_reason = "CUDA probe could not return a valid result."
if requested_device == "cuda":
- raise ApiError("CUDA was requested, but the fixed point-cloud GPU environment is unavailable.")
+ raise ApiError(f"CUDA was requested for {label}, but it is unavailable: {probe_reason}")
if not cpu_python.is_file():
- raise ApiError("3D point-cloud CPU virtual environment is unavailable. Run the capability setup first.")
- return {"python": str(cpu_python), "device": "cpu", "environment": POINTCLOUD_CPU_ENVIRONMENT, "torchVersion": "unknown"}
+ raise ApiError(f"{label} CPU virtual environment is unavailable. Run the capability setup first.")
+ return {"python": str(cpu_python), "device": "cpu", "environment": cpu_environment, "torchVersion": "unknown", "requestedDevice": "auto", "fallbackUsed": True, "fallbackReason": probe_reason}
-def object_detection_execution_environment(root: Path) -> dict[str, str]:
- """Choose the fixed object-detection CUDA environment only after probing it."""
- cpu_python = root / ".venvs" / OBJECT_DETECTION_CPU_ENVIRONMENT / "Scripts" / "python.exe"
- gpu_python = root / ".venvs" / OBJECT_DETECTION_GPU_ENVIRONMENT / "Scripts" / "python.exe"
- if gpu_python.is_file():
- try:
- probe = subprocess.run(
- [str(gpu_python), "-c", "import json, torch; print(json.dumps({'cuda': bool(torch.cuda.is_available()), 'torch': torch.__version__}))"],
- cwd=root,
- capture_output=True,
- text=True,
- timeout=20,
- check=False,
- )
- payload = json.loads(probe.stdout.strip().splitlines()[-1]) if probe.returncode == 0 and probe.stdout.strip() else {}
- if payload.get("cuda") is True and isinstance(payload.get("torch"), str):
- return {"python": str(gpu_python), "device": "cuda", "environment": OBJECT_DETECTION_GPU_ENVIRONMENT, "torchVersion": payload["torch"]}
- except (OSError, subprocess.SubprocessError, json.JSONDecodeError, IndexError):
- pass
- if not cpu_python.is_file():
- raise ApiError("Object-detection CPU virtual environment is unavailable. Run the capability setup first.")
- return {"python": str(cpu_python), "device": "cpu", "environment": OBJECT_DETECTION_CPU_ENVIRONMENT, "torchVersion": "unknown"}
+def pointcloud_execution_environment(root: Path, requested_device: str = "auto") -> dict[str, Any]:
+ return select_execution_environment(root, requested_device, POINTCLOUD_CPU_ENVIRONMENT, POINTCLOUD_GPU_ENVIRONMENT, "Point-cloud")
-def change_detection_execution_environment(root: Path) -> dict[str, str]:
- """Choose the fixed ChangeStar CUDA environment only after probing it."""
- cpu_python = root / ".venvs" / CHANGE_DETECTION_CPU_ENVIRONMENT / "Scripts" / "python.exe"
- gpu_python = root / ".venvs" / CHANGE_DETECTION_GPU_ENVIRONMENT / "Scripts" / "python.exe"
- if gpu_python.is_file():
- try:
- probe = subprocess.run(
- [str(gpu_python), "-c", "import json, torch; print(json.dumps({'cuda': bool(torch.cuda.is_available()), 'torch': torch.__version__}))"],
- cwd=root,
- capture_output=True,
- text=True,
- timeout=20,
- check=False,
- )
- payload = json.loads(probe.stdout.strip().splitlines()[-1]) if probe.returncode == 0 and probe.stdout.strip() else {}
- if payload.get("cuda") is True and isinstance(payload.get("torch"), str):
- return {"python": str(gpu_python), "device": "cuda", "environment": CHANGE_DETECTION_GPU_ENVIRONMENT, "torchVersion": payload["torch"]}
- except (OSError, subprocess.SubprocessError, json.JSONDecodeError, IndexError):
- pass
- if not cpu_python.is_file():
- raise ApiError("Change-detection CPU virtual environment is unavailable. Run the capability setup first.")
- return {"python": str(cpu_python), "device": "cpu", "environment": CHANGE_DETECTION_CPU_ENVIRONMENT, "torchVersion": "unknown"}
+def object_detection_execution_environment(root: Path, requested_device: str = "auto") -> dict[str, Any]:
+ return select_execution_environment(root, requested_device, OBJECT_DETECTION_CPU_ENVIRONMENT, OBJECT_DETECTION_GPU_ENVIRONMENT, "Object-detection")
+
+
+def change_detection_execution_environment(root: Path, requested_device: str = "auto") -> dict[str, Any]:
+ return select_execution_environment(root, requested_device, CHANGE_DETECTION_CPU_ENVIRONMENT, CHANGE_DETECTION_GPU_ENVIRONMENT, "Change-detection")
+
+
+def record_execution_metadata(path: Path, execution: dict[str, Any]) -> None:
+ """Preserve requested device, actual device and an automatic CPU fallback reason."""
+ metadata = load_json(path)
+ metadata["requested_device"] = execution["requestedDevice"]
+ metadata["device"] = execution["device"]
+ metadata["execution"] = {
+ "requested_device": execution["requestedDevice"],
+ "actual_device": execution["device"],
+ "environment": execution["environment"],
+ "torch_version": execution["torchVersion"],
+ "fallback_used": execution["fallbackUsed"],
+ "fallback_reason": execution["fallbackReason"],
+ }
+ path.write_text(json.dumps(metadata, ensure_ascii=False, indent=2), encoding="utf-8")
def trajectory_runs(root: Path) -> list[dict[str, Any]]:
@@ -511,8 +514,262 @@
"artifactRoot": case["artifactRoot"], "file": name, "url": f"/{case['artifactRoot']}/{name}",
"sha256": file_sha256(path), "pointCount": int(cloud.get("semantic_preview_points") or 0),
"sourceKind": str(cloud.get("semantic_annotation_source_kind") or "generated point-cloud preview"),
+ "detailAvailable": False, "detailFile": None,
})
+ sources.extend(standalone_pointcloud_annotation_sources(root))
+ sources.extend(multiview_pointcloud_annotation_sources(root))
return sources
+
+
+def standalone_pointcloud_annotation_sources(root: Path) -> list[dict[str, Any]]:
+ """Discover preview-only annotation uploads without pretending they are geometry runs."""
+ output_root = root / "shared" / "outputs" / "05-3d-pointcloud"
+ records: list[dict[str, Any]] = []
+ metadata_paths = [
+ *output_root.glob("runs/annotation-source-*/run_metadata.json"),
+ *output_root.glob("texture-baked-*/run_metadata.json"),
+ ]
+ for metadata_path in metadata_paths:
+ artifact = metadata_path.parent
+ metadata = load_json(metadata_path)
+ contract = metadata.get("annotation_source")
+ if metadata.get("capability") != "05-3d-pointcloud" or not metadata.get("annotation_source_job") or not isinstance(contract, dict):
+ continue
+ name, count, checksum = contract.get("file"), contract.get("point_count"), contract.get("sha256")
+ if not isinstance(name, str) or Path(name).name != name or not isinstance(count, int) or count < 1 or not isinstance(checksum, str):
+ continue
+ path = artifact / name
+ if not path.is_file() or path.suffix.lower() != ".ply" or file_sha256(path) != checksum or ply_vertex_count(path) != count:
+ continue
+ input_data = metadata.get("input") if isinstance(metadata.get("input"), dict) else {}
+ input_name = str(input_data.get("file") or name)
+ records.append({
+ "id": f"{artifact.name}:{name}", "runId": artifact.name, "label": f"{artifact.name} / {input_name}",
+ "artifactRoot": relative_path(root, artifact), "file": name, "url": f"/{relative_path(root, artifact)}/{name}",
+ "sha256": checksum, "pointCount": count,
+ "sourceKind": str(contract.get("kind") or "generated RGB/XYZ annotation preview"),
+ "sourceHasRgb": bool(input_data.get("has_rgb")),
+ "detailAvailable": False, "detailFile": None,
+ })
+ return sorted(records, key=lambda item: (item["runId"], item["id"]), reverse=True)
+
+
+def pointcloud_annotation_source(root: Path, source_id: str) -> dict[str, Any]:
+ if not isinstance(source_id, str) or not source_id or "/" in source_id or len(source_id) > 300:
+ raise ApiError("Invalid annotation source id.")
+ source = next((item for item in pointcloud_annotation_sources(root) if item["id"] == source_id), None)
+ if not source:
+ raise ApiError("The selected generated annotation source is unavailable.")
+ return source
+
+
+def pointcloud_annotation_detail(
+ root: Path,
+ source_id: str,
+ center: tuple[float, float, float],
+ radius: float,
+) -> bytes:
+ """Read one bounded, server-selected RGB detail window as a binary PLY."""
+ if not all(math.isfinite(value) and abs(value) <= 1_000_000_000.0 for value in center):
+ raise ApiError("Detail centre must contain finite local coordinates.")
+ if not math.isfinite(radius) or not 0 < radius <= MAX_POINTCLOUD_DETAIL_RADIUS:
+ raise ApiError(f"Detail radius must be between 0 and {MAX_POINTCLOUD_DETAIL_RADIUS:g}.")
+ source = pointcloud_annotation_source(root, source_id)
+ detail_name = source.get("detailFile")
+ if not source.get("detailAvailable") or not isinstance(detail_name, str) or Path(detail_name).name != detail_name:
+ raise ApiError("This annotation source has no local RGB detail layer.")
+ artifact = (root / str(source["artifactRoot"])).resolve()
+ output_root = (root / "shared" / "outputs" / "05-3d-pointcloud").resolve()
+ detail_path = (artifact / detail_name).resolve()
+ try:
+ artifact.relative_to(output_root)
+ detail_path.relative_to(artifact)
+ except ValueError as exc:
+ raise ApiError("The requested detail layer is outside the local point-cloud outputs.") from exc
+ if not detail_path.is_file() or detail_path.suffix.lower() not in {".las", ".laz", ".ply"}:
+ raise ApiError("The local RGB detail layer is unavailable.")
+ python = root / ".venvs" / POINTCLOUD_CPU_ENVIRONMENT / "Scripts" / "python.exe"
+ exporter = root / "capabilities" / "05-3d-pointcloud" / "export_pointcloud_detail.py"
+ if not python.is_file() or not exporter.is_file():
+ raise ApiError("Point-cloud detail exporter is unavailable. Run the capability setup first.")
+ command = [
+ str(python), str(exporter), "--input", str(detail_path), "--center",
+ *(f"{value:.12g}" for value in center), "--radius", f"{radius:.12g}",
+ "--max-points", "1200000",
+ ]
+ if not POINTCLOUD_DETAIL_REQUEST_LOCK.acquire(timeout=1):
+ raise ApiError("A point-cloud detail request is already running. Stop moving briefly and retry.")
+ try:
+ completed = subprocess.run(command, capture_output=True, timeout=120, check=False)
+ except subprocess.TimeoutExpired as exc:
+ raise ApiError("The local point-cloud detail request exceeded 120 seconds.") from exc
+ finally:
+ POINTCLOUD_DETAIL_REQUEST_LOCK.release()
+ if completed.returncode != 0:
+ message = completed.stderr.decode("utf-8", errors="replace").strip()
+ raise ApiError(message[:400] or "The local point-cloud detail exporter failed.")
+ payload = completed.stdout
+ if not payload.startswith(b"ply\nformat binary_little_endian 1.0\n") or len(payload) > 24 * 1024 * 1024:
+ raise ApiError("The local point-cloud detail response is invalid or exceeds its size limit.")
+ return payload
+
+
+def ply_vertex_count(path: Path) -> int | None:
+ """Read only the bounded PLY header; the point body can be hundreds of MB."""
+ try:
+ with path.open("rb") as stream:
+ header = bytearray()
+ while len(header) < 65_536:
+ line = stream.readline(4_096)
+ if not line:
+ return None
+ header.extend(line)
+ if line.rstrip(b"\r\n") == b"end_header":
+ break
+ else:
+ return None
+ vertex_count: int | None = None
+ for line in header.decode("ascii").splitlines():
+ fields = line.split()
+ if len(fields) == 3 and fields[:2] == ["element", "vertex"] and fields[2].isdigit():
+ vertex_count = int(fields[2])
+ return vertex_count
+ except (OSError, UnicodeDecodeError):
+ return None
+
+
+def npz_array_row_count(path: Path, array_name: str) -> int | None:
+ """Validate an NPZ array shape without importing a ML environment in the console."""
+ try:
+ with zipfile.ZipFile(path) as archive:
+ with archive.open(f"{array_name}.npy") as stream:
+ magic = stream.read(6)
+ version = stream.read(2)
+ if magic != b"\x93NUMPY" or len(version) != 2:
+ return None
+ header_size = 2 if version[0] == 1 else 4
+ header_length = int.from_bytes(stream.read(header_size), "little")
+ if header_length < 1 or header_length > 16_384:
+ return None
+ header = ast.literal_eval(stream.read(header_length).decode("latin1"))
+ shape = header.get("shape") if isinstance(header, dict) else None
+ if not isinstance(shape, tuple) or len(shape) != 2 or not all(isinstance(value, int) and value >= 0 for value in shape):
+ return None
+ return int(shape[0])
+ except (OSError, KeyError, ValueError, SyntaxError, zipfile.BadZipFile):
+ return None
+
+
+def multiview_pointcloud_annotation_sources(root: Path) -> list[dict[str, Any]]:
+ """Discover only complete, order-verified multi-view fusion annotation sources."""
+ output_root = root / "shared" / "outputs" / "05-3d-pointcloud"
+ records: list[dict[str, Any]] = []
+ for metadata_path in output_root.glob("multiview-feature-*/run_metadata.json"):
+ artifact = metadata_path.parent
+ metadata = load_json(metadata_path)
+ contract = metadata.get("annotation_source")
+ artifacts = metadata.get("artifacts")
+ if metadata.get("capability") != "05-3d-pointcloud" or not isinstance(contract, dict) or not isinstance(artifacts, dict):
+ continue
+ if contract.get("schema_version") != 1 or contract.get("kind") != "multiview_photo_feature_fusion":
+ continue
+ point_cloud = contract.get("point_cloud")
+ feature_dataset = contract.get("feature_dataset")
+ point_count = contract.get("point_count")
+ if (
+ not isinstance(point_cloud, str)
+ or not isinstance(feature_dataset, str)
+ or Path(point_cloud).name != point_cloud
+ or Path(feature_dataset).name != feature_dataset
+ or point_cloud != "multiview-annotation-source.ply"
+ or feature_dataset != "multiview-point-features.npz"
+ or artifacts.get("annotation_source") != point_cloud
+ or artifacts.get("feature_dataset") != feature_dataset
+ or not isinstance(point_count, int)
+ or point_count < 1
+ ):
+ continue
+ cloud_path = artifact / point_cloud
+ dataset_path = artifact / feature_dataset
+ if not cloud_path.is_file() or not dataset_path.is_file():
+ continue
+ if contract.get("point_cloud_sha256") != file_sha256(cloud_path) or contract.get("feature_dataset_sha256") != file_sha256(dataset_path):
+ continue
+ if ply_vertex_count(cloud_path) != point_count or npz_array_row_count(dataset_path, "xyz") != point_count or npz_array_row_count(dataset_path, "las_rgb") != point_count:
+ continue
+ records.append({
+ "id": f"{artifact.name}:{point_cloud}",
+ "runId": artifact.name,
+ "label": f"多视角照片特征融合样本 / {artifact.name}",
+ "artifactRoot": relative_path(root, artifact),
+ "file": point_cloud,
+ "url": f"/{relative_path(root, artifact)}/{point_cloud}",
+ "sha256": str(contract["point_cloud_sha256"]),
+ "pointCount": point_count,
+ "sourceKind": "多视角照片特征融合样本（原始 LAS RGB / XYZ；与特征数据同序）",
+ })
+ return sorted(records, key=lambda item: (item["runId"], item["id"]), reverse=True)
+
+
+def annotation_classes_path(root: Path) -> Path:
+ return root / "shared" / "outputs" / "05-3d-pointcloud" / "annotation-classes.json"
+
+
+def annotation_classes(root: Path) -> list[dict[str, Any]]:
+ """Return the local editable taxonomy, with stable defaults for old workspaces."""
+ saved = load_json(annotation_classes_path(root)).get("classes")
+ if not isinstance(saved, list):
+ return [dict(item) for item in DEFAULT_POINTCLOUD_ANNOTATION_CLASSES]
+ try:
+ return validate_annotation_classes(saved, allow_builtin=True)
+ except ApiError:
+ # A corrupt local taxonomy must not prevent existing annotations from opening.
+ return [dict(item) for item in DEFAULT_POINTCLOUD_ANNOTATION_CLASSES]
+
+
+def validate_annotation_classes(values: list[Any], *, allow_builtin: bool) -> list[dict[str, Any]]:
+ if not values or len(values) > 64:
+ raise ApiError("The annotation taxonomy must contain 1 to 64 classes.")
+ normalized: list[dict[str, Any]] = []
+ codes: set[int] = set()
+ keys: set[str] = set()
+ labels: set[str] = set()
+ built_in_codes = {item["code"] for item in DEFAULT_POINTCLOUD_ANNOTATION_CLASSES}
+ for item in values:
+ if not isinstance(item, dict):
+ raise ApiError("Each annotation class must be an object.")
+ code, key, label, color = item.get("code"), item.get("key"), item.get("label"), item.get("color")
+ if not isinstance(code, int) or not 1 <= code <= 255:
+ raise ApiError("Annotation class codes must be integers from 1 to 255 for LAS compatibility.")
+ if not isinstance(key, str) or not ANNOTATION_CLASS_KEY.fullmatch(key):
+ raise ApiError("Annotation class keys must use lowercase English letters, numbers, and underscores.")
+ if not isinstance(label, str) or not 1 <= len(label.strip()) <= 40:
+ raise ApiError("Annotation class labels must contain 1 to 40 characters.")
+ if not isinstance(color, list) or len(color) != 3 or not all(isinstance(value, int) and 0 <= value <= 255 for value in color):
+ raise ApiError("Annotation class colors must be three RGB integers from 0 to 255.")
+ if code in codes or key in keys or label.strip() in labels:
+ raise ApiError("Annotation class code, key, and label must each be unique.")
+ if code in built_in_codes and not allow_builtin:
+ raise ApiError("Built-in annotation classes cannot be replaced.")
+ codes.add(code); keys.add(key); labels.add(label.strip())
+ normalized.append({"code": code, "key": key, "label": label.strip(), "color": color, "builtIn": code in built_in_codes})
+ return sorted(normalized, key=lambda item: item["code"])
+
+
+def write_annotation_classes(root: Path, classes: list[dict[str, Any]]) -> None:
+ path = annotation_classes_path(root)
+ path.parent.mkdir(parents=True, exist_ok=True)
+ path.write_text(json.dumps({"schema_version": 1, "classes": classes}, ensure_ascii=False, indent=2), encoding="utf-8")
+
+
+def annotation_class_codes_in_use(root: Path) -> set[int]:
+ used: set[int] = set()
+ for path in (root / "shared" / "outputs" / "05-3d-pointcloud" / "annotations").glob("*/annotation.json"):
+ record = load_json(path)
+ for label in record.get("labels", []):
+ if isinstance(label, list) and len(label) == 2 and isinstance(label[1], int):
+ used.add(label[1])
+ return used
def pointcloud_annotations(root: Path) -> list[dict[str, Any]]:
@@ -526,16 +783,170 @@
return sorted(records, key=lambda item: (str(item["createdAt"]), str(item["id"])), reverse=True)
+def pointcloud_annotation_source_deletion_plan(root: Path, source_id: str) -> dict[str, Any]:
+ """Describe every generated artifact that will be removed for one source.
+
+ Source paths are discovered server-side. Files outside the established
+ point-cloud output/raw/processed layouts, including ``baseData``, are never
+ part of this plan.
+ """
+ source = next((item for item in pointcloud_annotation_sources(root) if item["id"] == source_id), None)
+ if not source:
+ raise ApiError("The selected annotation source is unavailable.")
+ output_root = (root / "shared" / "outputs" / "05-3d-pointcloud").resolve()
+ artifact = (root / str(source["artifactRoot"])).resolve()
+ try:
+ artifact.relative_to(output_root)
+ except ValueError as exc:
+ raise ApiError("The selected annotation source is outside the allowed output directory.") from exc
+ siblings = [item for item in pointcloud_annotation_sources(root) if item["runId"] == str(source["runId"])]
+ source_ids = {item["id"] for item in siblings}
+ annotation_root = output_root / "annotations"
+ annotations: list[Path] = []
+ for path in annotation_root.glob("*/annotation.json"):
+ if load_json(path).get("source_id") in source_ids:
+ annotations.append(path.parent)
+ annotation_paths = {path / "annotation.json" for path in annotations}
+ training_root = output_root / "training-runs"
+ training: list[Path] = []
+ for metrics_path in training_root.glob("*/metrics.json"):
+ value = load_json(metrics_path).get("annotation")
+ if not isinstance(value, str):
+ continue
+ try:
+ if Path(value).resolve() in annotation_paths:
+ training.append(metrics_path.parent)
+ except OSError:
+ continue
+ training_models = {path / "model.pt" for path in training}
+ inference_root = output_root / "model-inference-runs"
+ inference: list[Path] = []
+ for metadata_path in inference_root.glob("*/run_metadata.json"):
+ model = load_json(metadata_path).get("model")
+ model_path = model.get("path") if isinstance(model, dict) else None
+ if not isinstance(model_path, str):
+ continue
+ try:
+ if Path(model_path).resolve() in training_models:
+ inference.append(metadata_path.parent)
+ except OSError:
+ continue
+ run_id = str(source["runId"])
+ raw_root = root / "shared" / "data" / "raw" / "05-3d-pointcloud"
+ processed_root = root / "shared" / "data" / "processed" / "05-3d-pointcloud"
+ raw_candidates = [raw_root / "annotation-source-runs" / run_id, raw_root / "runs" / run_id]
+ processed_candidates = [processed_root / "annotation-source-runs" / run_id, processed_root / "runs" / run_id]
+ raw = [path for path in raw_candidates if path.is_dir()]
+ processed = [path for path in processed_candidates if path.is_dir()]
+ metadata = load_json(artifact / "run_metadata.json")
+ return {
+ "sourceId": source_id,
+ "label": source["label"],
+ "sourceKind": source["sourceKind"],
+ "runId": run_id,
+ "artifactRoot": str(source["artifactRoot"]),
+ "outputDirectories": 1,
+ "rawDirectories": len(raw),
+ "processedDirectories": len(processed),
+ "annotationRevisions": len(annotations),
+ "trainingRuns": len(training),
+ "inferenceRuns": len(inference),
+ "siblingSources": len(siblings),
+ "removesOriginalUpload": bool(raw),
+ "preservesExternalInputs": not bool(raw) and not bool(metadata.get("annotation_source_job")),
+ }
+
+
+def assert_removable_pointcloud_directory(root: Path, path: Path, allowed_root: Path) -> None:
+ resolved = path.resolve()
+ allowed = allowed_root.resolve()
+ try:
+ relative = resolved.relative_to(allowed)
+ except ValueError as exc:
+ raise ApiError("A deletion target is outside the allowed point-cloud workspace.") from exc
+ if not relative.parts or not resolved.is_dir():
+ raise ApiError("A deletion target is invalid.")
+
+
+def active_pointcloud_source_dependencies(run_id: str, annotation_ids: set[str], training_ids: set[str]) -> bool:
+ active = {"queued", "running"}
+ with POINTCLOUD_ANNOTATION_SOURCE_JOBS_LOCK:
+ if any(job.get("runId") == run_id and job.get("status") in active for job in POINTCLOUD_ANNOTATION_SOURCE_JOBS.values()):
+ return True
+ with POINTCLOUD_TRAINING_JOBS_LOCK:
+ if any(job.get("annotationId") in annotation_ids for job in POINTCLOUD_TRAINING_JOBS.values() if job.get("status") in active):
+ return True
+ with POINTCLOUD_INFERENCE_JOBS_LOCK:
+ if any(job.get("modelId") in training_ids and job.get("status") in active for job in POINTCLOUD_INFERENCE_JOBS.values()):
+ return True
+ return False
+
+
+def pointcloud_annotation_source_job(job_id: str) -> dict[str, Any] | None:
+ with POINTCLOUD_ANNOTATION_SOURCE_JOBS_LOCK:
+ value = POINTCLOUD_ANNOTATION_SOURCE_JOBS.get(job_id)
+ return dict(value) if value else None
+
+
+def execute_pointcloud_annotation_source_job(
+ root: Path,
+ job_id: str,
+ processed_path: Path,
+ output: Path,
+ source_sha256: str,
+ source_bytes: int,
+) -> None:
+ with POINTCLOUD_ANNOTATION_SOURCE_JOBS_LOCK:
+ POINTCLOUD_ANNOTATION_SOURCE_JOBS[job_id].update({"status": "running", "stage": "preparing_annotation_preview", "startedAt": datetime.now(UTC).isoformat()})
+ command = [
+ str(root / ".venvs" / POINTCLOUD_CPU_ENVIRONMENT / "Scripts" / "python.exe"),
+ str(root / "capabilities" / "05-3d-pointcloud" / "prepare_annotation_source.py"),
+ "--input", str(processed_path), "--output", str(output),
+ ]
+ try:
+ with RUN_LOCK:
+ completed = subprocess.run(command, cwd=root, capture_output=True, text=True, timeout=900, check=False)
+ if completed.returncode:
+ message = (completed.stderr or completed.stdout or "Unknown script error.").strip().splitlines()[-1]
+ raise RuntimeError(f"Processing failed: {message[:600]}")
+ metadata_path = output / "run_metadata.json"
+ if not metadata_path.is_file():
+ raise RuntimeError("Point-cloud processing finished without result metadata.")
+ metadata = load_json(metadata_path)
+ run_id = pointcloud_annotation_source_job(job_id)["runId"]
+ metadata["input_dir"] = relative_path(root, processed_path.parent)
+ metadata["raw_input_dir"] = relative_path(root, root / "shared" / "data" / "raw" / "05-3d-pointcloud" / "annotation-source-runs" / run_id)
+ metadata["source_sha256"] = {processed_path.name: source_sha256}
+ metadata["source_bytes"] = {processed_path.name: source_bytes}
+ metadata["annotation_source_job"] = True
+ metadata_path.write_text(json.dumps(metadata, ensure_ascii=False, indent=2), encoding="utf-8")
+ source = next((item for item in standalone_pointcloud_annotation_sources(root) if item["runId"] == run_id), None)
+ if not source:
+ raise RuntimeError("Annotation preview finished without a discoverable annotation source.")
+ with POINTCLOUD_ANNOTATION_SOURCE_JOBS_LOCK:
+ POINTCLOUD_ANNOTATION_SOURCE_JOBS[job_id].update({"status": "complete", "stage": "complete", "completedAt": datetime.now(UTC).isoformat(), "source": source})
+ except Exception as exc:
+ with POINTCLOUD_ANNOTATION_SOURCE_JOBS_LOCK:
+ POINTCLOUD_ANNOTATION_SOURCE_JOBS[job_id].update({"status": "failed", "stage": "failed", "completedAt": datetime.now(UTC).isoformat(), "error": str(exc)[:700]})
+
+
def pointcloud_training_job(job_id: str) -> dict[str, Any] | None:
with POINTCLOUD_TRAINING_JOBS_LOCK:
value = POINTCLOUD_TRAINING_JOBS.get(job_id)
return dict(value) if value else None
-def execute_pointcloud_training_job(root: Path, job_id: str, annotation: Path, output: Path, execution: dict[str, str]) -> None:
+def execute_pointcloud_training_job(root: Path, job_id: str, annotation: Path, output: Path, execution: dict[str, str], trainer: str = "rgb_xyz_baseline") -> None:
with POINTCLOUD_TRAINING_JOBS_LOCK:
POINTCLOUD_TRAINING_JOBS[job_id].update({"status": "running", "stage": "training", "startedAt": datetime.now(UTC).isoformat()})
- command = [execution["python"], str(root / "capabilities" / "05-3d-pointcloud" / "train_pointcloud_semantic_model.py"), "--annotation", str(annotation), "--output", str(output), "--device", execution["device"]]
+ scripts = {
+ "rgb_xyz_baseline": "train_pointcloud_semantic_model.py",
+ "multiview_local_attention_baseline": "train_multiview_point_transformer.py",
+ }
+ script = scripts.get(trainer)
+ if not script:
+ raise ApiError("Unsupported point-cloud training workflow.")
+ command = [execution["python"], str(root / "capabilities" / "05-3d-pointcloud" / script), "--annotation", str(annotation), "--output", str(output), "--device", execution["device"]]
try:
with RUN_LOCK:
completed = subprocess.run(command, cwd=root, capture_output=True, text=True, timeout=14_400, check=False)
@@ -547,8 +958,11 @@
preview = output / "predicted-semantic-preview.ply"
if not all(path.is_file() for path in (metrics, model, preview)):
raise ApiError("Training finished without model, metrics, and predicted preview artifacts.")
+ metadata = output / "run_metadata.json"
+ if metadata.is_file():
+ record_execution_metadata(metadata, execution)
with POINTCLOUD_TRAINING_JOBS_LOCK:
- POINTCLOUD_TRAINING_JOBS[job_id].update({"status": "complete", "stage": "complete", "completedAt": datetime.now(UTC).isoformat(), "artifactRoot": relative_path(root, output), "metrics": relative_path(root, metrics), "model": relative_path(root, model), "preview": relative_path(root, preview)})
+ POINTCLOUD_TRAINING_JOBS[job_id].update({"status": "complete", "stage": "complete", "completedAt": datetime.now(UTC).isoformat(), "artifactRoot": relative_path(root, output), "metrics": relative_path(root, metrics), "model": relative_path(root, model), "preview": relative_path(root, preview), "trainer": trainer})
except Exception as exc:
with POINTCLOUD_TRAINING_JOBS_LOCK:
POINTCLOUD_TRAINING_JOBS[job_id].update({"status": "failed", "stage": "failed", "completedAt": datetime.now(UTC).isoformat(), "error": str(exc)[:700]})
@@ -562,7 +976,7 @@
metrics_path = model_path.with_name("metrics.json")
metrics = load_json(metrics_path)
classes = metrics.get("classes")
- if metrics.get("capability") != "05-3d-pointcloud" or metrics.get("classification") != "B" or not isinstance(classes, dict):
+ if metrics.get("capability") != "05-3d-pointcloud" or metrics.get("classification") != "B" or metrics.get("model_input_kind", "rgb_xyz") != "rgb_xyz" or not isinstance(classes, dict):
continue
class_codes = sorted(str(code) for code in classes if str(code).isdigit())
if len(class_codes) < 2:
@@ -584,6 +998,69 @@
with POINTCLOUD_INFERENCE_JOBS_LOCK:
value = POINTCLOUD_INFERENCE_JOBS.get(job_id)
return dict(value) if value else None
+
+
+def latest_pointcloud_auto_annotation_job(root: Path, source_id: str, model_id: str) -> dict[str, Any] | None:
+ """Recover a completed local automatic-annotation run after a server restart."""
+ source = next((item for item in pointcloud_annotation_sources(root) if item["id"] == source_id), None)
+ model = next((item for item in pointcloud_semantic_models(root) if item["id"] == model_id), None)
+ if not source or not model:
+ raise ApiError("The selected annotation source or trained model is unavailable.")
+ output_root = root / "shared" / "outputs" / "05-3d-pointcloud" / "auto-annotation-runs"
+ records: list[dict[str, Any]] = []
+ model_sha256 = file_sha256(root / str(model["model"]))
+ for metadata_path in output_root.glob("*/run_metadata.json"):
+ artifact = metadata_path.parent
+ metadata = load_json(metadata_path)
+ prediction = metadata.get("prediction")
+ input_record = metadata.get("input")
+ model_record = metadata.get("model")
+ processing = metadata.get("processing")
+ automatic = prediction.get("automatic_annotation") if isinstance(prediction, dict) else None
+ preview = artifact / "predicted-semantic-preview.ply"
+ classified_las = artifact / "predicted-semantic-classified.las"
+ class_counts = artifact / "class-counts.csv"
+ summary = artifact / "prediction-summary.json"
+ candidates = artifact / "automatic-annotation-candidates.json"
+ if (
+ metadata.get("capability") != "05-3d-pointcloud"
+ or not isinstance(input_record, dict)
+ or not isinstance(model_record, dict)
+ or not isinstance(processing, dict)
+ or not isinstance(automatic, dict)
+ or input_record.get("sha256") != source.get("sha256")
+ or model_record.get("sha256") != model_sha256
+ or not all(path.is_file() for path in (preview, classified_las, class_counts, summary, candidates))
+ ):
+ continue
+ confidence = automatic.get("candidate_confidence")
+ if not isinstance(confidence, (int, float)):
+ continue
+ records.append({
+ "id": f"recovered-{artifact.name}",
+ "runId": artifact.name,
+ "modelId": model_id,
+ "sourceId": source_id,
+ "inputName": str(source["file"]),
+ "candidateConfidence": float(confidence),
+ "status": "complete",
+ "stage": "complete",
+ "requestedDevice": processing.get("requested_device"),
+ "device": processing.get("device", "cpu"),
+ "environment": processing.get("environment"),
+ "torchVersion": metadata.get("versions", {}).get("torch") if isinstance(metadata.get("versions"), dict) else None,
+ "fallbackUsed": processing.get("fallback_used", False),
+ "fallbackReason": processing.get("fallback_reason"),
+ "createdAt": str(metadata.get("created_at") or ""),
+ "artifactRoot": relative_path(root, artifact),
+ "metadata": relative_path(root, metadata_path),
+ "preview": relative_path(root, preview),
+ "classifiedLas": relative_path(root, classified_las),
+ "classCounts": relative_path(root, class_counts),
+ "summary": relative_path(root, summary),
+ "candidateFile": relative_path(root, candidates),
+ })
+ return max(records, key=lambda item: (item["createdAt"], item["runId"])) if records else None
def validate_pointcloud_model_input(path: Path) -> None:
@@ -611,10 +1088,12 @@
raise ApiError("LAS/LAZ 文件没有点记录。请选择包含实际 RGB 点位的完整点云文件，而不是空分块。")
-def execute_pointcloud_inference_job(root: Path, job_id: str, model: Path, source: Path, output: Path, execution: dict[str, str]) -> None:
+def execute_pointcloud_inference_job(root: Path, job_id: str, model: Path, source: Path, output: Path, execution: dict[str, str], annotation_source_id: str | None = None, candidate_confidence: float | None = None) -> None:
with POINTCLOUD_INFERENCE_JOBS_LOCK:
POINTCLOUD_INFERENCE_JOBS[job_id].update({"status": "running", "stage": "inference", "startedAt": datetime.now(UTC).isoformat()})
command = [execution["python"], str(root / "capabilities" / "05-3d-pointcloud" / "apply_pointcloud_semantic_model.py"), "--model", str(model), "--input", str(source), "--output", str(output), "--device", execution["device"]]
+ if annotation_source_id:
+ command.extend(["--annotation-source-id", annotation_source_id, "--candidate-confidence", f"{candidate_confidence or 0.95:.6f}"])
try:
with RUN_LOCK:
completed = subprocess.run(command, cwd=root, capture_output=True, text=True, timeout=14_400, check=False)
@@ -628,8 +1107,12 @@
summary = output / "prediction-summary.json"
if not all(path.is_file() for path in (metadata, preview, classified_las, counts, summary)):
raise ApiError("Model inference finished without all expected prediction artifacts.")
+ candidates = output / "automatic-annotation-candidates.json"
+ if annotation_source_id and not candidates.is_file():
+ raise ApiError("Automatic annotation finished without the expected candidate artifact.")
+ record_execution_metadata(metadata, execution)
with POINTCLOUD_INFERENCE_JOBS_LOCK:
- POINTCLOUD_INFERENCE_JOBS[job_id].update({"status": "complete", "stage": "complete", "completedAt": datetime.now(UTC).isoformat(), "artifactRoot": relative_path(root, output), "metadata": relative_path(root, metadata), "preview": relative_path(root, preview), "classifiedLas": relative_path(root, classified_las), "classCounts": relative_path(root, counts), "summary": relative_path(root, summary)})
+ POINTCLOUD_INFERENCE_JOBS[job_id].update({"status": "complete", "stage": "complete", "completedAt": datetime.now(UTC).isoformat(), "artifactRoot": relative_path(root, output), "metadata": relative_path(root, metadata), "preview": relative_path(root, preview), "classifiedLas": relative_path(root, classified_las), "classCounts": relative_path(root, counts), "summary": relative_path(root, summary), "candidateFile": relative_path(root, candidates) if annotation_source_id else None})
except Exception as exc:
with POINTCLOUD_INFERENCE_JOBS_LOCK:
POINTCLOUD_INFERENCE_JOBS[job_id].update({"status": "failed", "stage": "failed", "completedAt": datetime.now(UTC).isoformat(), "error": str(exc)[:700]})
@@ -700,6 +1183,161 @@
return sorted(records, key=lambda item: (item["createdAt"], item["id"]), reverse=True)
+# A console run is removable only when it was created in the fixed ``runs``
+# layout. Discovery also exposes baseline and validation artifacts, but those
+# are project evidence rather than disposable console-owned copies.
+RUN_DELETION_CAPABILITIES: dict[str, tuple[str, Any]] = {
+ "00-change-detection": ("00-change-detection", change_runs),
+ "01-object-detection": ("01-object-detection", detection_runs),
+ "02-semantic-mapping": ("02-semantic-mapping", semantic_runs),
+ "04-spatial-measurement": ("04-spatial-measurement", measurement_runs),
+ "05-3d-pointcloud": ("05-3d-pointcloud", pointcloud_runs),
+ "07-risk-rule-engine": ("07-risk-rule-engine", risk_rule_runs),
+ "09-anomaly-detection": ("09-anomaly-detection", anomaly_runs),
+ "15-trajectory-analysis": ("15-trajectory-analysis", trajectory_runs),
+}
+
+
+def valid_run_deletion_id(run_id: str) -> bool:
+ return bool(run_id) and len(run_id) <= 160 and "/" not in run_id and "\\" not in run_id and not SAFE_FILE_NAME.search(run_id)
+
+
+def existing_directory(path: Path, parent: Path) -> Path | None:
+ """Return an existing direct child directory, never a caller-supplied path."""
+ try:
+ resolved_parent = parent.resolve()
+ resolved = path.resolve()
+ resolved.relative_to(resolved_parent)
+ except (OSError, ValueError):
+ return None
+ return resolved if resolved.is_dir() else None
+
+
+def run_deletion_plan(root: Path, capability: str, run_id: str) -> dict[str, Any]:
+ if capability not in RUN_DELETION_CAPABILITIES:
+ raise ApiError("This capability does not expose removable console runs.")
+ if not valid_run_deletion_id(run_id):
+ raise ApiError("Invalid run id.")
+ capability_dir, discover = RUN_DELETION_CAPABILITIES[capability]
+ record = next((item for item in discover(root) if item.get("id") == run_id), None)
+ if not record:
+ raise ApiError("The selected result is unavailable.")
+
+ output_root = (root / "shared" / "outputs" / capability_dir).resolve()
+ expected_output = (output_root / "runs" / run_id).resolve()
+ artifact_value = record.get("artifactRoot")
+ try:
+ artifact = (root / str(artifact_value)).resolve()
+ except OSError as exc:
+ raise ApiError("The selected result has an invalid artifact location.") from exc
+ if artifact != expected_output:
+ return {
+ "capability": capability,
+ "runId": run_id,
+ "label": str(record.get("label") or run_id),
+ "removable": False,
+ "reason": "This is a built-in baseline, validation artifact, or external result. It was not created in the console-owned run layout.",
+ "outputDirectories": [],
+ "rawDirectories": [],
+ "processedDirectories": [],
+ "dependentDirectories": [],
+ "preservesExternalInputs": True,
+ }
+
+ output_directories = [expected_output] if existing_directory(expected_output, output_root / "runs") else []
+ raw_root = (root / "shared" / "data" / "raw" / capability_dir).resolve()
+ processed_root = (root / "shared" / "data" / "processed" / capability_dir).resolve()
+ raw_candidates = [raw_root / "runs" / run_id]
+ processed_candidates = [processed_root / run_id, processed_root / "runs" / run_id]
+ raw_directories = [path for candidate in raw_candidates if (path := existing_directory(candidate, raw_root))]
+ processed_directories = [path for candidate in processed_candidates if (path := existing_directory(candidate, processed_root))]
+ dependent_directories: list[Path] = []
+ if capability == "00-change-detection":
+ scan_root = output_root / "parameter-scans" / run_id
+ if path := existing_directory(scan_root, output_root / "parameter-scans"):
+ dependent_directories.append(path)
+ return {
+ "capability": capability,
+ "runId": run_id,
+ "label": str(record.get("label") or run_id),
+ "removable": bool(output_directories),
+ "reason": None if output_directories else "The console-owned output directory is missing, so no deletion is performed.",
+ "outputDirectories": [relative_path(root, path) for path in output_directories],
+ "rawDirectories": [relative_path(root, path) for path in raw_directories],
+ "processedDirectories": [relative_path(root, path) for path in processed_directories],
+ "dependentDirectories": [relative_path(root, path) for path in dependent_directories],
+ "preservesExternalInputs": True,
+ }
+
+
+def delete_console_run(root: Path, capability: str, run_id: str) -> dict[str, Any]:
+ plan = run_deletion_plan(root, capability, run_id)
+ if not plan["removable"]:
+ raise ApiError(str(plan["reason"] or "The selected result cannot be removed."))
+ directories = [*plan["dependentDirectories"], *plan["processedDirectories"], *plan["rawDirectories"], *plan["outputDirectories"]]
+ for relative in directories:
+ location = (root / relative).resolve()
+ # Every entry was created by run_deletion_plan from fixed roots above.
+ if location.is_dir():
+ shutil.rmtree(location)
+ return {"capability": capability, "runId": run_id, "removedDirectories": directories, "preservesExternalInputs": True}
+
+
+def pointcloud_semantic_model_deletion_plan(root: Path, model_id: str) -> dict[str, Any]:
+ if not valid_run_deletion_id(model_id):
+ raise ApiError("Invalid semantic model id.")
+ model = next((item for item in pointcloud_semantic_models(root) if item["id"] == model_id), None)
+ if not model:
+ raise ApiError("The selected trained model is unavailable.")
+ output_root = (root / "shared" / "outputs" / "05-3d-pointcloud").resolve()
+ model_root = (output_root / "training-runs" / model_id).resolve()
+ model_path = (root / str(model["model"])).resolve()
+ if model_path != model_root / "model.pt" or not existing_directory(model_root, output_root / "training-runs"):
+ raise ApiError("The selected trained model is outside the console-owned training layout.")
+ model_sha256 = file_sha256(model_path)
+ inference_directories: list[Path] = []
+ for metadata_path in (output_root / "model-inference-runs").glob("*/run_metadata.json"):
+ metadata = load_json(metadata_path)
+ value = metadata.get("model")
+ candidate = value.get("path") if isinstance(value, dict) else None
+ try:
+ if isinstance(candidate, str) and Path(candidate).resolve() == model_path:
+ inference_directories.append(metadata_path.parent.resolve())
+ except OSError:
+ continue
+ auto_directories: list[Path] = []
+ for metadata_path in (output_root / "auto-annotation-runs").glob("*/run_metadata.json"):
+ value = load_json(metadata_path).get("model")
+ if isinstance(value, dict) and value.get("sha256") == model_sha256:
+ auto_directories.append(metadata_path.parent.resolve())
+ raw_root = (root / "shared" / "data" / "raw" / "05-3d-pointcloud" / "model-inference-runs").resolve()
+ processed_root = (root / "shared" / "data" / "processed" / "05-3d-pointcloud" / "model-inference-runs").resolve()
+ raw_directories = [path for directory in inference_directories if (path := existing_directory(raw_root / directory.name, raw_root))]
+ processed_directories = [path for directory in inference_directories if (path := existing_directory(processed_root / directory.name, processed_root))]
+ return {
+ "modelId": model_id,
+ "label": str(model["label"]),
+ "removable": True,
+ "trainingDirectories": [relative_path(root, model_root)],
+ "inferenceDirectories": [relative_path(root, path) for path in inference_directories],
+ "autoAnnotationDirectories": [relative_path(root, path) for path in auto_directories],
+ "rawDirectories": [relative_path(root, path) for path in raw_directories],
+ "processedDirectories": [relative_path(root, path) for path in processed_directories],
+ "preservesAnnotationRevisions": True,
+ "preservesExternalInputs": True,
+ }
+
+
+def delete_pointcloud_semantic_model(root: Path, model_id: str) -> dict[str, Any]:
+ plan = pointcloud_semantic_model_deletion_plan(root, model_id)
+ directories = [*plan["autoAnnotationDirectories"], *plan["inferenceDirectories"], *plan["processedDirectories"], *plan["rawDirectories"], *plan["trainingDirectories"]]
+ for relative in directories:
+ location = (root / relative).resolve()
+ if location.is_dir():
+ shutil.rmtree(location)
+ return {"modelId": model_id, "removedDirectories": directories, "preservesAnnotationRevisions": True, "preservesExternalInputs": True}
+
+
def anomaly_job(job_id: str) -> dict[str, Any] | None:
with ANOMALY_JOB_LOCK:
value = ANOMALY_JOBS.get(job_id)
@@ -709,7 +1347,21 @@
def photo_reconstruction_job(job_id: str) -> dict[str, Any] | None:
with PHOTO_RECONSTRUCTION_JOBS_LOCK:
value = PHOTO_RECONSTRUCTION_JOBS.get(job_id)
- return dict(value) if value else None
+ job = dict(value) if value else None
+ if not job:
+ return None
+ progress_path = job.pop("progressPath", None)
+ if job["status"] == "complete":
+ job["progress"] = {"percent": 100, "stage": "complete", "message": "照片重建已完成，结果已加入案例库。", "inputImages": job["inputImages"], "estimate": True}
+ return job
+ if job["status"] == "failed":
+ job["progress"] = {"percent": 0, "stage": "failed", "message": job.get("error") or "照片重建失败。", "inputImages": job["inputImages"], "estimate": True}
+ return job
+ if isinstance(progress_path, str):
+ progress = load_json(Path(progress_path))
+ if progress:
+ job["progress"] = progress
+ return job
def run_background_command(command: list[str], root: Path, timeout: int) -> None:
@@ -737,6 +1389,7 @@
"--input", str(processed_root), "--output", str(sparse_output),
"--max-image-size", "2000", "--max-features", "18000",
"--camera-model", "OPENCV",
+ "--progress-file", str(processed_root / "photo_reconstruction_progress.json"),
]
if use_position_priors:
sparse_command.extend(["--matching-mode", "spatial", "--matching-neighbors", "4", "--use-position-priors", "--prior-position-loss-scale-m", "0.05"])
@@ -749,6 +1402,7 @@
"--openmvs-bin", str(root / "shared" / "tools" / "openmvs-2.4.0" / "vc17" / "x64" / "Release"),
"--threads", "12", "--max-resolution", "2400", "--dense-resolution-level", "0",
"--dense-number-views", "8", "--dense-number-views-fuse", "2", "--target-faces", "800000",
+ "--progress-file", str(processed_root / "photo_reconstruction_progress.json"),
]
try:
with PHOTO_RECONSTRUCTION_JOBS_LOCK:
@@ -869,6 +1523,17 @@
def do_GET(self) -> None: # noqa: N802 - inherited standard-library method name
path = urlsplit(self.path).path
+ run_deletion_prefix = "/api/runs/"
+ if path.startswith(run_deletion_prefix) and path.endswith("/deletion-plan"):
+ try:
+ parts = path[len(run_deletion_prefix):].split("/")
+ if len(parts) != 3 or parts[2] != "deletion-plan":
+ raise ApiError("Invalid result deletion-plan endpoint.")
+ capability, run_id = (unquote(parts[0]), unquote(parts[1]))
+ self.send_json(HTTPStatus.OK, {"plan": run_deletion_plan(self.root, capability, run_id)})
+ except ApiError as exc:
+ self.send_json(HTTPStatus.BAD_REQUEST, {"error": str(exc)})
+ return
if path == "/api/change-detection/runs":
self.send_json(HTTPStatus.OK, {"runs": change_runs(self.root)})
return
@@ -905,11 +1570,73 @@
if path == "/api/3d-pointcloud/annotation-sources":
self.send_json(HTTPStatus.OK, {"sources": pointcloud_annotation_sources(self.root)})
return
+ if path == "/api/3d-pointcloud/annotation-detail":
+ try:
+ query = parse_qs(urlsplit(self.path).query)
+ source_id = query.get("sourceId", [None])[0]
+ values = [query.get(axis, [None])[0] for axis in ("x", "y", "z", "radius")]
+ if not isinstance(source_id, str) or any(value is None for value in values):
+ raise ApiError("Detail request must include sourceId, x, y, z, and radius.")
+ try:
+ x, y, z, radius = (float(value) for value in values)
+ except (TypeError, ValueError) as exc:
+ raise ApiError("Detail coordinates and radius must be numbers.") from exc
+ payload = pointcloud_annotation_detail(self.root, source_id, (x, y, z), radius)
+ self.send_response(HTTPStatus.OK)
+ self.send_header("Content-Type", "application/octet-stream")
+ self.send_header("Content-Length", str(len(payload)))
+ self.send_header("Cache-Control", "no-store")
+ self.end_headers()
+ self.wfile.write(payload)
+ except ApiError as exc:
+ self.send_json(HTTPStatus.BAD_REQUEST, {"error": str(exc)})
+ return
+ deletion_plan_prefix = "/api/3d-pointcloud/annotation-source-deletion-plans/"
+ if path.startswith(deletion_plan_prefix):
+ try:
+ source_id = unquote(path[len(deletion_plan_prefix):])
+ self.send_json(HTTPStatus.OK, {"plan": pointcloud_annotation_source_deletion_plan(self.root, source_id)})
+ except ApiError as exc:
+ self.send_json(HTTPStatus.BAD_REQUEST, {"error": str(exc)})
+ return
+ if path == "/api/3d-pointcloud/annotation-classes":
+ self.send_json(HTTPStatus.OK, {"classes": annotation_classes(self.root)})
+ return
if path == "/api/3d-pointcloud/annotations":
self.send_json(HTTPStatus.OK, {"annotations": pointcloud_annotations(self.root)})
return
if path == "/api/3d-pointcloud/semantic-models":
self.send_json(HTTPStatus.OK, {"models": pointcloud_semantic_models(self.root)})
+ return
+ semantic_model_plan_prefix = "/api/3d-pointcloud/semantic-model-deletion-plans/"
+ if path.startswith(semantic_model_plan_prefix):
+ try:
+ model_id = unquote(path[len(semantic_model_plan_prefix):])
+ self.send_json(HTTPStatus.OK, {"plan": pointcloud_semantic_model_deletion_plan(self.root, model_id)})
+ except ApiError as exc:
+ self.send_json(HTTPStatus.BAD_REQUEST, {"error": str(exc)})
+ return
+ if path == "/api/3d-pointcloud/auto-annotation-runs/latest":
+ try:
+ query = parse_qs(urlsplit(self.path).query)
+ source_id = query.get("sourceId", [None])[0]
+ model_id = query.get("modelId", [None])[0]
+ if not isinstance(source_id, str) or not isinstance(model_id, str):
+ raise ApiError("Automatic-annotation recovery needs sourceId and modelId.")
+ self.send_json(HTTPStatus.OK, {"job": latest_pointcloud_auto_annotation_job(self.root, source_id, model_id)})
+ except ApiError as exc:
+ self.send_json(HTTPStatus.BAD_REQUEST, {"error": str(exc)})
+ return
+ review_prefix = "/api/3d-pointcloud/auto-annotation-review-drafts/"
+ if path.startswith(review_prefix):
+ try:
+ run_id = unquote(path[len(review_prefix):])
+ if not run_id or SAFE_FILE_NAME.search(run_id):
+ raise ApiError("Invalid automatic-annotation run id.")
+ review_path = self.root / "shared" / "outputs" / "05-3d-pointcloud" / "auto-annotation-runs" / run_id / "review-corrections.json"
+ self.send_json(HTTPStatus.OK, {"review": load_json(review_path) if review_path.is_file() else None})
+ except ApiError as exc:
+ self.send_json(HTTPStatus.BAD_REQUEST, {"error": str(exc)})
return
if path.startswith("/api/3d-pointcloud/model-inference-jobs/"):
job_id = path.rstrip("/").rsplit("/", 1)[-1]
@@ -920,6 +1647,11 @@
job_id = path.rstrip("/").rsplit("/", 1)[-1]
job = pointcloud_training_job(job_id)
self.send_json(HTTPStatus.OK if job else HTTPStatus.NOT_FOUND, {"job": job} if job else {"error": "Unknown point-cloud training job."})
+ return
+ if path.startswith("/api/3d-pointcloud/annotation-source-jobs/"):
+ job_id = path.rstrip("/").rsplit("/", 1)[-1]
+ job = pointcloud_annotation_source_job(job_id)
+ self.send_json(HTTPStatus.OK if job else HTTPStatus.NOT_FOUND, {"job": job} if job else {"error": "Unknown point-cloud annotation source job."})
return
if path.startswith("/api/3d-pointcloud/photo-reconstruction-jobs/"):
job_id = path.rstrip("/").rsplit("/", 1)[-1]
@@ -976,11 +1708,26 @@
if path == "/api/3d-pointcloud/annotations":
self.send_json(HTTPStatus.CREATED, {"annotation": self.create_pointcloud_annotation(payload)})
return
+ if path == "/api/3d-pointcloud/annotation-classes":
+ self.send_json(HTTPStatus.CREATED, {"class": self.create_pointcloud_annotation_class(payload)})
+ return
+ if path == "/api/3d-pointcloud/annotation-source-runs":
+ self.send_json(HTTPStatus.ACCEPTED, {"job": self.create_pointcloud_annotation_source_run(payload)})
+ return
if path == "/api/3d-pointcloud/training-runs":
self.send_json(HTTPStatus.ACCEPTED, {"job": self.create_pointcloud_training_run(payload)})
return
if path == "/api/3d-pointcloud/model-inference-runs":
self.send_json(HTTPStatus.ACCEPTED, {"job": self.create_pointcloud_model_inference_run(payload)})
+ return
+ if path == "/api/3d-pointcloud/auto-annotation-runs":
+ self.send_json(HTTPStatus.ACCEPTED, {"job": self.create_pointcloud_auto_annotation_run(payload)})
+ return
+ if path == "/api/3d-pointcloud/auto-annotation-acceptances":
+ self.send_json(HTTPStatus.CREATED, {"annotation": self.accept_pointcloud_auto_annotation(payload)})
+ return
+ if path == "/api/3d-pointcloud/auto-annotation-review-drafts":
+ self.send_json(HTTPStatus.CREATED, {"review": self.save_pointcloud_auto_annotation_review(payload)})
return
if path == "/api/3d-pointcloud/photo-reconstruction-runs":
self.send_json(HTTPStatus.ACCEPTED, {"job": self.create_photo_reconstruction_run(payload)})
@@ -1002,6 +1749,55 @@
def do_DELETE(self) -> None: # noqa: N802 - annotation revisions are explicitly user-removable
path = urlsplit(self.path).path
+ run_prefix = "/api/runs/"
+ if path.startswith(run_prefix):
+ try:
+ parts = path[len(run_prefix):].split("/")
+ if len(parts) != 2:
+ raise ApiError("Invalid result deletion endpoint.")
+ capability, run_id = (unquote(parts[0]), unquote(parts[1]))
+ self.send_json(HTTPStatus.OK, {"removed": delete_console_run(self.root, capability, run_id)})
+ except ApiError as exc:
+ self.send_json(HTTPStatus.BAD_REQUEST, {"error": str(exc)})
+ except Exception as exc: # pragma: no cover - defensive deletion boundary
+ self.log_error("console run deletion failed: %s", exc)
+ self.send_json(HTTPStatus.INTERNAL_SERVER_ERROR, {"error": "Result deletion failed. Check the console terminal for details."})
+ return
+ semantic_model_prefix = "/api/3d-pointcloud/semantic-models/"
+ if path.startswith(semantic_model_prefix):
+ try:
+ model_id = unquote(path[len(semantic_model_prefix):])
+ self.send_json(HTTPStatus.OK, {"removed": delete_pointcloud_semantic_model(self.root, model_id)})
+ except ApiError as exc:
+ self.send_json(HTTPStatus.BAD_REQUEST, {"error": str(exc)})
+ except Exception as exc: # pragma: no cover - defensive deletion boundary
+ self.log_error("semantic model deletion failed: %s", exc)
+ self.send_json(HTTPStatus.INTERNAL_SERVER_ERROR, {"error": "Semantic model deletion failed. Check the console terminal for details."})
+ return
+ source_prefix = "/api/3d-pointcloud/annotation-sources/"
+ if path.startswith(source_prefix):
+ try:
+ source_id = unquote(path[len(source_prefix):])
+ if not source_id or "/" in source_id or len(source_id) > 300:
+ raise ApiError("Invalid annotation source id.")
+ self.send_json(HTTPStatus.OK, {"removed": self.delete_pointcloud_annotation_source(source_id)})
+ except ApiError as exc:
+ self.send_json(HTTPStatus.BAD_REQUEST, {"error": str(exc)})
+ except Exception as exc: # pragma: no cover - defensive deletion boundary
+ self.log_error("annotation source deletion failed: %s", exc)
+ self.send_json(HTTPStatus.INTERNAL_SERVER_ERROR, {"error": "Annotation source deletion failed. Check the console terminal for details."})
+ return
+ class_prefix = "/api/3d-pointcloud/annotation-classes/"
+ if path.startswith(class_prefix):
+ try:
+ code_value = path[len(class_prefix):]
+ if not code_value.isdigit():
+ raise ApiError("Invalid annotation class code.")
+ deleted = self.delete_pointcloud_annotation_class(int(code_value))
+ self.send_json(HTTPStatus.OK, {"deletedCode": deleted})
+ except ApiError as exc:
+ self.send_json(HTTPStatus.BAD_REQUEST, {"error": str(exc)})
+ return
prefix = "/api/3d-pointcloud/annotations/"
if not path.startswith(prefix):
self.send_json(HTTPStatus.NOT_FOUND, {"error": "Unknown local API endpoint."})
@@ -1208,10 +2004,14 @@
def create_detection_run(self, payload: dict[str, Any]) -> dict[str, Any]:
uploads = payload.get("images")
+ requested_device = payload.get("device", "auto")
if not isinstance(uploads, list) or not uploads:
raise ApiError("Object-detection request must include at least one image.")
if len(uploads) > MAX_IMAGES_PER_RUN:
raise ApiError(f"A local run accepts at most {MAX_IMAGES_PER_RUN} images.")
+ if not isinstance(requested_device, str):
+ raise ApiError("Object-detection device must be auto, cpu, or cuda.")
+ execution = object_detection_execution_environment(self.root, requested_device)
decoded = [decode_upload(item, {".jpg", ".jpeg", ".png"}) for item in uploads]
if len({name.casefold() for name, _ in decoded}) != len(decoded):
raise ApiError("Uploaded image names must be unique within one run.")
@@ -1221,16 +2021,18 @@
for name, content in decoded:
(raw_root / name).write_bytes(content)
output = self.root / "shared" / "outputs" / "01-object-detection" / "runs" / run_id
- execution = object_detection_execution_environment(self.root)
with RUN_LOCK:
self.run_command([execution["python"], str(self.root / "capabilities" / "01-object-detection" / "run_detection.py"), "--input", str(raw_root), "--output", str(output), "--device", execution["device"]], 1200)
- if not (output / "run_metadata.json").is_file():
+ metadata_path = output / "run_metadata.json"
+ if not metadata_path.is_file():
raise ApiError("Detection script finished without the expected result metadata.")
+ record_execution_metadata(metadata_path, execution)
return next(item for item in detection_runs(self.root) if item["id"] == run_id)
def create_change_run(self, payload: dict[str, Any]) -> dict[str, Any]:
files = payload.get("files")
uploads = payload.get("uploads")
+ requested_device = payload.get("device", "auto")
staged: dict[str, tuple[str, Path]] = {}
if isinstance(uploads, dict):
staged["before"] = self.resolve_change_upload(uploads.get("before"), "before")
@@ -1255,6 +2057,9 @@
max_dimension = int(max_dimension_value)
if max_dimension != CHANGE_MAX_DIMENSION_AUTO and not CHANGE_MAX_DIMENSION_MIN <= max_dimension <= CHANGE_MAX_DIMENSION_MAX:
raise ApiError("Change-detection resolution must be 0 or between 512 and 4096.")
+ if not isinstance(requested_device, str):
+ raise ApiError("Change-detection device must be auto, cpu, or cuda.")
+ execution = change_detection_execution_environment(self.root, requested_device)
if staged:
before_name, after_name = staged["before"][0], staged["after"][0]
else:
@@ -1273,7 +2078,6 @@
after_path.write_bytes(decoded_after[1])
processed_root = self.root / "shared" / "data" / "processed" / "00-change-detection" / run_id
output = self.root / "shared" / "outputs" / "00-change-detection" / "runs" / run_id
- execution = change_detection_execution_environment(self.root)
with RUN_LOCK:
self.run_command(
[
@@ -1417,6 +2221,7 @@
def create_change_scan(self, payload: dict[str, Any]) -> dict[str, Any]:
uploads = payload.get("uploads")
+ requested_device = payload.get("device", "auto")
if not isinstance(uploads, dict):
raise ApiError("Parameter scan must contain staged before and after uploads.")
staged = {
@@ -1433,6 +2238,9 @@
max_dimension = int(max_dimension_value)
if max_dimension != CHANGE_MAX_DIMENSION_AUTO and not CHANGE_MAX_DIMENSION_MIN <= max_dimension <= CHANGE_MAX_DIMENSION_MAX:
raise ApiError("Change-detection resolution must be 0 or between 512 and 4096.")
+ if not isinstance(requested_device, str):
+ raise ApiError("Change-detection device must be auto, cpu, or cuda.")
+ execution = change_detection_execution_environment(self.root, requested_device)
run_id = make_run_id("scan")
raw_root = self.root / "shared" / "data" / "raw" / "00-change-detection" / "runs" / run_id
before_path = raw_root / "before" / staged["before"][0]
@@ -1453,10 +2261,16 @@
"minimumAreas": areas,
"processingMode": processing_mode,
"maxDimension": max_dimension,
+ "requestedDevice": execution["requestedDevice"],
+ "device": execution["device"],
+ "environment": execution["environment"],
+ "torchVersion": execution["torchVersion"],
+ "fallbackUsed": execution["fallbackUsed"],
+ "fallbackReason": execution["fallbackReason"],
}
thread = threading.Thread(
target=self._run_change_scan,
- args=(run_id, before_path, after_path, processed_root, inference_output, scan_output, thresholds, areas, processing_mode, max_dimension),
+ args=(run_id, before_path, after_path, processed_root, inference_output, scan_output, thresholds, areas, processing_mode, max_dimension, execution),
daemon=True,
name=f"change-scan-{run_id}",
)
@@ -1480,11 +2294,11 @@
areas: list[int],
processing_mode: str,
max_dimension: int,
+ execution: dict[str, Any],
) -> None:
- execution = change_detection_execution_environment(self.root)
python = execution["python"]
try:
- self._update_scan_job(run_id, status="running", phase="inference", device=execution["device"], environment=execution["environment"], torchVersion=execution["torchVersion"])
+ self._update_scan_job(run_id, status="running", phase="inference")
with RUN_LOCK:
self.run_command(
[
@@ -1502,6 +2316,7 @@
SCAN_JOB_TIMEOUT,
)
inference_metadata_path = inference_output / "run_metadata.json"
+ record_execution_metadata(inference_metadata_path, execution)
inference_metadata = load_json(inference_metadata_path)
inference_metadata["kind"] = "parameter-scan-inference"
inference_metadata["scan_job_id"] = run_id
@@ -1540,6 +2355,9 @@
"device": execution["device"],
"environment": execution["environment"],
"torch_version": execution["torchVersion"],
+ "requested_device": execution["requestedDevice"],
+ "fallback_used": execution["fallbackUsed"],
+ "fallback_reason": execution["fallbackReason"],
"raw_input_dir": relative_path(self.root, before_path.parent.parent),
"processed_input_dir": relative_path(self.root, processed_root),
}
@@ -1741,6 +2559,7 @@
metadata_path = output / "run_metadata.json"
if not metadata_path.is_file():
raise ApiError("3D point-cloud script finished without the expected result metadata.")
+ record_execution_metadata(metadata_path, execution)
metadata = load_json(metadata_path)
metadata["input_dir"] = relative_path(self.root, processed_root)
metadata["raw_input_dir"] = relative_path(self.root, raw_root)
@@ -1759,14 +2578,14 @@
source = next((item for item in pointcloud_annotation_sources(self.root) if item["id"] == source_id), None)
if not source:
raise ApiError("The selected generated annotation source is unavailable.")
- if len(labels) > MAX_ANNOTATION_LABELS:
- raise ApiError(f"An annotation revision accepts at most {MAX_ANNOTATION_LABELS} labelled points.")
+ active_classes = annotation_classes(self.root)
+ class_by_code = {item["code"]: item for item in active_classes}
compact: dict[int, int] = {}
for item in labels:
if not isinstance(item, list) or len(item) != 2 or not all(isinstance(value, int) for value in item):
raise ApiError("Each annotation label must be [pointIndex, classCode].")
index, code = item
- if index < 0 or index >= int(source["pointCount"]) or code not in POINTCLOUD_CLASS_CODES:
+ if index < 0 or index >= int(source["pointCount"]) or code not in class_by_code:
raise ApiError("Annotation contains an out-of-range point index or unsupported class code.")
compact[index] = code
if not compact:
@@ -1775,18 +2594,128 @@
location = self.root / "shared" / "outputs" / "05-3d-pointcloud" / "annotations" / annotation_id
location.mkdir(parents=True, exist_ok=False)
source_path = self.root / str(source["artifactRoot"]) / str(source["file"])
- class_counts = {str(code): sum(value == code for value in compact.values()) for code in sorted(POINTCLOUD_CLASS_CODES)}
+ class_counts = {str(code): sum(value == code for value in compact.values()) for code in sorted(class_by_code)}
document = {
"schema_version": 1, "id": annotation_id, "created_at": datetime.now(UTC).isoformat(),
"source_id": source_id, "source_path": str(source_path.resolve()), "source_sha256": source["sha256"],
"source_run_id": source["runId"], "point_count": int(source["pointCount"]),
"labels": [[index, code] for index, code in sorted(compact.items())], "class_counts": class_counts,
- "class_schema": {str(code): {"code": code} for code in sorted(POINTCLOUD_CLASS_CODES)},
+ "class_schema": {str(code): item for code, item in sorted(class_by_code.items())},
"provenance": "human_confirmed_point_labels_only",
}
path = location / "annotation.json"
path.write_text(json.dumps(document, ensure_ascii=False, indent=2), encoding="utf-8")
return {"id": annotation_id, "sourceId": source_id, "path": relative_path(self.root, path), "labelCount": len(compact), "classCounts": class_counts, "createdAt": document["created_at"]}
+
+ def create_pointcloud_annotation_class(self, payload: dict[str, Any]) -> dict[str, Any]:
+ key, label, color = payload.get("key"), payload.get("label"), payload.get("color")
+ if not isinstance(key, str) or not isinstance(label, str) or not isinstance(color, list):
+ raise ApiError("A class key, label, and RGB color are required.")
+ current = annotation_classes(self.root)
+ used_codes = {item["code"] for item in current}
+ code = next((value for value in range(17, 256) if value not in used_codes), None)
+ if code is None:
+ raise ApiError("All LAS-compatible annotation class codes are already in use.")
+ candidate = {"code": code, "key": key, "label": label, "color": color, "builtIn": False}
+ updated = validate_annotation_classes([*current, candidate], allow_builtin=True)
+ write_annotation_classes(self.root, updated)
+ return next(item for item in updated if item["code"] == code)
+
+ def delete_pointcloud_annotation_class(self, code: int) -> int:
+ current = annotation_classes(self.root)
+ item = next((value for value in current if value["code"] == code), None)
+ if not item:
+ raise ApiError("The annotation class is unavailable.")
+ if item["builtIn"]:
+ raise ApiError("Built-in annotation classes cannot be deleted.")
+ if code in annotation_class_codes_in_use(self.root):
+ raise ApiError("This annotation class is used by a saved annotation revision and cannot be deleted.")
+ write_annotation_classes(self.root, [value for value in current if value["code"] != code])
+ return code
+
+ def create_pointcloud_annotation_source_run(self, payload: dict[str, Any]) -> dict[str, Any]:
+ name, staged_path, expected_sha256 = self.resolve_pointcloud_upload(payload.get("pointCloud"))
+ run_id = make_run_id("annotation-source")
+ raw_root = self.root / "shared" / "data" / "raw" / "05-3d-pointcloud" / "annotation-source-runs" / run_id
+ processed_root = self.root / "shared" / "data" / "processed" / "05-3d-pointcloud" / "annotation-source-runs" / run_id
+ output = self.root / "shared" / "outputs" / "05-3d-pointcloud" / "runs" / run_id
+ raw_root.mkdir(parents=True, exist_ok=False)
+ processed_root.mkdir(parents=True, exist_ok=False)
+ raw_path = raw_root / name
+ shutil.copyfile(staged_path, raw_path)
+ source_sha256 = file_sha256(raw_path)
+ if expected_sha256 and source_sha256 != expected_sha256:
+ raise ApiError("Uploaded point-cloud checksum changed while staging.")
+ processed_path = processed_root / name
+ shutil.copyfile(raw_path, processed_path)
+ shutil.rmtree(staged_path.parent)
+ python = self.root / ".venvs" / POINTCLOUD_CPU_ENVIRONMENT / "Scripts" / "python.exe"
+ if not python.is_file():
+ raise ApiError("3D point-cloud CPU virtual environment is unavailable. Run the capability setup first.")
+ job_id = uuid4().hex
+ job = {"id": job_id, "runId": run_id, "inputName": name, "status": "queued", "stage": "queued", "createdAt": datetime.now(UTC).isoformat(), "sourceSha256": source_sha256}
+ with POINTCLOUD_ANNOTATION_SOURCE_JOBS_LOCK:
+ POINTCLOUD_ANNOTATION_SOURCE_JOBS[job_id] = job
+ thread = threading.Thread(target=execute_pointcloud_annotation_source_job, args=(self.root, job_id, processed_path, output, source_sha256, raw_path.stat().st_size), daemon=True, name=f"annotation-source-{job_id[:8]}")
+ thread.start()
+ return dict(job)
+
+ def delete_pointcloud_annotation_source(self, source_id: str) -> dict[str, Any]:
+ plan = pointcloud_annotation_source_deletion_plan(self.root, source_id)
+ source = next(item for item in pointcloud_annotation_sources(self.root) if item["id"] == source_id)
+ run_id = str(plan["runId"])
+ sibling_ids = {item["id"] for item in pointcloud_annotation_sources(self.root) if item["runId"] == run_id}
+ output_root = (self.root / "shared" / "outputs" / "05-3d-pointcloud").resolve()
+ annotation_root = output_root / "annotations"
+ annotations = [path.parent for path in annotation_root.glob("*/annotation.json") if load_json(path).get("source_id") in sibling_ids]
+ annotation_paths = {path / "annotation.json" for path in annotations}
+ training_root = output_root / "training-runs"
+ training = []
+ for metrics_path in training_root.glob("*/metrics.json"):
+ value = load_json(metrics_path).get("annotation")
+ try:
+ if isinstance(value, str) and Path(value).resolve() in annotation_paths:
+ training.append(metrics_path.parent)
+ except OSError:
+ continue
+ training_models = {path / "model.pt" for path in training}
+ inference_root = output_root / "model-inference-runs"
+ inference = []
+ for metadata_path in inference_root.glob("*/run_metadata.json"):
+ model = load_json(metadata_path).get("model")
+ value = model.get("path") if isinstance(model, dict) else None
+ try:
+ if isinstance(value, str) and Path(value).resolve() in training_models:
+ inference.append(metadata_path.parent)
+ except OSError:
+ continue
+ if active_pointcloud_source_dependencies(run_id, {path.name for path in annotations}, {path.name for path in training}):
+ raise ApiError("This data source has a queued or running dependent task. Wait for it to finish before removing the full data chain.")
+ artifact = (self.root / str(source["artifactRoot"])).resolve()
+ raw_root = self.root / "shared" / "data" / "raw" / "05-3d-pointcloud"
+ processed_root = self.root / "shared" / "data" / "processed" / "05-3d-pointcloud"
+ raw = [path for path in (raw_root / "annotation-source-runs" / run_id, raw_root / "runs" / run_id) if path.is_dir()]
+ processed = [path for path in (processed_root / "annotation-source-runs" / run_id, processed_root / "runs" / run_id) if path.is_dir()]
+ targets = [*inference, *training, *annotations, artifact, *raw, *processed]
+ unique: list[tuple[Path, Path]] = []
+ seen: set[Path] = set()
+ for target in targets:
+ resolved = target.resolve()
+ if resolved in seen:
+ continue
+ seen.add(resolved)
+ if target in raw:
+ allowed = raw_root
+ elif target in processed:
+ allowed = processed_root
+ else:
+ allowed = output_root
+ assert_removable_pointcloud_directory(self.root, target, allowed)
+ unique.append((target, allowed))
+ # Dependents first; every target was resolved against a fixed local root.
+ for target, _ in unique:
+ shutil.rmtree(target)
+ return {"sourceId": source_id, "runId": run_id, "removed": {"outputDirectories": int(plan["outputDirectories"]), "rawDirectories": int(plan["rawDirectories"]), "processedDirectories": int(plan["processedDirectories"]), "annotationRevisions": len(annotations), "trainingRuns": len(training), "inferenceRuns": len(inference), "siblingSources": int(plan["siblingSources"])}, "preservedExternalInputs": bool(plan["preservesExternalInputs"])}
def create_pointcloud_training_run(self, payload: dict[str, Any]) -> dict[str, Any]:
annotation_id = payload.get("annotationId")
@@ -1799,19 +2728,27 @@
record = load_json(annotation)
if record.get("schema_version") != 1:
raise ApiError("The selected annotation revision is unavailable.")
+ source_id = record.get("source_id")
+ source = next((item for item in pointcloud_annotation_sources(self.root) if item["id"] == source_id), None)
+ if not source:
+ raise ApiError("The annotation source is unavailable or its verification contract no longer passes.")
+ if source.get("sourceHasRgb") is False:
+ raise ApiError("This annotation source has no readable RGB values. It can be reviewed visually but cannot train the current RGB semantic model.")
+ trainer = "multiview_local_attention_baseline" if str(source.get("sourceKind", "")).startswith("多视角照片特征融合") else "rgb_xyz_baseline"
execution = pointcloud_execution_environment(self.root, device)
job_id = uuid4().hex
- output = self.root / "shared" / "outputs" / "05-3d-pointcloud" / "training-runs" / make_run_id("semantic-model")
- job = {"id": job_id, "annotationId": annotation_id, "status": "queued", "stage": "queued", "requestedDevice": device, "device": execution["device"], "environment": execution["environment"], "torchVersion": execution["torchVersion"], "createdAt": datetime.now(UTC).isoformat()}
+ output = self.root / "shared" / "outputs" / "05-3d-pointcloud" / "training-runs" / make_run_id("multiview-point-attention" if trainer == "multiview_local_attention_baseline" else "semantic-model")
+ job = {"id": job_id, "annotationId": annotation_id, "trainer": trainer, "status": "queued", "stage": "queued", "requestedDevice": execution["requestedDevice"], "device": execution["device"], "environment": execution["environment"], "torchVersion": execution["torchVersion"], "fallbackUsed": execution["fallbackUsed"], "fallbackReason": execution["fallbackReason"], "createdAt": datetime.now(UTC).isoformat()}
with POINTCLOUD_TRAINING_JOBS_LOCK:
POINTCLOUD_TRAINING_JOBS[job_id] = job
- thread = threading.Thread(target=execute_pointcloud_training_job, args=(self.root, job_id, annotation, output, execution), daemon=True, name=f"pointcloud-training-{job_id[:8]}")
+ thread = threading.Thread(target=execute_pointcloud_training_job, args=(self.root, job_id, annotation, output, execution, trainer), daemon=True, name=f"pointcloud-training-{job_id[:8]}")
thread.start()
return dict(job)
def create_pointcloud_model_inference_run(self, payload: dict[str, Any]) -> dict[str, Any]:
model_id = payload.get("modelId")
upload = payload.get("pointCloud")
+ device = payload.get("device", "auto")
if not isinstance(model_id, str) or SAFE_FILE_NAME.search(model_id) or len(model_id) > 120:
raise ApiError("Invalid trained model id.")
model_record = next((item for item in pointcloud_semantic_models(self.root) if item["id"] == model_id), None)
@@ -1822,7 +2759,9 @@
if Path(name).suffix.lower() not in suffixes:
raise ApiError("Model inference requires a PLY, PCD, XYZ, LAS, or LAZ point cloud.")
validate_pointcloud_model_input(staged_path)
- execution = pointcloud_execution_environment(self.root)
+ if not isinstance(device, str):
+ raise ApiError("Point-cloud device must be auto, cpu, or cuda.")
+ execution = pointcloud_execution_environment(self.root, device)
run_id = make_run_id("semantic-inference")
raw_root = self.root / "shared" / "data" / "raw" / "05-3d-pointcloud" / "model-inference-runs" / run_id
processed_root = self.root / "shared" / "data" / "processed" / "05-3d-pointcloud" / "model-inference-runs" / run_id
@@ -1845,12 +2784,149 @@
except ValueError as exc:
raise ApiError("Selected model is outside the allowed training output directory.") from exc
job_id = uuid4().hex
- job = {"id": job_id, "runId": run_id, "modelId": model_id, "inputName": name, "status": "queued", "stage": "queued", "requestedDevice": "auto", "device": execution["device"], "environment": execution["environment"], "torchVersion": execution["torchVersion"], "createdAt": datetime.now(UTC).isoformat(), "sourceSha256": actual_sha256, "rawInput": relative_path(self.root, raw_path), "processedInput": relative_path(self.root, processed_path)}
+ job = {"id": job_id, "runId": run_id, "modelId": model_id, "inputName": name, "status": "queued", "stage": "queued", "requestedDevice": execution["requestedDevice"], "device": execution["device"], "environment": execution["environment"], "torchVersion": execution["torchVersion"], "fallbackUsed": execution["fallbackUsed"], "fallbackReason": execution["fallbackReason"], "createdAt": datetime.now(UTC).isoformat(), "sourceSha256": actual_sha256, "rawInput": relative_path(self.root, raw_path), "processedInput": relative_path(self.root, processed_path)}
with POINTCLOUD_INFERENCE_JOBS_LOCK:
POINTCLOUD_INFERENCE_JOBS[job_id] = job
thread = threading.Thread(target=execute_pointcloud_inference_job, args=(self.root, job_id, model_path, processed_path, output, execution), daemon=True, name=f"pointcloud-inference-{job_id[:8]}")
thread.start()
return dict(job)
+
+ def create_pointcloud_auto_annotation_run(self, payload: dict[str, Any]) -> dict[str, Any]:
+ model_id, source_id, device, confidence = payload.get("modelId"), payload.get("sourceId"), payload.get("device", "auto"), payload.get("candidateConfidence", 0.95)
+ if not isinstance(model_id, str) or SAFE_FILE_NAME.search(model_id) or len(model_id) > 120:
+ raise ApiError("Invalid trained model id.")
+ if not isinstance(source_id, str):
+ raise ApiError("Automatic annotation needs a selected local annotation source.")
+ if not isinstance(device, str) or device not in COMPUTE_DEVICES:
+ raise ApiError("Point-cloud device must be auto, cpu, or cuda.")
+ if not isinstance(confidence, (int, float)) or isinstance(confidence, bool) or not 0.5 <= float(confidence) < 1.0:
+ raise ApiError("Candidate confidence must be between 0.5 and 1.0.")
+ model_record = next((item for item in pointcloud_semantic_models(self.root) if item["id"] == model_id), None)
+ source = next((item for item in pointcloud_annotation_sources(self.root) if item["id"] == source_id), None)
+ if not model_record or not source:
+ raise ApiError("The selected model or annotation source is unavailable.")
+ if source.get("sourceHasRgb") is False:
+ raise ApiError("Automatic annotation needs observed RGB point features.")
+ model_path = (self.root / str(model_record["model"])).resolve()
+ source_path = (self.root / str(source["artifactRoot"]) / str(source["file"])).resolve()
+ training_root = (self.root / "shared" / "outputs" / "05-3d-pointcloud" / "training-runs").resolve()
+ output_root = (self.root / "shared" / "outputs" / "05-3d-pointcloud").resolve()
+ try:
+ model_path.relative_to(training_root)
+ source_path.relative_to(output_root)
+ except ValueError as exc:
+ raise ApiError("Selected automatic-annotation inputs are outside local workbench outputs.") from exc
+ if not model_path.is_file() or not source_path.is_file() or file_sha256(source_path) != source["sha256"]:
+ raise ApiError("Selected automatic-annotation input no longer passes its verification contract.")
+ execution = pointcloud_execution_environment(self.root, device)
+ run_id = make_run_id("auto-annotation")
+ output = output_root / "auto-annotation-runs" / run_id
+ job_id = uuid4().hex
+ job = {"id": job_id, "runId": run_id, "modelId": model_id, "sourceId": source_id, "inputName": str(source["file"]), "candidateConfidence": float(confidence), "status": "queued", "stage": "queued", "requestedDevice": execution["requestedDevice"], "device": execution["device"], "environment": execution["environment"], "torchVersion": execution["torchVersion"], "fallbackUsed": execution["fallbackUsed"], "fallbackReason": execution["fallbackReason"], "createdAt": datetime.now(UTC).isoformat(), "sourceSha256": source["sha256"]}
+ with POINTCLOUD_INFERENCE_JOBS_LOCK:
+ POINTCLOUD_INFERENCE_JOBS[job_id] = job
+ thread = threading.Thread(target=execute_pointcloud_inference_job, args=(self.root, job_id, model_path, source_path, output, execution, source_id, float(confidence)), daemon=True, name=f"auto-annotation-{job_id[:8]}")
+ thread.start()
+ return dict(job)
+
+ def save_pointcloud_auto_annotation_review(self, payload: dict[str, Any]) -> dict[str, Any]:
+ run_id, source_id, corrections = payload.get("runId"), payload.get("sourceId"), payload.get("corrections")
+ if not isinstance(run_id, str) or SAFE_FILE_NAME.search(run_id) or len(run_id) > 120:
+ raise ApiError("Invalid automatic-annotation run id.")
+ if not isinstance(source_id, str) or not isinstance(corrections, list):
+ raise ApiError("Automatic candidate review needs a source id and corrections array.")
+ source = pointcloud_annotation_source(self.root, source_id)
+ output_root = (self.root / "shared" / "outputs" / "05-3d-pointcloud" / "auto-annotation-runs").resolve()
+ artifact = (output_root / run_id).resolve()
+ candidate_path = artifact / "automatic-annotation-candidates.json"
+ try:
+ artifact.relative_to(output_root)
+ except ValueError as exc:
+ raise ApiError("Automatic candidate review path is outside local workbench outputs.") from exc
+ candidate = load_json(candidate_path)
+ if candidate.get("schema_version") != 1 or candidate.get("source_id") != source_id or candidate.get("source_sha256") != source["sha256"]:
+ raise ApiError("Automatic candidate provenance no longer matches the selected source.")
+ active_codes = {item["code"] for item in annotation_classes(self.root)}
+ compact: dict[int, int] = {}
+ for item in corrections:
+ if not isinstance(item, list) or len(item) != 2 or not all(isinstance(value, int) for value in item):
+ raise ApiError("Each candidate correction must be [pointIndex, classCodeOrZero].")
+ index, code = item
+ if index < 0 or index >= int(source["pointCount"]) or (code != 0 and code not in active_codes):
+ raise ApiError("Candidate review contains an out-of-range point index or unsupported class code.")
+ compact[index] = code
+ review = {
+ "schema_version": 1,
+ "run_id": run_id,
+ "source_id": source_id,
+ "source_sha256": source["sha256"],
+ "candidate_confidence": candidate.get("candidate_confidence"),
+ "corrections": [[index, code] for index, code in sorted(compact.items())],
+ "saved_at": datetime.now(UTC).isoformat(),
+ }
+ (artifact / "review-corrections.json").write_text(json.dumps(review, ensure_ascii=False, indent=2), encoding="utf-8")
+ return {"runId": run_id, "sourceId": source_id, "correctionCount": len(compact), "savedAt": review["saved_at"]}
+
+ def accept_pointcloud_auto_annotation(self, payload: dict[str, Any]) -> dict[str, Any]:
+ run_id, source_id, base_annotation_id = payload.get("runId"), payload.get("sourceId"), payload.get("baseAnnotationId")
+ if not isinstance(run_id, str) or SAFE_FILE_NAME.search(run_id) or len(run_id) > 120:
+ raise ApiError("Invalid automatic-annotation run id.")
+ if not isinstance(source_id, str):
+ raise ApiError("Automatic-annotation acceptance needs a source id.")
+ source = pointcloud_annotation_source(self.root, source_id)
+ output_root = (self.root / "shared" / "outputs" / "05-3d-pointcloud" / "auto-annotation-runs").resolve()
+ candidate_path = (output_root / run_id / "automatic-annotation-candidates.json").resolve()
+ try:
+ candidate_path.relative_to(output_root)
+ except ValueError as exc:
+ raise ApiError("Automatic candidate path is outside local workbench outputs.") from exc
+ candidate = load_json(candidate_path)
+ if candidate.get("schema_version") != 1 or candidate.get("source_id") != source_id or candidate.get("source_sha256") != source["sha256"]:
+ raise ApiError("Automatic candidate provenance no longer matches the selected source.")
+ raw_labels = candidate.get("labels")
+ if not isinstance(raw_labels, list):
+ raise ApiError("Automatic candidate labels are unavailable.")
+ merged: dict[int, int] = {}
+ for item in raw_labels:
+ if not isinstance(item, list) or len(item) != 3 or not isinstance(item[0], int) or not isinstance(item[1], int):
+ raise ApiError("Automatic candidate labels are invalid.")
+ merged[item[0]] = item[1]
+ review_path = candidate_path.parent / "review-corrections.json"
+ review_correction_count = 0
+ if review_path.is_file():
+ review = load_json(review_path)
+ if review.get("schema_version") != 1 or review.get("run_id") != run_id or review.get("source_id") != source_id or review.get("source_sha256") != source["sha256"]:
+ raise ApiError("Candidate review draft provenance no longer matches the selected source.")
+ corrections = review.get("corrections")
+ if not isinstance(corrections, list):
+ raise ApiError("Candidate review corrections are invalid.")
+ valid_codes = {item["code"] for item in annotation_classes(self.root)}
+ for item in corrections:
+ if not isinstance(item, list) or len(item) != 2 or not all(isinstance(value, int) for value in item):
+ raise ApiError("Candidate review corrections are invalid.")
+ index, code = item
+ if index < 0 or index >= int(source["pointCount"]) or (code != 0 and code not in valid_codes):
+ raise ApiError("Candidate review contains an out-of-range point index or unsupported class code.")
+ if code == 0:
+ merged.pop(index, None)
+ else:
+ merged[index] = code
+ review_correction_count += 1
+ if isinstance(base_annotation_id, str) and base_annotation_id:
+ base_path = self.root / "shared" / "outputs" / "05-3d-pointcloud" / "annotations" / base_annotation_id / "annotation.json"
+ base = load_json(base_path)
+ if base.get("source_id") != source_id:
+ raise ApiError("The base human annotation belongs to another source.")
+ for item in base.get("labels", []):
+ if isinstance(item, list) and len(item) == 2 and all(isinstance(value, int) for value in item):
+ merged[item[0]] = item[1]
+ annotation = self.create_pointcloud_annotation({"sourceId": source_id, "labels": [[index, code] for index, code in merged.items()]})
+ annotation_path = self.root / str(annotation["path"])
+ document = load_json(annotation_path)
+ document["provenance"] = "user_confirmed_high_confidence_model_candidates_with_human_labels_preferred"
+ document["automatic_annotation"] = {"run_id": run_id, "candidate_confidence": candidate.get("candidate_confidence"), "candidate_count": candidate.get("candidate_count"), "review_correction_count": review_correction_count, "base_annotation_id": base_annotation_id or None}
+ annotation_path.write_text(json.dumps(document, ensure_ascii=False, indent=2), encoding="utf-8")
+ return annotation
def create_photo_reconstruction_run(self, payload: dict[str, Any]) -> dict[str, Any]:
uploads = payload.get("photos")
@@ -1891,7 +2967,9 @@
for _, staged_path, _ in photos:
shutil.rmtree(staged_path.parent)
- job = {"id": job_id, "runId": run_id, "status": "queued", "stage": "queued", "createdAt": datetime.now(UTC).isoformat(), "inputImages": len(photos), "usePositionPriors": use_position_priors}
+ progress_path = processed_root / "photo_reconstruction_progress.json"
+ progress_path.write_text(json.dumps({"percent": 0, "stage": "queued", "message": "照片已保存，正在等待 CPU 重建资源。", "inputImages": len(photos), "updatedAt": datetime.now(UTC).isoformat(), "estimate": True}, ensure_ascii=False), encoding="utf-8")
+ job = {"id": job_id, "runId": run_id, "status": "queued", "stage": "queued", "createdAt": datetime.now(UTC).isoformat(), "inputImages": len(photos), "usePositionPriors": use_position_priors, "progressPath": str(progress_path)}
with PHOTO_RECONSTRUCTION_JOBS_LOCK:
PHOTO_RECONSTRUCTION_JOBS[job_id] = job
thread = threading.Thread(
@@ -1901,7 +2979,7 @@
name=f"photo-reconstruction-{run_id}",
)
thread.start()
- return dict(job)
+ return photo_reconstruction_job(job_id) or {key: value for key, value in job.items() if key != "progressPath"}
def create_risk_rule_run(self, payload: dict[str, Any]) -> dict[str, Any]:
files = payload.get("files")
--
Gitblit v1.9.3