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 | 258 ++++++++++++++++++++++++++++++++++++++++++++++++---
1 files changed, 241 insertions(+), 17 deletions(-)
diff --git a/scripts/serve_workbench_console.py b/scripts/serve_workbench_console.py
index 9219509..d32a5b9 100644
--- a/scripts/serve_workbench_console.py
+++ b/scripts/serve_workbench_console.py
@@ -5,6 +5,7 @@
import argparse
import base64
import binascii
+import hashlib
import json
import os
import re
@@ -29,6 +30,7 @@
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
@@ -50,14 +52,17 @@
"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):
@@ -102,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]]:
@@ -306,6 +319,120 @@
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")
@@ -355,6 +482,14 @@
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/")
@@ -388,6 +523,9 @@
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)})
@@ -399,11 +537,15 @@
def do_PUT(self) -> None: # noqa: N802 - binary upload endpoint
path = urlsplit(self.path).path
- if not path.startswith("/api/change-detection/uploads/"):
+ 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:
- self.send_json(HTTPStatus.CREATED, self.receive_change_upload(path))
+ 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
@@ -412,7 +554,7 @@
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]:
@@ -439,18 +581,32 @@
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("Invalid change-detection 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 {"before", "after"}:
- raise ApiError("Change-detection upload role must be before or after.")
+ 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)
- except Exception as exc:
- raise ApiError("The uploaded filename is invalid.") from exc
+ 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():
@@ -458,11 +614,12 @@
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" / "00-change-detection" / "uploads" / upload_id
+ 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:
@@ -470,27 +627,36 @@
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}), encoding="utf-8")
+ (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}
+ 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"Change-detection uploads must include a {role} uploadId.")
+ raise ApiError(f"{label} uploads must include a {role} uploadId.")
upload_id = payload["uploadId"]
if not SAFE_UPLOAD_ID.fullmatch(upload_id):
- raise ApiError("Invalid change-detection upload id.")
- staging = self.root / "shared" / "data" / "raw" / "00-change-detection" / "uploads" / 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} TIFF upload is unavailable or incomplete.")
+ 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]:
@@ -899,6 +1065,64 @@
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:
--
Gitblit v1.9.3