diff --git a/apps/control-station/src/components/RerunViewport.tsx b/apps/control-station/src/components/RerunViewport.tsx index 0668e58..7662983 100644 --- a/apps/control-station/src/components/RerunViewport.tsx +++ b/apps/control-station/src/components/RerunViewport.tsx @@ -4,6 +4,7 @@ import type { SceneSettings } from "../sceneSettings"; import type { RecordedAdmissionPhase } from "../core/observation/recordedSessionAdmission"; export type RerunViewportStatus = "idle" | "loading" | "ready" | "error"; +export type RecordedRerunView = "spatial" | "perception" | "metrics"; export interface RerunSelection { entityPath: string; @@ -55,6 +56,8 @@ export interface RerunViewportProps { | "palette" | "customColor" >; + recordedView?: RecordedRerunView; + onPerceptionAvailabilityChange?: (available: boolean) => void; } export interface RecordedRrdArtifactDescriptor { @@ -80,7 +83,9 @@ interface RecordedRerunIdentity { const RECORDED_RRD_PATH = /^\/api\/v1\/observation-sessions\/[A-Za-z0-9][A-Za-z0-9._:-]{0,127}\/recording\.rrd$/; const RECORDED_BLUEPRINT_PATH = /^\/api\/v1\/observation-sessions\/[A-Za-z0-9][A-Za-z0-9._:-]{0,127}\/blueprint\.rrd$/; +const RECORDED_PERCEPTION_PATH = /^\/api\/v1\/observation-sessions\/[A-Za-z0-9][A-Za-z0-9._:-]{0,127}\/perception\.rrd$/; const MAX_BLUEPRINT_BYTES = 1_048_576; +const MAX_PERCEPTION_BYTES = 512 * 1024 * 1024; const BUFFER_END_TOLERANCE_NS = 1_000_000; const RECORDED_OPEN_MIN_TIMEOUT_MS = 120_000; const RECORDED_OPEN_MAX_TIMEOUT_MS = 1_800_000; @@ -405,6 +410,17 @@ export function resolveRecordedBlueprintUrl(sourceUrl: string, origin: string): return endpoint.origin === base.origin ? endpoint.href : null; } +export function resolveRecordedPerceptionUrl(sourceUrl: string, origin: string): string | null { + const normalized = sourceUrl.trim(); + if (!RECORDED_RRD_PATH.test(normalized)) return null; + const base = new URL(origin); + const endpoint = new URL( + normalized.replace(/\/recording\.rrd$/, "/perception.rrd"), + `${base.origin}/`, + ); + return endpoint.origin === base.origin ? endpoint.href : null; +} + export async function fetchRecordedBlueprintRrd( endpointUrl: string, settings: Pick< @@ -421,10 +437,12 @@ export async function fetchRecordedBlueprintRrd( { origin, signal, + activeView = "spatial", fetcher = globalThis.fetch, }: { origin: string; signal?: AbortSignal; + activeView?: RecordedRerunView; fetcher?: typeof globalThis.fetch; }, ): Promise { @@ -443,6 +461,7 @@ export async function fetchRecordedBlueprintRrd( settings.pointSize > 32 || !["turbo", "viridis", "plasma", "grayscale", "custom"].includes(settings.palette) || !/^#[0-9A-Fa-f]{6}$/.test(settings.customColor) || + !["spatial", "perception", "metrics"].includes(activeView) || identity.applicationId !== "nodedc_mission_core_recorded" || !/^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$/.test(identity.recordingId) ) { @@ -465,6 +484,7 @@ export async function fetchRecordedBlueprintRrd( point_size: settings.pointSize, palette: settings.palette, custom_color: settings.customColor, + active_view: activeView, }), signal, }); @@ -491,6 +511,69 @@ export async function fetchRecordedBlueprintRrd( return payload; } +export async function fetchRecordedPerceptionRrd( + endpointUrl: string, + identity: RecordedRerunIdentity, + { + origin, + signal, + fetcher = globalThis.fetch, + }: { + origin: string; + signal?: AbortSignal; + fetcher?: typeof globalThis.fetch; + }, +): Promise { + const base = new URL(origin); + const endpoint = new URL(endpointUrl, base.origin); + if ( + endpoint.origin !== base.origin || + endpoint.search || + endpoint.hash || + !RECORDED_PERCEPTION_PATH.test(endpoint.pathname) || + identity.applicationId !== "nodedc_mission_core_recorded" || + !/^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$/.test(identity.recordingId) + ) { + throw new Error("Unsafe recorded perception request"); + } + const response = await fetcher(endpoint.href, { + method: "POST", + credentials: "same-origin", + headers: { + Accept: "application/vnd.rerun.rrd", + "Content-Type": "application/json", + }, + body: JSON.stringify({ + application_id: identity.applicationId, + recording_id: identity.recordingId, + }), + signal, + }); + if (response.status === 204) return null; + const contentType = response.headers.get("Content-Type")?.split(";", 1)[0].trim(); + const declaredLength = Number(response.headers.get("Content-Length")); + if ( + !response.ok || + contentType !== "application/vnd.rerun.rrd" || + !Number.isSafeInteger(declaredLength) || + declaredLength < 4 || + declaredLength > MAX_PERCEPTION_BYTES + ) { + throw new Error("Invalid recorded perception response"); + } + const payload = new Uint8Array(await response.arrayBuffer()); + if ( + payload.byteLength !== declaredLength || + payload[0] !== 0x52 || + payload[1] !== 0x52 || + payload[2] !== 0x46 || + payload[3] !== 0x32 + ) { + throw new Error("Invalid recorded perception RRD"); + } + return payload; +} + export function RerunViewport({ sourceUrl, recordedArtifact = null, @@ -504,19 +587,26 @@ export function RerunViewport({ onPlaybackChange, onPlaybackControllerChange, sceneSettings, + recordedView = "spatial", + onPerceptionAvailabilityChange, }: RerunViewportProps) { const hostRef = useRef(null); const [status, setStatus] = useState(sourceUrl ? "loading" : "idle"); const [recordingBufferProgress, setRecordingBufferProgress] = useState(null); const [retryNonce, setRetryNonce] = useState(0); const blueprintChannelRef = useRef(null); + const perceptionChannelRef = useRef(null); const recordedIdentityRef = useRef(null); const presentationGateRef = useRef(presentationGate); presentationGateRef.current = presentationGate; const [blueprintChannelRevision, setBlueprintChannelRevision] = useState(0); + const [perceptionChannelRevision, setPerceptionChannelRevision] = useState(0); const recordedBlueprintUrl = sourceUrl ? resolveRecordedBlueprintUrl(sourceUrl, window.location.origin) : null; + const recordedPerceptionUrl = sourceUrl + ? resolveRecordedPerceptionUrl(sourceUrl, window.location.origin) + : null; const presentationStatus = rerunPresentationStatus( status, presentationGate, @@ -580,6 +670,7 @@ export function RerunViewport({ let playbackState: RerunPlaybackState | null = null; const recordedAutoplay = createRecordedAutoplayGate(); let blueprintChannel: RerunBlueprintChannel | null = null; + let perceptionChannel: RerunBlueprintChannel | null = null; const unsubscribers: Array<() => void> = []; const unsubscribeAll = () => { while (unsubscribers.length > 0) { @@ -665,12 +756,20 @@ export function RerunViewport({ if (blueprintChannelRef.current === blueprintChannel) { blueprintChannelRef.current = null; } + if (perceptionChannelRef.current === perceptionChannel) { + perceptionChannelRef.current = null; + } recordedIdentityRef.current = null; try { blueprintChannel?.channel.close(); } catch { // The viewer may already have closed all auxiliary channels. } + try { + perceptionChannel?.channel.close(); + } catch { + // The viewer may already have closed all auxiliary channels. + } try { if (viewer.ready) viewer.close(resolvedSource); } catch { @@ -700,7 +799,11 @@ export function RerunViewport({ unsubscribers.push( viewer.on("recording_open", (event) => { - if (disposed || recordingOpened) return; + if ( + disposed || + recordingOpened || + (isRecordedSource && event.application_id !== "nodedc_mission_core_recorded") + ) return; recordingOpened = true; if ( recordedBlueprintUrl && @@ -717,6 +820,7 @@ export function RerunViewport({ recordingId: event.recording_id, }; setBlueprintChannelRevision((revision) => revision + 1); + setPerceptionChannelRevision((revision) => revision + 1); } } if (!isRecordedSource) clearLiveRecordingOpenTimer(); @@ -961,6 +1065,12 @@ export function RerunViewport({ blueprintChannelRef.current = blueprintChannel; setBlueprintChannelRevision((revision) => revision + 1); } + if (recordedPerceptionUrl) { + const channel = viewer.open_channel("missioncore/recorded-perception"); + perceptionChannel = { endpointUrl: recordedPerceptionUrl, channel }; + perceptionChannelRef.current = perceptionChannel; + setPerceptionChannelRevision((revision) => revision + 1); + } if (!recordingOpened && !isRecordedSource) { recordingOpenTimer = window.setTimeout(() => { @@ -1012,6 +1122,50 @@ export function RerunViewport({ sourceUrl, ]); + useEffect(() => { + onPerceptionAvailabilityChange?.(false); + }, [onPerceptionAvailabilityChange, recordedPerceptionUrl]); + + useEffect(() => { + if (!recordedPerceptionUrl) return; + const active = perceptionChannelRef.current; + const identity = recordedIdentityRef.current; + if ( + !active || + !identity || + active.endpointUrl !== recordedPerceptionUrl || + !active.channel.ready + ) return; + const abort = new AbortController(); + void fetchRecordedPerceptionRrd(recordedPerceptionUrl, identity, { + origin: window.location.origin, + signal: abort.signal, + }).then((payload) => { + if (payload === null) { + onPerceptionAvailabilityChange?.(false); + return; + } + if ( + abort.signal.aborted || + perceptionChannelRef.current !== active || + recordedIdentityRef.current !== identity || + !active.channel.ready + ) { + return; + } + active.channel.send_rrd(payload); + onPerceptionAvailabilityChange?.(true); + }).catch(() => { + onPerceptionAvailabilityChange?.(false); + // The base recording remains available when no admitted perception layer exists. + }); + return () => abort.abort(); + }, [ + onPerceptionAvailabilityChange, + perceptionChannelRevision, + recordedPerceptionUrl, + ]); + useEffect(() => { if (!recordedBlueprintUrl || !sceneSettings) return; const active = blueprintChannelRef.current; @@ -1026,6 +1180,7 @@ export function RerunViewport({ void fetchRecordedBlueprintRrd(recordedBlueprintUrl, sceneSettings, identity, { origin: window.location.origin, signal: abort.signal, + activeView: recordedView, }).then((payload) => { if ( abort.signal.aborted || @@ -1044,6 +1199,7 @@ export function RerunViewport({ }, [ blueprintChannelRevision, recordedBlueprintUrl, + recordedView, sceneSettings?.accumulationSeconds, sceneSettings?.customColor, sceneSettings?.palette, diff --git a/apps/control-station/src/workspaces/Workspaces.tsx b/apps/control-station/src/workspaces/Workspaces.tsx index d5f7557..ceb9be1 100644 --- a/apps/control-station/src/workspaces/Workspaces.tsx +++ b/apps/control-station/src/workspaces/Workspaces.tsx @@ -45,6 +45,7 @@ import { type RerunPlaybackState, type RerunSelection, type RerunViewportStatus, + type RecordedRerunView, } from "../components/RerunViewport"; import { capabilityStatusLabel, @@ -294,6 +295,8 @@ function SpatialWorkspace({ const [selection, setSelection] = useState(null); const [playbackState, setPlaybackState] = useState(null); const [playbackController, setPlaybackController] = useState(null); + const [recordedRerunView, setRecordedRerunView] = useState("spatial"); + const [perceptionAvailable, setPerceptionAvailable] = useState(false); const recordedSource = state?.sourceMode === "replay" || /\.rrd(?:$|[?#])/i.test(sourceUrl); const recordedSessionGate: RecordedAdmissionPhase = recordedSource ? recordedSessionAdmission?.phase ?? "loading" @@ -369,6 +372,15 @@ function SpatialWorkspace({ (next: RerunPlaybackController | null) => setPlaybackController(next), [], ); + const onPerceptionAvailabilityChange = useCallback((available: boolean) => { + setPerceptionAvailable(available); + if (!available) setRecordedRerunView("spatial"); + }, []); + + useEffect(() => { + setPerceptionAvailable(false); + setRecordedRerunView("spatial"); + }, [sourceUrl]); useEffect(() => { if (pointCloudVisible && sourceUrl.trim()) return; @@ -377,6 +389,8 @@ function SpatialWorkspace({ setSelection(null); setPlaybackState(null); setPlaybackController(null); + setPerceptionAvailable(false); + setRecordedRerunView("spatial"); }, [pointCloudVisible, sourceUrl]); useEffect(() => { @@ -419,6 +433,17 @@ function SpatialWorkspace({ СЦЕНА 3D · RERUN
+ {recordedSource && perceptionAvailable ? ( + + ) : null} @@ -451,6 +476,8 @@ function SpatialWorkspace({ ? state?.observationTimeline?.range?.endSeconds : undefined} sceneSettings={sceneSettings} + recordedView={recordedRerunView} + onPerceptionAvailabilityChange={onPerceptionAvailabilityChange} onStatusChange={onStatusChange} onSelectionChange={onSelectionChange} onPlaybackChange={onPlaybackChange} @@ -630,7 +657,7 @@ function SpatialWorkspace({ Траектория Преобразования Камеры в 3D - Объекты / маски + Объекты / рамки Компоновка
diff --git a/apps/control-station/test/observationSources.test.mjs b/apps/control-station/test/observationSources.test.mjs index 95db8db..603cb5c 100644 --- a/apps/control-station/test/observationSources.test.mjs +++ b/apps/control-station/test/observationSources.test.mjs @@ -23,6 +23,8 @@ let formatAccumulationDuration; let resolveRerunSourceUrl; let resolveRecordedBlueprintUrl; let fetchRecordedBlueprintRrd; +let resolveRecordedPerceptionUrl; +let fetchRecordedPerceptionRrd; let isRecordedPlaybackFullyBuffered; let recordedObservationSources; let selectRecordedMediaEpoch; @@ -62,6 +64,8 @@ before(async () => { resolveRerunSourceUrl, resolveRecordedBlueprintUrl, fetchRecordedBlueprintRrd, + resolveRecordedPerceptionUrl, + fetchRecordedPerceptionRrd, isRecordedPlaybackFullyBuffered, } = await server.ssrLoadModule( "/src/components/RerunViewport.tsx", @@ -265,6 +269,7 @@ test("recorded blueprint fetch is bounded, strict and sends only display setting { applicationId: "nodedc_mission_core_recorded", recordingId: "recording-001" }, { origin: "http://127.0.0.1:5174", + activeView: "perception", fetcher: async (input, init) => { calls.push({ input: String(input), init, body: JSON.parse(String(init.body)) }); return new Response(payload, { @@ -287,6 +292,7 @@ test("recorded blueprint fetch is bounded, strict and sends only display setting point_size: 4.5, palette: "custom", custom_color: "#35d7c1", + active_view: "perception", }); await assert.rejects( @@ -308,6 +314,49 @@ test("recorded blueprint fetch is bounded, strict and sends only display setting ); }); +test("recorded perception fetch admits one complete same-origin RRD or no layer", async () => { + const endpoint = resolveRecordedPerceptionUrl( + "/api/v1/observation-sessions/session-1/recording.rrd", + "http://127.0.0.1:5174", + ); + assert.equal( + endpoint, + "http://127.0.0.1:5174/api/v1/observation-sessions/session-1/perception.rrd", + ); + const payload = Uint8Array.from([0x52, 0x52, 0x46, 0x32, 0x01]); + const result = await fetchRecordedPerceptionRrd( + endpoint, + { applicationId: "nodedc_mission_core_recorded", recordingId: "recording-001" }, + { + origin: "http://127.0.0.1:5174", + fetcher: async (_input, init) => { + assert.deepEqual(JSON.parse(String(init.body)), { + application_id: "nodedc_mission_core_recorded", + recording_id: "recording-001", + }); + return new Response(payload, { + status: 200, + headers: { + "Content-Type": "application/vnd.rerun.rrd", + "Content-Length": String(payload.byteLength), + }, + }); + }, + }, + ); + assert.deepEqual([...result], [...payload]); + + const absent = await fetchRecordedPerceptionRrd( + endpoint, + { applicationId: "nodedc_mission_core_recorded", recordingId: "recording-001" }, + { + origin: "http://127.0.0.1:5174", + fetcher: async () => new Response(null, { status: 204 }), + }, + ); + assert.equal(absent, null); +}); + test("recorded replay creates an isolated source catalog without live device bindings", () => { const sources = recordedObservationSources({ kind: "rerun-recording", diff --git a/src/k1link/compute/__init__.py b/src/k1link/compute/__init__.py index e39194d..2a1fff4 100644 --- a/src/k1link/compute/__init__.py +++ b/src/k1link/compute/__init__.py @@ -6,10 +6,24 @@ from .jobs import ( prepare_camera_compute_job, validate_camera_compute_job, ) +from .results import ( + DetectionFrame, + ObjectDetection, + RecordedPerceptionOverlayError, + RecordedPerceptionOverlayStore, + RecordedPerceptionResult, + validate_recorded_perception_result, +) __all__ = [ "COMPUTE_JOB_SCHEMA", "CameraComputeJob", "prepare_camera_compute_job", "validate_camera_compute_job", + "DetectionFrame", + "ObjectDetection", + "RecordedPerceptionOverlayError", + "RecordedPerceptionOverlayStore", + "RecordedPerceptionResult", + "validate_recorded_perception_result", ] diff --git a/src/k1link/compute/jobs.py b/src/k1link/compute/jobs.py index 6c647a8..8cecfb1 100644 --- a/src/k1link/compute/jobs.py +++ b/src/k1link/compute/jobs.py @@ -30,6 +30,9 @@ class CameraComputeJob: job_id: str job_root: Path manifest_path: Path + session_id: str + source_id: str + codec_epoch: int input_sha256: str input_byte_length: int segment_count: int @@ -170,11 +173,14 @@ def validate_camera_compute_job(job_root: Path) -> CameraComputeJob: raise SessionIntegrityError("compute job input generation is inconsistent") source_id = input_document.get("source_id") + session_id = input_document.get("session_id") codec_epoch = input_document.get("codec_epoch") timeline = input_document.get("timeline") files = input_document.get("files") if ( - not isinstance(source_id, str) + not isinstance(session_id, str) + or _SAFE_COMPONENT.fullmatch(session_id) is None + or not isinstance(source_id, str) or _SAFE_COMPONENT.fullmatch(source_id) is None or not isinstance(codec_epoch, int) or isinstance(codec_epoch, bool) @@ -235,6 +241,9 @@ def validate_camera_compute_job(job_root: Path) -> CameraComputeJob: job_id=root.name, job_root=root, manifest_path=manifest_path, + session_id=session_id, + source_id=source_id, + codec_epoch=codec_epoch, input_sha256=input_sha256, input_byte_length=total_bytes, segment_count=segment_count, diff --git a/src/k1link/compute/results.py b/src/k1link/compute/results.py new file mode 100644 index 0000000..b704662 --- /dev/null +++ b/src/k1link/compute/results.py @@ -0,0 +1,656 @@ +from __future__ import annotations + +import hashlib +import json +import math +import os +import re +import stat +import subprocess +import threading +from dataclasses import dataclass +from pathlib import Path +from typing import Any, TypeGuard + +import numpy as np +import rerun as rr + +from k1link.artifacts import write_json_atomic +from k1link.sessions import SessionIntegrityError + +from .jobs import CameraComputeJob, validate_camera_compute_job + +COMPUTE_RESULT_SCHEMA = "missioncore.compute-result/v1" +COMPUTE_RESULT_IDENTITY_SCHEMA = "missioncore.compute-result-identity/v1" +OBJECT_DETECTIONS_SCHEMA = "missioncore.object-detections/v1" +SESSION_TIMELINE = "session_time" + +MAX_RESULT_JSON_BYTES = 32 * 1024 * 1024 +MAX_RESULT_FRAMES = 10_000 +MAX_DETECTIONS_PER_FRAME = 10_000 +MAX_OVERLAY_SOURCE_BYTES = 512 * 1024 * 1024 +MAX_OVERLAY_DECODED_BYTES = 512 * 1024 * 1024 +MAX_OVERLAY_RRD_BYTES = 512 * 1024 * 1024 +MAX_JOB_SCAN = 512 + +_SAFE_RESULT_ID = re.compile(r"^result-[a-f0-9]{64}$") +_SAFE_RECORDING_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$") +_SHA256 = re.compile(r"^[a-f0-9]{64}$") + + +@dataclass(frozen=True, slots=True) +class ObjectDetection: + class_id: int + class_name: str + score: float + bbox_xyxy: tuple[float, float, float, float] + + +@dataclass(frozen=True, slots=True) +class DetectionFrame: + frame_index: int + session_time_ns: int + detections: tuple[ObjectDetection, ...] + + +@dataclass(frozen=True, slots=True) +class RecordedPerceptionResult: + result_id: str + result_root: Path + job: CameraComputeJob + created_at_utc: str + frames: tuple[DetectionFrame, ...] + + +class RecordedPerceptionOverlayError(RuntimeError): + """A validated compute result could not be projected into Rerun.""" + + +def validate_recorded_perception_result( + job_root: Path, + result_root: Path, +) -> RecordedPerceptionResult: + """Validate one content-addressed worker result against its exact job.""" + + job = validate_camera_compute_job(job_root) + root = result_root.expanduser().resolve(strict=True) + if not root.is_dir() or _SAFE_RESULT_ID.fullmatch(root.name) is None: + raise SessionIntegrityError("compute result root is invalid") + + result = _read_json_object(root / "result.json", root) + detections = _read_json_object(root / "detections.json", root) + identity_sha256 = result.get("identity_sha256") + pipeline = result.get("pipeline") + model = result.get("model") + parameters = result.get("parameters") + if ( + result.get("schema_version") != COMPUTE_RESULT_SCHEMA + or result.get("result_id") != root.name + or not isinstance(identity_sha256, str) + or _SHA256.fullmatch(identity_sha256) is None + or root.name != f"result-{identity_sha256}" + or result.get("job_id") != job.job_id + or result.get("input_sha256") != job.input_sha256 + or not isinstance(pipeline, dict) + or not isinstance(model, dict) + or not isinstance(parameters, dict) + ): + raise SessionIntegrityError("compute result identity is inconsistent") + identity = { + "schema_version": COMPUTE_RESULT_IDENTITY_SCHEMA, + "job_id": job.job_id, + "input_sha256": job.input_sha256, + "pipeline": pipeline, + "model": { + "id": model.get("id"), + "version": model.get("version"), + "sha256": model.get("sha256"), + }, + "parameters": parameters, + } + if hashlib.sha256(_canonical_json(identity)).hexdigest() != identity_sha256: + raise SessionIntegrityError("compute result generation is inconsistent") + + artifacts = result.get("artifacts") + detection_path = root / "detections.json" + detection_stat = _confined_regular_file(detection_path, root) + if not isinstance(artifacts, list) or len(artifacts) != 1: + raise SessionIntegrityError("compute result artifact descriptor is invalid") + artifact = artifacts[0] + if ( + not isinstance(artifact, dict) + or artifact.get("kind") != "object-detections" + or artifact.get("path") != "detections.json" + or artifact.get("schema_version") != OBJECT_DETECTIONS_SCHEMA + or artifact.get("byte_length") != detection_stat.st_size + or artifact.get("sha256") != _sha256_file(detection_path) + ): + raise SessionIntegrityError("compute result artifact identity changed") + + if ( + detections.get("schema_version") != OBJECT_DETECTIONS_SCHEMA + or detections.get("result_id") != root.name + or detections.get("job_id") != job.job_id + or detections.get("input_sha256") != job.input_sha256 + or detections.get("timestamp_basis") != "session-time-seconds" + ): + raise SessionIntegrityError("object detections are not bound to the result") + raw_frames = detections.get("frames") + if not isinstance(raw_frames, list) or not 1 <= len(raw_frames) <= MAX_RESULT_FRAMES: + raise SessionIntegrityError("object detection frame set is outside bounds") + + frames: list[DetectionFrame] = [] + detection_count = 0 + previous_time_ns = -1 + for expected_index, value in enumerate(raw_frames): + frame = _validate_detection_frame(value, expected_index, job) + if frame.session_time_ns <= previous_time_ns: + raise SessionIntegrityError("object detection timestamps are not increasing") + previous_time_ns = frame.session_time_ns + frames.append(frame) + detection_count += len(frame.detections) + + metrics = result.get("metrics") + if ( + not isinstance(metrics, dict) + or metrics.get("frames_processed") != len(frames) + or metrics.get("detections") != detection_count + ): + raise SessionIntegrityError("compute result metrics do not match its artifact") + created_at_utc = result.get("created_at_utc") + if not isinstance(created_at_utc, str) or len(created_at_utc) > 64: + raise SessionIntegrityError("compute result creation time is invalid") + return RecordedPerceptionResult( + result_id=root.name, + result_root=root, + job=job, + created_at_utc=created_at_utc, + frames=tuple(frames), + ) + + +class RecordedPerceptionOverlayStore: + """Discover admitted local results and cache complete overlay RRD files.""" + + def __init__( + self, + *, + jobs_root: Path, + results_root: Path, + cache_root: Path, + ffmpeg_path: Path, + ffprobe_path: Path, + ) -> None: + self.jobs_root = jobs_root.expanduser().absolute() + self.results_root = results_root.expanduser().absolute() + self.cache_root = cache_root.expanduser().absolute() + self.ffmpeg_path = ffmpeg_path.expanduser().resolve(strict=True) + self.ffprobe_path = ffprobe_path.expanduser().resolve(strict=True) + self._lock = threading.Lock() + + def render( + self, + session_id: str, + *, + application_id: str, + recording_id: str, + ) -> bytes | None: + if _SAFE_RECORDING_ID.fullmatch(session_id) is None: + raise ValueError("observation session id is invalid") + if application_id != "nodedc_mission_core_recorded": + raise ValueError("recorded perception application id is invalid") + if _SAFE_RECORDING_ID.fullmatch(recording_id) is None: + raise ValueError("recorded perception recording id is invalid") + with self._lock: + result = self._latest_result(session_id) + if result is None: + return None + cache_root = _private_directory(self.cache_root) + session_cache = _private_child_directory(cache_root, session_id) + cache_dir = _private_child_directory(session_cache, result.result_id) + output = cache_dir / f"{recording_id}.rrd" + sidecar = output.with_suffix(".rrd.cache.json") + cached = _read_cached_overlay(output, sidecar, result) + if cached is not None: + return cached + payload = _render_overlay( + result, + application_id=application_id, + recording_id=recording_id, + ffmpeg_path=self.ffmpeg_path, + ffprobe_path=self.ffprobe_path, + ) + temporary = output.with_name(f".{output.name}.{os.getpid()}.tmp") + try: + with temporary.open("xb") as stream: + stream.write(payload) + stream.flush() + os.fsync(stream.fileno()) + os.chmod(temporary, 0o600) + os.replace(temporary, output) + write_json_atomic( + sidecar, + { + "schema_version": "missioncore.perception-overlay-cache/v1", + "result_id": result.result_id, + "recording_id": recording_id, + "byte_length": len(payload), + "sha256": hashlib.sha256(payload).hexdigest(), + }, + ) + finally: + temporary.unlink(missing_ok=True) + return payload + + def _latest_result(self, session_id: str) -> RecordedPerceptionResult | None: + try: + job_roots = sorted(self.jobs_root.iterdir()) + except FileNotFoundError: + return None + if len(job_roots) > MAX_JOB_SCAN: + raise RecordedPerceptionOverlayError("compute job catalog is outside bounds") + matches: list[RecordedPerceptionResult] = [] + for job_root in job_roots: + if job_root.is_symlink(): + continue + try: + job = validate_camera_compute_job(job_root) + except (OSError, SessionIntegrityError): + continue + if job.session_id != session_id: + continue + result_parent = self.results_root / job.job_id + try: + result_roots = sorted( + path + for path in result_parent.iterdir() + if path.is_dir() + and not path.is_symlink() + and _SAFE_RESULT_ID.fullmatch(path.name) is not None + ) + except FileNotFoundError: + continue + for result_root in result_roots: + matches.append(validate_recorded_perception_result(job_root, result_root)) + if not matches: + return None + return max(matches, key=lambda value: (value.created_at_utc, value.result_id)) + + +def _validate_detection_frame( + value: object, + expected_index: int, + job: CameraComputeJob, +) -> DetectionFrame: + if not isinstance(value, dict) or value.get("frame_index") != expected_index: + raise SessionIntegrityError("object detection frame index is inconsistent") + session_seconds = value.get("session_seconds") + epoch_seconds = value.get("epoch_seconds") + raw_detections = value.get("detections") + if ( + not _finite_number(session_seconds) + or not _finite_number(epoch_seconds) + or float(epoch_seconds) < 0 + or float(session_seconds) < job.timeline_start_seconds - 0.001 + or float(session_seconds) > job.timeline_end_seconds + 0.001 + or not isinstance(raw_detections, list) + or len(raw_detections) > MAX_DETECTIONS_PER_FRAME + ): + raise SessionIntegrityError("object detection frame is invalid") + found = tuple(_validate_detection(item) for item in raw_detections) + return DetectionFrame( + frame_index=expected_index, + session_time_ns=round(float(session_seconds) * 1_000_000_000), + detections=found, + ) + + +def _validate_detection(value: object) -> ObjectDetection: + if not isinstance(value, dict): + raise SessionIntegrityError("object detection is invalid") + class_id = value.get("class_id") + class_name = value.get("class_name") + score = value.get("score") + bbox = value.get("bbox_xyxy") + if ( + not isinstance(class_id, int) + or isinstance(class_id, bool) + or not 0 <= class_id <= 10_000 + or not isinstance(class_name, str) + or not class_name + or len(class_name) > 128 + or not _finite_number(score) + or not 0 <= float(score) <= 1 + or not isinstance(bbox, list) + or len(bbox) != 4 + or not all(_finite_number(item) for item in bbox) + ): + raise SessionIntegrityError("object detection fields are invalid") + x1, y1, x2, y2 = (float(item) for item in bbox) + if min(x1, y1) < 0 or x2 < x1 or y2 < y1 or max(x2, y2) > 100_000: + raise SessionIntegrityError("object detection bounds are invalid") + return ObjectDetection(class_id, class_name, float(score), (x1, y1, x2, y2)) + + +def _render_overlay( + result: RecordedPerceptionResult, + *, + application_id: str, + recording_id: str, + ffmpeg_path: Path, + ffprobe_path: Path, +) -> bytes: + if validate_camera_compute_job(result.job.job_root) != result.job: + raise RecordedPerceptionOverlayError("camera compute job changed before projection") + source = _camera_source_bytes(result.job) + width, height, frame_count = _probe_video(source, ffprobe_path) + if frame_count != len(result.frames): + raise RecordedPerceptionOverlayError("camera and detection frame counts differ") + _validate_image_bounds(result, width, height) + frame_bytes = width * height * 3 + expected_bytes = frame_bytes * len(result.frames) + if expected_bytes > MAX_OVERLAY_DECODED_BYTES: + raise RecordedPerceptionOverlayError("decoded camera overlay is outside bounds") + decoded = _decode_rgb(source, ffmpeg_path, frame_count) + if len(decoded) != expected_bytes: + raise RecordedPerceptionOverlayError("decoded camera frame count changed") + + recording = rr.RecordingStream( + application_id, + recording_id=recording_id, + send_properties=False, + ) + stream = rr.binary_stream(recording) + try: + recording.log( + "/perception/camera/metadata", + rr.AnyValues( + result_id=result.result_id, + job_id=result.job.job_id, + source_id=result.job.source_id, + pipeline="recorded-camera-coco-detection/v1", + warning="Generic detections are evidence, not a driving decision.", + ), + static=True, + ) + for frame in result.frames: + offset = frame.frame_index * frame_bytes + image = np.frombuffer( + decoded, + dtype=np.uint8, + count=frame_bytes, + offset=offset, + ).reshape((height, width, 3)) + recording.set_time( + SESSION_TIMELINE, + duration=np.timedelta64(frame.session_time_ns, "ns"), + ) + recording.log( + "/perception/camera/image", + rr.Image(image, color_model="RGB"), + ) + if frame.detections: + recording.log( + "/perception/camera/detections", + rr.Boxes2D( + array=[detection.bbox_xyxy for detection in frame.detections], + array_format=rr.Box2DFormat.XYXY, + labels=[ + f"{detection.class_name} · {detection.score:.0%}" + for detection in frame.detections + ], + colors=[ + _class_color(detection.class_id) + for detection in frame.detections + ], + show_labels=True, + ), + ) + else: + recording.log( + "/perception/camera/detections", + rr.Clear(recursive=False), + ) + payload = stream.read(flush=True, flush_timeout_sec=120.0) + except Exception as exc: + raise RecordedPerceptionOverlayError("failed to serialize perception overlay") from exc + finally: + recording.disconnect() + if payload is None or not payload.startswith(b"RRF2") or len(payload) > MAX_OVERLAY_RRD_BYTES: + raise RecordedPerceptionOverlayError("serialized perception overlay is invalid") + return payload + + +def _camera_source_bytes(job: CameraComputeJob) -> bytes: + epoch = ( + job.job_root + / "input" + / "camera" + / job.source_id + / f"epoch-{job.codec_epoch}" + ) + paths = [epoch / "init.mp4"] + [ + epoch / "segments" / f"{sequence}.m4s" + for sequence in range(1, job.segment_count + 1) + ] + total = sum(path.stat().st_size for path in paths) + if total > MAX_OVERLAY_SOURCE_BYTES: + raise RecordedPerceptionOverlayError("camera overlay source is outside bounds") + return b"".join(path.read_bytes() for path in paths) + + +def _probe_video(source: bytes, ffprobe_path: Path) -> tuple[int, int, int]: + completed = _run_media_tool( + [ + str(ffprobe_path), + "-v", + "error", + "-count_frames", + "-select_streams", + "v:0", + "-show_entries", + "stream=width,height,nb_read_frames", + "-of", + "json", + "-i", + "pipe:0", + ], + source, + ) + try: + document = json.loads(completed.stdout) + stream = document["streams"][0] + width = int(stream["width"]) + height = int(stream["height"]) + frame_count = int(stream["nb_read_frames"]) + except (IndexError, KeyError, TypeError, ValueError, json.JSONDecodeError) as exc: + raise RecordedPerceptionOverlayError("camera dimensions are unavailable") from exc + if ( + width < 1 + or height < 1 + or width * height > 4_194_304 + or not 1 <= frame_count <= MAX_RESULT_FRAMES + ): + raise RecordedPerceptionOverlayError("camera dimensions are outside bounds") + return width, height, frame_count + + +def _decode_rgb(source: bytes, ffmpeg_path: Path, frame_count: int) -> bytes: + return _run_media_tool( + [ + str(ffmpeg_path), + "-v", + "error", + "-i", + "pipe:0", + "-map", + "0:v:0", + "-frames:v", + str(frame_count), + "-fps_mode", + "passthrough", + "-pix_fmt", + "rgb24", + "-f", + "rawvideo", + "pipe:1", + ], + source, + ).stdout + + +def _run_media_tool(argv: list[str], payload: bytes) -> subprocess.CompletedProcess[bytes]: + try: + metadata = Path(argv[0]).lstat() + if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISREG(metadata.st_mode): + raise RecordedPerceptionOverlayError("media tool is not a regular file") + completed = subprocess.run( + argv, + input=payload, + capture_output=True, + check=False, + timeout=120, + ) + except (OSError, subprocess.SubprocessError) as exc: + raise RecordedPerceptionOverlayError("media tool failed") from exc + if completed.returncode != 0: + raise RecordedPerceptionOverlayError("camera media could not be decoded") + return completed + + +def _read_cached_overlay( + output: Path, + sidecar: Path, + result: RecordedPerceptionResult, +) -> bytes | None: + try: + value = _read_json_object(sidecar, sidecar.parent) + _confined_regular_file(output, output.parent) + payload = output.read_bytes() + except (OSError, SessionIntegrityError): + return None + if ( + value.get("schema_version") != "missioncore.perception-overlay-cache/v1" + or value.get("result_id") != result.result_id + or value.get("recording_id") != output.stem + or value.get("byte_length") != len(payload) + or value.get("sha256") != hashlib.sha256(payload).hexdigest() + or not payload.startswith(b"RRF2") + or len(payload) > MAX_OVERLAY_RRD_BYTES + ): + return None + return payload + + +def _validate_image_bounds( + result: RecordedPerceptionResult, + width: int, + height: int, +) -> None: + for frame in result.frames: + for detection in frame.detections: + x1, y1, x2, y2 = detection.bbox_xyxy + if x1 > width or x2 > width or y1 > height or y2 > height: + raise RecordedPerceptionOverlayError( + "object detection escapes the decoded camera image" + ) + + +def _private_directory(path: Path) -> Path: + path.mkdir(mode=0o700, parents=True, exist_ok=True) + try: + metadata = path.lstat() + except OSError as exc: + raise RecordedPerceptionOverlayError("perception cache is unavailable") from exc + if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISDIR(metadata.st_mode): + raise RecordedPerceptionOverlayError("perception cache is not a real directory") + resolved = path.resolve(strict=True) + os.chmod(resolved, 0o700) + return resolved + + +def _private_child_directory(parent: Path, name: str) -> Path: + child = parent / name + child.mkdir(mode=0o700, exist_ok=True) + try: + metadata = child.lstat() + resolved = child.resolve(strict=True) + except OSError as exc: + raise RecordedPerceptionOverlayError("perception cache child is unavailable") from exc + if ( + stat.S_ISLNK(metadata.st_mode) + or not stat.S_ISDIR(metadata.st_mode) + or resolved.parent != parent + ): + raise RecordedPerceptionOverlayError("perception cache child escapes its root") + os.chmod(resolved, 0o700) + return resolved + + +def _class_color(class_id: int) -> list[int]: + palette = ( + (185, 255, 74, 255), + (67, 191, 255, 255), + (255, 193, 92, 255), + (222, 110, 255, 255), + (255, 103, 117, 255), + ) + return list(palette[class_id % len(palette)]) + + +def _read_json_object(path: Path, root: Path) -> dict[str, Any]: + metadata = _confined_regular_file(path, root) + if not 0 < metadata.st_size <= MAX_RESULT_JSON_BYTES: + raise SessionIntegrityError("compute result JSON is outside bounds") + try: + value = json.loads(path.read_text(encoding="utf-8-sig")) + except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc: + raise SessionIntegrityError("compute result JSON is unavailable") from exc + if not isinstance(value, dict): + raise SessionIntegrityError("compute result JSON is not an object") + return value + + +def _confined_regular_file(path: Path, root: Path) -> os.stat_result: + try: + resolved_root = root.resolve(strict=True) + resolved = path.resolve(strict=True) + metadata = path.lstat() + except OSError as exc: + raise SessionIntegrityError("compute result artifact is unavailable") from exc + if ( + stat.S_ISLNK(metadata.st_mode) + or not stat.S_ISREG(metadata.st_mode) + or not resolved.is_relative_to(resolved_root) + ): + raise SessionIntegrityError("compute result artifact is not a confined regular file") + return metadata + + +def _sha256_file(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as stream: + while chunk := stream.read(1024 * 1024): + digest.update(chunk) + return digest.hexdigest() + + +def _canonical_json(value: object) -> bytes: + try: + return json.dumps( + value, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + allow_nan=False, + ).encode("utf-8") + except (TypeError, ValueError) as exc: + raise SessionIntegrityError("compute result identity cannot be encoded") from exc + + +def _finite_number(value: object) -> TypeGuard[int | float]: + return ( + isinstance(value, (int, float)) + and not isinstance(value, bool) + and math.isfinite(float(value)) + ) diff --git a/src/k1link/device_plugins/xgrids_k1/rrd_export.py b/src/k1link/device_plugins/xgrids_k1/rrd_export.py index a5d0ab4..8183b9f 100644 --- a/src/k1link/device_plugins/xgrids_k1/rrd_export.py +++ b/src/k1link/device_plugins/xgrids_k1/rrd_export.py @@ -8,7 +8,7 @@ from collections.abc import Callable from contextlib import suppress from dataclasses import dataclass from pathlib import Path -from typing import TypedDict +from typing import Literal, TypedDict from uuid import UUID, uuid4 import numpy as np @@ -60,7 +60,9 @@ JS_MAX_SAFE_INTEGER = (1 << 53) - 1 RECORDED_SPATIAL_VIEW_ID = UUID("5f5f11d5-3b0a-4a81-887b-2be767cba1c0") RECORDED_ROOT_CONTAINER_ID = UUID("b02f2aca-8471-4dcb-b786-53df5a320fc8") RECORDED_POINTS_VISUALIZER_ID = UUID("ca037ec0-8761-4417-86ee-846fa2875303") +RECORDED_CAMERA_VIEW_ID = UUID("5c1db75b-07cd-479a-903d-f4f2ed554513") RECORDED_METRICS_VIEW_ID = UUID("f973fc11-0867-4732-ad3c-97008621fab7") +RecordedView = Literal["spatial", "perception", "metrics"] class RrdExportSummary(TypedDict): @@ -496,6 +498,7 @@ def _recorded_blueprint( settings: RerunSceneSettings, *, include_initial_playback_state: bool = True, + active_view: RecordedView = "spatial", ) -> rrb.Blueprint: accumulation = max(0.0, settings.accumulation_seconds) # An omitted range means Rerun's native latest-at query: the most recent @@ -551,12 +554,24 @@ def _recorded_blueprint( time_ranges=time_ranges, ) spatial_view.id = RECORDED_SPATIAL_VIEW_ID + camera_view = rrb.Spatial2DView( + origin="/perception/camera", + name="Камера · распознавание", + background=[7, 8, 10, 255], + ) + camera_view.id = RECORDED_CAMERA_VIEW_ID metrics_view = rrb.TimeSeriesView( origin="/metrics/device", name="Маршрут и время", ) metrics_view.id = RECORDED_METRICS_VIEW_ID - root_container = rrb.Tabs(spatial_view, metrics_view, active_tab=0) + active_tab = {"spatial": 0, "perception": 1, "metrics": 2}[active_view] + root_container = rrb.Tabs( + spatial_view, + camera_view, + metrics_view, + active_tab=active_tab, + ) root_container.id = RECORDED_ROOT_CONTAINER_ID if include_initial_playback_state: diff --git a/src/k1link/viewer/recorded.py b/src/k1link/viewer/recorded.py index 9b6b7c5..59f3cd7 100644 --- a/src/k1link/viewer/recorded.py +++ b/src/k1link/viewer/recorded.py @@ -3,6 +3,7 @@ from __future__ import annotations from contextlib import suppress +from typing import Literal from uuid import UUID import rerun as rr @@ -16,6 +17,9 @@ SESSION_TIMELINE = "session_time" RECORDED_SPATIAL_VIEW_ID = UUID("5f5f11d5-3b0a-4a81-887b-2be767cba1c0") RECORDED_ROOT_CONTAINER_ID = UUID("b02f2aca-8471-4dcb-b786-53df5a320fc8") RECORDED_POINTS_VISUALIZER_ID = UUID("ca037ec0-8761-4417-86ee-846fa2875303") +RECORDED_CAMERA_VIEW_ID = UUID("5c1db75b-07cd-479a-903d-f4f2ed554513") +RECORDED_METRICS_VIEW_ID = UUID("f973fc11-0867-4732-ad3c-97008621fab7") +RecordedView = Literal["spatial", "perception", "metrics"] class RecordedBlueprintError(RuntimeError): @@ -26,6 +30,7 @@ def recorded_blueprint( settings: RerunSceneSettings, *, include_initial_playback_state: bool = True, + active_view: RecordedView = "spatial", ) -> rrb.Blueprint: accumulation = max(0.0, settings.accumulation_seconds) time_ranges: list[rr.VisibleTimeRange] | None = None @@ -67,7 +72,24 @@ def recorded_blueprint( time_ranges=time_ranges, ) spatial_view.id = RECORDED_SPATIAL_VIEW_ID - root_container = rrb.Tabs(spatial_view) + camera_view = rrb.Spatial2DView( + origin="/perception/camera", + name="Камера · распознавание", + background=[7, 8, 10, 255], + ) + camera_view.id = RECORDED_CAMERA_VIEW_ID + metrics_view = rrb.TimeSeriesView( + origin="/metrics/device", + name="Маршрут и время", + ) + metrics_view.id = RECORDED_METRICS_VIEW_ID + active_tab = {"spatial": 0, "perception": 1, "metrics": 2}[active_view] + root_container = rrb.Tabs( + spatial_view, + camera_view, + metrics_view, + active_tab=active_tab, + ) root_container.id = RECORDED_ROOT_CONTAINER_ID if include_initial_playback_state: @@ -95,6 +117,7 @@ def recorded_blueprint_rrd( *, application_id: str = APPLICATION_ID, recording_id: str, + active_view: RecordedView = "spatial", ) -> bytes: """Serialize a bounded active blueprint update without recorded data.""" @@ -109,6 +132,7 @@ def recorded_blueprint_rrd( recorded_blueprint( settings, include_initial_playback_state=False, + active_view=active_view, ), make_active=True, make_default=False, diff --git a/src/k1link/web/app.py b/src/k1link/web/app.py index e7db064..4ea561a 100644 --- a/src/k1link/web/app.py +++ b/src/k1link/web/app.py @@ -1,6 +1,7 @@ from __future__ import annotations import asyncio +import shutil from collections.abc import AsyncIterator, Iterable from contextlib import asynccontextmanager, suppress from pathlib import Path @@ -13,6 +14,7 @@ from fastapi.staticfiles import StaticFiles from pydantic import ValidationError from k1link import __version__ +from k1link.compute import RecordedPerceptionOverlayStore from k1link.sessions import ( MaterializedRecording, RecordedMediaInspector, @@ -51,6 +53,19 @@ session_recording_materializer = SessionRecordingMaterializer( session_recorded_media_inspector = RecordedMediaInspector( session_store.data_dir / "recorded-media-preparations" ) +_ffmpeg = shutil.which("ffmpeg") +_ffprobe = shutil.which("ffprobe") +session_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=Path(_ffmpeg), + ffprobe_path=Path(_ffprobe), + ) + if _ffmpeg is not None and _ffprobe is not None + else None +) def _prepare_recorded_media_for_launch( @@ -304,6 +319,7 @@ app.include_router( 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, ) ) diff --git a/src/k1link/web/session_api.py b/src/k1link/web/session_api.py index a9dedd6..6ffde06 100644 --- a/src/k1link/web/session_api.py +++ b/src/k1link/web/session_api.py @@ -13,6 +13,7 @@ from pydantic import BaseModel, ConfigDict, Field, StrictBool, field_validator, from starlette.concurrency import run_in_threadpool from starlette.types import Receive, Scope, Send +from k1link.compute import RecordedPerceptionOverlayError from k1link.sessions import ( RECORDED_MEDIA_MANIFEST_SCHEMA, LayoutConflictError, @@ -102,6 +103,16 @@ class RecordedBlueprintRequest(StrictApiModel): point_size: float = Field(default=2.5, strict=True, ge=0.1, le=32.0) palette: Literal["turbo", "viridis", "plasma", "grayscale", "custom"] = "turbo" custom_color: str = Field(default="#f7f8f4", pattern=r"^#[0-9A-Fa-f]{6}$") + active_view: Literal["spatial", "perception", "metrics"] = "spatial" + + +class RecordedPerceptionRequest(StrictApiModel): + application_id: Literal["nodedc_mission_core_recorded"] + recording_id: str = Field( + min_length=1, + max_length=128, + pattern=r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$", + ) class SceneSettingsDocument(StrictApiModel): @@ -233,6 +244,16 @@ class CatalogRefresher(Protocol): def __call__(self) -> object: ... +class RecordedPerceptionOverlayProvider(Protocol): + def render( + self, + session_id: str, + *, + application_id: str, + recording_id: str, + ) -> bytes | None: ... + + def build_session_router( store: SessionStore, *, @@ -241,6 +262,7 @@ def build_session_router( recording_materializer: RecordingMaterializer | None = None, recording_preparation_manager: SessionRecordingPreparationManager | None = None, media_inspector: RecordedMediaInspector | None = None, + perception_overlay_provider: RecordedPerceptionOverlayProvider | None = None, allow_synchronous_recording_fallback: bool = False, replay_action_id: str = DEFAULT_REPLAY_ACTION_ID, ) -> APIRouter: @@ -755,6 +777,7 @@ def build_session_router( ), application_id=RECORDED_APPLICATION_ID, recording_id=request.recording_id, + active_view=request.active_view, ) except SessionNotFoundError as exc: raise HTTPException(status_code=404, detail=str(exc)) from exc @@ -780,6 +803,56 @@ def build_session_router( }, ) + @router.post("/api/v1/observation-sessions/{session_id}/perception.rrd") + async def get_observation_session_perception( + session_id: str, + request: RecordedPerceptionRequest, + ) -> Response: + if perception_overlay_provider is None: + return Response(status_code=204, headers={"Cache-Control": "no-store"}) + try: + await run_in_threadpool( + _prepare_replay, + store, + catalog_refresher, + session_id, + 1.0, + False, + False, + ) + payload = await run_in_threadpool( + perception_overlay_provider.render, + session_id, + application_id=request.application_id, + recording_id=request.recording_id, + ) + except SessionNotFoundError as exc: + raise HTTPException(status_code=404, detail=str(exc)) from exc + except (SessionNotReplayableError, SessionIntegrityError) as exc: + raise HTTPException(status_code=409, detail=str(exc)) from exc + except RecordedPerceptionOverlayError as exc: + raise HTTPException( + status_code=500, + detail="Не удалось подготовить слой распознавания.", + ) from exc + except ValueError as exc: + raise HTTPException( + status_code=422, + detail="Некорректный идентификатор сессии.", + ) from exc + if payload is None: + return Response(status_code=204, headers={"Cache-Control": "no-store"}) + return Response( + content=payload, + media_type="application/vnd.rerun.rrd", + headers={ + "Cache-Control": "private, no-cache, no-transform", + "Content-Length": str(len(payload)), + "X-Content-Type-Options": "nosniff", + "Content-Disposition": 'inline; filename="perception.rrd"', + }, + ) + @router.get("/api/v1/observation-sessions/{session_id}/media/{artifact_id}/manifest") def get_recorded_media_manifest( session_id: str, diff --git a/tests/test_compute_results.py b/tests/test_compute_results.py new file mode 100644 index 0000000..a5c5def --- /dev/null +++ b/tests/test_compute_results.py @@ -0,0 +1,212 @@ +from __future__ import annotations + +import hashlib +import json +from pathlib import Path +from typing import Any + +import pytest + +import k1link.compute.results as result_module +from k1link.compute import ( + RecordedPerceptionOverlayStore, + prepare_camera_compute_job, + validate_recorded_perception_result, +) +from k1link.sessions import SessionIntegrityError +from k1link.web.camera_archive import CameraArchiveWriter + + +def _canonical_json(value: object) -> bytes: + return json.dumps( + value, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + allow_nan=False, + ).encode("utf-8") + + +def _job(tmp_path: Path) -> Any: + origin_epoch_ns = 1_000_000_000 + origin_monotonic_ns = 2_000_000_000 + session = tmp_path / "session-1" + session.mkdir() + + def box(box_type: bytes, payload: bytes = b"") -> bytes: + return (8 + len(payload)).to_bytes(4, "big") + box_type + payload + + def full_box(box_type: bytes, payload: bytes = b"", *, flags: int = 0) -> bytes: + return box(box_type, bytes([0]) + flags.to_bytes(3, "big") + payload) + + track_id = 1 + tkhd = full_box(b"tkhd", b"\x00" * 8 + track_id.to_bytes(4, "big") + b"\x00" * 4) + mdhd = full_box( + b"mdhd", + b"\x00" * 8 + (1_000).to_bytes(4, "big") + b"\x00" * 4, + ) + hdlr = full_box(b"hdlr", b"\x00" * 4 + b"vide") + trak = box(b"trak", tkhd + box(b"mdia", mdhd + hdlr)) + trex = full_box( + b"trex", + track_id.to_bytes(4, "big") + + (1).to_bytes(4, "big") + + (500).to_bytes(4, "big") + + b"\x00" * 8, + ) + init = box(b"ftyp", b"isom") + box( + b"moov", + trak + box(b"mvex", trex) + box(b"avcC", b"\x01\x64\x00\x28"), + ) + tfhd = full_box(b"tfhd", track_id.to_bytes(4, "big"), flags=0x020000) + tfdt = full_box(b"tfdt", (0).to_bytes(4, "big")) + trun = full_box(b"trun", (1).to_bytes(4, "big")) + fragment = box(b"moof", box(b"traf", tfhd + tfdt + trun)) + box(b"mdat", b"frame") + writer = CameraArchiveWriter(session, "sensor.camera.left", 1) + writer.append( + "init", + init, + host_epoch_ns=origin_epoch_ns + 100_000_000, + host_monotonic_ns=origin_monotonic_ns + 100_000_000, + ) + writer.append( + "media", + fragment, + host_epoch_ns=origin_epoch_ns + 200_000_000, + host_monotonic_ns=origin_monotonic_ns + 200_000_000, + ) + writer.close() + return prepare_camera_compute_job( + session_root=session, + source_id="sensor.camera.left", + codec_epoch=1, + origin_epoch_ns=origin_epoch_ns, + origin_monotonic_ns=origin_monotonic_ns, + output_root=tmp_path / "jobs", + ) + + +def _result(tmp_path: Path, job: Any) -> Path: + parameters = { + "score_threshold": 0.25, + "nms_threshold": 0.45, + "input_shape": [1, 3, 640, 640], + } + pipeline = {"id": "recorded-camera-coco-detection", "version": 1} + model = { + "id": "synthetic", + "version": 1, + "sha256": "a" * 64, + "source_url": "https://example.invalid/model.onnx", + "license": "test-only", + "classes": "synthetic", + } + identity = { + "schema_version": "missioncore.compute-result-identity/v1", + "job_id": job.job_id, + "input_sha256": job.input_sha256, + "pipeline": pipeline, + "model": {key: model[key] for key in ("id", "version", "sha256")}, + "parameters": parameters, + } + identity_sha256 = hashlib.sha256(_canonical_json(identity)).hexdigest() + result_id = f"result-{identity_sha256}" + root = tmp_path / "results" / job.job_id / result_id + root.mkdir(parents=True) + detections = { + "schema_version": "missioncore.object-detections/v1", + "result_id": result_id, + "job_id": job.job_id, + "input_sha256": job.input_sha256, + "timestamp_basis": "session-time-seconds", + "frames": [ + { + "frame_index": 0, + "epoch_seconds": 0.0, + "session_seconds": job.timeline_start_seconds, + "detections": [ + { + "class_id": 0, + "class_name": "person", + "score": 0.75, + "bbox_xyxy": [10.0, 20.0, 30.0, 60.0], + } + ], + } + ], + } + detection_payload = json.dumps(detections, indent=2).encode() + b"\n" + (root / "detections.json").write_bytes(detection_payload) + result = { + "schema_version": "missioncore.compute-result/v1", + "result_id": result_id, + "identity_sha256": identity_sha256, + "job_id": job.job_id, + "input_sha256": job.input_sha256, + "created_at_utc": "2026-07-19T00:00:00.000Z", + "pipeline": pipeline, + "runtime": {"kind": "synthetic"}, + "model": model, + "parameters": parameters, + "metrics": {"frames_processed": 1, "detections": 1}, + "artifacts": [ + { + "kind": "object-detections", + "path": "detections.json", + "byte_length": len(detection_payload), + "sha256": hashlib.sha256(detection_payload).hexdigest(), + "schema_version": "missioncore.object-detections/v1", + } + ], + "warnings": [], + } + (root / "result.json").write_text(json.dumps(result, indent=2) + "\n", encoding="utf-8") + return root + + +def test_result_is_bound_to_job_and_cache_reuses_complete_overlay( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + job = _job(tmp_path) + result_root = _result(tmp_path, job) + result = validate_recorded_perception_result(job.job_root, result_root) + assert result.job.session_id == "session-1" + assert result.frames[0].detections[0].class_name == "person" + + calls = 0 + + def render(*_: object, **__: object) -> bytes: + nonlocal calls + calls += 1 + return b"RRF2synthetic-overlay" + + monkeypatch.setattr(result_module, "_render_overlay", render) + store = RecordedPerceptionOverlayStore( + jobs_root=tmp_path / "jobs", + results_root=tmp_path / "results", + cache_root=tmp_path / "cache", + ffmpeg_path=Path(__file__), + ffprobe_path=Path(__file__), + ) + first = store.render( + "session-1", + application_id="nodedc_mission_core_recorded", + recording_id="recording-1", + ) + second = store.render( + "session-1", + application_id="nodedc_mission_core_recorded", + recording_id="recording-1", + ) + assert first == second == b"RRF2synthetic-overlay" + assert calls == 1 + + +def test_result_rejects_changed_detection_payload(tmp_path: Path) -> None: + job = _job(tmp_path) + result_root = _result(tmp_path, job) + detection_path = result_root / "detections.json" + detection_path.write_bytes(detection_path.read_bytes().replace(b"0.75", b"0.76")) + with pytest.raises(SessionIntegrityError, match="artifact identity changed"): + validate_recorded_perception_result(job.job_root, result_root) diff --git a/tests/test_rrd_export.py b/tests/test_rrd_export.py index dc1b367..ec44650 100644 --- a/tests/test_rrd_export.py +++ b/tests/test_rrd_export.py @@ -17,6 +17,7 @@ from k1link.device_plugins.xgrids_k1.mqtt.capture import ( RAW_MAGIC, ) from k1link.device_plugins.xgrids_k1.rrd_export import ( + RECORDED_CAMERA_VIEW_ID, RECORDED_METRICS_VIEW_ID, RECORDED_POINTS_VISUALIZER_ID, RECORDED_ROOT_CONTAINER_ID, @@ -232,8 +233,10 @@ def test_dynamic_blueprint_reuses_scene_ids_without_playback_mutation() -> None: first_view = first.root_container.contents[0] second_view = second.root_container.contents[0] assert first_view.id == second_view.id == RECORDED_SPATIAL_VIEW_ID - assert first.root_container.contents[1].id == RECORDED_METRICS_VIEW_ID - assert second.root_container.contents[1].id == RECORDED_METRICS_VIEW_ID + assert first.root_container.contents[1].id == RECORDED_CAMERA_VIEW_ID + assert second.root_container.contents[1].id == RECORDED_CAMERA_VIEW_ID + assert first.root_container.contents[2].id == RECORDED_METRICS_VIEW_ID + assert second.root_container.contents[2].id == RECORDED_METRICS_VIEW_ID first_point_behavior, first_point_visualizer = first_view.visualizer_overrides["/world/points"] second_point_behavior, second_point_visualizer = second_view.visualizer_overrides[ diff --git a/tests/test_session_api.py b/tests/test_session_api.py index 2169850..87a80cb 100644 --- a/tests/test_session_api.py +++ b/tests/test_session_api.py @@ -35,6 +35,7 @@ from k1link.web.camera_archive import CameraArchiveWriter from k1link.web.session_api import ( LayoutPutRequest, RecordedBlueprintRequest, + RecordedPerceptionRequest, ReplayRequest, build_session_router, ) @@ -1036,9 +1037,11 @@ def test_recorded_blueprint_endpoint_is_small_strict_and_session_scoped( store = SessionStore(repository, data_dir=tmp_path / "data") store.reconcile_archive(xgrids_k1_archive_source(sessions)) observed_settings: list[RerunSceneSettings] = [] + observed_kwargs: list[dict[str, Any]] = [] def capture_settings(settings: RerunSceneSettings, **kwargs: Any) -> bytes: observed_settings.append(settings) + observed_kwargs.append(kwargs) return recorded_blueprint_rrd(settings, **kwargs) monkeypatch.setattr( @@ -1065,6 +1068,7 @@ def test_recorded_blueprint_endpoint_is_small_strict_and_session_scoped( point_size=6.25, palette="custom", custom_color="#112233", + active_view="perception", ), ) ) @@ -1078,6 +1082,7 @@ def test_recorded_blueprint_endpoint_is_small_strict_and_session_scoped( assert len(observed_settings) == 1 assert observed_settings[0].show_points is False assert observed_settings[0].show_trajectory is True + assert observed_kwargs[0]["active_view"] == "perception" with pytest.raises(ValueError): RecordedBlueprintRequest.model_validate( @@ -1106,6 +1111,67 @@ def test_recorded_blueprint_endpoint_is_small_strict_and_session_scoped( assert missing.value.status_code == 404 +def test_recorded_perception_endpoint_returns_one_complete_optional_overlay( + tmp_path: Path, +) -> None: + repository = tmp_path / "repo" + sessions = repository / "sessions" + session = make_legacy_session(sessions, "20260716T205632Z_viewer_live") + store = SessionStore(repository, data_dir=tmp_path / "data") + store.reconcile_archive(xgrids_k1_archive_source(sessions)) + calls: list[tuple[str, str, str]] = [] + + class Provider: + def render( + self, + session_id: str, + *, + application_id: str, + recording_id: str, + ) -> bytes: + calls.append((session_id, application_id, recording_id)) + return b"RRF2perception" + + router = build_session_router(store, perception_overlay_provider=Provider()) + perception_route = endpoint( + router, + "/api/v1/observation-sessions/{session_id}/perception.rrd", + "POST", + ) + response = asyncio.run( + perception_route( + session_id=session.name, + request=RecordedPerceptionRequest( + application_id="nodedc_mission_core_recorded", + recording_id="recording-001", + ), + ) + ) + assert response.body == b"RRF2perception" + assert response.media_type == "application/vnd.rerun.rrd" + assert response.headers["content-length"] == str(len(response.body)) + assert calls == [ + (session.name, "nodedc_mission_core_recorded", "recording-001") + ] + + empty_router = build_session_router(store) + empty_route = endpoint( + empty_router, + "/api/v1/observation-sessions/{session_id}/perception.rrd", + "POST", + ) + empty = asyncio.run( + empty_route( + session_id=session.name, + request=RecordedPerceptionRequest( + application_id="nodedc_mission_core_recorded", + recording_id="recording-001", + ), + ) + ) + assert empty.status_code == 204 + + def test_session_router_exposes_opaque_recorded_media_manifest_and_ranges( tmp_path: Path, ) -> None: