From 7cc239cee1a9af4e2e8a0f3d5b7a00a074b17214 Mon Sep 17 00:00:00 2001
From: 罗广辉 <guanghui.luo@foxmail.com>
Date: Wed, 19 Aug 2026 15:43:13 +0800
Subject: [PATCH] feat: 复用检测
---
scripts/serve_workbench_console.py | 829 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++-
1 files changed, 814 insertions(+), 15 deletions(-)
diff --git a/scripts/serve_workbench_console.py b/scripts/serve_workbench_console.py
index 83126a9..b0e1b0b 100644
--- a/scripts/serve_workbench_console.py
+++ b/scripts/serve_workbench_console.py
@@ -1,44 +1,846 @@
-"""Serve the local GeoAI Workbench console from the repository root."""
+"""Serve the local GeoAI Workbench console and its narrow local-run APIs."""
from __future__ import annotations
import argparse
+import base64
+import binascii
+import hashlib
+import json
import os
-from pathlib import PurePosixPath
+import re
+import shutil
+import subprocess
+import threading
+from datetime import UTC, datetime
from http import HTTPStatus
from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
-from pathlib import Path
-from urllib.parse import unquote, urlsplit
+from pathlib import Path, PurePosixPath
+from typing import Any
+from urllib.parse import parse_qs, unquote, urlsplit
+from uuid import uuid4
DEFAULT_HOST = "127.0.0.1"
DEFAULT_PORT = 6173
+# Uploads are sent as Base64 JSON. Keep the request limit above two 1 GiB
+# files after encoding while retaining a per-file bound for local experiments.
+MAX_REQUEST_BYTES = 3072 * 1024 * 1024
+MAX_FILE_BYTES = 1024 * 1024 * 1024
+MAX_IMAGES_PER_RUN = 12
+MAX_SEGMENTATION_IMAGES_PER_RUN = 6
+MAX_MEASUREMENT_RASTERS_PER_RUN = 4
+MAX_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"}
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"[^\w.-]+", re.UNICODE)
+SAFE_UPLOAD_ID = re.compile(r"^[0-9a-f]{32}$")
+RUN_LOCK = threading.Lock()
+ANOMALY_JOB_LOCK = threading.Lock()
+ANOMALY_JOBS: dict[str, dict[str, Any]] = {}
+
+
+class ApiError(ValueError):
+ """A request error that can be shown to the local console user."""
+
+
+def safe_file_name(value: str, expected_suffixes: set[str]) -> str:
+ name = Path(value).name
+ suffix = Path(name).suffix.lower()
+ if suffix not in expected_suffixes:
+ raise ApiError(f"Unsupported file type: {suffix or '(none)'}.")
+ stem = SAFE_FILE_NAME.sub("_", Path(name).stem).strip("._") or "upload"
+ return f"{stem[:80]}{suffix}"
+
+
+def decode_upload(payload: dict[str, Any], expected_suffixes: set[str]) -> tuple[str, bytes]:
+ if not isinstance(payload, dict) or not isinstance(payload.get("name"), str) or not isinstance(payload.get("content"), str):
+ raise ApiError("Each uploaded file must include name and Base64 content.")
+ name = safe_file_name(payload["name"], expected_suffixes)
+ try:
+ content = base64.b64decode(payload["content"], validate=True)
+ except (binascii.Error, ValueError) as exc:
+ raise ApiError(f"Invalid Base64 file content for {name}.") from exc
+ if not content:
+ raise ApiError(f"Uploaded file is empty: {name}.")
+ if len(content) > MAX_FILE_BYTES:
+ raise ApiError(f"Uploaded file exceeds {MAX_FILE_BYTES // (1024 * 1024)} MB: {name}.")
+ return name, content
+
+
+def make_run_id(prefix: str) -> str:
+ return f"{prefix}-{datetime.now(UTC):%Y%m%d-%H%M%S}-{uuid4().hex[:6]}"
+
+
+def relative_path(root: Path, path: Path) -> str:
+ return path.relative_to(root).as_posix()
+
+
+def load_json(path: Path) -> dict[str, Any]:
+ try:
+ payload = json.loads(path.read_text(encoding="utf-8"))
+ except (OSError, json.JSONDecodeError):
+ return {}
+ return payload if isinstance(payload, dict) else {}
+
+
+def file_sha256(path: Path) -> str:
+ digest = hashlib.sha256()
+ with path.open("rb") as stream:
+ for chunk in iter(lambda: stream.read(8 * 1024 * 1024), b""):
+ digest.update(chunk)
+ return digest.hexdigest()
+
+
+def trajectory_runs(root: Path) -> list[dict[str, Any]]:
+ output_root = root / "shared" / "outputs" / "15-trajectory-analysis"
+ records: list[dict[str, Any]] = []
+ for metadata_path in output_root.rglob("run_metadata.json"):
+ artifact = metadata_path.parent
+ if not (artifact / "trajectory_summary.csv").is_file() or not (artifact / "events.json").is_file():
+ continue
+ metadata = load_json(metadata_path)
+ case_id = str(metadata.get("case_id") or artifact.name)
+ is_real = case_id == "tian-dun-flight-19578"
+ records.append(
+ {
+ "id": case_id,
+ "label": "田墩实飞" if is_real else case_id,
+ "note": "区域相交是来源数据的空间结果，不是违规结论。" if is_real else "本地实验运行结果，可继续查看结构化输出。",
+ "artifactRoot": relative_path(root, artifact),
+ "showSpatialContext": (artifact / "zones.geojson").is_file() and (artifact / "reference_routes.geojson").is_file(),
+ "showFlyableZones": (artifact / "flyable_zones.geojson").is_file(),
+ "createdAt": str(metadata.get("created_at") or ""),
+ }
+ )
+ return sorted(records, key=lambda item: (item["createdAt"], item["id"]), reverse=True)
+
+
+def detection_runs(root: Path) -> list[dict[str, Any]]:
+ output_root = root / "shared" / "outputs" / "01-object-detection"
+ records: list[dict[str, Any]] = []
+ for metadata_path in output_root.rglob("run_metadata.json"):
+ artifact = metadata_path.parent
+ if not (artifact / "detections.json").is_file():
+ continue
+ metadata = load_json(metadata_path)
+ input_value = str(metadata.get("input_dir") or "")
+ input_dir = Path(input_value) if input_value else root / "shared" / "data" / "raw" / "01-object-detection"
+ try:
+ input_root = relative_path(root, input_dir.resolve())
+ except ValueError:
+ continue
+ run_id = artifact.name if artifact != output_root else "baseline"
+ records.append(
+ {
+ "id": run_id,
+ "label": "既有基线结果" if run_id == "baseline" else run_id,
+ "note": "CPU 基线：人员与常见车辆；树木不在当前模型有效类别内。",
+ "artifactRoot": relative_path(root, artifact),
+ "inputRoot": input_root,
+ "createdAt": str(metadata.get("created_at") or ""),
+ }
+ )
+ return sorted(records, key=lambda item: (item["createdAt"], item["id"]), reverse=True)
+
+
+def change_runs(root: Path) -> list[dict[str, Any]]:
+ output_root = root / "shared" / "outputs" / "00-change-detection"
+ records: list[dict[str, Any]] = []
+ for metadata_path in output_root.rglob("run_metadata.json"):
+ artifact = metadata_path.parent
+ metadata = load_json(metadata_path)
+ artifacts = metadata.get("artifacts")
+ if metadata.get("capability") != "00-change-detection" or metadata.get("schema_version") != 1 or 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 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):
- """Read-only static handler rooted at the workbench repository."""
+ """Static UI plus fixed, local-only ingestion and experiment commands."""
+
+ server_version = "GeoAIWorkbench/1.0"
+
+ @property
+ def root(self) -> Path:
+ return Path(self.directory).resolve()
def do_GET(self) -> None: # noqa: N802 - inherited standard-library method name
- if urlsplit(self.path).path == "/":
+ 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/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)
self.send_header("Location", "/apps/workbench-console/")
self.end_headers()
return
super().do_GET()
+ def do_POST(self) -> None: # noqa: N802 - inherited standard-library method name
+ path = urlsplit(self.path).path
+ try:
+ payload = self.read_json_body()
+ if path == "/api/change-detection/runs":
+ self.send_json(HTTPStatus.CREATED, {"run": self.create_change_run(payload)})
+ return
+ if path == "/api/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:
+ self.send_json(HTTPStatus.BAD_REQUEST, {"error": str(exc)})
+ except subprocess.TimeoutExpired:
+ self.send_json(HTTPStatus.GATEWAY_TIMEOUT, {"error": "The local run exceeded its time limit; no existing result was overwritten."})
+ except Exception as exc: # pragma: no cover - defensive server boundary
+ self.log_error("local run failed: %s", exc)
+ self.send_json(HTTPStatus.INTERNAL_SERVER_ERROR, {"error": "Local run failed. Check the console terminal for details."})
+
+ def do_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, PUT, OPTIONS")
+ self.end_headers()
+
+ def read_json_body(self) -> dict[str, Any]:
+ content_length = self.headers.get("Content-Length")
+ if content_length is None or not content_length.isdigit():
+ raise ApiError("A JSON request body with Content-Length is required.")
+ size = int(content_length)
+ if size <= 0 or size > MAX_REQUEST_BYTES:
+ raise ApiError(f"Request must be between 1 byte and {MAX_REQUEST_BYTES // (1024 * 1024)} MB.")
+ if "application/json" not in self.headers.get("Content-Type", ""):
+ raise ApiError("Content-Type must be application/json.")
+ try:
+ payload = json.loads(self.rfile.read(size).decode("utf-8"))
+ except (UnicodeDecodeError, json.JSONDecodeError) as exc:
+ raise ApiError("Request body is not valid UTF-8 JSON.") from exc
+ if not isinstance(payload, dict):
+ raise ApiError("JSON request body must be an object.")
+ return payload
+
+ def run_command(self, command: list[str], timeout: int) -> None:
+ completed = subprocess.run(command, cwd=self.root, capture_output=True, text=True, timeout=timeout, check=False)
+ if completed.returncode:
+ message = (completed.stderr or completed.stdout or "Unknown script error.").strip().splitlines()[-1]
+ raise ApiError(f"Processing failed: {message[:600]}")
+
+ def receive_change_upload(self, path: str) -> dict[str, Any]:
+ return self.receive_binary_upload(path, "00-change-detection", {"before", "after"}, "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")
+ if not isinstance(files, dict):
+ raise ApiError("Trajectory request must contain a files object.")
+ required = {
+ "flight": {".xlsx"},
+ "route": {".kmz"},
+ "restricted": {".geojson"},
+ }
+ decoded = {key: decode_upload(files.get(key), suffixes) for key, suffixes in required.items()}
+ flyable = decode_upload(files["flyable"], {".gzip"}) if files.get("flyable") else None
+ run_id = make_run_id("trajectory")
+ raw_root = self.root / "shared" / "data" / "raw" / "15-trajectory-analysis" / "runs" / run_id
+ paths = {"flight": raw_root / "tracks" / decoded["flight"][0], "route": raw_root / "routes" / decoded["route"][0], "restricted": raw_root / "areas" / decoded["restricted"][0]}
+ for key, path in paths.items():
+ path.parent.mkdir(parents=True, exist_ok=True)
+ path.write_bytes(decoded[key][1])
+ if flyable:
+ flyable_path = raw_root / "areas" / flyable[0]
+ flyable_path.write_bytes(flyable[1])
+ processed = self.root / "shared" / "data" / "processed" / "15-trajectory-analysis" / run_id
+ output_parent = self.root / "shared" / "outputs" / "15-trajectory-analysis" / "runs" / run_id
+ python = self.root / ".venvs" / "15-trajectory-analysis" / "Scripts" / "python.exe"
+ if not python.is_file():
+ raise ApiError("Trajectory virtual environment is unavailable. Run the capability setup first.")
+ with RUN_LOCK:
+ self.run_command([str(python), str(self.root / "capabilities" / "15-trajectory-analysis" / "prepare_real_flight.py"), "--raw-dir", str(raw_root), "--output", str(processed), "--case-id", run_id], 300)
+ self.run_command([str(python), str(self.root / "capabilities" / "15-trajectory-analysis" / "run_trajectory_analysis.py"), "--input", str(processed / f"{run_id}.case.json"), "--output", str(output_parent)], 300)
+ artifact = output_parent / run_id
+ if not (artifact / "run_metadata.json").is_file():
+ raise ApiError("Trajectory script finished without the expected result metadata.")
+ return next(item for item in trajectory_runs(self.root) if item["id"] == run_id)
+
+ def create_detection_run(self, payload: dict[str, Any]) -> dict[str, Any]:
+ uploads = payload.get("images")
+ if not isinstance(uploads, list) or not uploads:
+ raise ApiError("Object-detection request must include at least one image.")
+ if len(uploads) > MAX_IMAGES_PER_RUN:
+ raise ApiError(f"A local run accepts at most {MAX_IMAGES_PER_RUN} images.")
+ decoded = [decode_upload(item, {".jpg", ".jpeg", ".png"}) for item in uploads]
+ if len({name.casefold() for name, _ in decoded}) != len(decoded):
+ raise ApiError("Uploaded image names must be unique within one run.")
+ run_id = make_run_id("detection")
+ raw_root = self.root / "shared" / "data" / "raw" / "01-object-detection" / "runs" / run_id
+ raw_root.mkdir(parents=True, exist_ok=False)
+ for name, content in decoded:
+ (raw_root / name).write_bytes(content)
+ output = self.root / "shared" / "outputs" / "01-object-detection" / "runs" / run_id
+ python = self.root / ".venvs" / "01-object-detection" / "Scripts" / "python.exe"
+ if not python.is_file():
+ raise ApiError("Object-detection virtual environment is unavailable. Run the capability setup first.")
+ with RUN_LOCK:
+ self.run_command([str(python), str(self.root / "capabilities" / "01-object-detection" / "run_detection.py"), "--input", str(raw_root), "--output", str(output)], 1200)
+ if not (output / "run_metadata.json").is_file():
+ 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 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)
+ self.send_header("Content-Type", "application/json; charset=utf-8")
+ self.send_header("Content-Length", str(len(body)))
+ self.end_headers()
+ self.wfile.write(body)
+
def translate_path(self, path: str) -> str:
"""Expose only the static UI and artifacts required by the local console."""
decoded_path = unquote(urlsplit(path).path).lstrip("/")
requested = PurePosixPath(decoded_path)
- is_allowed = any(
- decoded_path == prefix or decoded_path.startswith(f"{prefix}/")
- for prefix in ALLOWED_PATH_PREFIXES
- )
+ is_allowed = any(decoded_path == prefix or decoded_path.startswith(f"{prefix}/") for prefix in ALLOWED_PATH_PREFIXES)
if ".." in requested.parts or not is_allowed:
return os.fspath(Path(self.directory) / ".console-forbidden")
if decoded_path == "apps/workbench-console" or decoded_path.startswith("apps/workbench-console/"):
@@ -71,13 +873,10 @@
app_dir = root / "apps" / "workbench-console"
if not (root / "shared").is_dir() or not (app_dir / "dist" / "index.html").is_file():
raise SystemExit(f"Not a GeoAI Workbench root: {root}")
-
- handler = lambda *handler_args, **handler_kwargs: WorkbenchConsoleHandler( # noqa: E731
- *handler_args, directory=os.fspath(root), **handler_kwargs
- )
+ handler = lambda *handler_args, **handler_kwargs: WorkbenchConsoleHandler(*handler_args, directory=os.fspath(root), **handler_kwargs) # noqa: E731
server = ThreadingHTTPServer((args.host, args.port), handler)
print(f"GeoAI Workbench console: http://{args.host}:{args.port}")
- print(f"Serving built console and read-only artifacts from: {root}")
+ print("Local runs use fixed capability scripts and create a new run directory.")
try:
server.serve_forever()
except KeyboardInterrupt:
--
Gitblit v1.9.3