"""Serve the local GeoAI Workbench console and its narrow local-run APIs.""" from __future__ import annotations import argparse import base64 import binascii import json import os 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, 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 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() 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 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 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.""" 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 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) 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 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/"): 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) self.send_header("Allow", "GET, POST, 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]: 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") 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_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) 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/"): console_relative = requested.parts[2:] return os.fspath(Path(self.directory) / "apps" / "workbench-console" / "dist" / Path(*console_relative)) return os.fspath(Path(self.directory).joinpath(*requested.parts)) def end_headers(self) -> None: self.send_header("Cache-Control", "no-store") self.send_header("X-Content-Type-Options", "nosniff") super().end_headers() def parse_args() -> argparse.Namespace: root = Path(__file__).resolve().parents[1] parser = argparse.ArgumentParser(description="Serve the local GeoAI Workbench console.") parser.add_argument("--host", default=DEFAULT_HOST, help="Bind address. Defaults to loopback only.") parser.add_argument("--port", type=int, default=DEFAULT_PORT, help="TCP port in the 6000-6999 range.") parser.add_argument("--root", type=Path, default=root, help="Workbench repository root to serve.") return parser.parse_args() def main() -> int: args = parse_args() if not 6000 <= args.port <= 6999: raise SystemExit("Port must be in the 6000-6999 range.") if args.host not in {"127.0.0.1", "localhost", "::1"}: raise SystemExit("This console is local-only. Use 127.0.0.1, localhost, or ::1.") root = args.root.resolve() 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(*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("Local runs use fixed capability scripts and create a new run directory.") try: server.serve_forever() except KeyboardInterrupt: print("\nConsole stopped.") finally: server.server_close() return 0 if __name__ == "__main__": raise SystemExit(main())