From 2ae460fc4a4c2419cf44329783d49a739e2a04ea Mon Sep 17 00:00:00 2001
From: shuishen <1109946754@qq.com>
Date: Wed, 19 Aug 2026 16:58:50 +0800
Subject: [PATCH] Merge branch 'master' of http://139.196.74.78:10010/r/geoai/geoai-workbench
---
scripts/serve_workbench_console.py | 908 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++
1 files changed, 903 insertions(+), 5 deletions(-)
diff --git a/scripts/serve_workbench_console.py b/scripts/serve_workbench_console.py
index 302eb37..d32a5b9 100644
--- a/scripts/serve_workbench_console.py
+++ b/scripts/serve_workbench_console.py
@@ -5,9 +5,11 @@
import argparse
import base64
import binascii
+import hashlib
import json
import os
import re
+import shutil
import subprocess
import threading
from datetime import UTC, datetime
@@ -15,22 +17,52 @@
from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path, PurePosixPath
from typing import Any
-from urllib.parse import unquote, urlsplit
+from urllib.parse import parse_qs, unquote, urlsplit
from uuid import uuid4
DEFAULT_HOST = "127.0.0.1"
DEFAULT_PORT = 6173
-MAX_REQUEST_BYTES = 128 * 1024 * 1024
-MAX_FILE_BYTES = 96 * 1024 * 1024
+# 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_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",
)
-SAFE_FILE_NAME = re.compile(r"[^A-Za-z0-9._-]+")
+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]] = {}
class ApiError(ValueError):
@@ -75,6 +107,14 @@
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 trajectory_runs(root: Path) -> list[dict[str, Any]]:
@@ -129,6 +169,278 @@
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
+ run_id = artifact.name
+ records.append(
+ {
+ "id": run_id,
+ "label": run_id,
+ "note": "ChangeStar CPU 变化栅格与 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 ""),
+ }
+ )
+ 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 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)
+
+
+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 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."""
@@ -140,11 +452,43 @@
def do_GET(self) -> None: # noqa: N802 - inherited standard-library method name
path = urlsplit(self.path).path
+ 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/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)
@@ -157,11 +501,30 @@
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/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:
@@ -172,9 +535,26 @@
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_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/"):
+ 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)
+ 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, OPTIONS")
+ self.send_header("Allow", "GET, POST, PUT, OPTIONS")
self.end_headers()
def read_json_body(self) -> dict[str, Any]:
@@ -199,6 +579,85 @@
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"}, "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")
+
+ def receive_binary_upload(
+ self,
+ path: str,
+ capability: str,
+ allowed_roles: 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, {".jpg", ".jpeg", ".png", ".tif", ".tiff"})
+ 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_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")
@@ -257,6 +716,445 @@
raise ApiError("Detection script finished without the expected result metadata.")
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")
+ 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 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
+ python = self.root / ".venvs" / "00-change-detection" / "Scripts" / "python.exe"
+ if not python.is_file():
+ raise ApiError("Change-detection virtual environment is unavailable. Run the capability setup first.")
+ with RUN_LOCK:
+ self.run_command(
+ [
+ str(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,
+ "--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"),
+ ("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 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),
+ "artifacts": {
+ "probability_raster": "change_probability.tif",
+ "raw_mask_raster": "change_mask.tif",
+ "mask_raster": "change_mask.tif",
+ "overlay": "change_overlay.jpg",
+ "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")
+ 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.")
+ 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,
+ }
+ 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),
+ 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,
+ ) -> None:
+ python = self.root / ".venvs" / "00-change-detection" / "Scripts" / "python.exe"
+ try:
+ if not python.is_file():
+ raise ApiError("Change-detection virtual environment is unavailable. Run the capability setup first.")
+ self._update_scan_job(run_id, status="running", phase="inference")
+ with RUN_LOCK:
+ self.run_command(
+ [
+ str(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,
+ "--processed-output", str(processed_root),
+ "--output", str(inference_output),
+ ],
+ SCAN_JOB_TIMEOUT,
+ )
+ inference_metadata_path = inference_output / "run_metadata.json"
+ 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 = [
+ str(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 = [
+ str(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,
+ "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 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