From fbb068ec702338d609c1ca6eddbdb9f182d8f211 Mon Sep 17 00:00:00 2001
From: shuishen <1109946754@qq.com>
Date: Mon, 24 Aug 2026 11:37:16 +0800
Subject: [PATCH] feat: extend local GeoAI capability workflows
---
scripts/serve_workbench_console.py | 692 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++
1 files changed, 684 insertions(+), 8 deletions(-)
diff --git a/scripts/serve_workbench_console.py b/scripts/serve_workbench_console.py
index d32a5b9..94ba5cb 100644
--- a/scripts/serve_workbench_console.py
+++ b/scripts/serve_workbench_console.py
@@ -30,6 +30,10 @@
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 = 30
+PHOTO_RECONSTRUCTION_TIMEOUT = 7200
+RISK_RULE_REQUIRED_FILES = {"observations", "zones", "rules"}
MAX_ANOMALY_IMAGES_PER_ROLE = 6
CHANGE_THRESHOLD_DEFAULT = 0.5
CHANGE_THRESHOLD_MIN = 0.01
@@ -53,6 +57,7 @@
"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}$")
@@ -63,6 +68,14 @@
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()
+MAX_ANNOTATION_LABELS = 400_000
+POINTCLOUD_CLASS_CODES = {1, 2, 5, 6, 15, 16}
class ApiError(ValueError):
@@ -197,9 +210,12 @@
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
- records.append(
- {
+ record = {
"id": run_id,
"label": run_id,
"note": "ChangeStar CPU 变化栅格与 GeoAI 像素坐标图斑；结果需人工复核。",
@@ -208,7 +224,15 @@
"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)
@@ -319,6 +343,209 @@
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"),
+ })
+ return sources
+
+
+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_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, device: str) -> None:
+ with POINTCLOUD_TRAINING_JOBS_LOCK:
+ POINTCLOUD_TRAINING_JOBS[job_id].update({"status": "running", "stage": "training", "startedAt": datetime.now(UTC).isoformat()})
+ python = root / ".venvs" / "05-3d-pointcloud" / "Scripts" / "python.exe"
+ command = [str(python), str(root / "capabilities" / "05-3d-pointcloud" / "train_pointcloud_semantic_model.py"), "--annotation", str(annotation), "--output", str(output), "--device", 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.")
+ 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)})
+ 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 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 execute_pointcloud_inference_job(root: Path, job_id: str, model: Path, source: Path, output: Path) -> None:
+ with POINTCLOUD_INFERENCE_JOBS_LOCK:
+ POINTCLOUD_INFERENCE_JOBS[job_id].update({"status": "running", "stage": "inference", "startedAt": datetime.now(UTC).isoformat()})
+ python = root / ".venvs" / "05-3d-pointcloud" / "Scripts" / "python.exe"
+ command = [str(python), str(root / "capabilities" / "05-3d-pointcloud" / "apply_pointcloud_semantic_model.py"), "--model", str(model), "--input", str(source), "--output", str(output), "--device", "cpu"]
+ 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.")
+ 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)})
+ 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()
@@ -363,6 +590,82 @@
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)
+ return dict(value) if value else None
+
+
+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",
+ ]
+ 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",
+ ]
+ 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]:
@@ -482,6 +785,36 @@
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/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
+ 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/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
@@ -523,6 +856,24 @@
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/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/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
@@ -535,14 +886,46 @@
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
+ 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/"):
+ 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)
@@ -554,7 +937,7 @@
def do_OPTIONS(self) -> None: # noqa: N802
self.send_response(HTTPStatus.NO_CONTENT)
- self.send_header("Allow", "GET, POST, PUT, OPTIONS")
+ self.send_header("Allow", "GET, POST, PUT, DELETE, OPTIONS")
self.end_headers()
def read_json_body(self) -> dict[str, Any]:
@@ -581,16 +964,23 @@
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"}, "change-detection")
+ 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"}, "anomaly-detection")
+ 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]
@@ -607,7 +997,7 @@
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, {".jpg", ".jpeg", ".png", ".tif", ".tiff"})
+ 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.")
@@ -643,6 +1033,16 @@
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]:
@@ -843,6 +1243,10 @@
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"),
@@ -851,6 +1255,8 @@
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")
@@ -876,11 +1282,17 @@
"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,
@@ -1155,6 +1567,270 @@
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" / "05-3d-pointcloud" / "Scripts" / "python.exe"
+ if not python.is_file():
+ raise ApiError("3D point-cloud 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.")
+ 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.")
+ if len(labels) > MAX_ANNOTATION_LABELS:
+ raise ApiError(f"An annotation revision accepts at most {MAX_ANNOTATION_LABELS} labelled points.")
+ compact: dict[int, int] = {}
+ for item in labels:
+ if not isinstance(item, list) or len(item) != 2 or not all(isinstance(value, int) for value in item):
+ raise ApiError("Each annotation label must be [pointIndex, classCode].")
+ index, code = item
+ if index < 0 or index >= int(source["pointCount"]) or code not in POINTCLOUD_CLASS_CODES:
+ 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(POINTCLOUD_CLASS_CODES)}
+ document = {
+ "schema_version": 1, "id": annotation_id, "created_at": datetime.now(UTC).isoformat(),
+ "source_id": source_id, "source_path": str(source_path.resolve()), "source_sha256": source["sha256"],
+ "source_run_id": source["runId"], "point_count": int(source["pointCount"]),
+ "labels": [[index, code] for index, code in sorted(compact.items())], "class_counts": class_counts,
+ "class_schema": {str(code): {"code": code} for code in sorted(POINTCLOUD_CLASS_CODES)},
+ "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_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.")
+ python = self.root / ".venvs" / "05-3d-pointcloud" / "Scripts" / "python.exe"
+ if not python.is_file():
+ raise ApiError("3D point-cloud virtual environment is unavailable. Run the capability setup first.")
+ job_id = uuid4().hex
+ output = self.root / "shared" / "outputs" / "05-3d-pointcloud" / "training-runs" / make_run_id("semantic-model")
+ job = {"id": job_id, "annotationId": annotation_id, "status": "queued", "stage": "queued", "device": device, "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, device), 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")
+ 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.")
+ python = self.root / ".venvs" / "05-3d-pointcloud" / "Scripts" / "python.exe"
+ if not python.is_file():
+ raise ApiError("3D point-cloud virtual environment is unavailable. Run the capability setup first.")
+ 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", "device": "cpu", "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), daemon=True, name=f"pointcloud-inference-{job_id[:8]}")
+ thread.start()
+ return dict(job)
+
+ 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)
+
+ job = {"id": job_id, "runId": run_id, "status": "queued", "stage": "queued", "createdAt": datetime.now(UTC).isoformat(), "inputImages": len(photos), "usePositionPriors": use_position_priors}
+ 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 dict(job)
+
+ 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)
--
Gitblit v1.9.3