From 2402f64ceea4d8fdf472bba83bfafefbda99e3e2 Mon Sep 17 00:00:00 2001
From: 罗广辉 <guanghui.luo@foxmail.com>
Date: Tue, 18 Aug 2026 18:06:25 +0800
Subject: [PATCH] feat: 说明修改
---
scripts/serve_workbench_console.py | 349 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++
1 files changed, 346 insertions(+), 3 deletions(-)
diff --git a/scripts/serve_workbench_console.py b/scripts/serve_workbench_console.py
index 302eb37..e6fd1b1 100644
--- a/scripts/serve_workbench_console.py
+++ b/scripts/serve_workbench_console.py
@@ -8,6 +8,7 @@
import json
import os
import re
+import shutil
import subprocess
import threading
from datetime import UTC, datetime
@@ -15,21 +16,37 @@
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
+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",
)
SAFE_FILE_NAME = re.compile(r"[^A-Za-z0-9._-]+")
+SAFE_UPLOAD_ID = re.compile(r"^[0-9a-f]{32}$")
RUN_LOCK = threading.Lock()
@@ -129,6 +146,101 @@
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 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 +252,23 @@
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/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 == "/":
self.send_response(HTTPStatus.FOUND)
@@ -157,11 +281,20 @@
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
self.send_json(HTTPStatus.NOT_FOUND, {"error": "Unknown local API endpoint."})
except ApiError as exc:
@@ -171,6 +304,19 @@
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/"):
+ self.send_json(HTTPStatus.NOT_FOUND, {"error": "Unknown local API endpoint."})
+ return
+ try:
+ self.send_json(HTTPStatus.CREATED, self.receive_change_upload(path))
+ 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)
@@ -199,6 +345,57 @@
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]:
+ upload_id = path.rstrip("/").rsplit("/", 1)[-1]
+ if not SAFE_UPLOAD_ID.fullmatch(upload_id):
+ raise ApiError("Invalid change-detection 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.")
+ name = self.headers.get("X-Upload-Name", "")
+ 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" / "00-change-detection" / "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
+ 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)
+ remaining -= len(chunk)
+ part.replace(target)
+ (staging / f"{role}.json").write_text(json.dumps({"role": role, "name": safe_name, "size": size}), 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}
+
+ def resolve_change_upload(self, payload: Any, role: 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.")
+ 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
+ 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.")
+ return name, path
def create_trajectory_run(self, payload: dict[str, Any]) -> dict[str, Any]:
files = payload.get("files")
@@ -257,6 +454,152 @@
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_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