263 lines
9.5 KiB
Python
263 lines
9.5 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from collections.abc import AsyncIterator
|
|
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.sessions import (
|
|
MaterializedRecording,
|
|
RecordedMediaInspector,
|
|
RecordedMediaManifest,
|
|
RecordingPreparationQueueFull,
|
|
ReplayCommand,
|
|
SessionRecordingMaterializer,
|
|
SessionRecordingPreparationManager,
|
|
SessionStore,
|
|
recover_stale_active_session_marker,
|
|
resolve_missioncore_evidence_dir,
|
|
)
|
|
from k1link.web.camera_archive import recover_incomplete_camera_archives
|
|
from k1link.web.device_plugin_composition import load_installed_device_plugins
|
|
from k1link.web.plugin_catalog import DevicePluginCatalog, PluginCatalogError
|
|
from k1link.web.plugin_runtime import (
|
|
STATE_READ_ACTION_ID,
|
|
DevicePluginActionRequest,
|
|
DevicePluginDispatcher,
|
|
PluginActionNotFoundError,
|
|
PluginExecutionError,
|
|
PluginNotFoundError,
|
|
)
|
|
from k1link.web.session_api import build_session_router
|
|
|
|
REPOSITORY_ROOT = Path(__file__).resolve().parents[3]
|
|
INVALID_REQUEST_DETAIL = "Некорректные параметры запроса."
|
|
|
|
|
|
legacy_observation_sessions_root = REPOSITORY_ROOT / "sessions"
|
|
observation_sessions_root = resolve_missioncore_evidence_dir(REPOSITORY_ROOT)
|
|
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)
|
|
session_recorded_media_inspector = RecordedMediaInspector(
|
|
session_store.data_dir / "recorded-media-preparations"
|
|
)
|
|
|
|
|
|
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)
|
|
)
|
|
|
|
|
|
session_recording_preparation_manager = SessionRecordingPreparationManager(
|
|
session_recording_materializer,
|
|
ready_preparer=_prepare_recorded_media_for_launch,
|
|
)
|
|
|
|
|
|
def refresh_observation_catalog() -> tuple[str, ...]:
|
|
"""Discover completed or recoverable local evidence without copying payloads."""
|
|
|
|
imported = [
|
|
*session_store.import_legacy_viewer_live(legacy_observation_sessions_root),
|
|
*session_store.import_legacy_viewer_live(observation_sessions_root),
|
|
]
|
|
return tuple(dict.fromkeys(imported))
|
|
|
|
|
|
def enqueue_replayable_recordings() -> tuple[str, ...]:
|
|
"""Reconcile finalized catalog sessions into the durable RRD work queue."""
|
|
|
|
enqueued: list[str] = []
|
|
cursor: str | None = None
|
|
while True:
|
|
page = session_store.list_recent(limit=100, cursor=cursor)
|
|
for summary in page.items:
|
|
if not summary.replayable or summary.status not in {
|
|
"ready",
|
|
"interrupted",
|
|
"failed",
|
|
}:
|
|
continue
|
|
try:
|
|
command = session_store.prepare_replay(summary.session_id)
|
|
session_recording_preparation_manager.enqueue(
|
|
command,
|
|
retry_interrupted=True,
|
|
)
|
|
except RecordingPreparationQueueFull:
|
|
return tuple(enqueued)
|
|
except Exception:
|
|
# One stale/corrupt catalog row must not starve every valid
|
|
# finalized session that follows it in the reconciliation
|
|
# page. The row remains visible as failed/interrupted evidence
|
|
# and can be repaired independently.
|
|
continue
|
|
enqueued.append(summary.session_id)
|
|
cursor = page.next_cursor
|
|
if cursor is None:
|
|
return tuple(enqueued)
|
|
|
|
|
|
async def _recording_preparation_reconciler() -> None:
|
|
"""Discover newly completed/recovered captures without blocking requests."""
|
|
|
|
while True:
|
|
try:
|
|
await asyncio.to_thread(refresh_observation_catalog)
|
|
await asyncio.to_thread(enqueue_replayable_recordings)
|
|
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.
|
|
await asyncio.to_thread(recover_stale_active_session_marker, observation_sessions_root)
|
|
for sessions_root in (
|
|
legacy_observation_sessions_root,
|
|
observation_sessions_root,
|
|
):
|
|
await asyncio.to_thread(recover_incomplete_camera_archives, sessions_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]:
|
|
return {
|
|
"ok": True,
|
|
"status": "ok",
|
|
"service": "mission-core-control-plane",
|
|
"version": __version__,
|
|
}
|
|
|
|
|
|
@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.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
|
|
|
|
|
|
@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:
|
|
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,
|
|
)
|
|
)
|
|
|
|
|
|
frontend_dist = REPOSITORY_ROOT / "apps" / "control-station" / "dist"
|
|
if frontend_dist.is_dir():
|
|
app.mount("/", StaticFiles(directory=frontend_dist, html=True), name="frontend")
|