501 lines
17 KiB
Python
501 lines
17 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import os
|
|
import shutil
|
|
from collections.abc import AsyncIterator, Iterable
|
|
from contextlib import asynccontextmanager, suppress
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from fastapi import FastAPI, HTTPException, Request, WebSocket, WebSocketDisconnect
|
|
from fastapi.exceptions import RequestValidationError
|
|
from fastapi.responses import JSONResponse
|
|
from fastapi.staticfiles import StaticFiles
|
|
from pydantic import ValidationError
|
|
|
|
from k1link import __version__
|
|
from k1link.compute import (
|
|
IntegratedPerceptionOverlayStore,
|
|
RecordedCalibratedFusionStore,
|
|
RecordedPerceptionEpochStore,
|
|
RecordedPerceptionOverlayMux,
|
|
RecordedPerceptionOverlayStore,
|
|
)
|
|
from k1link.sessions import (
|
|
MaterializedRecording,
|
|
RecordedMediaInspector,
|
|
RecordedMediaManifest,
|
|
RecordingPreparationQueueFull,
|
|
ReplayCommand,
|
|
SessionRecordingMaterializer,
|
|
SessionRecordingPreparationManager,
|
|
SessionStore,
|
|
)
|
|
from k1link.web.device_plugin_composition import load_installed_device_plugins
|
|
from k1link.web.environment_api import build_environment_router
|
|
from k1link.web.laboratory_api import build_laboratory_router
|
|
from k1link.web.lidar_api import build_lidar_router
|
|
from k1link.web.plugin_catalog import DevicePluginCatalog, PluginCatalogError
|
|
from k1link.web.plugin_runtime import (
|
|
STATE_READ_ACTION_ID,
|
|
DevicePluginActionRequest,
|
|
DevicePluginDispatcher,
|
|
PluginActionNotFoundError,
|
|
PluginExecutionError,
|
|
PluginNotFoundError,
|
|
PluginRuntimeUnavailableError,
|
|
)
|
|
from k1link.web.polygon_api import build_polygon_router, configured_polygon_runs_root
|
|
from k1link.web.session_api import build_session_router
|
|
|
|
REPOSITORY_ROOT = Path(__file__).resolve().parents[3]
|
|
INVALID_REQUEST_DETAIL = "Некорректные параметры запроса."
|
|
|
|
|
|
def _resolve_media_tool(name: str) -> Path | None:
|
|
"""Resolve media tools under interactive shells and minimal launchd PATHs."""
|
|
|
|
discovered = shutil.which(name)
|
|
candidates = (
|
|
Path(discovered) if discovered is not None else None,
|
|
Path("/opt/homebrew/bin") / name,
|
|
Path("/usr/local/bin") / name,
|
|
Path("/usr/bin") / name,
|
|
)
|
|
for candidate in candidates:
|
|
if candidate is not None and candidate.is_file() and os.access(candidate, os.X_OK):
|
|
return candidate
|
|
return None
|
|
|
|
|
|
plugin_environment = load_installed_device_plugins(REPOSITORY_ROOT)
|
|
plugin_catalog: DevicePluginCatalog = plugin_environment.catalog
|
|
plugin_dispatcher: DevicePluginDispatcher = plugin_environment.dispatcher
|
|
session_store = SessionStore(REPOSITORY_ROOT)
|
|
session_recording_materializer = SessionRecordingMaterializer(
|
|
session_store.data_dir,
|
|
exporters=plugin_environment.recording_exporters,
|
|
)
|
|
session_recorded_media_inspector = RecordedMediaInspector(
|
|
session_store.data_dir / "recorded-media-preparations"
|
|
)
|
|
_ffmpeg = _resolve_media_tool("ffmpeg")
|
|
_ffprobe = _resolve_media_tool("ffprobe")
|
|
session_legacy_perception_overlay_store = (
|
|
RecordedPerceptionOverlayStore(
|
|
jobs_root=REPOSITORY_ROOT / ".runtime" / "compute-jobs",
|
|
results_root=REPOSITORY_ROOT / ".runtime" / "compute-results",
|
|
cache_root=session_store.data_dir / "perception-overlays",
|
|
ffmpeg_path=_ffmpeg,
|
|
ffprobe_path=_ffprobe,
|
|
)
|
|
if _ffmpeg is not None and _ffprobe is not None
|
|
else None
|
|
)
|
|
session_calibrated_fusion_store = RecordedCalibratedFusionStore(
|
|
jobs_root=REPOSITORY_ROOT / ".runtime" / "compute-jobs",
|
|
perception_results_root=REPOSITORY_ROOT / ".runtime" / "compute-results",
|
|
fusion_results_root=REPOSITORY_ROOT / ".runtime" / "compute-fusions",
|
|
cache_root=session_store.data_dir / "calibrated-fusion-overlays",
|
|
)
|
|
session_previous_perception_overlay_store = RecordedPerceptionOverlayMux(
|
|
session_calibrated_fusion_store, session_legacy_perception_overlay_store
|
|
)
|
|
session_integrated_perception_store = (
|
|
IntegratedPerceptionOverlayStore(
|
|
jobs_root=REPOSITORY_ROOT / ".runtime" / "compute-jobs",
|
|
results_root=(
|
|
REPOSITORY_ROOT
|
|
/ ".runtime"
|
|
/ "compute-experiments"
|
|
/ "e10"
|
|
/ "worker-results"
|
|
),
|
|
lidar_packs_root=(
|
|
REPOSITORY_ROOT
|
|
/ ".runtime"
|
|
/ "compute-experiments"
|
|
/ "e10"
|
|
/ "lidar-packs"
|
|
),
|
|
cache_root=session_store.data_dir / "integrated-perception-overlays",
|
|
ffmpeg_path=_ffmpeg,
|
|
)
|
|
if _ffmpeg is not None
|
|
else None
|
|
)
|
|
session_perception_overlay_store = RecordedPerceptionOverlayMux(
|
|
session_integrated_perception_store or session_previous_perception_overlay_store,
|
|
(
|
|
session_previous_perception_overlay_store
|
|
if session_integrated_perception_store is not None
|
|
else None
|
|
),
|
|
)
|
|
session_perception_epoch_store = (
|
|
RecordedPerceptionEpochStore(
|
|
jobs_root=REPOSITORY_ROOT / ".runtime" / "compute-jobs",
|
|
results_root=REPOSITORY_ROOT / ".runtime" / "compute-results",
|
|
ffprobe_path=_ffprobe,
|
|
)
|
|
if _ffprobe is not None
|
|
else None
|
|
)
|
|
|
|
|
|
def _prepare_recorded_media_for_launch(
|
|
command: ReplayCommand,
|
|
_: MaterializedRecording,
|
|
) -> tuple[RecordedMediaManifest, ...]:
|
|
"""Prepare all camera descriptors inside the background job."""
|
|
|
|
return tuple(
|
|
session_recorded_media_inspector.inspect(artifact, command)
|
|
for artifact in session_store.list_recorded_media(command.session_id)
|
|
)
|
|
|
|
|
|
def _restore_recorded_media_for_launch(
|
|
command: ReplayCommand,
|
|
_: MaterializedRecording,
|
|
) -> tuple[RecordedMediaManifest, ...] | None:
|
|
"""Restore only previously published media descriptors."""
|
|
|
|
restored: list[RecordedMediaManifest] = []
|
|
for artifact in session_store.list_recorded_media(command.session_id):
|
|
manifest = session_recorded_media_inspector.restore_prepared(artifact, command)
|
|
if manifest is None:
|
|
return None
|
|
restored.append(manifest)
|
|
return tuple(restored)
|
|
|
|
|
|
session_recording_preparation_manager = SessionRecordingPreparationManager(
|
|
session_recording_materializer,
|
|
ready_preparer=_prepare_recorded_media_for_launch,
|
|
ready_restorer=_restore_recorded_media_for_launch,
|
|
)
|
|
|
|
|
|
def refresh_observation_catalog() -> tuple[str, ...]:
|
|
"""Discover completed or recoverable local evidence without copying payloads."""
|
|
|
|
imported = [
|
|
session_id
|
|
for archive in plugin_environment.observation_archives
|
|
for session_id in session_store.reconcile_archive(archive)
|
|
]
|
|
return tuple(dict.fromkeys(imported))
|
|
|
|
|
|
def finalized_replayable_recording_ids() -> tuple[str, ...]:
|
|
"""List finalized replayable catalog identities without scheduling work."""
|
|
|
|
finalized: list[str] = []
|
|
cursor: str | None = None
|
|
while True:
|
|
page = session_store.list_recent(limit=100, cursor=cursor)
|
|
finalized.extend(
|
|
summary.session_id
|
|
for summary in page.items
|
|
if summary.replayable and summary.status in {"ready", "interrupted", "failed"}
|
|
)
|
|
cursor = page.next_cursor
|
|
if cursor is None:
|
|
return tuple(finalized)
|
|
|
|
|
|
def enqueue_replayable_recordings(session_ids: Iterable[str]) -> tuple[str, ...]:
|
|
"""Schedule only explicitly selected newly finalized sessions."""
|
|
|
|
enqueued: list[str] = []
|
|
for session_id in dict.fromkeys(session_ids):
|
|
try:
|
|
command = session_store.prepare_replay(session_id)
|
|
session_recording_preparation_manager.enqueue(
|
|
command,
|
|
retry_interrupted=True,
|
|
)
|
|
except RecordingPreparationQueueFull:
|
|
return tuple(enqueued)
|
|
except Exception:
|
|
# One stale/corrupt row must not starve a later newly completed
|
|
# session. Historical cold caches remain operator-triggered.
|
|
continue
|
|
enqueued.append(session_id)
|
|
return tuple(enqueued)
|
|
|
|
|
|
def newly_finalized_recording_ids(
|
|
known_finalized: set[str] | None,
|
|
current_finalized: Iterable[str],
|
|
) -> tuple[str, ...]:
|
|
"""Return only post-startup completions; the first scan is a baseline."""
|
|
|
|
if known_finalized is None:
|
|
return ()
|
|
return tuple(sorted(set(current_finalized) - known_finalized))
|
|
|
|
|
|
async def _recording_preparation_reconciler() -> None:
|
|
"""Prepare sessions finalized during this process, never historical rows."""
|
|
|
|
known_finalized: set[str] | None = None
|
|
while True:
|
|
try:
|
|
await asyncio.to_thread(refresh_observation_catalog)
|
|
finalized = set(await asyncio.to_thread(finalized_replayable_recording_ids))
|
|
newly_finalized = newly_finalized_recording_ids(
|
|
known_finalized,
|
|
finalized,
|
|
)
|
|
if newly_finalized:
|
|
await asyncio.to_thread(
|
|
enqueue_replayable_recordings,
|
|
newly_finalized,
|
|
)
|
|
known_finalized = finalized
|
|
except Exception:
|
|
# A transient filesystem/catalog failure must not permanently
|
|
# disable preparation of sessions completed later in the run.
|
|
pass
|
|
await asyncio.sleep(2.0)
|
|
|
|
|
|
@asynccontextmanager
|
|
async def app_lifespan(_: FastAPI) -> AsyncIterator[None]:
|
|
reconciler: asyncio.Task[None] | None = None
|
|
try:
|
|
session_recording_preparation_manager.start()
|
|
# Recovery is intentionally a one-shot startup phase. The archive
|
|
# helper owns a cross-process lease, while ordinary catalog requests
|
|
# only perform discovery and therefore never touch a live writer.
|
|
for archive in plugin_environment.observation_archives:
|
|
await asyncio.to_thread(archive.recover, archive.root)
|
|
# Full evidence discovery, hashing and RRD queue reconciliation can be
|
|
# expensive on field captures. Start it immediately in the background
|
|
# instead of holding the ASGI startup gate.
|
|
reconciler = asyncio.create_task(_recording_preparation_reconciler())
|
|
yield
|
|
finally:
|
|
if reconciler is not None:
|
|
reconciler.cancel()
|
|
with suppress(asyncio.CancelledError):
|
|
await reconciler
|
|
await asyncio.to_thread(session_recording_preparation_manager.close)
|
|
plugin_environment.close()
|
|
|
|
|
|
app = FastAPI(
|
|
title="NODEDC MISSION CORE API",
|
|
version=__version__,
|
|
docs_url="/api/docs",
|
|
redoc_url=None,
|
|
openapi_url="/api/openapi.json",
|
|
lifespan=app_lifespan,
|
|
)
|
|
|
|
|
|
@app.exception_handler(RequestValidationError)
|
|
async def request_validation_error_handler(
|
|
_: Request,
|
|
__: RequestValidationError,
|
|
) -> JSONResponse:
|
|
"""Return validation failures without reflecting request values or credentials."""
|
|
|
|
return JSONResponse(status_code=422, content={"detail": INVALID_REQUEST_DETAIL})
|
|
|
|
|
|
@app.get("/api/health")
|
|
def health() -> dict[str, Any]:
|
|
runtime_health = plugin_environment.runtime_health
|
|
runtimes_ready = all(item["status"] == "ready" for item in runtime_health)
|
|
return {
|
|
"ok": runtimes_ready,
|
|
"status": "ok" if runtimes_ready else "degraded",
|
|
"service": "mission-core-control-plane",
|
|
"version": __version__,
|
|
"plugin_runtimes": {
|
|
"ready": sum(item["status"] == "ready" for item in runtime_health),
|
|
"total": len(runtime_health),
|
|
},
|
|
}
|
|
|
|
|
|
@app.get("/api/v1/device-plugins")
|
|
def get_device_plugins() -> dict[str, Any]:
|
|
try:
|
|
return {"items": plugin_catalog.plugin_documents()}
|
|
except PluginCatalogError as exc:
|
|
raise HTTPException(status_code=500, detail=str(exc)) from exc
|
|
|
|
|
|
@app.get("/api/v1/device-models")
|
|
def get_device_models() -> dict[str, Any]:
|
|
try:
|
|
return {"items": plugin_catalog.model_documents()}
|
|
except PluginCatalogError as exc:
|
|
raise HTTPException(status_code=500, detail=str(exc)) from exc
|
|
|
|
|
|
@app.get("/api/v1/device-plugin-runtimes")
|
|
def get_device_plugin_runtimes() -> dict[str, Any]:
|
|
return {"items": list(plugin_environment.runtime_health)}
|
|
|
|
|
|
@app.post("/api/v1/device-plugins/{plugin_id}/actions/{action_id}")
|
|
async def invoke_device_plugin_action(
|
|
plugin_id: str,
|
|
action_id: str,
|
|
request: DevicePluginActionRequest,
|
|
) -> dict[str, Any]:
|
|
try:
|
|
state = await plugin_dispatcher.invoke(plugin_id, action_id, request.input)
|
|
return {"state": state}
|
|
except (PluginNotFoundError, PluginActionNotFoundError) as exc:
|
|
raise HTTPException(status_code=404, detail=str(exc)) from exc
|
|
except ValidationError as exc:
|
|
raise HTTPException(status_code=422, detail=INVALID_REQUEST_DETAIL) from exc
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
|
except PluginExecutionError as exc:
|
|
raise HTTPException(status_code=502, detail=str(exc)) from exc
|
|
except PluginRuntimeUnavailableError as exc:
|
|
raise HTTPException(status_code=503, detail=str(exc)) from exc
|
|
|
|
|
|
@app.websocket("/api/v1/device-plugins/{plugin_id}/events")
|
|
async def device_plugin_events(websocket: WebSocket, plugin_id: str) -> None:
|
|
await websocket.accept()
|
|
sequence = 0
|
|
try:
|
|
while True:
|
|
state = await plugin_dispatcher.invoke(plugin_id, STATE_READ_ACTION_ID, {})
|
|
sequence += 1
|
|
await websocket.send_json({"pluginId": plugin_id, "sequence": sequence, "state": state})
|
|
await asyncio.sleep(0.5)
|
|
except WebSocketDisconnect:
|
|
return
|
|
except (PluginNotFoundError, PluginActionNotFoundError):
|
|
await websocket.close(code=1008, reason="Device plugin is not available")
|
|
except (PluginExecutionError, PluginRuntimeUnavailableError):
|
|
await websocket.close(code=1011, reason="Device plugin state stream failed")
|
|
|
|
|
|
for legacy_router in plugin_environment.legacy_routers:
|
|
app.include_router(legacy_router)
|
|
|
|
app.include_router(
|
|
build_session_router(
|
|
session_store,
|
|
# Production discovery belongs to the startup/background reconciler.
|
|
# HTTP list/replay paths must never rescan evidence roots inline.
|
|
catalog_refresher=None,
|
|
recording_materializer=session_recording_materializer,
|
|
recording_preparation_manager=session_recording_preparation_manager,
|
|
media_inspector=session_recorded_media_inspector,
|
|
perception_overlay_provider=session_perception_overlay_store,
|
|
perception_media_provider=session_perception_epoch_store,
|
|
point_color_renderers=plugin_environment.point_color_renderers,
|
|
)
|
|
)
|
|
app.include_router(
|
|
build_environment_router(
|
|
root_provider=lambda: session_store.data_dir / "ui-environment"
|
|
)
|
|
)
|
|
app.include_router(
|
|
build_polygon_router(
|
|
root_provider=lambda: configured_polygon_runs_root()
|
|
or REPOSITORY_ROOT / ".runtime" / "polygon-runs"
|
|
)
|
|
)
|
|
app.include_router(
|
|
build_lidar_router(
|
|
root_provider=lambda: (
|
|
REPOSITORY_ROOT
|
|
/ ".runtime"
|
|
/ "compute-experiments"
|
|
/ "lidar-replay-v2"
|
|
/ "packs"
|
|
),
|
|
ground_root_provider=lambda: (
|
|
REPOSITORY_ROOT
|
|
/ ".runtime"
|
|
/ "compute-experiments"
|
|
/ "lidar-ground-v1"
|
|
/ "benchmarks"
|
|
),
|
|
field_review_root_provider=lambda: (
|
|
REPOSITORY_ROOT
|
|
/ ".runtime"
|
|
/ "compute-experiments"
|
|
/ "lidar-field-review-v1"
|
|
/ "reviews"
|
|
),
|
|
local_surface_root_provider=lambda: (
|
|
REPOSITORY_ROOT
|
|
/ ".runtime"
|
|
/ "compute-experiments"
|
|
/ "k1-local-surface-v1"
|
|
/ "models"
|
|
),
|
|
e10_source_root_provider=lambda: (
|
|
REPOSITORY_ROOT
|
|
/ ".runtime"
|
|
/ "compute-experiments"
|
|
/ "e10"
|
|
/ "lidar-packs"
|
|
),
|
|
dataset_admission_provider=lambda: (
|
|
REPOSITORY_ROOT / ".runtime" / "dataset-gateway" / "admission.json"
|
|
),
|
|
dataset_preview_provider=lambda: (
|
|
REPOSITORY_ROOT / ".runtime" / "dataset-gateway" / "preview.json"
|
|
),
|
|
dataset_rellis_preview_provider=lambda: (
|
|
REPOSITORY_ROOT / ".runtime" / "dataset-gateway" / "rellis-preview.json"
|
|
),
|
|
dataset_rellis_admission_provider=lambda: (
|
|
REPOSITORY_ROOT / ".runtime" / "dataset-gateway" / "rellis-admission.json"
|
|
),
|
|
dataset_ground_preview_provider=lambda: (
|
|
REPOSITORY_ROOT / ".runtime" / "dataset-gateway" / "ground-comparison.json"
|
|
),
|
|
)
|
|
)
|
|
app.include_router(
|
|
build_laboratory_router(
|
|
e29_root_provider=lambda: (
|
|
REPOSITORY_ROOT / ".runtime" / "compute-experiments" / "e29" / "results"
|
|
),
|
|
local_surface_root_provider=lambda: (
|
|
REPOSITORY_ROOT
|
|
/ ".runtime"
|
|
/ "compute-experiments"
|
|
/ "k1-local-surface-v1"
|
|
/ "models"
|
|
),
|
|
source_pack_root_provider=lambda: (
|
|
REPOSITORY_ROOT
|
|
/ ".runtime"
|
|
/ "compute-experiments"
|
|
/ "e10"
|
|
/ "lidar-packs"
|
|
),
|
|
source_result_root_provider=lambda: (
|
|
REPOSITORY_ROOT
|
|
/ ".runtime"
|
|
/ "compute-experiments"
|
|
/ "e10"
|
|
/ "worker-results"
|
|
),
|
|
)
|
|
)
|
|
|
|
|
|
frontend_dist = REPOSITORY_ROOT / "apps" / "control-station" / "dist"
|
|
if frontend_dist.is_dir():
|
|
app.mount("/", StaticFiles(directory=frontend_dist, html=True), name="frontend")
|