"""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 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 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
|
MAX_IMAGES_PER_RUN = 12
|
ALLOWED_PATH_PREFIXES = (
|
"apps/workbench-console",
|
"shared/outputs",
|
"shared/data/raw/01-object-detection",
|
)
|
SAFE_FILE_NAME = re.compile(r"[^A-Za-z0-9._-]+")
|
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)
|
|
|
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/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 == "/":
|
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/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
|
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_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 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 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())
|
