feat: 说明修改

罗广辉

2026-08-18 2402f64ceea4d8fdf472bba83bfafefbda99e3e2

 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)