feat:语义分割demo处理

shuishen

2 days ago 91e2fe47a57cd39b612d54177b548e4ec3432783

 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)