"""Serve the local GeoAI Workbench console and its narrow local-run APIs."""
|
|
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
|
from pathlib import Path, PurePosixPath
|
from typing import Any
|
from urllib.parse import parse_qs, unquote, urlsplit
|
from uuid import uuid4
|
|
|
DEFAULT_HOST = "127.0.0.1"
|
DEFAULT_PORT = 6173
|
# Uploads are sent as Base64 JSON. Keep the request limit above two 1 GiB
|
# files after encoding while retaining a per-file bound for local experiments.
|
MAX_REQUEST_BYTES = 3072 * 1024 * 1024
|
MAX_FILE_BYTES = 1024 * 1024 * 1024
|
MAX_IMAGES_PER_RUN = 12
|
MAX_SEGMENTATION_IMAGES_PER_RUN = 6
|
MAX_MEASUREMENT_RASTERS_PER_RUN = 4
|
MAX_POINTCLOUDS_PER_RUN = 2
|
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
|
CHANGE_THRESHOLD_MIN = 0.01
|
CHANGE_THRESHOLD_MAX = 0.99
|
CHANGE_MAX_DIMENSION_DEFAULT = 1024
|
CHANGE_MAX_DIMENSION_AUTO = 0
|
CHANGE_MAX_DIMENSION_MIN = 512
|
CHANGE_MAX_DIMENSION_MAX = 4096
|
CHANGE_PROCESSING_MODE_DEFAULT = "auto"
|
CHANGE_PROCESSING_MODES = {"auto", "image", "geotiff"}
|
SCAN_DEFAULT_THRESHOLDS = [0.3, 0.4, 0.5]
|
SCAN_DEFAULT_AREAS = [64, 256, 686]
|
SCAN_MAX_THRESHOLDS = 6
|
SCAN_MAX_AREAS = 6
|
SCAN_MAX_COMBINATIONS = 24
|
SCAN_JOB_TIMEOUT = 1800
|
ALLOWED_PATH_PREFIXES = (
|
"apps/workbench-console",
|
"shared/outputs",
|
"shared/data/raw/00-change-detection",
|
"shared/data/raw/01-object-detection",
|
"shared/data/raw/02-semantic-mapping",
|
"shared/data/raw/09-anomaly-detection",
|
"shared/data/raw/05-3d-pointcloud",
|
)
|
SAFE_FILE_NAME = re.compile(r"[^\w.-]+", re.UNICODE)
|
SAFE_UPLOAD_ID = re.compile(r"^[0-9a-f]{32}$")
|
SAFE_SCAN_ID = re.compile(r"^[A-Za-z0-9._-]{1,100}$")
|
SAFE_SCAN_RESULT_ID = re.compile(r"^threshold-\d+(?:\.\d+)?_area-\d+$")
|
RUN_LOCK = threading.Lock()
|
SCAN_JOBS: dict[str, dict[str, Any]] = {}
|
SCAN_JOBS_LOCK = threading.Lock()
|
ANOMALY_JOB_LOCK = threading.Lock()
|
ANOMALY_JOBS: dict[str, dict[str, Any]] = {}
|
PHOTO_RECONSTRUCTION_JOBS: dict[str, dict[str, Any]] = {}
|
PHOTO_RECONSTRUCTION_JOBS_LOCK = threading.Lock()
|
POINTCLOUD_TRAINING_JOBS: dict[str, dict[str, Any]] = {}
|
POINTCLOUD_TRAINING_JOBS_LOCK = threading.Lock()
|
POINTCLOUD_INFERENCE_JOBS: dict[str, dict[str, Any]] = {}
|
POINTCLOUD_INFERENCE_JOBS_LOCK = threading.Lock()
|
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):
|
"""A request error that can be shown to the local console user."""
|
|
|
def safe_file_name(value: str, expected_suffixes: set[str]) -> str:
|
name = Path(value).name
|
suffix = Path(name).suffix.lower()
|
if suffix not in expected_suffixes:
|
raise ApiError(f"Unsupported file type: {suffix or '(none)'}.")
|
stem = SAFE_FILE_NAME.sub("_", Path(name).stem).strip("._") or "upload"
|
return f"{stem[:80]}{suffix}"
|
|
|
def decode_upload(payload: dict[str, Any], expected_suffixes: set[str]) -> tuple[str, bytes]:
|
if not isinstance(payload, dict) or not isinstance(payload.get("name"), str) or not isinstance(payload.get("content"), str):
|
raise ApiError("Each uploaded file must include name and Base64 content.")
|
name = safe_file_name(payload["name"], expected_suffixes)
|
try:
|
content = base64.b64decode(payload["content"], validate=True)
|
except (binascii.Error, ValueError) as exc:
|
raise ApiError(f"Invalid Base64 file content for {name}.") from exc
|
if not content:
|
raise ApiError(f"Uploaded file is empty: {name}.")
|
if len(content) > MAX_FILE_BYTES:
|
raise ApiError(f"Uploaded file exceeds {MAX_FILE_BYTES // (1024 * 1024)} MB: {name}.")
|
return name, content
|
|
|
def make_run_id(prefix: str) -> str:
|
return f"{prefix}-{datetime.now(UTC):%Y%m%d-%H%M%S}-{uuid4().hex[:6]}"
|
|
|
def relative_path(root: Path, path: Path) -> str:
|
# 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]:
|
try:
|
payload = json.loads(path.read_text(encoding="utf-8"))
|
except (OSError, json.JSONDecodeError):
|
return {}
|
return payload if isinstance(payload, dict) else {}
|
|
|
def file_sha256(path: Path) -> str:
|
digest = hashlib.sha256()
|
with path.open("rb") as stream:
|
for chunk in iter(lambda: stream.read(8 * 1024 * 1024), b""):
|
digest.update(chunk)
|
return digest.hexdigest()
|
|
|
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}
|
|
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__}))"],
|
cwd=root,
|
capture_output=True,
|
text=True,
|
timeout=20,
|
check=False,
|
)
|
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):
|
probe_reason = "CUDA probe could not return a valid result."
|
if requested_device == "cuda":
|
raise ApiError(f"CUDA was requested for {label}, but it is unavailable: {probe_reason}")
|
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": "auto", "fallbackUsed": True, "fallbackReason": probe_reason}
|
|
|
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 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]]:
|
output_root = root / "shared" / "outputs" / "15-trajectory-analysis"
|
records: list[dict[str, Any]] = []
|
for metadata_path in output_root.rglob("run_metadata.json"):
|
artifact = metadata_path.parent
|
if not (artifact / "trajectory_summary.csv").is_file() or not (artifact / "events.json").is_file():
|
continue
|
metadata = load_json(metadata_path)
|
case_id = str(metadata.get("case_id") or artifact.name)
|
is_real = case_id == "tian-dun-flight-19578"
|
records.append(
|
{
|
"id": case_id,
|
"label": "田墩实飞" if is_real else case_id,
|
"note": "区域相交是来源数据的空间结果，不是违规结论。" if is_real else "本地实验运行结果，可继续查看结构化输出。",
|
"artifactRoot": relative_path(root, artifact),
|
"showSpatialContext": (artifact / "zones.geojson").is_file() and (artifact / "reference_routes.geojson").is_file(),
|
"showFlyableZones": (artifact / "flyable_zones.geojson").is_file(),
|
"createdAt": str(metadata.get("created_at") or ""),
|
}
|
)
|
return sorted(records, key=lambda item: (item["createdAt"], item["id"]), reverse=True)
|
|
|
def detection_runs(root: Path) -> list[dict[str, Any]]:
|
output_root = root / "shared" / "outputs" / "01-object-detection"
|
records: list[dict[str, Any]] = []
|
for metadata_path in output_root.rglob("run_metadata.json"):
|
artifact = metadata_path.parent
|
if not (artifact / "detections.json").is_file():
|
continue
|
metadata = load_json(metadata_path)
|
input_value = str(metadata.get("input_dir") or "")
|
input_dir = Path(input_value) if input_value else root / "shared" / "data" / "raw" / "01-object-detection"
|
try:
|
input_root = relative_path(root, input_dir.resolve())
|
except ValueError:
|
continue
|
run_id = artifact.name if artifact != output_root else "baseline"
|
device = str(metadata.get("device") or "cpu").lower()
|
execution = "GPU" if device.startswith("cuda") else "CPU"
|
records.append(
|
{
|
"id": run_id,
|
"label": "既有基线结果" if run_id == "baseline" else run_id,
|
"note": f"{execution} 基线：人员与常见车辆；树木不在当前模型有效类别内。",
|
"artifactRoot": relative_path(root, artifact),
|
"inputRoot": input_root,
|
"createdAt": str(metadata.get("created_at") or ""),
|
}
|
)
|
return sorted(records, key=lambda item: (item["createdAt"], item["id"]), reverse=True)
|
|
|
def change_runs(root: Path) -> list[dict[str, Any]]:
|
output_root = root / "shared" / "outputs" / "00-change-detection"
|
records: list[dict[str, Any]] = []
|
for metadata_path in output_root.rglob("run_metadata.json"):
|
artifact = metadata_path.parent
|
metadata = load_json(metadata_path)
|
artifacts = metadata.get("artifacts")
|
if metadata.get("capability") != "00-change-detection" or metadata.get("schema_version") != 1 or metadata.get("kind") == "parameter-scan-inference" or not isinstance(artifacts, dict):
|
continue
|
if not (artifact / str(artifacts.get("overlay") or "")).is_file() or not (artifact / str(artifacts.get("vector") or "")).is_file():
|
continue
|
raw_root_value = str(metadata.get("raw_input_dir") or "shared/data/raw/00-change-detection/validation-20260817")
|
raw_root = root / Path(raw_root_value)
|
input_files = metadata.get("input_files")
|
if not isinstance(input_files, list) or len(input_files) != 2:
|
continue
|
before_value = str(metadata.get("raw_before") or (Path(raw_root_value) / str(input_files[0])).as_posix())
|
after_value = str(metadata.get("raw_after") or (Path(raw_root_value) / str(input_files[1])).as_posix())
|
try:
|
before_path = (root / before_value).resolve()
|
after_path = (root / after_value).resolve()
|
allowed_raw = (root / "shared" / "data" / "raw" / "00-change-detection").resolve()
|
before_path.relative_to(allowed_raw)
|
after_path.relative_to(allowed_raw)
|
except ValueError:
|
continue
|
if not before_path.is_file() or not after_path.is_file():
|
continue
|
registered_before_name = str(artifacts.get("before_processed_preview") or "")
|
registered_after_name = str(artifacts.get("after_registered_preview") or "")
|
registered_before_path = artifact / registered_before_name if registered_before_name else None
|
registered_after_path = artifact / registered_after_name if registered_after_name else None
|
run_id = artifact.name
|
device = str(metadata.get("device") or "cpu").lower()
|
execution = "GPU" if device.startswith("cuda") else "CPU"
|
record = {
|
"id": run_id,
|
"label": run_id,
|
"note": f"ChangeStar {execution} 变化栅格与 GeoAI 像素坐标图斑；结果需人工复核。",
|
"artifactRoot": relative_path(root, artifact),
|
"beforeImage": relative_path(root, before_path),
|
"afterImage": relative_path(root, after_path),
|
"createdAt": str(metadata.get("created_at") or ""),
|
}
|
if registered_before_path and registered_after_path and registered_before_path.is_file() and registered_after_path.is_file():
|
record["registeredBeforeImage"] = relative_path(root, registered_before_path)
|
record["registeredAfterImage"] = relative_path(root, registered_after_path)
|
# Existing console consumers use afterImage as the vector-overlay
|
# base. Point it at the registered grid so polygons and highlights
|
# share the same pixel coordinates as the model output.
|
record["rawAfterImage"] = record["afterImage"]
|
record["afterImage"] = record["registeredAfterImage"]
|
records.append(record)
|
return sorted(records, key=lambda item: (item["createdAt"], item["id"]), reverse=True)
|
|
|
def change_parameter_scans(root: Path) -> list[dict[str, Any]]:
|
"""Discover read-only parameter scans produced from an existing change run."""
|
output_root = root / "shared" / "outputs" / "00-change-detection"
|
records: list[dict[str, Any]] = []
|
for summary_path in output_root.rglob("scan_summary.json"):
|
scan_root = summary_path.parent
|
summary = load_json(summary_path)
|
results: list[dict[str, Any]] = []
|
for item in summary.get("results", []):
|
if not isinstance(item, dict) or not isinstance(item.get("directory"), str):
|
continue
|
directory = scan_root / item["directory"]
|
overlay = directory / "overlay_preview.jpg"
|
mask = directory / "change_mask.tif"
|
regions = directory / "regions.json"
|
if not overlay.is_file() or not mask.is_file() or not regions.is_file():
|
continue
|
results.append(
|
{
|
"id": item["directory"],
|
"label": f"T={float(item.get('threshold', 0.5)):.2f} / 面积={int(item.get('minimum_area_pixels', 0))} px",
|
"threshold": item.get("threshold"),
|
"minimumAreaPixels": item.get("minimum_area_pixels"),
|
"cleanedComponents": item.get("cleaned_components"),
|
"changedPixels": item.get("changed_pixels"),
|
"changedPixelRatio": item.get("changed_pixel_ratio"),
|
"vectorFeatureCount": item.get("vector_feature_count"),
|
"fullVectorFeatureCount": (load_json(directory / "full_result.json").get("full_vector_feature_count") if (directory / "full_result.json").is_file() else None),
|
"rectangleFeatureCount": (
|
load_json(directory / "full_result.json").get("rectangle_vector_feature_count")
|
if (directory / "full_result.json").is_file() and load_json(directory / "full_result.json").get("rectangle_vector_feature_count") is not None
|
else len(load_json(directory / "changes_rectangles.geojson").get("features", [])) if (directory / "changes_rectangles.geojson").is_file() else None
|
),
|
"overlay": relative_path(root, overlay),
|
"mask": relative_path(root, mask),
|
"regions": relative_path(root, regions),
|
"vector": (relative_path(root, directory / "changes.geojson") if (directory / "changes.geojson").is_file() else None),
|
"rectangleVector": (relative_path(root, directory / "changes_rectangles.geojson") if (directory / "changes_rectangles.geojson").is_file() else None),
|
"rectangleVectorWgs84": (relative_path(root, directory / "changes_rectangles_wgs84.geojson") if (directory / "changes_rectangles_wgs84.geojson").is_file() else None),
|
}
|
)
|
# A failed job may have been repaired or materialized later. Keep it
|
# discoverable whenever at least one complete candidate exists; only
|
# hide scans that still have no usable result.
|
if not results:
|
continue
|
scan_metadata = load_json(scan_root / "scan_metadata.json")
|
contact_sheet = scan_root / "parameter_scan_contact_sheet.jpg"
|
records.append(
|
{
|
"id": scan_root.name,
|
"label": f"低成本参数扫描 · {scan_root.name}",
|
"note": f"复用已有变化概率结果，不重新运行 ChangeStar；源运行：{summary.get('source_run', '未知')}",
|
"artifactRoot": relative_path(root, scan_root),
|
"sourceRun": summary.get("source_run"),
|
"userSubmitted": bool(scan_metadata),
|
"contactSheet": relative_path(root, contact_sheet) if contact_sheet.is_file() else None,
|
"results": results,
|
}
|
)
|
return sorted(records, key=lambda item: item["id"], reverse=True)
|
|
|
def semantic_runs(root: Path) -> list[dict[str, Any]]:
|
output_root = root / "shared" / "outputs" / "02-semantic-mapping"
|
records: list[dict[str, Any]] = []
|
for metadata_path in output_root.rglob("run_metadata.json"):
|
artifact = metadata_path.parent
|
metadata = load_json(metadata_path)
|
if metadata.get("capability") != "02-semantic-mapping" or not isinstance(metadata.get("images"), list):
|
continue
|
run_id = artifact.name if artifact != output_root else "baseline"
|
records.append(
|
{
|
"id": run_id,
|
"label": "语义分割基线" if run_id == "baseline" else run_id,
|
"note": f"{metadata.get('task_name') or '通用颜色规则基线'}，输出栅格掩膜与 GeoAI 矢量结果。",
|
"artifactRoot": relative_path(root, artifact),
|
"inputRoot": str(metadata.get("input_dir") or "shared/data/processed/02-semantic-mapping"),
|
"rawInputRoot": str(metadata.get("raw_input_dir") or "shared/data/raw/02-semantic-mapping"),
|
"createdAt": str(metadata.get("created_at") or ""),
|
}
|
)
|
return sorted(records, key=lambda item: (item["createdAt"], item["id"]), reverse=True)
|
|
|
def measurement_runs(root: Path) -> list[dict[str, Any]]:
|
output_root = root / "shared" / "outputs" / "04-spatial-measurement"
|
records: list[dict[str, Any]] = []
|
for metadata_path in output_root.rglob("run_metadata.json"):
|
artifact = metadata_path.parent
|
metadata = load_json(metadata_path)
|
if metadata.get("capability") != "04-spatial-measurement" or not isinstance(metadata.get("images"), list):
|
continue
|
run_id = artifact.name
|
records.append(
|
{
|
"id": run_id,
|
"label": run_id,
|
"note": "GeoAI 栅格转矢量后进行对象计数、面积和周长测量。",
|
"artifactRoot": relative_path(root, artifact),
|
"createdAt": str(metadata.get("created_at") or ""),
|
}
|
)
|
return sorted(records, key=lambda item: (item["createdAt"], item["id"]), reverse=True)
|
|
|
def pointcloud_runs(root: Path) -> list[dict[str, Any]]:
|
output_root = root / "shared" / "outputs" / "05-3d-pointcloud"
|
records: list[dict[str, Any]] = []
|
for metadata_path in output_root.rglob("run_metadata.json"):
|
artifact = metadata_path.parent
|
metadata = load_json(metadata_path)
|
dense_photo_reconstruction = metadata.get("dense_photo_reconstruction")
|
if metadata.get("capability") == "05-3d-pointcloud" and isinstance(dense_photo_reconstruction, dict):
|
textured_model = dense_photo_reconstruction.get("textured_model_file")
|
dense_point_cloud = dense_photo_reconstruction.get("dense_point_cloud_file")
|
mesh = dense_photo_reconstruction.get("mesh_file")
|
if all(isinstance(item, str) and (artifact / item).is_file() for item in (textured_model, dense_point_cloud, mesh)):
|
run_id = artifact.name
|
records.append(
|
{
|
"id": run_id,
|
"label": str(metadata.get("display_name") or run_id),
|
"note": "CPU 稠密 MVS：显示经过深度融合、网格化和纹理化的局部模型；不是测绘级坐标、DSM、正射图或语义识别结论。",
|
"artifactRoot": relative_path(root, artifact),
|
"createdAt": str(metadata.get("created_at") or ""),
|
}
|
)
|
continue
|
photo_reconstruction = metadata.get("photo_reconstruction")
|
if metadata.get("capability") == "05-3d-pointcloud" and isinstance(photo_reconstruction, dict):
|
preview = photo_reconstruction.get("preview_file")
|
point_cloud = photo_reconstruction.get("point_cloud_file")
|
if isinstance(preview, str) and isinstance(point_cloud, str) and (artifact / preview).is_file() and (artifact / point_cloud).is_file():
|
run_id = artifact.name
|
records.append(
|
{
|
"id": run_id,
|
"label": str(metadata.get("display_name") or run_id),
|
"note": "CPU 稀疏 SfM：显示可复核点云、相机位姿与误差；不是稠密重建、DSM、语义识别或测绘精度结论。",
|
"artifactRoot": relative_path(root, artifact),
|
"createdAt": str(metadata.get("created_at") or ""),
|
}
|
)
|
continue
|
point_clouds = metadata.get("point_clouds")
|
if metadata.get("capability") != "05-3d-pointcloud" or not isinstance(point_clouds, list) or not point_clouds:
|
continue
|
if any(not isinstance(item, dict) or not (artifact / str(item.get("preview_file") or "")).is_file() or not (artifact / str(item.get("vector_file") or "")).is_file() for item in point_clouds):
|
continue
|
run_id = artifact.name
|
records.append(
|
{
|
"id": run_id,
|
"label": str(metadata.get("display_name") or run_id),
|
"note": "CPU 语义规则基线：地面、植被、构筑物以及电线/杆塔候选，需要人工复核；不提供测绘精度或资产台账结论。" if any(isinstance(item, dict) and item.get("semantic_summary_file") for item in point_clouds) else "CPU 几何基线：地面/高出地物分离、DSM、近似网格和 GeoAI 足迹；不提供语义类别或测绘精度结论。",
|
"artifactRoot": relative_path(root, artifact),
|
"createdAt": str(metadata.get("created_at") or ""),
|
}
|
)
|
return sorted(records, key=lambda item: (item["createdAt"], item["id"]), reverse=True)
|
|
|
def pointcloud_annotation_sources(root: Path) -> list[dict[str, Any]]:
|
"""Expose only generated, fixed preview PLYs suitable for manual labels."""
|
sources: list[dict[str, Any]] = []
|
for case in pointcloud_runs(root):
|
artifact = root / str(case["artifactRoot"])
|
metadata = load_json(artifact / "run_metadata.json")
|
for cloud in metadata.get("point_clouds", []):
|
if not isinstance(cloud, dict):
|
continue
|
name = cloud.get("semantic_annotation_source_point_cloud")
|
if not isinstance(name, str) or Path(name).name != name:
|
continue
|
path = artifact / name
|
if not path.is_file() or path.suffix.lower() != ".ply":
|
continue
|
sources.append({
|
"id": f"{case['id']}:{name}", "runId": case["id"], "label": f"{case['label']} / {cloud.get('file', name)}",
|
"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]]:
|
output = root / "shared" / "outputs" / "05-3d-pointcloud" / "annotations"
|
records: list[dict[str, Any]] = []
|
for path in output.glob("*/annotation.json"):
|
data = load_json(path)
|
if data.get("schema_version") != 1 or not isinstance(data.get("id"), str):
|
continue
|
records.append({"id": data["id"], "sourceId": data.get("source_id"), "createdAt": data.get("created_at"), "labelCount": len(data.get("labels", [])), "classCounts": data.get("class_counts", {}), "path": relative_path(root, path)})
|
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], 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()})
|
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)
|
if completed.returncode:
|
message = (completed.stderr or completed.stdout or "Unknown training error.").strip().splitlines()[-1]
|
raise ApiError(message[:600])
|
metrics = output / "metrics.json"
|
model = output / "model.pt"
|
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), "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]})
|
|
|
def pointcloud_semantic_models(root: Path) -> list[dict[str, Any]]:
|
"""Expose only complete locally trained models, never arbitrary model paths."""
|
output_root = root / "shared" / "outputs" / "05-3d-pointcloud" / "training-runs"
|
records: list[dict[str, Any]] = []
|
for model_path in output_root.glob("*/model.pt"):
|
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 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:
|
continue
|
test = metrics.get("test") if isinstance(metrics.get("test"), dict) else {}
|
report = test.get("report") if isinstance(test.get("report"), dict) else {}
|
summary: dict[str, float] = {}
|
for code in class_codes:
|
definition = classes.get(code)
|
key = definition.get("key") if isinstance(definition, dict) else None
|
score = report.get(key) if isinstance(key, str) else None
|
if isinstance(score, dict) and isinstance(score.get("f1-score"), (int, float)):
|
summary[key] = round(float(score["f1-score"]), 3)
|
records.append({"id": model_path.parent.name, "label": model_path.parent.name, "artifactRoot": relative_path(root, model_path.parent), "model": relative_path(root, model_path), "metrics": relative_path(root, metrics_path), "createdAt": str(metrics.get("created_at") or ""), "classes": classes, "testF1": summary})
|
return sorted(records, key=lambda item: (item["createdAt"], item["id"]), reverse=True)
|
|
|
def pointcloud_inference_job(job_id: str) -> dict[str, Any] | None:
|
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:
|
"""Reject obviously incomplete uploads before consuming a CPU inference job."""
|
size = path.stat().st_size
|
if size < 1:
|
raise ApiError("点云文件为空。请选择包含 RGB 点位的完整点云导出文件。")
|
if path.suffix.lower() not in {".las", ".laz"}:
|
return
|
if size < 227:
|
raise ApiError("LAS/LAZ 文件不完整（小于有效 LAS 文件头）。请选择完整点云文件，不要上传空白或未完成下载的分块。")
|
with path.open("rb") as stream:
|
header = stream.read(375)
|
if len(header) < 227 or header[:4] != b"LASF":
|
raise ApiError("LAS/LAZ 文件头无效。请选择完整的 LAS/LAZ 点云导出文件。")
|
header_size = int.from_bytes(header[94:96], "little")
|
if header_size < 227 or header_size > size:
|
raise ApiError("LAS/LAZ 文件头不完整。请选择完整的 LAS/LAZ 点云导出文件。")
|
version_minor = header[25]
|
point_count_offset, point_count_size = (247, 8) if version_minor >= 4 else (107, 4)
|
if len(header) < point_count_offset + point_count_size:
|
raise ApiError("LAS/LAZ 文件头不完整。请选择完整的 LAS/LAZ 点云导出文件。")
|
point_count = int.from_bytes(header[point_count_offset:point_count_offset + point_count_size], "little")
|
if point_count < 1:
|
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], 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)
|
if completed.returncode:
|
message = (completed.stderr or completed.stdout or "Unknown model inference error.").strip().splitlines()[-1]
|
raise ApiError(message[:600])
|
metadata = output / "run_metadata.json"
|
preview = output / "predicted-semantic-preview.ply"
|
classified_las = output / "predicted-semantic-classified.las"
|
counts = output / "class-counts.csv"
|
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), "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]})
|
|
|
def risk_rule_runs(root: Path) -> list[dict[str, Any]]:
|
output_root = root / "shared" / "outputs" / "07-risk-rule-engine"
|
records: list[dict[str, Any]] = []
|
for metadata_path in output_root.rglob("run_metadata.json"):
|
artifact = metadata_path.parent
|
metadata = load_json(metadata_path)
|
artifacts = metadata.get("artifacts")
|
if metadata.get("capability") != "07-risk-rule-engine" or not isinstance(artifacts, dict):
|
continue
|
required = ("risk_raster", "risk_preview", "risk_vector", "risk_scores_csv", "summary")
|
if any(not isinstance(artifacts.get(key), str) or not (artifact / artifacts[key]).is_file() for key in required):
|
continue
|
run_id = artifact.name
|
records.append(
|
{
|
"id": run_id,
|
"label": str(metadata.get("display_name") or run_id),
|
"note": "可审计空间规则评分，仅供人工复核，不构成事件或处置结论。",
|
"artifactRoot": relative_path(root, artifact),
|
"createdAt": str(metadata.get("created_at") or ""),
|
}
|
)
|
return sorted(records, key=lambda item: (item["createdAt"], item["id"]), reverse=True)
|
|
|
def anomaly_runs(root: Path) -> list[dict[str, Any]]:
|
output_root = root / "shared" / "outputs" / "09-anomaly-detection"
|
allowed_raw = (root / "shared" / "data" / "raw" / "09-anomaly-detection").resolve()
|
records: list[dict[str, Any]] = []
|
for metadata_path in output_root.rglob("run_metadata.json"):
|
artifact = metadata_path.parent
|
metadata = load_json(metadata_path)
|
images = metadata.get("images")
|
if metadata.get("capability") != "09-anomaly-detection" or not isinstance(images, list):
|
continue
|
raw_input_value = str(metadata.get("raw_input_dir") or "")
|
raw_reference_value = str(metadata.get("raw_reference_dir") or "")
|
if not raw_input_value or not raw_reference_value:
|
continue
|
try:
|
raw_input = (root / raw_input_value).resolve()
|
raw_reference = (root / raw_reference_value).resolve()
|
raw_input.relative_to(allowed_raw)
|
raw_reference.relative_to(allowed_raw)
|
except ValueError:
|
continue
|
if not raw_input.is_dir() or not raw_reference.is_dir():
|
continue
|
if any(not (artifact / str(item.get("overlay_file") or "")).is_file() for item in images if isinstance(item, dict)):
|
continue
|
run_id = artifact.name
|
records.append(
|
{
|
"id": run_id,
|
"label": str(metadata.get("display_name") or run_id),
|
"note": str(metadata.get("case_note") or "规则基线与 Isolation Forest 的视觉离群候选，只供人工复核。"),
|
"artifactRoot": relative_path(root, artifact),
|
"inputRoot": relative_path(root, raw_input),
|
"referenceRoot": relative_path(root, raw_reference),
|
"createdAt": str(metadata.get("created_at") or ""),
|
}
|
)
|
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)
|
return dict(value) if value else None
|
|
|
def photo_reconstruction_job(job_id: str) -> dict[str, Any] | None:
|
with PHOTO_RECONSTRUCTION_JOBS_LOCK:
|
value = PHOTO_RECONSTRUCTION_JOBS.get(job_id)
|
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:
|
completed = subprocess.run(command, cwd=root, capture_output=True, text=True, timeout=timeout, check=False)
|
if completed.returncode:
|
message = (completed.stderr or completed.stdout or "Unknown script error.").strip().splitlines()[-1]
|
raise ApiError(f"Processing failed: {message[:600]}")
|
|
|
def execute_photo_reconstruction_job(
|
root: Path,
|
job_id: str,
|
run_id: str,
|
raw_root: Path,
|
processed_root: Path,
|
sparse_output: Path,
|
output: Path,
|
source_sha256: dict[str, str],
|
source_bytes: dict[str, int],
|
use_position_priors: bool,
|
) -> None:
|
python = root / ".venvs" / "05-3d-pointcloud" / "Scripts" / "python.exe"
|
sparse_command = [
|
str(python), str(root / "capabilities" / "05-3d-pointcloud" / "run_photo_reconstruction.py"),
|
"--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"])
|
else:
|
sparse_command.extend(["--matching-mode", "exhaustive"])
|
dense_command = [
|
str(python), str(root / "capabilities" / "05-3d-pointcloud" / "run_cpu_dense_reconstruction.py"),
|
"--input", str(processed_root), "--sparse-model", str(sparse_output / "sparse_model" / "0"),
|
"--output", str(output),
|
"--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:
|
PHOTO_RECONSTRUCTION_JOBS[job_id].update({"status": "running", "stage": "sparse_sfm", "startedAt": datetime.now(UTC).isoformat()})
|
with RUN_LOCK:
|
run_background_command(sparse_command, root, 1800)
|
sparse_metadata = load_json(sparse_output / "run_metadata.json")
|
if not (sparse_output / "sparse_model" / "0").is_dir():
|
raise ApiError("Sparse photo reconstruction finished without the expected COLMAP model.")
|
with PHOTO_RECONSTRUCTION_JOBS_LOCK:
|
PHOTO_RECONSTRUCTION_JOBS[job_id].update({"stage": "dense_mvs"})
|
run_background_command(dense_command, root, PHOTO_RECONSTRUCTION_TIMEOUT)
|
metadata_path = output / "run_metadata.json"
|
if not metadata_path.is_file():
|
raise ApiError("CPU dense reconstruction finished without the expected result metadata.")
|
metadata = load_json(metadata_path)
|
metadata["photo_reconstruction"] = sparse_metadata.get("photo_reconstruction", {})
|
metadata["input_dir"] = relative_path(root, processed_root)
|
metadata["raw_input_dir"] = relative_path(root, raw_root)
|
metadata["source_sha256"] = source_sha256
|
metadata["source_bytes"] = source_bytes
|
metadata["console_photo_reconstruction"] = {"use_position_priors": use_position_priors, "matching_mode": "spatial" if use_position_priors else "exhaustive"}
|
metadata["display_name"] = f"用户照片 CPU 稠密重建（{len(source_sha256)} 图）"
|
metadata["case_note"] = "用户上传的同架次 JPG/JPEG 照片经 CPU SfM/MVS 重建；需要人工检查几何与纹理质量，不是测绘级成果。"
|
metadata_path.write_text(json.dumps(metadata, ensure_ascii=False, indent=2), encoding="utf-8")
|
definition = next(item for item in pointcloud_runs(root) if item["id"] == run_id)
|
with PHOTO_RECONSTRUCTION_JOBS_LOCK:
|
PHOTO_RECONSTRUCTION_JOBS[job_id].update({"status": "complete", "stage": "complete", "run": definition, "finishedAt": datetime.now(UTC).isoformat()})
|
except Exception as exc: # pragma: no cover - background boundary
|
with PHOTO_RECONSTRUCTION_JOBS_LOCK:
|
PHOTO_RECONSTRUCTION_JOBS[job_id].update({"status": "failed", "stage": "failed", "error": str(exc), "finishedAt": datetime.now(UTC).isoformat()})
|
|
|
def validate_anomaly_parameters(payload: dict[str, Any]) -> tuple[int, int, float, int]:
|
tile_size = payload.get("tileSize", 256)
|
stride = payload.get("stride", 128)
|
threshold_quantile = payload.get("thresholdQuantile", 0.995)
|
random_state = payload.get("randomState", 42)
|
if isinstance(tile_size, bool) or not isinstance(tile_size, int) or not 128 <= tile_size <= 1024:
|
raise ApiError("Tile size must be an integer between 128 and 1024.")
|
if isinstance(stride, bool) or not isinstance(stride, int) or not 32 <= stride <= tile_size:
|
raise ApiError("Stride must be an integer between 32 and tile size.")
|
if isinstance(threshold_quantile, bool) or not isinstance(threshold_quantile, (int, float)) or not 0.9 <= float(threshold_quantile) <= 0.9999:
|
raise ApiError("Threshold quantile must be between 0.9 and 0.9999.")
|
if isinstance(random_state, bool) or not isinstance(random_state, int) or not 0 <= random_state <= 2_147_483_647:
|
raise ApiError("Random state must be a non-negative integer.")
|
return tile_size, stride, float(threshold_quantile), random_state
|
|
|
def execute_anomaly_job(
|
root: Path,
|
job_id: str,
|
run_id: str,
|
raw_reference: Path,
|
raw_input: Path,
|
processed_reference: Path,
|
processed_input: Path,
|
output: Path,
|
tile_size: int,
|
stride: int,
|
threshold_quantile: float,
|
random_state: int,
|
) -> None:
|
with ANOMALY_JOB_LOCK:
|
ANOMALY_JOBS[job_id]["status"] = "running"
|
python = root / ".venvs" / "09-anomaly-detection" / "Scripts" / "python.exe"
|
command = [
|
str(python),
|
str(root / "capabilities" / "09-anomaly-detection" / "run_anomaly_detection.py"),
|
"--reference", str(processed_reference),
|
"--input", str(processed_input),
|
"--output", str(output),
|
"--tile-size", str(tile_size),
|
"--stride", str(stride),
|
"--threshold-quantile", f"{threshold_quantile:.6f}",
|
"--random-state", str(random_state),
|
"--spatial-mode", "auto",
|
]
|
try:
|
with RUN_LOCK:
|
completed = subprocess.run(command, cwd=root, capture_output=True, text=True, timeout=1800, check=False)
|
if completed.returncode:
|
message = (completed.stderr or completed.stdout or "Unknown script error.").strip().splitlines()[-1]
|
raise ApiError(f"Processing failed: {message[:600]}")
|
metadata_path = output / "run_metadata.json"
|
if not metadata_path.is_file():
|
raise ApiError("Anomaly-detection script finished without the expected result metadata.")
|
metadata = load_json(metadata_path)
|
metadata["raw_input_dir"] = relative_path(root, raw_input)
|
metadata["raw_reference_dir"] = relative_path(root, raw_reference)
|
metadata["processed_input_dir"] = relative_path(root, processed_input)
|
metadata["processed_reference_dir"] = relative_path(root, processed_reference)
|
metadata_path.write_text(json.dumps(metadata, ensure_ascii=False, indent=2), encoding="utf-8")
|
definition = next(item for item in anomaly_runs(root) if item["id"] == run_id)
|
with ANOMALY_JOB_LOCK:
|
ANOMALY_JOBS[job_id].update({"status": "complete", "run": definition, "finishedAt": datetime.now(UTC).isoformat()})
|
except Exception as exc: # pragma: no cover - background boundary
|
with ANOMALY_JOB_LOCK:
|
ANOMALY_JOBS[job_id].update({"status": "failed", "error": str(exc), "finishedAt": datetime.now(UTC).isoformat()})
|
|
|
def semantic_tasks(root: Path) -> list[dict[str, Any]]:
|
catalog = load_json(root / "capabilities" / "02-semantic-mapping" / "configs" / "task-catalog.json")
|
tasks = catalog.get("tasks")
|
if not isinstance(tasks, list):
|
return []
|
return [item for item in tasks if isinstance(item, dict) and isinstance(item.get("id"), str)]
|
|
|
class WorkbenchConsoleHandler(SimpleHTTPRequestHandler):
|
"""Static UI plus fixed, local-only ingestion and experiment commands."""
|
|
server_version = "GeoAIWorkbench/1.0"
|
|
@property
|
def root(self) -> Path:
|
return Path(self.directory).resolve()
|
|
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
|
if path == "/api/change-detection/scans":
|
self.send_json(HTTPStatus.OK, {"scans": change_parameter_scans(self.root)})
|
return
|
if path.startswith("/api/change-detection/scan-jobs/"):
|
job_id = path.rstrip("/").rsplit("/", 1)[-1]
|
with SCAN_JOBS_LOCK:
|
job = dict(SCAN_JOBS.get(job_id, {}))
|
if not job:
|
self.send_json(HTTPStatus.NOT_FOUND, {"error": "Unknown change-detection scan job."})
|
else:
|
self.send_json(HTTPStatus.OK, {"job": job})
|
return
|
if path == "/api/trajectory/runs":
|
self.send_json(HTTPStatus.OK, {"runs": trajectory_runs(self.root)})
|
return
|
if path == "/api/object-detection/runs":
|
self.send_json(HTTPStatus.OK, {"runs": detection_runs(self.root)})
|
return
|
if path == "/api/semantic-mapping/runs":
|
self.send_json(HTTPStatus.OK, {"runs": semantic_runs(self.root)})
|
return
|
if path == "/api/semantic-mapping/tasks":
|
self.send_json(HTTPStatus.OK, {"tasks": semantic_tasks(self.root)})
|
return
|
if path == "/api/spatial-measurement/runs":
|
self.send_json(HTTPStatus.OK, {"runs": measurement_runs(self.root)})
|
return
|
if path == "/api/3d-pointcloud/runs":
|
self.send_json(HTTPStatus.OK, {"runs": pointcloud_runs(self.root)})
|
return
|
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]
|
job = pointcloud_inference_job(job_id)
|
self.send_json(HTTPStatus.OK if job else HTTPStatus.NOT_FOUND, {"job": job} if job else {"error": "Unknown point-cloud model inference job."})
|
return
|
if path.startswith("/api/3d-pointcloud/training-jobs/"):
|
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]
|
job = photo_reconstruction_job(job_id)
|
self.send_json(HTTPStatus.OK if job else HTTPStatus.NOT_FOUND, {"job": job} if job else {"error": "Unknown photo-reconstruction job."})
|
return
|
if path == "/api/risk-rule-engine/runs":
|
self.send_json(HTTPStatus.OK, {"runs": risk_rule_runs(self.root)})
|
return
|
if path == "/api/anomaly-detection/runs":
|
self.send_json(HTTPStatus.OK, {"runs": anomaly_runs(self.root)})
|
return
|
if path.startswith("/api/anomaly-detection/jobs/"):
|
job_id = path.rstrip("/").rsplit("/", 1)[-1]
|
job = anomaly_job(job_id)
|
self.send_json(HTTPStatus.OK if job else HTTPStatus.NOT_FOUND, {"job": job} if job else {"error": "Unknown anomaly-detection job."})
|
return
|
if path == "/":
|
self.send_response(HTTPStatus.FOUND)
|
self.send_header("Location", "/apps/workbench-console/")
|
self.end_headers()
|
return
|
super().do_GET()
|
|
def do_POST(self) -> None: # noqa: N802 - inherited standard-library method name
|
path = urlsplit(self.path).path
|
try:
|
payload = self.read_json_body()
|
if path == "/api/change-detection/runs":
|
self.send_json(HTTPStatus.CREATED, {"run": self.create_change_run(payload)})
|
return
|
if path == "/api/change-detection/scans":
|
self.send_json(HTTPStatus.ACCEPTED, {"job": self.create_change_scan(payload)})
|
return
|
if path.startswith("/api/change-detection/scans/") and path.endswith("/promote"):
|
scan_id = path.split("/")[-2]
|
self.send_json(HTTPStatus.CREATED, {"run": self.promote_change_scan(scan_id, payload)})
|
return
|
if path == "/api/trajectory/runs":
|
self.send_json(HTTPStatus.CREATED, {"run": self.create_trajectory_run(payload)})
|
return
|
if path == "/api/object-detection/runs":
|
self.send_json(HTTPStatus.CREATED, {"run": self.create_detection_run(payload)})
|
return
|
if path == "/api/semantic-mapping/runs":
|
self.send_json(HTTPStatus.CREATED, {"run": self.create_semantic_run(payload)})
|
return
|
if path == "/api/spatial-measurement/runs":
|
self.send_json(HTTPStatus.CREATED, {"run": self.create_measurement_run(payload)})
|
return
|
if path == "/api/3d-pointcloud/runs":
|
self.send_json(HTTPStatus.CREATED, {"run": self.create_pointcloud_run(payload)})
|
return
|
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)})
|
return
|
if path == "/api/risk-rule-engine/runs":
|
self.send_json(HTTPStatus.CREATED, {"run": self.create_risk_rule_run(payload)})
|
return
|
if path == "/api/anomaly-detection/runs":
|
self.send_json(HTTPStatus.ACCEPTED, {"job": self.create_anomaly_run(payload)})
|
return
|
self.send_json(HTTPStatus.NOT_FOUND, {"error": "Unknown local API endpoint."})
|
except ApiError as exc:
|
self.send_json(HTTPStatus.BAD_REQUEST, {"error": str(exc)})
|
except subprocess.TimeoutExpired:
|
self.send_json(HTTPStatus.GATEWAY_TIMEOUT, {"error": "The local run exceeded its time limit; no existing result was overwritten."})
|
except Exception as exc: # pragma: no cover - defensive server boundary
|
self.log_error("local run failed: %s", exc)
|
self.send_json(HTTPStatus.INTERNAL_SERVER_ERROR, {"error": "Local run failed. Check the console terminal for details."})
|
|
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."})
|
return
|
try:
|
annotation_id = path[len(prefix):]
|
if not annotation_id or "/" in annotation_id or SAFE_FILE_NAME.search(annotation_id) or len(annotation_id) > 120:
|
raise ApiError("Invalid annotation id.")
|
record = next((item for item in pointcloud_annotations(self.root) if item["id"] == annotation_id), None)
|
if not record:
|
raise ApiError("The selected annotation revision is unavailable.")
|
annotation_root = (self.root / "shared" / "outputs" / "05-3d-pointcloud" / "annotations").resolve()
|
location = (annotation_root / annotation_id).resolve()
|
location.relative_to(annotation_root)
|
if not (location / "annotation.json").is_file():
|
raise ApiError("The selected annotation revision is incomplete.")
|
shutil.rmtree(location)
|
self.send_json(HTTPStatus.OK, {"deletedId": annotation_id})
|
except ApiError as exc:
|
self.send_json(HTTPStatus.BAD_REQUEST, {"error": str(exc)})
|
except ValueError:
|
self.send_json(HTTPStatus.BAD_REQUEST, {"error": "Invalid annotation location."})
|
except Exception as exc: # pragma: no cover - defensive server boundary
|
self.log_error("annotation deletion failed: %s", exc)
|
self.send_json(HTTPStatus.INTERNAL_SERVER_ERROR, {"error": "Annotation deletion failed. Check the console terminal for details."})
|
|
def do_PUT(self) -> None: # noqa: N802 - binary upload endpoint
|
path = urlsplit(self.path).path
|
if not path.startswith("/api/change-detection/uploads/") and not path.startswith("/api/anomaly-detection/uploads/") and not path.startswith("/api/3d-pointcloud/photo-uploads/") and not path.startswith("/api/3d-pointcloud/pointcloud-uploads/"):
|
self.send_json(HTTPStatus.NOT_FOUND, {"error": "Unknown local API endpoint."})
|
return
|
try:
|
if path.startswith("/api/anomaly-detection/uploads/"):
|
result = self.receive_anomaly_upload(path)
|
elif path.startswith("/api/3d-pointcloud/photo-uploads/"):
|
result = self.receive_photo_reconstruction_upload(path)
|
elif path.startswith("/api/3d-pointcloud/pointcloud-uploads/"):
|
result = self.receive_pointcloud_upload(path)
|
else:
|
result = self.receive_change_upload(path)
|
self.send_json(HTTPStatus.CREATED, result)
|
except ApiError as exc:
|
self.send_json(HTTPStatus.BAD_REQUEST, {"error": str(exc)})
|
except Exception as exc: # pragma: no cover - defensive server boundary
|
self.log_error("binary upload failed: %s", exc)
|
self.send_json(HTTPStatus.INTERNAL_SERVER_ERROR, {"error": "Binary upload failed. Check the console terminal for details."})
|
|
def do_OPTIONS(self) -> None: # noqa: N802
|
self.send_response(HTTPStatus.NO_CONTENT)
|
self.send_header("Allow", "GET, POST, PUT, DELETE, OPTIONS")
|
self.end_headers()
|
|
def read_json_body(self) -> dict[str, Any]:
|
content_length = self.headers.get("Content-Length")
|
if content_length is None or not content_length.isdigit():
|
raise ApiError("A JSON request body with Content-Length is required.")
|
size = int(content_length)
|
if size <= 0 or size > MAX_REQUEST_BYTES:
|
raise ApiError(f"Request must be between 1 byte and {MAX_REQUEST_BYTES // (1024 * 1024)} MB.")
|
if "application/json" not in self.headers.get("Content-Type", ""):
|
raise ApiError("Content-Type must be application/json.")
|
try:
|
payload = json.loads(self.rfile.read(size).decode("utf-8"))
|
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
|
raise ApiError("Request body is not valid UTF-8 JSON.") from exc
|
if not isinstance(payload, dict):
|
raise ApiError("JSON request body must be an object.")
|
return payload
|
|
def run_command(self, command: list[str], timeout: int) -> None:
|
completed = subprocess.run(command, cwd=self.root, capture_output=True, text=True, timeout=timeout, check=False)
|
if completed.returncode:
|
message = (completed.stderr or completed.stdout or "Unknown script error.").strip().splitlines()[-1]
|
raise ApiError(f"Processing failed: {message[:600]}")
|
|
def receive_change_upload(self, path: str) -> dict[str, Any]:
|
return self.receive_binary_upload(path, "00-change-detection", {"before", "after"}, {".jpg", ".jpeg", ".png", ".tif", ".tiff"}, "change-detection")
|
|
def receive_anomaly_upload(self, path: str) -> dict[str, Any]:
|
return self.receive_binary_upload(path, "09-anomaly-detection", {"reference", "input"}, {".jpg", ".jpeg", ".png", ".tif", ".tiff"}, "anomaly-detection")
|
|
def receive_photo_reconstruction_upload(self, path: str) -> dict[str, Any]:
|
return self.receive_binary_upload(path, "05-3d-pointcloud", {"photo"}, {".jpg", ".jpeg"}, "photo reconstruction")
|
|
def receive_pointcloud_upload(self, path: str) -> dict[str, Any]:
|
return self.receive_binary_upload(path, "05-3d-pointcloud", {"pointcloud"}, {".ply", ".pcd", ".xyz", ".xyzn", ".xyzrgb", ".las", ".laz"}, "point-cloud")
|
|
def receive_binary_upload(
|
self,
|
path: str,
|
capability: str,
|
allowed_roles: set[str],
|
suffixes: set[str],
|
label: str,
|
) -> dict[str, Any]:
|
upload_id = path.rstrip("/").rsplit("/", 1)[-1]
|
if not SAFE_UPLOAD_ID.fullmatch(upload_id):
|
raise ApiError(f"Invalid {label} upload id.")
|
query = parse_qs(urlsplit(self.path).query)
|
role = query.get("role", [""])[0]
|
if role not in allowed_roles:
|
raise ApiError(f"Invalid {label} upload role.")
|
encoded_name = self.headers.get("X-Upload-Name", "")
|
if len(encoded_name) > 2048:
|
raise ApiError("Encoded upload name is too long.")
|
try:
|
name = unquote(encoded_name, encoding="utf-8", errors="strict")
|
except UnicodeError as exc:
|
raise ApiError("Upload name is not valid UTF-8 percent encoding.") from exc
|
safe_name = safe_file_name(name, suffixes)
|
content_length = self.headers.get("Content-Length")
|
if content_length is None or not content_length.isdigit():
|
raise ApiError("Binary upload requires a Content-Length header.")
|
size = int(content_length)
|
if size <= 0 or size > MAX_FILE_BYTES:
|
raise ApiError(f"Uploaded file must be between 1 byte and {MAX_FILE_BYTES // (1024 * 1024)} MB: {safe_name}.")
|
staging = self.root / "shared" / "data" / "raw" / capability / "uploads" / upload_id
|
staging.mkdir(parents=True, exist_ok=False)
|
part = staging / f"{role}.part"
|
target = staging / f"{role}{Path(safe_name).suffix.lower()}"
|
remaining = size
|
digest = hashlib.sha256()
|
try:
|
with part.open("wb") as stream:
|
while remaining:
|
chunk = self.rfile.read(min(8 * 1024 * 1024, remaining))
|
if not chunk:
|
raise ApiError("Binary upload ended before Content-Length was reached.")
|
stream.write(chunk)
|
digest.update(chunk)
|
remaining -= len(chunk)
|
part.replace(target)
|
(staging / f"{role}.json").write_text(json.dumps({"role": role, "name": safe_name, "size": size, "sha256": digest.hexdigest()}), encoding="utf-8")
|
except Exception:
|
part.unlink(missing_ok=True)
|
target.unlink(missing_ok=True)
|
raise
|
return {"uploadId": upload_id, "role": role, "name": safe_name, "size": size, "sha256": digest.hexdigest()}
|
|
def resolve_change_upload(self, payload: Any, role: str) -> tuple[str, Path]:
|
return self.resolve_binary_upload(payload, role, "00-change-detection", "change-detection")
|
|
def resolve_anomaly_upload(self, payload: Any, role: str) -> tuple[str, Path, str]:
|
name, path = self.resolve_binary_upload(payload, role, "09-anomaly-detection", "anomaly-detection")
|
manifest = load_json(path.parent / f"{role}.json")
|
return name, path, str(manifest.get("sha256") or "")
|
|
def resolve_photo_reconstruction_upload(self, payload: Any) -> tuple[str, Path, str]:
|
name, path = self.resolve_binary_upload(payload, "photo", "05-3d-pointcloud", "photo reconstruction")
|
manifest = load_json(path.parent / "photo.json")
|
return name, path, str(manifest.get("sha256") or "")
|
|
def resolve_pointcloud_upload(self, payload: Any) -> tuple[str, Path, str]:
|
name, path = self.resolve_binary_upload(payload, "pointcloud", "05-3d-pointcloud", "point-cloud")
|
manifest = load_json(path.parent / "pointcloud.json")
|
return name, path, str(manifest.get("sha256") or "")
|
|
def resolve_binary_upload(self, payload: Any, role: str, capability: str, label: str) -> tuple[str, Path]:
|
if not isinstance(payload, dict) or not isinstance(payload.get("uploadId"), str):
|
raise ApiError(f"{label} uploads must include a {role} uploadId.")
|
upload_id = payload["uploadId"]
|
if not SAFE_UPLOAD_ID.fullmatch(upload_id):
|
raise ApiError(f"Invalid {label} upload id.")
|
staging = self.root / "shared" / "data" / "raw" / capability / "uploads" / upload_id
|
manifest = load_json(staging / f"{role}.json")
|
name = str(manifest.get("name") or "")
|
path = staging / f"{role}{Path(name).suffix.lower()}"
|
if manifest.get("role") != role or not name or not path.is_file():
|
raise ApiError(f"The staged {role} upload is unavailable or incomplete.")
|
return name, path
|
|
def create_trajectory_run(self, payload: dict[str, Any]) -> dict[str, Any]:
|
files = payload.get("files")
|
if not isinstance(files, dict):
|
raise ApiError("Trajectory request must contain a files object.")
|
required = {
|
"flight": {".xlsx"},
|
"route": {".kmz"},
|
"restricted": {".geojson"},
|
}
|
decoded = {key: decode_upload(files.get(key), suffixes) for key, suffixes in required.items()}
|
flyable = decode_upload(files["flyable"], {".gzip"}) if files.get("flyable") else None
|
run_id = make_run_id("trajectory")
|
raw_root = self.root / "shared" / "data" / "raw" / "15-trajectory-analysis" / "runs" / run_id
|
paths = {"flight": raw_root / "tracks" / decoded["flight"][0], "route": raw_root / "routes" / decoded["route"][0], "restricted": raw_root / "areas" / decoded["restricted"][0]}
|
for key, path in paths.items():
|
path.parent.mkdir(parents=True, exist_ok=True)
|
path.write_bytes(decoded[key][1])
|
if flyable:
|
flyable_path = raw_root / "areas" / flyable[0]
|
flyable_path.write_bytes(flyable[1])
|
processed = self.root / "shared" / "data" / "processed" / "15-trajectory-analysis" / run_id
|
output_parent = self.root / "shared" / "outputs" / "15-trajectory-analysis" / "runs" / run_id
|
python = self.root / ".venvs" / "15-trajectory-analysis" / "Scripts" / "python.exe"
|
if not python.is_file():
|
raise ApiError("Trajectory virtual environment is unavailable. Run the capability setup first.")
|
with RUN_LOCK:
|
self.run_command([str(python), str(self.root / "capabilities" / "15-trajectory-analysis" / "prepare_real_flight.py"), "--raw-dir", str(raw_root), "--output", str(processed), "--case-id", run_id], 300)
|
self.run_command([str(python), str(self.root / "capabilities" / "15-trajectory-analysis" / "run_trajectory_analysis.py"), "--input", str(processed / f"{run_id}.case.json"), "--output", str(output_parent)], 300)
|
artifact = output_parent / run_id
|
if not (artifact / "run_metadata.json").is_file():
|
raise ApiError("Trajectory script finished without the expected result metadata.")
|
return next(item for item in trajectory_runs(self.root) if item["id"] == run_id)
|
|
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.")
|
run_id = make_run_id("detection")
|
raw_root = self.root / "shared" / "data" / "raw" / "01-object-detection" / "runs" / run_id
|
raw_root.mkdir(parents=True, exist_ok=False)
|
for name, content in decoded:
|
(raw_root / name).write_bytes(content)
|
output = self.root / "shared" / "outputs" / "01-object-detection" / "runs" / run_id
|
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)
|
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")
|
staged["after"] = self.resolve_change_upload(uploads.get("after"), "after")
|
elif isinstance(files, dict):
|
decoded_before = decode_upload(files.get("before"), {".jpg", ".jpeg", ".png", ".tif", ".tiff"})
|
decoded_after = decode_upload(files.get("after"), {".jpg", ".jpeg", ".png", ".tif", ".tiff"})
|
else:
|
raise ApiError("Change-detection request must contain before and after files or uploads.")
|
threshold_value = payload.get("threshold", CHANGE_THRESHOLD_DEFAULT)
|
if isinstance(threshold_value, bool) or not isinstance(threshold_value, (int, float)):
|
raise ApiError("Change-detection threshold must be a number between 0.01 and 0.99.")
|
threshold = float(threshold_value)
|
if not CHANGE_THRESHOLD_MIN <= threshold <= CHANGE_THRESHOLD_MAX:
|
raise ApiError("Change-detection threshold must be between 0.01 and 0.99.")
|
processing_mode = payload.get("processingMode", CHANGE_PROCESSING_MODE_DEFAULT)
|
if not isinstance(processing_mode, str) or processing_mode not in CHANGE_PROCESSING_MODES:
|
raise ApiError("Change-detection processing mode must be auto, image, or geotiff.")
|
max_dimension_value = payload.get("maxDimension", CHANGE_MAX_DIMENSION_AUTO)
|
if isinstance(max_dimension_value, bool) or not isinstance(max_dimension_value, int):
|
raise ApiError("Change-detection resolution must be an integer: 0 or between 512 and 4096.")
|
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:
|
before_name, after_name = decoded_before[0], decoded_after[0]
|
run_id = make_run_id("change")
|
raw_root = self.root / "shared" / "data" / "raw" / "00-change-detection" / "runs" / run_id
|
before_path = raw_root / "before" / before_name
|
after_path = raw_root / "after" / after_name
|
before_path.parent.mkdir(parents=True, exist_ok=False)
|
after_path.parent.mkdir(parents=True, exist_ok=False)
|
if staged:
|
shutil.copyfile(staged["before"][1], before_path)
|
shutil.copyfile(staged["after"][1], after_path)
|
else:
|
before_path.write_bytes(decoded_before[1])
|
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
|
with RUN_LOCK:
|
self.run_command(
|
[
|
execution["python"],
|
str(self.root / "capabilities" / "00-change-detection" / "run_change_detection.py"),
|
"--before", str(before_path),
|
"--after", str(after_path),
|
"--threshold", f"{threshold:.4f}",
|
"--max-dimension", str(max_dimension),
|
"--processing-mode", processing_mode,
|
"--device", execution["device"],
|
"--processed-output", str(processed_root),
|
"--output", str(output),
|
],
|
1200,
|
)
|
metadata_path = output / "run_metadata.json"
|
if not metadata_path.is_file():
|
raise ApiError("Change-detection script finished without the expected result metadata.")
|
metadata = load_json(metadata_path)
|
metadata["raw_input_dir"] = relative_path(self.root, raw_root)
|
metadata["processed_input_dir"] = relative_path(self.root, processed_root)
|
metadata["raw_before"] = relative_path(self.root, before_path)
|
metadata["raw_after"] = relative_path(self.root, after_path)
|
metadata_path.write_text(json.dumps(metadata, ensure_ascii=False, indent=2), encoding="utf-8")
|
return next(item for item in change_runs(self.root) if item["id"] == run_id)
|
|
def _scan_parameters(self, payload: dict[str, Any]) -> tuple[list[float], list[int]]:
|
raw_thresholds = payload.get("thresholds", SCAN_DEFAULT_THRESHOLDS)
|
raw_areas = payload.get("minimumAreas", SCAN_DEFAULT_AREAS)
|
if not isinstance(raw_thresholds, list) or not raw_thresholds or len(raw_thresholds) > SCAN_MAX_THRESHOLDS:
|
raise ApiError(f"Parameter scan thresholds must contain 1-{SCAN_MAX_THRESHOLDS} values.")
|
if not isinstance(raw_areas, list) or not raw_areas or len(raw_areas) > SCAN_MAX_AREAS:
|
raise ApiError(f"Parameter scan minimum areas must contain 1-{SCAN_MAX_AREAS} values.")
|
thresholds: list[float] = []
|
for value in raw_thresholds:
|
if isinstance(value, bool) or not isinstance(value, (int, float)):
|
raise ApiError("Each scan threshold must be a number between 0.01 and 0.99.")
|
number = round(float(value), 4)
|
if not CHANGE_THRESHOLD_MIN <= number <= CHANGE_THRESHOLD_MAX:
|
raise ApiError("Each scan threshold must be between 0.01 and 0.99.")
|
if number not in thresholds:
|
thresholds.append(number)
|
areas: list[int] = []
|
for value in raw_areas:
|
if isinstance(value, bool) or not isinstance(value, int) or not 16 <= value <= 200000:
|
raise ApiError("Each scan minimum area must be an integer between 16 and 200000 pixels.")
|
if value not in areas:
|
areas.append(value)
|
if len(thresholds) * len(areas) > SCAN_MAX_COMBINATIONS:
|
raise ApiError(f"A parameter scan accepts at most {SCAN_MAX_COMBINATIONS} combinations.")
|
return thresholds, areas
|
|
def _scan_root(self, scan_id: str) -> Path:
|
if not SAFE_SCAN_ID.fullmatch(scan_id):
|
raise ApiError("Invalid parameter-scan id.")
|
scan_root = self.root / "shared" / "outputs" / "00-change-detection" / "parameter-scans" / scan_id
|
if not scan_root.is_dir() or not (scan_root / "scan_summary.json").is_file():
|
raise ApiError("The parameter-scan result is unavailable.")
|
return scan_root
|
|
def promote_change_scan(self, scan_id: str, payload: dict[str, Any]) -> dict[str, Any]:
|
scan_root = self._scan_root(scan_id)
|
result_id = payload.get("resultId")
|
if not isinstance(result_id, str) or not SAFE_SCAN_RESULT_ID.fullmatch(result_id):
|
raise ApiError("A valid parameter-scan resultId is required.")
|
result_dir = scan_root / result_id
|
summary = load_json(result_dir / "summary.json")
|
full_result = load_json(result_dir / "full_result.json")
|
inference_run_id = str(load_json(scan_root / "scan_summary.json").get("source_run") or "")
|
inference_output = self.root / "shared" / "outputs" / "00-change-detection" / "runs" / inference_run_id
|
inference_metadata = load_json(inference_output / "run_metadata.json")
|
if not result_dir.is_dir() or not (result_dir / "changes.geojson").is_file() or not inference_metadata:
|
raise ApiError("The selected scan result is incomplete and cannot be promoted.")
|
run_id = make_run_id("change")
|
output = self.root / "shared" / "outputs" / "00-change-detection" / "runs" / run_id
|
output.mkdir(parents=True, exist_ok=False)
|
for source_name, destination_name in (
|
("change_probability.tif", "change_probability.tif"),
|
("change_mask.tif", "change_mask.tif"),
|
("generic_difference_mask.tif", "generic_difference_mask.tif"),
|
("change_model_overlay.jpg", "change_model_overlay.jpg"),
|
("before_processed_preview.jpg", "before_processed_preview.jpg"),
|
("after_registered_preview.jpg", "after_registered_preview.jpg"),
|
("changes.geojson", "changes.geojson"),
|
("changes_rectangles.geojson", "changes_rectangles.geojson"),
|
("changes_rectangles_wgs84.geojson", "changes_rectangles_wgs84.geojson"),
|
):
|
source = result_dir / source_name if source_name.startswith("change_mask") or source_name.startswith("changes") else inference_output / source_name
|
if source.is_file():
|
shutil.copyfile(source, output / destination_name)
|
overlay_source = result_dir / "overlay_preview.jpg"
|
if (inference_output / "change_overlay.jpg").is_file() and float(summary.get("threshold", 0.5)) == float(inference_metadata.get("thresholds", {}).get("change_probability", 0.5)):
|
overlay_source = inference_output / "change_overlay.jpg"
|
if not overlay_source.is_file():
|
raise ApiError("The selected scan preview is unavailable.")
|
shutil.copyfile(overlay_source, output / "change_overlay.jpg")
|
metadata = dict(inference_metadata)
|
scan_raw_root = self.root / "shared" / "data" / "raw" / "00-change-detection" / "runs" / scan_id
|
before_raw = scan_raw_root / "before" / str(inference_metadata.get("input_files", ["before.tif", "after.tif"])[0])
|
after_raw = scan_raw_root / "after" / str(inference_metadata.get("input_files", ["before.tif", "after.tif"])[1])
|
rectangle_count = int(full_result.get("rectangle_vector_feature_count") or 0)
|
if rectangle_count == 0 and (result_dir / "changes_rectangles.geojson").is_file():
|
rectangle_count = len(load_json(result_dir / "changes_rectangles.geojson").get("features", []))
|
metadata.update(
|
{
|
"kind": "formal-change-run",
|
"created_at": datetime.now(UTC).isoformat(),
|
"thresholds": {"change_probability": float(summary.get("threshold", 0.5)), "minimum_component_pixels": int(summary.get("minimum_area_pixels", 16))},
|
"raw_changed_pixels": int(summary.get("raw_changed_pixels", 0)),
|
"changed_pixels": int(summary.get("changed_pixels", 0)),
|
"changed_pixel_ratio": float(summary.get("changed_pixel_ratio", 0)),
|
"vector_feature_count": int(full_result.get("full_vector_feature_count", summary.get("vector_feature_count", 0))),
|
"rectangle_feature_count": rectangle_count,
|
"promoted_from_scan": scan_id,
|
"promoted_result": result_id,
|
"raw_input_dir": relative_path(self.root, scan_raw_root),
|
"raw_before": relative_path(self.root, before_raw),
|
"raw_after": relative_path(self.root, after_raw),
|
"processed_input_dir": relative_path(self.root, self.root / "shared" / "data" / "processed" / "00-change-detection" / scan_id),
|
"generic_difference": inference_metadata.get("generic_difference"),
|
"artifacts": {
|
"probability_raster": "change_probability.tif",
|
"raw_mask_raster": "change_mask.tif",
|
"mask_raster": "change_mask.tif",
|
"generic_difference_mask": "generic_difference_mask.tif" if (output / "generic_difference_mask.tif").is_file() else None,
|
"overlay": "change_overlay.jpg",
|
"model_overlay": "change_model_overlay.jpg" if (output / "change_model_overlay.jpg").is_file() else None,
|
"before_processed_preview": "before_processed_preview.jpg" if (output / "before_processed_preview.jpg").is_file() else None,
|
"after_registered_preview": "after_registered_preview.jpg" if (output / "after_registered_preview.jpg").is_file() else None,
|
"vector": "changes.geojson",
|
"rectangle_vector": "changes_rectangles.geojson",
|
"rectangle_vector_wgs84": "changes_rectangles_wgs84.geojson" if (output / "changes_rectangles_wgs84.geojson").is_file() else None,
|
"features": "full_result.json",
|
},
|
}
|
)
|
(output / "full_result.json").write_text(json.dumps(full_result, ensure_ascii=False, indent=2), encoding="utf-8")
|
(output / "run_metadata.json").write_text(json.dumps(metadata, ensure_ascii=False, indent=2), encoding="utf-8")
|
return next(item for item in change_runs(self.root) if item["id"] == run_id)
|
|
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 = {
|
"before": self.resolve_change_upload(uploads.get("before"), "before"),
|
"after": self.resolve_change_upload(uploads.get("after"), "after"),
|
}
|
thresholds, areas = self._scan_parameters(payload)
|
processing_mode = payload.get("processingMode", CHANGE_PROCESSING_MODE_DEFAULT)
|
if not isinstance(processing_mode, str) or processing_mode not in CHANGE_PROCESSING_MODES:
|
raise ApiError("Change-detection processing mode must be auto, image, or geotiff.")
|
max_dimension_value = payload.get("maxDimension", CHANGE_MAX_DIMENSION_AUTO)
|
if isinstance(max_dimension_value, bool) or not isinstance(max_dimension_value, int):
|
raise ApiError("Change-detection resolution must be an integer: 0 or between 512 and 4096.")
|
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]
|
after_path = raw_root / "after" / staged["after"][0]
|
before_path.parent.mkdir(parents=True, exist_ok=False)
|
after_path.parent.mkdir(parents=True, exist_ok=False)
|
shutil.copyfile(staged["before"][1], before_path)
|
shutil.copyfile(staged["after"][1], after_path)
|
processed_root = self.root / "shared" / "data" / "processed" / "00-change-detection" / run_id
|
inference_output = self.root / "shared" / "outputs" / "00-change-detection" / "runs" / run_id
|
scan_output = self.root / "shared" / "outputs" / "00-change-detection" / "parameter-scans" / run_id
|
with SCAN_JOBS_LOCK:
|
SCAN_JOBS[run_id] = {
|
"id": run_id,
|
"status": "queued",
|
"createdAt": datetime.now(UTC).isoformat(),
|
"thresholds": thresholds,
|
"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, execution),
|
daemon=True,
|
name=f"change-scan-{run_id}",
|
)
|
thread.start()
|
return dict(SCAN_JOBS[run_id])
|
|
def _update_scan_job(self, job_id: str, **values: Any) -> None:
|
with SCAN_JOBS_LOCK:
|
if job_id in SCAN_JOBS:
|
SCAN_JOBS[job_id].update(values)
|
|
def _run_change_scan(
|
self,
|
run_id: str,
|
before_path: Path,
|
after_path: Path,
|
processed_root: Path,
|
inference_output: Path,
|
scan_output: Path,
|
thresholds: list[float],
|
areas: list[int],
|
processing_mode: str,
|
max_dimension: int,
|
execution: dict[str, Any],
|
) -> None:
|
python = execution["python"]
|
try:
|
self._update_scan_job(run_id, status="running", phase="inference")
|
with RUN_LOCK:
|
self.run_command(
|
[
|
python,
|
str(self.root / "capabilities" / "00-change-detection" / "run_change_detection.py"),
|
"--before", str(before_path),
|
"--after", str(after_path),
|
"--threshold", "0.5000",
|
"--max-dimension", str(max_dimension),
|
"--processing-mode", processing_mode,
|
"--device", execution["device"],
|
"--processed-output", str(processed_root),
|
"--output", str(inference_output),
|
],
|
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
|
inference_metadata_path.write_text(json.dumps(inference_metadata, ensure_ascii=False, indent=2), encoding="utf-8")
|
self._update_scan_job(run_id, phase="parameter-scan")
|
command = [
|
python,
|
str(self.root / "capabilities" / "00-change-detection" / "scan_change_detection_parameters.py"),
|
"--run-dir", str(inference_output),
|
"--output", str(scan_output),
|
]
|
for threshold in thresholds:
|
command.extend(["--threshold", f"{threshold:.4f}"])
|
for area in areas:
|
command.extend(["--minimum-area", str(area)])
|
self.run_command(command, SCAN_JOB_TIMEOUT)
|
self._update_scan_job(run_id, phase="vectorization")
|
vector_command = [
|
python,
|
str(self.root / "capabilities" / "00-change-detection" / "materialize_parameter_scan_candidates.py"),
|
"--scan-dir", str(scan_output),
|
]
|
for threshold in thresholds:
|
for area in areas:
|
vector_command.extend(["--candidate", f"threshold-{threshold:.2f}_area-{area}"])
|
self.run_command(vector_command, SCAN_JOB_TIMEOUT)
|
metadata = {
|
"capability": "00-change-detection",
|
"kind": "parameter-scan",
|
"source_run": run_id,
|
"created_at": datetime.now(UTC).isoformat(),
|
"thresholds": thresholds,
|
"minimum_areas": areas,
|
"processing_mode": processing_mode,
|
"max_dimension": max_dimension,
|
"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),
|
}
|
scan_output.mkdir(parents=True, exist_ok=True)
|
(scan_output / "scan_metadata.json").write_text(json.dumps(metadata, ensure_ascii=False, indent=2), encoding="utf-8")
|
self._update_scan_job(run_id, status="completed", phase="done", scanId=run_id)
|
except Exception as exc: # background errors are returned through polling
|
scan_output.mkdir(parents=True, exist_ok=True)
|
(scan_output / "scan_failed.json").write_text(json.dumps({"job_id": run_id, "error": str(exc)[:600]}, ensure_ascii=False, indent=2), encoding="utf-8")
|
self._update_scan_job(run_id, status="failed", phase="error", error=str(exc)[:600])
|
|
def create_semantic_run(self, payload: dict[str, Any]) -> dict[str, Any]:
|
task_id = str(payload.get("taskId") or "color_baseline")
|
task = next((item for item in semantic_tasks(self.root) if item["id"] == task_id), None)
|
if task is None:
|
raise ApiError(f"Unknown semantic-mapping task: {task_id}.")
|
if task.get("selectable") is not True:
|
raise ApiError(f"Semantic-mapping task is not runnable yet: {task_id}.")
|
uploads = payload.get("images")
|
if not isinstance(uploads, list) or not uploads:
|
raise ApiError("Semantic-mapping request must include at least one image.")
|
if len(uploads) > MAX_SEGMENTATION_IMAGES_PER_RUN:
|
raise ApiError(f"A semantic-mapping run accepts at most {MAX_SEGMENTATION_IMAGES_PER_RUN} images.")
|
decoded = [decode_upload(item, {".jpg", ".jpeg", ".png", ".tif", ".tiff"}) 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.")
|
run_id = make_run_id("semantic")
|
raw_root = self.root / "shared" / "data" / "raw" / "02-semantic-mapping" / "runs" / run_id
|
processed_root = self.root / "shared" / "data" / "processed" / "02-semantic-mapping" / run_id
|
raw_root.mkdir(parents=True, exist_ok=False)
|
processed_root.mkdir(parents=True, exist_ok=False)
|
for name, content in decoded:
|
(raw_root / name).write_bytes(content)
|
(processed_root / name).write_bytes(content)
|
output = self.root / "shared" / "outputs" / "02-semantic-mapping" / "runs" / run_id
|
python = self.root / ".venvs" / "02-semantic-mapping" / "Scripts" / "python.exe"
|
if not python.is_file():
|
raise ApiError("Semantic-mapping virtual environment is unavailable. Run the capability setup first.")
|
with RUN_LOCK:
|
self.run_command([str(python), str(self.root / "capabilities" / "02-semantic-mapping" / "run_semantic_segmentation.py"), "--input", str(processed_root), "--output", str(output)], 900)
|
metadata_path = output / "run_metadata.json"
|
if not metadata_path.is_file():
|
raise ApiError("Semantic-mapping script finished without the expected result metadata.")
|
metadata = load_json(metadata_path)
|
metadata["task_id"] = task_id
|
metadata["task_name"] = str(task.get("name") or task_id)
|
metadata["input_dir"] = relative_path(self.root, processed_root)
|
metadata["raw_input_dir"] = relative_path(self.root, raw_root)
|
metadata_path.write_text(json.dumps(metadata, ensure_ascii=False, indent=2), encoding="utf-8")
|
return next(item for item in semantic_runs(self.root) if item["id"] == run_id)
|
|
def create_anomaly_run(self, payload: dict[str, Any]) -> dict[str, Any]:
|
tile_size, stride, threshold_quantile, random_state = validate_anomaly_parameters(payload)
|
uploads = payload.get("uploads")
|
if not isinstance(uploads, dict):
|
raise ApiError("Anomaly-detection request must contain reference and input uploads.")
|
reference_values = uploads.get("reference")
|
input_values = uploads.get("input")
|
if not isinstance(reference_values, list) or not reference_values:
|
raise ApiError("Select at least one normal reference image.")
|
if not isinstance(input_values, list) or not input_values:
|
raise ApiError("Select at least one image to inspect.")
|
if len(reference_values) > MAX_ANOMALY_IMAGES_PER_ROLE or len(input_values) > MAX_ANOMALY_IMAGES_PER_ROLE:
|
raise ApiError(f"An anomaly-detection run accepts at most {MAX_ANOMALY_IMAGES_PER_ROLE} images in each group.")
|
references = [self.resolve_anomaly_upload(value, "reference") for value in reference_values]
|
inputs = [self.resolve_anomaly_upload(value, "input") for value in input_values]
|
if len({name.casefold() for name, _, _ in references}) != len(references):
|
raise ApiError("Normal reference image names must be unique within one run.")
|
if len({name.casefold() for name, _, _ in inputs}) != len(inputs):
|
raise ApiError("Input image names must be unique within one run.")
|
|
python = self.root / ".venvs" / "09-anomaly-detection" / "Scripts" / "python.exe"
|
if not python.is_file():
|
raise ApiError("Anomaly-detection virtual environment is unavailable. Run the capability setup first.")
|
run_id = make_run_id("anomaly")
|
job_id = uuid4().hex
|
raw_root = self.root / "shared" / "data" / "raw" / "09-anomaly-detection" / "runs" / run_id
|
raw_reference = raw_root / "reference"
|
raw_input = raw_root / "input"
|
processed_root = self.root / "shared" / "data" / "processed" / "09-anomaly-detection" / run_id
|
processed_reference = processed_root / "reference"
|
processed_input = processed_root / "input"
|
output = self.root / "shared" / "outputs" / "09-anomaly-detection" / "runs" / run_id
|
for directory in (raw_reference, raw_input, processed_reference, processed_input):
|
directory.mkdir(parents=True, exist_ok=False)
|
for group, raw_dir, processed_dir in ((references, raw_reference, processed_reference), (inputs, raw_input, processed_input)):
|
for name, staged_path, expected_sha256 in group:
|
raw_path = raw_dir / name
|
processed_path = processed_dir / name
|
shutil.copyfile(staged_path, raw_path)
|
if expected_sha256 and file_sha256(raw_path) != expected_sha256:
|
raise ApiError(f"Uploaded file checksum changed while staging: {name}.")
|
shutil.copyfile(raw_path, processed_path)
|
for _, staged_path, _ in references + inputs:
|
shutil.rmtree(staged_path.parent)
|
|
created_at = datetime.now(UTC).isoformat()
|
job = {"id": job_id, "runId": run_id, "status": "queued", "createdAt": created_at}
|
with ANOMALY_JOB_LOCK:
|
ANOMALY_JOBS[job_id] = job
|
thread = threading.Thread(
|
target=execute_anomaly_job,
|
args=(self.root, job_id, run_id, raw_reference, raw_input, processed_reference, processed_input, output, tile_size, stride, threshold_quantile, random_state),
|
daemon=True,
|
name=f"anomaly-{run_id}",
|
)
|
thread.start()
|
return dict(job)
|
|
def create_measurement_run(self, payload: dict[str, Any]) -> dict[str, Any]:
|
uploads = payload.get("rasters")
|
if not isinstance(uploads, list) or not uploads:
|
raise ApiError("Spatial-measurement request must include at least one label raster.")
|
if len(uploads) > MAX_MEASUREMENT_RASTERS_PER_RUN:
|
raise ApiError(f"A spatial-measurement run accepts at most {MAX_MEASUREMENT_RASTERS_PER_RUN} rasters.")
|
decoded = [decode_upload(item, {".png", ".tif", ".tiff"}) for item in uploads]
|
if len({name.casefold() for name, _ in decoded}) != len(decoded):
|
raise ApiError("Uploaded raster names must be unique within one run.")
|
run_id = make_run_id("measurement")
|
raw_root = self.root / "shared" / "data" / "raw" / "04-spatial-measurement" / "runs" / run_id
|
processed_root = self.root / "shared" / "data" / "processed" / "04-spatial-measurement" / run_id
|
raw_root.mkdir(parents=True, exist_ok=False)
|
processed_root.mkdir(parents=True, exist_ok=False)
|
for name, content in decoded:
|
(raw_root / name).write_bytes(content)
|
(processed_root / name).write_bytes(content)
|
output = self.root / "shared" / "outputs" / "04-spatial-measurement" / "runs" / run_id
|
python = self.root / ".venvs" / "04-spatial-measurement" / "Scripts" / "python.exe"
|
if not python.is_file():
|
raise ApiError("Spatial-measurement virtual environment is unavailable. Run the capability setup first.")
|
with RUN_LOCK:
|
self.run_command([str(python), str(self.root / "capabilities" / "04-spatial-measurement" / "run_spatial_measurement.py"), "--input", str(processed_root), "--output", str(output)], 900)
|
metadata_path = output / "run_metadata.json"
|
if not metadata_path.is_file():
|
raise ApiError("Spatial-measurement script finished without the expected result metadata.")
|
metadata = load_json(metadata_path)
|
metadata["input_dir"] = relative_path(self.root, processed_root)
|
metadata["raw_input_dir"] = relative_path(self.root, raw_root)
|
metadata_path.write_text(json.dumps(metadata, ensure_ascii=False, indent=2), encoding="utf-8")
|
return next(item for item in measurement_runs(self.root) if item["id"] == run_id)
|
|
def create_pointcloud_run(self, payload: dict[str, Any]) -> dict[str, Any]:
|
uploads = payload.get("pointClouds")
|
source_dense_run_id = payload.get("sourceDenseRunId")
|
if uploads is None and not isinstance(source_dense_run_id, str):
|
raise ApiError("3D point-cloud request must include at least one PLY, PCD, XYZ, LAS, or LAZ file.")
|
if uploads is not None and (not isinstance(uploads, list) or not uploads):
|
raise ApiError("3D point-cloud request must include at least one PLY, PCD, XYZ, LAS, or LAZ file.")
|
if isinstance(uploads, list) and len(uploads) > MAX_POINTCLOUDS_PER_RUN:
|
raise ApiError(f"A 3D point-cloud run accepts at most {MAX_POINTCLOUDS_PER_RUN} files.")
|
suffixes = {".ply", ".pcd", ".xyz", ".xyzn", ".xyzrgb", ".las", ".laz"}
|
staged_upload_dirs: list[Path] = []
|
if isinstance(source_dense_run_id, str):
|
if SAFE_FILE_NAME.search(source_dense_run_id) or len(source_dense_run_id) > 120:
|
raise ApiError("Invalid dense point-cloud source run id.")
|
source_case = next((item for item in pointcloud_runs(self.root) if item["id"] == source_dense_run_id), None)
|
if not source_case:
|
raise ApiError("The selected dense point-cloud source is unavailable.")
|
source_artifact = self.root / str(source_case["artifactRoot"])
|
source_metadata = load_json(source_artifact / "run_metadata.json")
|
dense = source_metadata.get("dense_photo_reconstruction")
|
source_file = dense.get("dense_point_cloud_file") if isinstance(dense, dict) else None
|
source_path = source_artifact / str(source_file or "")
|
if not isinstance(source_file, str) or source_path.suffix.lower() != ".ply" or not source_path.is_file():
|
raise ApiError("The selected run has no available dense PLY output.")
|
decoded = [(f"{source_dense_run_id}-dense.ply", source_path, file_sha256(source_path))]
|
elif all(isinstance(item, dict) and isinstance(item.get("content"), str) for item in uploads):
|
decoded = [(name, content, "") for name, content in (decode_upload(item, suffixes) for item in uploads)]
|
else:
|
decoded = [self.resolve_pointcloud_upload(item) for item in uploads]
|
staged_upload_dirs = [path.parent for _, path, _ in decoded]
|
if len({name.casefold() for name, _, _ in decoded}) != len(decoded):
|
raise ApiError("Uploaded point-cloud names must be unique within one run.")
|
run_id = make_run_id("pointcloud")
|
raw_root = self.root / "shared" / "data" / "raw" / "05-3d-pointcloud" / "runs" / run_id
|
processed_root = self.root / "shared" / "data" / "processed" / "05-3d-pointcloud" / run_id
|
raw_root.mkdir(parents=True, exist_ok=False)
|
processed_root.mkdir(parents=True, exist_ok=False)
|
source_sha256: dict[str, str] = {}
|
source_bytes: dict[str, int] = {}
|
for name, staged_or_content, expected_sha256 in decoded:
|
raw_path = raw_root / name
|
if isinstance(staged_or_content, bytes):
|
raw_path.write_bytes(staged_or_content)
|
else:
|
shutil.copyfile(staged_or_content, raw_path)
|
if expected_sha256 and file_sha256(raw_path) != expected_sha256:
|
raise ApiError(f"Uploaded point-cloud checksum changed while staging: {name}.")
|
source_sha256[name] = file_sha256(raw_path)
|
source_bytes[name] = raw_path.stat().st_size
|
shutil.copyfile(raw_path, processed_root / name)
|
output = self.root / "shared" / "outputs" / "05-3d-pointcloud" / "runs" / run_id
|
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.")
|
command = [str(python), str(self.root / "capabilities" / "05-3d-pointcloud" / "run_pointcloud_understanding.py"), "--input", str(processed_root), "--output", str(output), "--ground-up-axis", "z"]
|
with RUN_LOCK:
|
self.run_command(command, 900)
|
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)
|
metadata["source_sha256"] = source_sha256
|
metadata["source_bytes"] = source_bytes
|
metadata_path.write_text(json.dumps(metadata, ensure_ascii=False, indent=2), encoding="utf-8")
|
for staging in staged_upload_dirs:
|
shutil.rmtree(staging)
|
return next(item for item in pointcloud_runs(self.root) if item["id"] == run_id)
|
|
def create_pointcloud_annotation(self, payload: dict[str, Any]) -> dict[str, Any]:
|
source_id = payload.get("sourceId")
|
labels = payload.get("labels")
|
if not isinstance(source_id, str) or not isinstance(labels, list):
|
raise ApiError("Annotation request must include a sourceId and labels array.")
|
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.")
|
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 class_by_code:
|
raise ApiError("Annotation contains an out-of-range point index or unsupported class code.")
|
compact[index] = code
|
if not compact:
|
raise ApiError("Save at least one user-confirmed point label.")
|
annotation_id = make_run_id("annotation")
|
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(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): 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")
|
device = payload.get("device", "auto")
|
if not isinstance(annotation_id, str) or SAFE_FILE_NAME.search(annotation_id) or len(annotation_id) > 120:
|
raise ApiError("Invalid annotation id.")
|
if device not in {"auto", "cpu", "cuda"}:
|
raise ApiError("Training device must be auto, cpu, or cuda.")
|
annotation = self.root / "shared" / "outputs" / "05-3d-pointcloud" / "annotations" / annotation_id / "annotation.json"
|
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("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, 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)
|
if not model_record:
|
raise ApiError("The selected trained model is unavailable or incomplete.")
|
name, staged_path, expected_sha256 = self.resolve_pointcloud_upload(upload)
|
suffixes = {".ply", ".pcd", ".xyz", ".xyzn", ".xyzrgb", ".las", ".laz"}
|
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)
|
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
|
output = self.root / "shared" / "outputs" / "05-3d-pointcloud" / "model-inference-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)
|
actual_sha256 = file_sha256(raw_path)
|
if expected_sha256 and actual_sha256 != expected_sha256:
|
raise ApiError("Uploaded point-cloud checksum changed while staging.")
|
processed_path = processed_root / name
|
shutil.copyfile(raw_path, processed_path)
|
# Only remove the staging copy after its immutable raw copy was verified.
|
shutil.rmtree(staged_path.parent)
|
model_path = self.root / str(model_record["model"])
|
training_root = (self.root / "shared" / "outputs" / "05-3d-pointcloud" / "training-runs").resolve()
|
try:
|
model_path.resolve().relative_to(training_root)
|
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": 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")
|
if not isinstance(uploads, list) or len(uploads) < 3:
|
raise ApiError("Photo reconstruction needs at least three JPG/JPEG photos from one coherent flight or camera sequence.")
|
if len(uploads) > MAX_PHOTO_RECONSTRUCTION_IMAGES_PER_RUN:
|
raise ApiError(f"A photo reconstruction run accepts at most {MAX_PHOTO_RECONSTRUCTION_IMAGES_PER_RUN} photos.")
|
use_position_priors = payload.get("usePositionPriors", False)
|
if not isinstance(use_position_priors, bool):
|
raise ApiError("Photo reconstruction usePositionPriors must be true or false.")
|
photos = [self.resolve_photo_reconstruction_upload(value) for value in uploads]
|
if len({name.casefold() for name, _, _ in photos}) != len(photos):
|
raise ApiError("Uploaded photo names must be unique within one run.")
|
python = self.root / ".venvs" / "05-3d-pointcloud" / "Scripts" / "python.exe"
|
openmvs = self.root / "shared" / "tools" / "openmvs-2.4.0" / "vc17" / "x64" / "Release"
|
if not python.is_file() or not (openmvs / "DensifyPointCloud.exe").is_file():
|
raise ApiError("Photo-reconstruction CPU environment is unavailable. Run the capability setup first.")
|
|
run_id = make_run_id("photo-reconstruction")
|
job_id = uuid4().hex
|
raw_root = self.root / "shared" / "data" / "raw" / "05-3d-pointcloud" / "runs" / run_id
|
processed_root = self.root / "shared" / "data" / "processed" / "05-3d-pointcloud" / run_id
|
sparse_output = processed_root / "sparse_sfm"
|
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)
|
source_sha256: dict[str, str] = {}
|
source_bytes: dict[str, int] = {}
|
for name, staged_path, expected_sha256 in photos:
|
raw_path = raw_root / name
|
shutil.copyfile(staged_path, raw_path)
|
actual_sha256 = file_sha256(raw_path)
|
if expected_sha256 and actual_sha256 != expected_sha256:
|
raise ApiError(f"Uploaded file checksum changed while staging: {name}.")
|
shutil.copyfile(raw_path, processed_root / name)
|
source_sha256[name] = actual_sha256
|
source_bytes[name] = raw_path.stat().st_size
|
for _, staged_path, _ in photos:
|
shutil.rmtree(staged_path.parent)
|
|
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(
|
target=execute_photo_reconstruction_job,
|
args=(self.root, job_id, run_id, raw_root, processed_root, sparse_output, output, source_sha256, source_bytes, use_position_priors),
|
daemon=True,
|
name=f"photo-reconstruction-{run_id}",
|
)
|
thread.start()
|
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")
|
if not isinstance(files, dict) or set(files) != RISK_RULE_REQUIRED_FILES:
|
raise ApiError("Risk-rule request must contain observations, zones and rules files.")
|
observations_name, observations_bytes = decode_upload(files["observations"], {".geojson"})
|
zones_name, zones_bytes = decode_upload(files["zones"], {".geojson"})
|
rules_name, rules_bytes = decode_upload(files["rules"], {".json"})
|
if len({observations_name.casefold(), zones_name.casefold(), rules_name.casefold()}) != 3:
|
raise ApiError("Risk-rule uploaded file names must be unique.")
|
run_id = make_run_id("risk")
|
raw_root = self.root / "shared" / "data" / "raw" / "07-risk-rule-engine" / "runs" / run_id
|
processed_root = self.root / "shared" / "data" / "processed" / "07-risk-rule-engine" / run_id
|
raw_root.mkdir(parents=True, exist_ok=False)
|
processed_root.mkdir(parents=True, exist_ok=False)
|
staged = ((observations_name, observations_bytes), (zones_name, zones_bytes), (rules_name, rules_bytes))
|
for name, content in staged:
|
(raw_root / name).write_bytes(content)
|
(processed_root / name).write_bytes(content)
|
output = self.root / "shared" / "outputs" / "07-risk-rule-engine" / "runs" / run_id
|
python = self.root / ".venvs" / "07-risk-rule-engine" / "Scripts" / "python.exe"
|
if not python.is_file():
|
raise ApiError("Risk-rule virtual environment is unavailable. Run the capability setup first.")
|
command = [
|
str(python), str(self.root / "capabilities" / "07-risk-rule-engine" / "run_risk_rule_engine.py"),
|
"--observations", str(processed_root / observations_name), "--zones", str(processed_root / zones_name),
|
"--rules", str(processed_root / rules_name), "--output", str(output),
|
]
|
with RUN_LOCK:
|
self.run_command(command, 600)
|
metadata_path = output / "run_metadata.json"
|
if not metadata_path.is_file():
|
raise ApiError("Risk-rule script finished without the expected result metadata.")
|
metadata = load_json(metadata_path)
|
metadata["input_dir"] = relative_path(self.root, processed_root)
|
metadata["raw_input_dir"] = relative_path(self.root, raw_root)
|
metadata["source_bytes"] = {name: len(content) for name, content in staged}
|
metadata_path.write_text(json.dumps(metadata, ensure_ascii=False, indent=2), encoding="utf-8")
|
return next(item for item in risk_rule_runs(self.root) if item["id"] == run_id)
|
|
def send_json(self, status: HTTPStatus, payload: dict[str, Any]) -> None:
|
body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
|
self.send_response(status)
|
self.send_header("Content-Type", "application/json; charset=utf-8")
|
self.send_header("Content-Length", str(len(body)))
|
self.end_headers()
|
self.wfile.write(body)
|
|
def translate_path(self, path: str) -> str:
|
"""Expose only the static UI and artifacts required by the local console."""
|
decoded_path = unquote(urlsplit(path).path).lstrip("/")
|
requested = PurePosixPath(decoded_path)
|
is_allowed = any(decoded_path == prefix or decoded_path.startswith(f"{prefix}/") for prefix in ALLOWED_PATH_PREFIXES)
|
if ".." in requested.parts or not is_allowed:
|
return os.fspath(Path(self.directory) / ".console-forbidden")
|
if decoded_path == "apps/workbench-console" or decoded_path.startswith("apps/workbench-console/"):
|
console_relative = requested.parts[2:]
|
return os.fspath(Path(self.directory) / "apps" / "workbench-console" / "dist" / Path(*console_relative))
|
return os.fspath(Path(self.directory).joinpath(*requested.parts))
|
|
def end_headers(self) -> None:
|
self.send_header("Cache-Control", "no-store")
|
self.send_header("X-Content-Type-Options", "nosniff")
|
super().end_headers()
|
|
|
def parse_args() -> argparse.Namespace:
|
root = Path(__file__).resolve().parents[1]
|
parser = argparse.ArgumentParser(description="Serve the local GeoAI Workbench console.")
|
parser.add_argument("--host", default=DEFAULT_HOST, help="Bind address. Defaults to loopback only.")
|
parser.add_argument("--port", type=int, default=DEFAULT_PORT, help="TCP port in the 6000-6999 range.")
|
parser.add_argument("--root", type=Path, default=root, help="Workbench repository root to serve.")
|
return parser.parse_args()
|
|
|
def main() -> int:
|
args = parse_args()
|
if not 6000 <= args.port <= 6999:
|
raise SystemExit("Port must be in the 6000-6999 range.")
|
if args.host not in {"127.0.0.1", "localhost", "::1"}:
|
raise SystemExit("This console is local-only. Use 127.0.0.1, localhost, or ::1.")
|
root = args.root.resolve()
|
app_dir = root / "apps" / "workbench-console"
|
if not (root / "shared").is_dir() or not (app_dir / "dist" / "index.html").is_file():
|
raise SystemExit(f"Not a GeoAI Workbench root: {root}")
|
handler = lambda *handler_args, **handler_kwargs: WorkbenchConsoleHandler(*handler_args, directory=os.fspath(root), **handler_kwargs) # noqa: E731
|
server = ThreadingHTTPServer((args.host, args.port), handler)
|
print(f"GeoAI Workbench console: http://{args.host}:{args.port}")
|
print("Local runs use fixed capability scripts and create a new run directory.")
|
try:
|
server.serve_forever()
|
except KeyboardInterrupt:
|
print("\nConsole stopped.")
|
finally:
|
server.server_close()
|
return 0
|
|
|
if __name__ == "__main__":
|
raise SystemExit(main())
|
