| | |
| | | import argparse |
| | | import base64 |
| | | import binascii |
| | | import hashlib |
| | | import json |
| | | import os |
| | | import re |
| | | import shutil |
| | | import subprocess |
| | | import threading |
| | | from datetime import UTC, datetime |
| | |
| | | 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 |
| | | MAX_ANOMALY_IMAGES_PER_ROLE = 6 |
| | | 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"} |
| | | SCAN_DEFAULT_THRESHOLDS = [0.3, 0.4, 0.5] |
| | | SCAN_DEFAULT_AREAS = [64, 256, 686] |
| | | SCAN_MAX_THRESHOLDS = 6 |
| | | SCAN_MAX_AREAS = 6 |
| | | SCAN_MAX_COMBINATIONS = 24 |
| | | SCAN_JOB_TIMEOUT = 1800 |
| | | 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", |
| | | "shared/data/raw/09-anomaly-detection", |
| | | ) |
| | | SAFE_FILE_NAME = re.compile(r"[^A-Za-z0-9._-]+") |
| | | SAFE_FILE_NAME = re.compile(r"[^\w.-]+", re.UNICODE) |
| | | SAFE_UPLOAD_ID = re.compile(r"^[0-9a-f]{32}$") |
| | | SAFE_SCAN_ID = re.compile(r"^[A-Za-z0-9._-]{1,100}$") |
| | | SAFE_SCAN_RESULT_ID = re.compile(r"^threshold-\d+(?:\.\d+)?_area-\d+$") |
| | | RUN_LOCK = threading.Lock() |
| | | SCAN_JOBS: dict[str, dict[str, Any]] = {} |
| | | SCAN_JOBS_LOCK = threading.Lock() |
| | | ANOMALY_JOB_LOCK = threading.Lock() |
| | | ANOMALY_JOBS: dict[str, dict[str, Any]] = {} |
| | | |
| | | |
| | | class ApiError(ValueError): |
| | |
| | | except (OSError, json.JSONDecodeError): |
| | | return {} |
| | | return payload if isinstance(payload, dict) else {} |
| | | |
| | | |
| | | def file_sha256(path: Path) -> str: |
| | | digest = hashlib.sha256() |
| | | with path.open("rb") as stream: |
| | | for chunk in iter(lambda: stream.read(8 * 1024 * 1024), b""): |
| | | digest.update(chunk) |
| | | return digest.hexdigest() |
| | | |
| | | |
| | | def trajectory_runs(root: Path) -> list[dict[str, Any]]: |
| | |
| | | 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 metadata.get("kind") == "parameter-scan-inference" 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 change_parameter_scans(root: Path) -> list[dict[str, Any]]: |
| | | """Discover read-only parameter scans produced from an existing change run.""" |
| | | output_root = root / "shared" / "outputs" / "00-change-detection" |
| | | records: list[dict[str, Any]] = [] |
| | | for summary_path in output_root.rglob("scan_summary.json"): |
| | | scan_root = summary_path.parent |
| | | summary = load_json(summary_path) |
| | | results: list[dict[str, Any]] = [] |
| | | for item in summary.get("results", []): |
| | | if not isinstance(item, dict) or not isinstance(item.get("directory"), str): |
| | | continue |
| | | directory = scan_root / item["directory"] |
| | | overlay = directory / "overlay_preview.jpg" |
| | | mask = directory / "change_mask.tif" |
| | | regions = directory / "regions.json" |
| | | if not overlay.is_file() or not mask.is_file() or not regions.is_file(): |
| | | continue |
| | | results.append( |
| | | { |
| | | "id": item["directory"], |
| | | "label": f"T={float(item.get('threshold', 0.5)):.2f} / 面积={int(item.get('minimum_area_pixels', 0))} px", |
| | | "threshold": item.get("threshold"), |
| | | "minimumAreaPixels": item.get("minimum_area_pixels"), |
| | | "cleanedComponents": item.get("cleaned_components"), |
| | | "changedPixels": item.get("changed_pixels"), |
| | | "changedPixelRatio": item.get("changed_pixel_ratio"), |
| | | "vectorFeatureCount": item.get("vector_feature_count"), |
| | | "fullVectorFeatureCount": (load_json(directory / "full_result.json").get("full_vector_feature_count") if (directory / "full_result.json").is_file() else None), |
| | | "rectangleFeatureCount": ( |
| | | load_json(directory / "full_result.json").get("rectangle_vector_feature_count") |
| | | if (directory / "full_result.json").is_file() and load_json(directory / "full_result.json").get("rectangle_vector_feature_count") is not None |
| | | else len(load_json(directory / "changes_rectangles.geojson").get("features", [])) if (directory / "changes_rectangles.geojson").is_file() else None |
| | | ), |
| | | "overlay": relative_path(root, overlay), |
| | | "mask": relative_path(root, mask), |
| | | "regions": relative_path(root, regions), |
| | | "vector": (relative_path(root, directory / "changes.geojson") if (directory / "changes.geojson").is_file() else None), |
| | | "rectangleVector": (relative_path(root, directory / "changes_rectangles.geojson") if (directory / "changes_rectangles.geojson").is_file() else None), |
| | | "rectangleVectorWgs84": (relative_path(root, directory / "changes_rectangles_wgs84.geojson") if (directory / "changes_rectangles_wgs84.geojson").is_file() else None), |
| | | } |
| | | ) |
| | | # A failed job may have been repaired or materialized later. Keep it |
| | | # discoverable whenever at least one complete candidate exists; only |
| | | # hide scans that still have no usable result. |
| | | if not results: |
| | | continue |
| | | scan_metadata = load_json(scan_root / "scan_metadata.json") |
| | | contact_sheet = scan_root / "parameter_scan_contact_sheet.jpg" |
| | | records.append( |
| | | { |
| | | "id": scan_root.name, |
| | | "label": f"低成本参数扫描 · {scan_root.name}", |
| | | "note": f"复用已有变化概率结果,不重新运行 ChangeStar;源运行:{summary.get('source_run', '未知')}", |
| | | "artifactRoot": relative_path(root, scan_root), |
| | | "sourceRun": summary.get("source_run"), |
| | | "userSubmitted": bool(scan_metadata), |
| | | "contactSheet": relative_path(root, contact_sheet) if contact_sheet.is_file() else None, |
| | | "results": results, |
| | | } |
| | | ) |
| | | return sorted(records, key=lambda item: 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 anomaly_runs(root: Path) -> list[dict[str, Any]]: |
| | | output_root = root / "shared" / "outputs" / "09-anomaly-detection" |
| | | allowed_raw = (root / "shared" / "data" / "raw" / "09-anomaly-detection").resolve() |
| | | records: list[dict[str, Any]] = [] |
| | | for metadata_path in output_root.rglob("run_metadata.json"): |
| | | artifact = metadata_path.parent |
| | | metadata = load_json(metadata_path) |
| | | images = metadata.get("images") |
| | | if metadata.get("capability") != "09-anomaly-detection" or not isinstance(images, list): |
| | | continue |
| | | raw_input_value = str(metadata.get("raw_input_dir") or "") |
| | | raw_reference_value = str(metadata.get("raw_reference_dir") or "") |
| | | if not raw_input_value or not raw_reference_value: |
| | | continue |
| | | try: |
| | | raw_input = (root / raw_input_value).resolve() |
| | | raw_reference = (root / raw_reference_value).resolve() |
| | | raw_input.relative_to(allowed_raw) |
| | | raw_reference.relative_to(allowed_raw) |
| | | except ValueError: |
| | | continue |
| | | if not raw_input.is_dir() or not raw_reference.is_dir(): |
| | | continue |
| | | if any(not (artifact / str(item.get("overlay_file") or "")).is_file() for item in images if isinstance(item, dict)): |
| | | continue |
| | | run_id = artifact.name |
| | | records.append( |
| | | { |
| | | "id": run_id, |
| | | "label": str(metadata.get("display_name") or run_id), |
| | | "note": str(metadata.get("case_note") or "规则基线与 Isolation Forest 的视觉离群候选,只供人工复核。"), |
| | | "artifactRoot": relative_path(root, artifact), |
| | | "inputRoot": relative_path(root, raw_input), |
| | | "referenceRoot": relative_path(root, raw_reference), |
| | | "createdAt": str(metadata.get("created_at") or ""), |
| | | } |
| | | ) |
| | | return sorted(records, key=lambda item: (item["createdAt"], item["id"]), reverse=True) |
| | | |
| | | |
| | | def anomaly_job(job_id: str) -> dict[str, Any] | None: |
| | | with ANOMALY_JOB_LOCK: |
| | | value = ANOMALY_JOBS.get(job_id) |
| | | return dict(value) if value else None |
| | | |
| | | |
| | | def validate_anomaly_parameters(payload: dict[str, Any]) -> tuple[int, int, float, int]: |
| | | tile_size = payload.get("tileSize", 256) |
| | | stride = payload.get("stride", 128) |
| | | threshold_quantile = payload.get("thresholdQuantile", 0.995) |
| | | random_state = payload.get("randomState", 42) |
| | | if isinstance(tile_size, bool) or not isinstance(tile_size, int) or not 128 <= tile_size <= 1024: |
| | | raise ApiError("Tile size must be an integer between 128 and 1024.") |
| | | if isinstance(stride, bool) or not isinstance(stride, int) or not 32 <= stride <= tile_size: |
| | | raise ApiError("Stride must be an integer between 32 and tile size.") |
| | | if isinstance(threshold_quantile, bool) or not isinstance(threshold_quantile, (int, float)) or not 0.9 <= float(threshold_quantile) <= 0.9999: |
| | | raise ApiError("Threshold quantile must be between 0.9 and 0.9999.") |
| | | if isinstance(random_state, bool) or not isinstance(random_state, int) or not 0 <= random_state <= 2_147_483_647: |
| | | raise ApiError("Random state must be a non-negative integer.") |
| | | return tile_size, stride, float(threshold_quantile), random_state |
| | | |
| | | |
| | | def execute_anomaly_job( |
| | | root: Path, |
| | | job_id: str, |
| | | run_id: str, |
| | | raw_reference: Path, |
| | | raw_input: Path, |
| | | processed_reference: Path, |
| | | processed_input: Path, |
| | | output: Path, |
| | | tile_size: int, |
| | | stride: int, |
| | | threshold_quantile: float, |
| | | random_state: int, |
| | | ) -> None: |
| | | with ANOMALY_JOB_LOCK: |
| | | ANOMALY_JOBS[job_id]["status"] = "running" |
| | | python = root / ".venvs" / "09-anomaly-detection" / "Scripts" / "python.exe" |
| | | command = [ |
| | | str(python), |
| | | str(root / "capabilities" / "09-anomaly-detection" / "run_anomaly_detection.py"), |
| | | "--reference", str(processed_reference), |
| | | "--input", str(processed_input), |
| | | "--output", str(output), |
| | | "--tile-size", str(tile_size), |
| | | "--stride", str(stride), |
| | | "--threshold-quantile", f"{threshold_quantile:.6f}", |
| | | "--random-state", str(random_state), |
| | | "--spatial-mode", "auto", |
| | | ] |
| | | try: |
| | | with RUN_LOCK: |
| | | completed = subprocess.run(command, cwd=root, capture_output=True, text=True, timeout=1800, 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]}") |
| | | metadata_path = output / "run_metadata.json" |
| | | if not metadata_path.is_file(): |
| | | raise ApiError("Anomaly-detection script finished without the expected result metadata.") |
| | | metadata = load_json(metadata_path) |
| | | metadata["raw_input_dir"] = relative_path(root, raw_input) |
| | | metadata["raw_reference_dir"] = relative_path(root, raw_reference) |
| | | metadata["processed_input_dir"] = relative_path(root, processed_input) |
| | | metadata["processed_reference_dir"] = relative_path(root, processed_reference) |
| | | metadata_path.write_text(json.dumps(metadata, ensure_ascii=False, indent=2), encoding="utf-8") |
| | | definition = next(item for item in anomaly_runs(root) if item["id"] == run_id) |
| | | with ANOMALY_JOB_LOCK: |
| | | ANOMALY_JOBS[job_id].update({"status": "complete", "run": definition, "finishedAt": datetime.now(UTC).isoformat()}) |
| | | except Exception as exc: # pragma: no cover - background boundary |
| | | with ANOMALY_JOB_LOCK: |
| | | ANOMALY_JOBS[job_id].update({"status": "failed", "error": str(exc), "finishedAt": datetime.now(UTC).isoformat()}) |
| | | |
| | | |
| | | 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.""" |
| | | |
| | |
| | | |
| | | 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/change-detection/scans": |
| | | self.send_json(HTTPStatus.OK, {"scans": change_parameter_scans(self.root)}) |
| | | return |
| | | if path.startswith("/api/change-detection/scan-jobs/"): |
| | | job_id = path.rstrip("/").rsplit("/", 1)[-1] |
| | | with SCAN_JOBS_LOCK: |
| | | job = dict(SCAN_JOBS.get(job_id, {})) |
| | | if not job: |
| | | self.send_json(HTTPStatus.NOT_FOUND, {"error": "Unknown change-detection scan job."}) |
| | | else: |
| | | self.send_json(HTTPStatus.OK, {"job": job}) |
| | | 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 == "/api/anomaly-detection/runs": |
| | | self.send_json(HTTPStatus.OK, {"runs": anomaly_runs(self.root)}) |
| | | return |
| | | if path.startswith("/api/anomaly-detection/jobs/"): |
| | | job_id = path.rstrip("/").rsplit("/", 1)[-1] |
| | | job = anomaly_job(job_id) |
| | | self.send_json(HTTPStatus.OK if job else HTTPStatus.NOT_FOUND, {"job": job} if job else {"error": "Unknown anomaly-detection job."}) |
| | | return |
| | | if path == "/": |
| | | self.send_response(HTTPStatus.FOUND) |
| | |
| | | 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/change-detection/scans": |
| | | self.send_json(HTTPStatus.ACCEPTED, {"job": self.create_change_scan(payload)}) |
| | | return |
| | | if path.startswith("/api/change-detection/scans/") and path.endswith("/promote"): |
| | | scan_id = path.split("/")[-2] |
| | | self.send_json(HTTPStatus.CREATED, {"run": self.promote_change_scan(scan_id, 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 |
| | | if path == "/api/anomaly-detection/runs": |
| | | self.send_json(HTTPStatus.ACCEPTED, {"job": self.create_anomaly_run(payload)}) |
| | | return |
| | | self.send_json(HTTPStatus.NOT_FOUND, {"error": "Unknown local API endpoint."}) |
| | | except ApiError as exc: |
| | |
| | | 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/") and not path.startswith("/api/anomaly-detection/uploads/"): |
| | | self.send_json(HTTPStatus.NOT_FOUND, {"error": "Unknown local API endpoint."}) |
| | | return |
| | | try: |
| | | if path.startswith("/api/anomaly-detection/uploads/"): |
| | | result = self.receive_anomaly_upload(path) |
| | | else: |
| | | result = self.receive_change_upload(path) |
| | | self.send_json(HTTPStatus.CREATED, result) |
| | | 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.send_header("Allow", "GET, POST, PUT, OPTIONS") |
| | | self.end_headers() |
| | | |
| | | def read_json_body(self) -> dict[str, Any]: |
| | |
| | | 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]: |
| | | return self.receive_binary_upload(path, "00-change-detection", {"before", "after"}, "change-detection") |
| | | |
| | | def receive_anomaly_upload(self, path: str) -> dict[str, Any]: |
| | | return self.receive_binary_upload(path, "09-anomaly-detection", {"reference", "input"}, "anomaly-detection") |
| | | |
| | | def receive_binary_upload( |
| | | self, |
| | | path: str, |
| | | capability: str, |
| | | allowed_roles: set[str], |
| | | label: str, |
| | | ) -> dict[str, Any]: |
| | | upload_id = path.rstrip("/").rsplit("/", 1)[-1] |
| | | if not SAFE_UPLOAD_ID.fullmatch(upload_id): |
| | | raise ApiError(f"Invalid {label} upload id.") |
| | | query = parse_qs(urlsplit(self.path).query) |
| | | role = query.get("role", [""])[0] |
| | | if role not in allowed_roles: |
| | | raise ApiError(f"Invalid {label} upload role.") |
| | | encoded_name = self.headers.get("X-Upload-Name", "") |
| | | if len(encoded_name) > 2048: |
| | | raise ApiError("Encoded upload name is too long.") |
| | | try: |
| | | name = unquote(encoded_name, encoding="utf-8", errors="strict") |
| | | except UnicodeError as exc: |
| | | raise ApiError("Upload name is not valid UTF-8 percent encoding.") from exc |
| | | 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" / capability / "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 |
| | | digest = hashlib.sha256() |
| | | 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) |
| | | digest.update(chunk) |
| | | remaining -= len(chunk) |
| | | part.replace(target) |
| | | (staging / f"{role}.json").write_text(json.dumps({"role": role, "name": safe_name, "size": size, "sha256": digest.hexdigest()}), 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, "sha256": digest.hexdigest()} |
| | | |
| | | def resolve_change_upload(self, payload: Any, role: str) -> tuple[str, Path]: |
| | | return self.resolve_binary_upload(payload, role, "00-change-detection", "change-detection") |
| | | |
| | | def resolve_anomaly_upload(self, payload: Any, role: str) -> tuple[str, Path, str]: |
| | | name, path = self.resolve_binary_upload(payload, role, "09-anomaly-detection", "anomaly-detection") |
| | | manifest = load_json(path.parent / f"{role}.json") |
| | | return name, path, str(manifest.get("sha256") or "") |
| | | |
| | | def resolve_binary_upload(self, payload: Any, role: str, capability: str, label: str) -> tuple[str, Path]: |
| | | if not isinstance(payload, dict) or not isinstance(payload.get("uploadId"), str): |
| | | raise ApiError(f"{label} uploads must include a {role} uploadId.") |
| | | upload_id = payload["uploadId"] |
| | | if not SAFE_UPLOAD_ID.fullmatch(upload_id): |
| | | raise ApiError(f"Invalid {label} upload id.") |
| | | staging = self.root / "shared" / "data" / "raw" / capability / "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} upload is unavailable or incomplete.") |
| | | return name, path |
| | | |
| | | def create_trajectory_run(self, payload: dict[str, Any]) -> dict[str, Any]: |
| | | files = payload.get("files") |
| | |
| | | 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 _scan_parameters(self, payload: dict[str, Any]) -> tuple[list[float], list[int]]: |
| | | raw_thresholds = payload.get("thresholds", SCAN_DEFAULT_THRESHOLDS) |
| | | raw_areas = payload.get("minimumAreas", SCAN_DEFAULT_AREAS) |
| | | if not isinstance(raw_thresholds, list) or not raw_thresholds or len(raw_thresholds) > SCAN_MAX_THRESHOLDS: |
| | | raise ApiError(f"Parameter scan thresholds must contain 1-{SCAN_MAX_THRESHOLDS} values.") |
| | | if not isinstance(raw_areas, list) or not raw_areas or len(raw_areas) > SCAN_MAX_AREAS: |
| | | raise ApiError(f"Parameter scan minimum areas must contain 1-{SCAN_MAX_AREAS} values.") |
| | | thresholds: list[float] = [] |
| | | for value in raw_thresholds: |
| | | if isinstance(value, bool) or not isinstance(value, (int, float)): |
| | | raise ApiError("Each scan threshold must be a number between 0.01 and 0.99.") |
| | | number = round(float(value), 4) |
| | | if not CHANGE_THRESHOLD_MIN <= number <= CHANGE_THRESHOLD_MAX: |
| | | raise ApiError("Each scan threshold must be between 0.01 and 0.99.") |
| | | if number not in thresholds: |
| | | thresholds.append(number) |
| | | areas: list[int] = [] |
| | | for value in raw_areas: |
| | | if isinstance(value, bool) or not isinstance(value, int) or not 16 <= value <= 200000: |
| | | raise ApiError("Each scan minimum area must be an integer between 16 and 200000 pixels.") |
| | | if value not in areas: |
| | | areas.append(value) |
| | | if len(thresholds) * len(areas) > SCAN_MAX_COMBINATIONS: |
| | | raise ApiError(f"A parameter scan accepts at most {SCAN_MAX_COMBINATIONS} combinations.") |
| | | return thresholds, areas |
| | | |
| | | def _scan_root(self, scan_id: str) -> Path: |
| | | if not SAFE_SCAN_ID.fullmatch(scan_id): |
| | | raise ApiError("Invalid parameter-scan id.") |
| | | scan_root = self.root / "shared" / "outputs" / "00-change-detection" / "parameter-scans" / scan_id |
| | | if not scan_root.is_dir() or not (scan_root / "scan_summary.json").is_file(): |
| | | raise ApiError("The parameter-scan result is unavailable.") |
| | | return scan_root |
| | | |
| | | def promote_change_scan(self, scan_id: str, payload: dict[str, Any]) -> dict[str, Any]: |
| | | scan_root = self._scan_root(scan_id) |
| | | result_id = payload.get("resultId") |
| | | if not isinstance(result_id, str) or not SAFE_SCAN_RESULT_ID.fullmatch(result_id): |
| | | raise ApiError("A valid parameter-scan resultId is required.") |
| | | result_dir = scan_root / result_id |
| | | summary = load_json(result_dir / "summary.json") |
| | | full_result = load_json(result_dir / "full_result.json") |
| | | inference_run_id = str(load_json(scan_root / "scan_summary.json").get("source_run") or "") |
| | | inference_output = self.root / "shared" / "outputs" / "00-change-detection" / "runs" / inference_run_id |
| | | inference_metadata = load_json(inference_output / "run_metadata.json") |
| | | if not result_dir.is_dir() or not (result_dir / "changes.geojson").is_file() or not inference_metadata: |
| | | raise ApiError("The selected scan result is incomplete and cannot be promoted.") |
| | | run_id = make_run_id("change") |
| | | output = self.root / "shared" / "outputs" / "00-change-detection" / "runs" / run_id |
| | | output.mkdir(parents=True, exist_ok=False) |
| | | for source_name, destination_name in ( |
| | | ("change_probability.tif", "change_probability.tif"), |
| | | ("change_mask.tif", "change_mask.tif"), |
| | | ("changes.geojson", "changes.geojson"), |
| | | ("changes_rectangles.geojson", "changes_rectangles.geojson"), |
| | | ("changes_rectangles_wgs84.geojson", "changes_rectangles_wgs84.geojson"), |
| | | ): |
| | | source = result_dir / source_name if source_name.startswith("change_mask") or source_name.startswith("changes") else inference_output / source_name |
| | | if source.is_file(): |
| | | shutil.copyfile(source, output / destination_name) |
| | | overlay_source = result_dir / "overlay_preview.jpg" |
| | | if not overlay_source.is_file(): |
| | | raise ApiError("The selected scan preview is unavailable.") |
| | | shutil.copyfile(overlay_source, output / "change_overlay.jpg") |
| | | metadata = dict(inference_metadata) |
| | | scan_raw_root = self.root / "shared" / "data" / "raw" / "00-change-detection" / "runs" / scan_id |
| | | before_raw = scan_raw_root / "before" / str(inference_metadata.get("input_files", ["before.tif", "after.tif"])[0]) |
| | | after_raw = scan_raw_root / "after" / str(inference_metadata.get("input_files", ["before.tif", "after.tif"])[1]) |
| | | rectangle_count = int(full_result.get("rectangle_vector_feature_count") or 0) |
| | | if rectangle_count == 0 and (result_dir / "changes_rectangles.geojson").is_file(): |
| | | rectangle_count = len(load_json(result_dir / "changes_rectangles.geojson").get("features", [])) |
| | | metadata.update( |
| | | { |
| | | "kind": "formal-change-run", |
| | | "created_at": datetime.now(UTC).isoformat(), |
| | | "thresholds": {"change_probability": float(summary.get("threshold", 0.5)), "minimum_component_pixels": int(summary.get("minimum_area_pixels", 16))}, |
| | | "raw_changed_pixels": int(summary.get("raw_changed_pixels", 0)), |
| | | "changed_pixels": int(summary.get("changed_pixels", 0)), |
| | | "changed_pixel_ratio": float(summary.get("changed_pixel_ratio", 0)), |
| | | "vector_feature_count": int(full_result.get("full_vector_feature_count", summary.get("vector_feature_count", 0))), |
| | | "rectangle_feature_count": rectangle_count, |
| | | "promoted_from_scan": scan_id, |
| | | "promoted_result": result_id, |
| | | "raw_input_dir": relative_path(self.root, scan_raw_root), |
| | | "raw_before": relative_path(self.root, before_raw), |
| | | "raw_after": relative_path(self.root, after_raw), |
| | | "artifacts": { |
| | | "probability_raster": "change_probability.tif", |
| | | "raw_mask_raster": "change_mask.tif", |
| | | "mask_raster": "change_mask.tif", |
| | | "overlay": "change_overlay.jpg", |
| | | "vector": "changes.geojson", |
| | | "rectangle_vector": "changes_rectangles.geojson", |
| | | "rectangle_vector_wgs84": "changes_rectangles_wgs84.geojson" if (output / "changes_rectangles_wgs84.geojson").is_file() else None, |
| | | "features": "full_result.json", |
| | | }, |
| | | } |
| | | ) |
| | | (output / "full_result.json").write_text(json.dumps(full_result, ensure_ascii=False, indent=2), encoding="utf-8") |
| | | (output / "run_metadata.json").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_change_scan(self, payload: dict[str, Any]) -> dict[str, Any]: |
| | | uploads = payload.get("uploads") |
| | | if not isinstance(uploads, dict): |
| | | raise ApiError("Parameter scan must contain staged before and after uploads.") |
| | | staged = { |
| | | "before": self.resolve_change_upload(uploads.get("before"), "before"), |
| | | "after": self.resolve_change_upload(uploads.get("after"), "after"), |
| | | } |
| | | thresholds, areas = self._scan_parameters(payload) |
| | | 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.") |
| | | run_id = make_run_id("scan") |
| | | raw_root = self.root / "shared" / "data" / "raw" / "00-change-detection" / "runs" / run_id |
| | | before_path = raw_root / "before" / staged["before"][0] |
| | | after_path = raw_root / "after" / staged["after"][0] |
| | | before_path.parent.mkdir(parents=True, exist_ok=False) |
| | | after_path.parent.mkdir(parents=True, exist_ok=False) |
| | | shutil.copyfile(staged["before"][1], before_path) |
| | | shutil.copyfile(staged["after"][1], after_path) |
| | | processed_root = self.root / "shared" / "data" / "processed" / "00-change-detection" / run_id |
| | | inference_output = self.root / "shared" / "outputs" / "00-change-detection" / "runs" / run_id |
| | | scan_output = self.root / "shared" / "outputs" / "00-change-detection" / "parameter-scans" / run_id |
| | | with SCAN_JOBS_LOCK: |
| | | SCAN_JOBS[run_id] = { |
| | | "id": run_id, |
| | | "status": "queued", |
| | | "createdAt": datetime.now(UTC).isoformat(), |
| | | "thresholds": thresholds, |
| | | "minimumAreas": areas, |
| | | "processingMode": processing_mode, |
| | | "maxDimension": max_dimension, |
| | | } |
| | | thread = threading.Thread( |
| | | target=self._run_change_scan, |
| | | args=(run_id, before_path, after_path, processed_root, inference_output, scan_output, thresholds, areas, processing_mode, max_dimension), |
| | | daemon=True, |
| | | name=f"change-scan-{run_id}", |
| | | ) |
| | | thread.start() |
| | | return dict(SCAN_JOBS[run_id]) |
| | | |
| | | def _update_scan_job(self, job_id: str, **values: Any) -> None: |
| | | with SCAN_JOBS_LOCK: |
| | | if job_id in SCAN_JOBS: |
| | | SCAN_JOBS[job_id].update(values) |
| | | |
| | | def _run_change_scan( |
| | | self, |
| | | run_id: str, |
| | | before_path: Path, |
| | | after_path: Path, |
| | | processed_root: Path, |
| | | inference_output: Path, |
| | | scan_output: Path, |
| | | thresholds: list[float], |
| | | areas: list[int], |
| | | processing_mode: str, |
| | | max_dimension: int, |
| | | ) -> None: |
| | | python = self.root / ".venvs" / "00-change-detection" / "Scripts" / "python.exe" |
| | | try: |
| | | if not python.is_file(): |
| | | raise ApiError("Change-detection virtual environment is unavailable. Run the capability setup first.") |
| | | self._update_scan_job(run_id, status="running", phase="inference") |
| | | 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", "0.5000", |
| | | "--max-dimension", str(max_dimension), |
| | | "--processing-mode", processing_mode, |
| | | "--processed-output", str(processed_root), |
| | | "--output", str(inference_output), |
| | | ], |
| | | SCAN_JOB_TIMEOUT, |
| | | ) |
| | | inference_metadata_path = inference_output / "run_metadata.json" |
| | | inference_metadata = load_json(inference_metadata_path) |
| | | inference_metadata["kind"] = "parameter-scan-inference" |
| | | inference_metadata["scan_job_id"] = run_id |
| | | inference_metadata_path.write_text(json.dumps(inference_metadata, ensure_ascii=False, indent=2), encoding="utf-8") |
| | | self._update_scan_job(run_id, phase="parameter-scan") |
| | | command = [ |
| | | str(python), |
| | | str(self.root / "capabilities" / "00-change-detection" / "scan_change_detection_parameters.py"), |
| | | "--run-dir", str(inference_output), |
| | | "--output", str(scan_output), |
| | | ] |
| | | for threshold in thresholds: |
| | | command.extend(["--threshold", f"{threshold:.4f}"]) |
| | | for area in areas: |
| | | command.extend(["--minimum-area", str(area)]) |
| | | self.run_command(command, SCAN_JOB_TIMEOUT) |
| | | self._update_scan_job(run_id, phase="vectorization") |
| | | vector_command = [ |
| | | str(python), |
| | | str(self.root / "capabilities" / "00-change-detection" / "materialize_parameter_scan_candidates.py"), |
| | | "--scan-dir", str(scan_output), |
| | | ] |
| | | for threshold in thresholds: |
| | | for area in areas: |
| | | vector_command.extend(["--candidate", f"threshold-{threshold:.2f}_area-{area}"]) |
| | | self.run_command(vector_command, SCAN_JOB_TIMEOUT) |
| | | metadata = { |
| | | "capability": "00-change-detection", |
| | | "kind": "parameter-scan", |
| | | "source_run": run_id, |
| | | "created_at": datetime.now(UTC).isoformat(), |
| | | "thresholds": thresholds, |
| | | "minimum_areas": areas, |
| | | "processing_mode": processing_mode, |
| | | "max_dimension": max_dimension, |
| | | "raw_input_dir": relative_path(self.root, before_path.parent.parent), |
| | | "processed_input_dir": relative_path(self.root, processed_root), |
| | | } |
| | | scan_output.mkdir(parents=True, exist_ok=True) |
| | | (scan_output / "scan_metadata.json").write_text(json.dumps(metadata, ensure_ascii=False, indent=2), encoding="utf-8") |
| | | self._update_scan_job(run_id, status="completed", phase="done", scanId=run_id) |
| | | except Exception as exc: # background errors are returned through polling |
| | | scan_output.mkdir(parents=True, exist_ok=True) |
| | | (scan_output / "scan_failed.json").write_text(json.dumps({"job_id": run_id, "error": str(exc)[:600]}, ensure_ascii=False, indent=2), encoding="utf-8") |
| | | self._update_scan_job(run_id, status="failed", phase="error", error=str(exc)[:600]) |
| | | |
| | | 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_anomaly_run(self, payload: dict[str, Any]) -> dict[str, Any]: |
| | | tile_size, stride, threshold_quantile, random_state = validate_anomaly_parameters(payload) |
| | | uploads = payload.get("uploads") |
| | | if not isinstance(uploads, dict): |
| | | raise ApiError("Anomaly-detection request must contain reference and input uploads.") |
| | | reference_values = uploads.get("reference") |
| | | input_values = uploads.get("input") |
| | | if not isinstance(reference_values, list) or not reference_values: |
| | | raise ApiError("Select at least one normal reference image.") |
| | | if not isinstance(input_values, list) or not input_values: |
| | | raise ApiError("Select at least one image to inspect.") |
| | | if len(reference_values) > MAX_ANOMALY_IMAGES_PER_ROLE or len(input_values) > MAX_ANOMALY_IMAGES_PER_ROLE: |
| | | raise ApiError(f"An anomaly-detection run accepts at most {MAX_ANOMALY_IMAGES_PER_ROLE} images in each group.") |
| | | references = [self.resolve_anomaly_upload(value, "reference") for value in reference_values] |
| | | inputs = [self.resolve_anomaly_upload(value, "input") for value in input_values] |
| | | if len({name.casefold() for name, _, _ in references}) != len(references): |
| | | raise ApiError("Normal reference image names must be unique within one run.") |
| | | if len({name.casefold() for name, _, _ in inputs}) != len(inputs): |
| | | raise ApiError("Input image names must be unique within one run.") |
| | | |
| | | python = self.root / ".venvs" / "09-anomaly-detection" / "Scripts" / "python.exe" |
| | | if not python.is_file(): |
| | | raise ApiError("Anomaly-detection virtual environment is unavailable. Run the capability setup first.") |
| | | run_id = make_run_id("anomaly") |
| | | job_id = uuid4().hex |
| | | raw_root = self.root / "shared" / "data" / "raw" / "09-anomaly-detection" / "runs" / run_id |
| | | raw_reference = raw_root / "reference" |
| | | raw_input = raw_root / "input" |
| | | processed_root = self.root / "shared" / "data" / "processed" / "09-anomaly-detection" / run_id |
| | | processed_reference = processed_root / "reference" |
| | | processed_input = processed_root / "input" |
| | | output = self.root / "shared" / "outputs" / "09-anomaly-detection" / "runs" / run_id |
| | | for directory in (raw_reference, raw_input, processed_reference, processed_input): |
| | | directory.mkdir(parents=True, exist_ok=False) |
| | | for group, raw_dir, processed_dir in ((references, raw_reference, processed_reference), (inputs, raw_input, processed_input)): |
| | | for name, staged_path, expected_sha256 in group: |
| | | raw_path = raw_dir / name |
| | | processed_path = processed_dir / name |
| | | shutil.copyfile(staged_path, raw_path) |
| | | if expected_sha256 and file_sha256(raw_path) != expected_sha256: |
| | | raise ApiError(f"Uploaded file checksum changed while staging: {name}.") |
| | | shutil.copyfile(raw_path, processed_path) |
| | | for _, staged_path, _ in references + inputs: |
| | | shutil.rmtree(staged_path.parent) |
| | | |
| | | created_at = datetime.now(UTC).isoformat() |
| | | job = {"id": job_id, "runId": run_id, "status": "queued", "createdAt": created_at} |
| | | with ANOMALY_JOB_LOCK: |
| | | ANOMALY_JOBS[job_id] = job |
| | | thread = threading.Thread( |
| | | target=execute_anomaly_job, |
| | | args=(self.root, job_id, run_id, raw_reference, raw_input, processed_reference, processed_input, output, tile_size, stride, threshold_quantile, random_state), |
| | | daemon=True, |
| | | name=f"anomaly-{run_id}", |
| | | ) |
| | | thread.start() |
| | | return dict(job) |
| | | |
| | | 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) |