418 lines
16 KiB
Python
418 lines
16 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
import re
|
|
from collections.abc import Callable
|
|
from pathlib import Path
|
|
from typing import Any, Final
|
|
|
|
from fastapi import APIRouter, HTTPException, Query, Response
|
|
|
|
from k1link.compute import (
|
|
LidarFieldReviewV1,
|
|
LidarGroundBenchmarkV1,
|
|
LidarGroundError,
|
|
LidarReplayError,
|
|
LidarReplayPackV2,
|
|
lidar_field_review_catalog_item,
|
|
lidar_ground_benchmark_catalog_item,
|
|
lidar_ground_frame_detail,
|
|
lidar_pack_catalog_item,
|
|
lidar_pack_detail,
|
|
)
|
|
|
|
LIDAR_CATALOG_SCHEMA: Final = "missioncore.lidar-replay-pack-catalog/v1"
|
|
LIDAR_GROUND_CATALOG_SCHEMA: Final = "missioncore.lidar-ground-benchmark-catalog/v1"
|
|
LIDAR_FIELD_REVIEW_CATALOG_SCHEMA: Final = "missioncore.lidar-field-review-catalog/v1"
|
|
_PACK_ID = re.compile(r"^lidar-replay-pack-[a-f0-9]{64}$")
|
|
_BENCHMARK_ID = re.compile(r"^ground-benchmark-[a-f0-9]{64}$")
|
|
_FIELD_REVIEW_ID = re.compile(r"^lidar-field-review-[a-f0-9]{64}$")
|
|
RootProvider = Callable[[], Path | None]
|
|
|
|
|
|
def configured_lidar_replay_root() -> Path | None:
|
|
value = os.environ.get("MISSIONCORE_LIDAR_REPLAY_ROOT", "").strip()
|
|
return Path(value).expanduser().absolute() if value else None
|
|
|
|
|
|
def configured_lidar_ground_root() -> Path | None:
|
|
value = os.environ.get("MISSIONCORE_LIDAR_GROUND_ROOT", "").strip()
|
|
return Path(value).expanduser().absolute() if value else None
|
|
|
|
|
|
def configured_lidar_field_review_root() -> Path | None:
|
|
value = os.environ.get("MISSIONCORE_LIDAR_FIELD_REVIEW_ROOT", "").strip()
|
|
return Path(value).expanduser().absolute() if value else None
|
|
|
|
|
|
def build_lidar_router(
|
|
*,
|
|
root_provider: RootProvider = configured_lidar_replay_root,
|
|
ground_root_provider: RootProvider = configured_lidar_ground_root,
|
|
field_review_root_provider: RootProvider = configured_lidar_field_review_root,
|
|
) -> APIRouter:
|
|
router = APIRouter(prefix="/api/v1/lidar", tags=["lidar"])
|
|
|
|
@router.get("/replay-packs")
|
|
def list_lidar_replay_packs(
|
|
limit: int = Query(default=20, ge=1, le=100),
|
|
) -> dict[str, Any]:
|
|
root = root_provider()
|
|
if root is None or not root.is_dir():
|
|
return {
|
|
"schema_version": LIDAR_CATALOG_SCHEMA,
|
|
"configured": root is not None,
|
|
"items": [],
|
|
"valid_total": 0,
|
|
"invalid_total": 0,
|
|
"duplicate_total": 0,
|
|
"access": "read-only",
|
|
}
|
|
items: list[dict[str, object]] = []
|
|
invalid_total = 0
|
|
duplicate_total = 0
|
|
seen_logical_content: set[str] = set()
|
|
candidates = sorted(
|
|
(
|
|
candidate
|
|
for candidate in root.iterdir()
|
|
if candidate.is_dir() and _PACK_ID.fullmatch(candidate.name) is not None
|
|
),
|
|
key=lambda candidate: candidate.stat().st_mtime_ns,
|
|
reverse=True,
|
|
)
|
|
for candidate in candidates:
|
|
try:
|
|
pack = LidarReplayPackV2(candidate)
|
|
try:
|
|
item = lidar_pack_catalog_item(pack)
|
|
logical_content = str(item["logical_content_sha256"])
|
|
if logical_content in seen_logical_content:
|
|
duplicate_total += 1
|
|
continue
|
|
seen_logical_content.add(logical_content)
|
|
item["created_at_utc"] = pack.manifest.get("created_at_utc")
|
|
items.append(item)
|
|
finally:
|
|
pack.close()
|
|
except (LidarReplayError, OSError):
|
|
invalid_total += 1
|
|
return {
|
|
"schema_version": LIDAR_CATALOG_SCHEMA,
|
|
"configured": True,
|
|
"items": items[:limit],
|
|
"valid_total": len(items),
|
|
"invalid_total": invalid_total,
|
|
"duplicate_total": duplicate_total,
|
|
"access": "read-only",
|
|
}
|
|
|
|
@router.get("/replay-packs/{pack_id}")
|
|
def get_lidar_replay_pack(pack_id: str) -> dict[str, object]:
|
|
if _PACK_ID.fullmatch(pack_id) is None:
|
|
raise HTTPException(status_code=404, detail="LiDAR replay pack не найден")
|
|
root = root_provider()
|
|
if root is None or not root.is_dir():
|
|
raise HTTPException(status_code=503, detail="LiDAR replay storage не настроен")
|
|
candidate = root / pack_id
|
|
if not candidate.is_dir():
|
|
raise HTTPException(status_code=404, detail="LiDAR replay pack не найден")
|
|
try:
|
|
pack = LidarReplayPackV2(candidate)
|
|
try:
|
|
return lidar_pack_detail(pack)
|
|
finally:
|
|
pack.close()
|
|
except (LidarReplayError, OSError) as exc:
|
|
raise HTTPException(
|
|
status_code=409,
|
|
detail="LiDAR replay pack не прошёл проверку целостности",
|
|
) from exc
|
|
|
|
@router.get("/ground-benchmarks")
|
|
def list_lidar_ground_benchmarks(
|
|
pack_id: str | None = Query(default=None),
|
|
limit: int = Query(default=20, ge=1, le=100),
|
|
) -> dict[str, Any]:
|
|
if pack_id is not None and _PACK_ID.fullmatch(pack_id) is None:
|
|
raise HTTPException(status_code=404, detail="LiDAR replay pack не найден")
|
|
root = ground_root_provider()
|
|
if root is None or not root.is_dir():
|
|
return {
|
|
"schema_version": LIDAR_GROUND_CATALOG_SCHEMA,
|
|
"configured": root is not None,
|
|
"items": [],
|
|
"valid_total": 0,
|
|
"invalid_total": 0,
|
|
"access": "read-only",
|
|
}
|
|
items: list[dict[str, object]] = []
|
|
invalid_total = 0
|
|
candidates = sorted(
|
|
(
|
|
candidate
|
|
for candidate in root.iterdir()
|
|
if candidate.is_dir() and _BENCHMARK_ID.fullmatch(candidate.name) is not None
|
|
),
|
|
key=lambda candidate: candidate.stat().st_mtime_ns,
|
|
reverse=True,
|
|
)
|
|
for candidate in candidates:
|
|
try:
|
|
benchmark = LidarGroundBenchmarkV1(candidate)
|
|
try:
|
|
if pack_id is None or benchmark.identity.get("replay_pack_id") == pack_id:
|
|
items.append(lidar_ground_benchmark_catalog_item(benchmark))
|
|
finally:
|
|
benchmark.close()
|
|
except (LidarGroundError, OSError):
|
|
invalid_total += 1
|
|
return {
|
|
"schema_version": LIDAR_GROUND_CATALOG_SCHEMA,
|
|
"configured": True,
|
|
"items": items[:limit],
|
|
"valid_total": len(items),
|
|
"invalid_total": invalid_total,
|
|
"access": "read-only",
|
|
}
|
|
|
|
@router.get("/ground-benchmarks/{benchmark_id}")
|
|
def get_lidar_ground_benchmark(benchmark_id: str) -> dict[str, object]:
|
|
if _BENCHMARK_ID.fullmatch(benchmark_id) is None:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail="LiDAR ground benchmark не найден",
|
|
)
|
|
root = ground_root_provider()
|
|
if root is None or not root.is_dir():
|
|
raise HTTPException(
|
|
status_code=503,
|
|
detail="LiDAR ground storage не настроен",
|
|
)
|
|
candidate = root / benchmark_id
|
|
if not candidate.is_dir():
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail="LiDAR ground benchmark не найден",
|
|
)
|
|
try:
|
|
benchmark = LidarGroundBenchmarkV1(candidate)
|
|
try:
|
|
return {
|
|
"schema_version": ("missioncore.lidar-ground-benchmark-detail/v1"),
|
|
"benchmark": lidar_ground_benchmark_catalog_item(benchmark),
|
|
"report": benchmark.report,
|
|
"access": "read-only",
|
|
}
|
|
finally:
|
|
benchmark.close()
|
|
except (LidarGroundError, OSError) as exc:
|
|
raise HTTPException(
|
|
status_code=409,
|
|
detail="LiDAR ground benchmark не прошёл проверку целостности",
|
|
) from exc
|
|
|
|
@router.get("/ground-benchmarks/{benchmark_id}/frames/{frame_index}")
|
|
def get_lidar_ground_frame(
|
|
benchmark_id: str,
|
|
frame_index: int,
|
|
) -> dict[str, object]:
|
|
if _BENCHMARK_ID.fullmatch(benchmark_id) is None or frame_index < 0:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail="LiDAR ground frame не найден",
|
|
)
|
|
ground_root = ground_root_provider()
|
|
replay_root = root_provider()
|
|
if ground_root is None or not ground_root.is_dir():
|
|
raise HTTPException(
|
|
status_code=503,
|
|
detail="LiDAR ground storage не настроен",
|
|
)
|
|
if replay_root is None or not replay_root.is_dir():
|
|
raise HTTPException(
|
|
status_code=503,
|
|
detail="LiDAR replay storage не настроен",
|
|
)
|
|
benchmark_path = ground_root / benchmark_id
|
|
if not benchmark_path.is_dir():
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail="LiDAR ground benchmark не найден",
|
|
)
|
|
try:
|
|
benchmark = LidarGroundBenchmarkV1(benchmark_path)
|
|
try:
|
|
replay_pack_id = benchmark.identity.get("replay_pack_id")
|
|
if (
|
|
not isinstance(replay_pack_id, str)
|
|
or _PACK_ID.fullmatch(replay_pack_id) is None
|
|
):
|
|
raise LidarGroundError("Ground benchmark replay identity is invalid")
|
|
replay_path = replay_root / replay_pack_id
|
|
if not replay_path.is_dir():
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail="Связанный LiDAR replay pack не найден",
|
|
)
|
|
replay = LidarReplayPackV2(replay_path)
|
|
try:
|
|
return lidar_ground_frame_detail(
|
|
benchmark,
|
|
replay,
|
|
frame_index,
|
|
)
|
|
finally:
|
|
replay.close()
|
|
finally:
|
|
benchmark.close()
|
|
except IndexError as exc:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail="LiDAR ground frame не найден",
|
|
) from exc
|
|
except HTTPException:
|
|
raise
|
|
except (LidarGroundError, LidarReplayError, OSError) as exc:
|
|
raise HTTPException(
|
|
status_code=409,
|
|
detail="LiDAR ground frame не прошёл проверку целостности",
|
|
) from exc
|
|
|
|
@router.get("/field-reviews")
|
|
def list_lidar_field_reviews(
|
|
limit: int = Query(default=10, ge=1, le=50),
|
|
) -> dict[str, Any]:
|
|
root = field_review_root_provider()
|
|
if root is None or not root.is_dir():
|
|
return {
|
|
"schema_version": LIDAR_FIELD_REVIEW_CATALOG_SCHEMA,
|
|
"configured": root is not None,
|
|
"items": [],
|
|
"valid_total": 0,
|
|
"invalid_total": 0,
|
|
"access": "read-only",
|
|
}
|
|
items: list[dict[str, object]] = []
|
|
invalid_total = 0
|
|
candidates = sorted(
|
|
(
|
|
candidate
|
|
for candidate in root.iterdir()
|
|
if candidate.is_dir() and _FIELD_REVIEW_ID.fullmatch(candidate.name) is not None
|
|
),
|
|
key=lambda candidate: candidate.stat().st_mtime_ns,
|
|
reverse=True,
|
|
)
|
|
for candidate in candidates:
|
|
try:
|
|
review = LidarFieldReviewV1(candidate)
|
|
try:
|
|
items.append(lidar_field_review_catalog_item(review))
|
|
finally:
|
|
review.close()
|
|
except (LidarGroundError, OSError):
|
|
invalid_total += 1
|
|
return {
|
|
"schema_version": LIDAR_FIELD_REVIEW_CATALOG_SCHEMA,
|
|
"configured": True,
|
|
"items": items[:limit],
|
|
"valid_total": len(items),
|
|
"invalid_total": invalid_total,
|
|
"access": "read-only",
|
|
}
|
|
|
|
@router.get("/field-reviews/{review_id}/windows/{window_index}")
|
|
def get_lidar_field_review_window(
|
|
review_id: str,
|
|
window_index: int,
|
|
) -> dict[str, object]:
|
|
if _FIELD_REVIEW_ID.fullmatch(review_id) is None or window_index < 0:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail="LiDAR field-review window не найден",
|
|
)
|
|
root = field_review_root_provider()
|
|
if root is None or not root.is_dir():
|
|
raise HTTPException(
|
|
status_code=503,
|
|
detail="LiDAR field-review storage не настроен",
|
|
)
|
|
candidate = root / review_id
|
|
if not candidate.is_dir():
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail="LiDAR field review не найден",
|
|
)
|
|
try:
|
|
review = LidarFieldReviewV1(candidate)
|
|
try:
|
|
return review.window_detail(window_index)
|
|
finally:
|
|
review.close()
|
|
except IndexError as exc:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail="LiDAR field-review window не найден",
|
|
) from exc
|
|
except (LidarGroundError, OSError) as exc:
|
|
raise HTTPException(
|
|
status_code=409,
|
|
detail="LiDAR field review не прошёл проверку целостности",
|
|
) from exc
|
|
|
|
@router.get("/field-reviews/{review_id}/windows/{window_index}/preview")
|
|
def get_lidar_field_review_preview(
|
|
review_id: str,
|
|
window_index: int,
|
|
) -> Response:
|
|
if _FIELD_REVIEW_ID.fullmatch(review_id) is None or window_index < 0:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail="LiDAR field-review preview не найден",
|
|
)
|
|
root = field_review_root_provider()
|
|
if root is None or not root.is_dir():
|
|
raise HTTPException(
|
|
status_code=503,
|
|
detail="LiDAR field-review storage не настроен",
|
|
)
|
|
candidate = root / review_id
|
|
if not candidate.is_dir():
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail="LiDAR field review не найден",
|
|
)
|
|
try:
|
|
review = LidarFieldReviewV1(candidate)
|
|
try:
|
|
windows = review.report.get("windows")
|
|
if not isinstance(windows, list) or not 0 <= window_index < len(windows):
|
|
raise IndexError(window_index)
|
|
window = windows[window_index]
|
|
if not isinstance(window, dict) or not isinstance(window.get("key"), str):
|
|
raise LidarGroundError("LiDAR field-review preview key is invalid")
|
|
preview = review.preview_paths.get(window["key"])
|
|
if preview is None:
|
|
raise LidarGroundError("LiDAR field-review preview is missing")
|
|
content = preview.read_bytes()
|
|
finally:
|
|
review.close()
|
|
except IndexError as exc:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail="LiDAR field-review preview не найден",
|
|
) from exc
|
|
except (LidarGroundError, OSError) as exc:
|
|
raise HTTPException(
|
|
status_code=409,
|
|
detail="LiDAR field-review preview не прошёл проверку",
|
|
) from exc
|
|
return Response(
|
|
content=content,
|
|
media_type="image/jpeg",
|
|
headers={"Cache-Control": "private, max-age=31536000, immutable"},
|
|
)
|
|
|
|
return router
|