From 91e2fe47a57cd39b612d54177b548e4ec3432783 Mon Sep 17 00:00:00 2001
From: shuishen <1109946754@qq.com>
Date: Mon, 17 Aug 2026 14:50:48 +0800
Subject: [PATCH] feat:语义分割demo处理
---
scripts/serve_workbench_console.py | 82 +++++++++++++++++++++++++++++++++++++++++
1 files changed, 82 insertions(+), 0 deletions(-)
diff --git a/scripts/serve_workbench_console.py b/scripts/serve_workbench_console.py
index 302eb37..08b333e 100644
--- a/scripts/serve_workbench_console.py
+++ b/scripts/serve_workbench_console.py
@@ -24,10 +24,12 @@
MAX_REQUEST_BYTES = 128 * 1024 * 1024
MAX_FILE_BYTES = 96 * 1024 * 1024
MAX_IMAGES_PER_RUN = 12
+MAX_SEGMENTATION_IMAGES_PER_RUN = 6
ALLOWED_PATH_PREFIXES = (
"apps/workbench-console",
"shared/outputs",
"shared/data/raw/01-object-detection",
+ "shared/data/raw/02-semantic-mapping",
)
SAFE_FILE_NAME = re.compile(r"[^A-Za-z0-9._-]+")
RUN_LOCK = threading.Lock()
@@ -129,6 +131,37 @@
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 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."""
@@ -146,6 +179,12 @@
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 == "/":
self.send_response(HTTPStatus.FOUND)
self.send_header("Location", "/apps/workbench-console/")
@@ -162,6 +201,9 @@
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
self.send_json(HTTPStatus.NOT_FOUND, {"error": "Unknown local API endpoint."})
except ApiError as exc:
@@ -257,6 +299,46 @@
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_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 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