SILVER/NODEDC_MISSION_CORE
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

perf(lab): stream sealed spatial playback tracks

Browse Source
This commit is contained in:
DCCONSTRUCTIONS
2026-08-27 15:18:45 +03:00
parent b70dc4298a
commit bf76304e00
10 changed files with 911 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/k1link/web/m49_tgs_full_shadow_api.py
+125 -10
View File
@@ -13,17 +13,44 @@ from typing import Final
import numpy as np
from fastapi import APIRouter, HTTPException, Query, Response
from fastapi.responses import FileResponse
from k1link.laboratory.m49_tgs_full_shadow import (
PREFIX,
M49TgsFullShadowError,
M49TgsFullShadowResult,
PREFIX,
read_m49_tgs_full_shadow,
)
RootProvider = Callable[[], Path | None]
RESULT_ID: Final = re.compile(rf"^{re.escape(PREFIX)}[a-f0-9]{{64}}$")
ENDPOINT_ROOT: Final = "/api/v1/laboratory/m49/tgs-full-shadow"
PLAYBACK_TRACKS: Final = {
"centers": (
"costmap-cell-centers-xy-m.npy",
"application/x-npy",
"<f4",
[2244, 2],
),
"states": (
"costmap-states.npy",
"application/x-npy",
"|u1",
[4489, 2244],
),
"z-bounds": (
"costmap-z-bounds-m.npy",
"application/x-npy",
"<f4",
[4489, 2244, 2],
),
"frames": (
"frames.ndjson",
"application/x-ndjson",
"ndjson",
[4489],
),
}
def build_m49_tgs_full_shadow_router(*, root_provider: RootProvider = lambda: None) -> APIRouter:
@@ -55,16 +82,46 @@ def build_m49_tgs_full_shadow_router(*, root_provider: RootProvider = lambda: No
if not candidate.is_dir() or RESULT_ID.fullmatch(candidate.name) is None:
continue
try:
results.append(_project(_read_cached(str(candidate.resolve()), _signature(candidate))))
results.append(
_project(_read_cached(str(candidate.resolve()), _signature(candidate)))
)
except (M49TgsFullShadowError, OSError, ValueError):
invalid += 1
results.sort(key=lambda row: (str(row["created_at_utc"]), str(row["result_id"])), reverse=True)
results.sort(
key=lambda row: (str(row["created_at_utc"]), str(row["result_id"])), reverse=True
)
return _catalog(results[:limit], configured=True, invalid_total=invalid)
@router.get("/{result_id}")
def get_result(result_id: str) -> dict[str, object]:
return _project(sealed(result_id))
@router.get("/{result_id}/playback/manifest")
def get_playback_manifest(result_id: str) -> dict[str, object]:
return _playback_manifest(sealed(result_id))
@router.get("/{result_id}/playback/tracks/{track_id}")
def get_playback_track(result_id: str, track_id: str) -> FileResponse:
result = sealed(result_id)
descriptor = PLAYBACK_TRACKS.get(track_id)
if descriptor is None:
raise HTTPException(status_code=404, detail="M49 playback track not found")
name, media_type, _dtype, _shape = descriptor
artifact = _artifact_descriptor(result, name)
return FileResponse(
result.root / name,
media_type=media_type,
headers={
"Cache-Control": "private, max-age=31536000, immutable",
# These playback artifacts are consumed on the local control
# station. Avoid spending multiple seconds recompressing
# already compact numeric tracks on every cold open.
"Content-Encoding": "identity",
"ETag": f'"{artifact["sha256"]}"',
"X-Content-Type-Options": "nosniff",
},
)
@router.get("/{result_id}/frames/{source_sequence}/spatial")
def get_spatial(result_id: str, source_sequence: int) -> Response:
if source_sequence < 0 or source_sequence >= 4489:
@@ -75,11 +132,16 @@ def build_m49_tgs_full_shadow_router(*, root_provider: RootProvider = lambda: No
str(result.root), result_id, source_sequence, _evidence_signature(result.root)
)
except (OSError, ValueError, KeyError, json.JSONDecodeError):
raise HTTPException(status_code=503, detail="M49 TGS full-shadow spatial evidence failed verification") from None
raise HTTPException(
status_code=503, detail="M49 TGS full-shadow spatial evidence failed verification"
) from None
return Response(
content=content,
media_type="application/json",
headers={"Cache-Control": "private, max-age=31536000, immutable", "X-Content-Type-Options": "nosniff"},
headers={
"Cache-Control": "private, max-age=31536000, immutable",
"X-Content-Type-Options": "nosniff",
},
)
@router.get("/{result_id}/spatial/chunk")
@@ -134,7 +196,9 @@ def _frame_json_cached(
root_path = Path(root)
frame_signature = (signature[-2], signature[-1])
frame = _frames(root, frame_signature)[source_sequence]
centers = np.load(root_path / "costmap-cell-centers-xy-m.npy", mmap_mode="r", allow_pickle=False)
centers = np.load(
root_path / "costmap-cell-centers-xy-m.npy", mmap_mode="r", allow_pickle=False
)
states_all = np.load(root_path / "costmap-states.npy", mmap_mode="r", allow_pickle=False)
z_all = np.load(root_path / "costmap-z-bounds-m.npy", mmap_mode="r", allow_pickle=False)
states = states_all[source_sequence]
@@ -169,13 +233,20 @@ def _frame_json_cached(
],
},
"metrics": copy.deepcopy(frame),
"state_codes": {"UNOBSERVED": 0, "GROUND_SUPPORT": 1, "NONGROUND_OCCUPIED": 2, "UNKNOWN_REJECTED": 3},
"state_codes": {
"UNOBSERVED": 0,
"GROUND_SUPPORT": 1,
"NONGROUND_OCCUPIED": 2,
"UNKNOWN_REJECTED": 3,
},
"aos_used": False,
"gpu_used": False,
"authority": {"navigation_or_safety_accepted": False, "visual_quality_accepted": False},
"access": "read-only",
}
return json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False).encode("utf-8")
return json.dumps(
payload, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False
).encode("utf-8")
@lru_cache(maxsize=16)
@@ -189,7 +260,9 @@ def _chunk_json_cached(
root_path = Path(root)
frame_signature = (signature[-2], signature[-1])
frames = _frames(root, frame_signature)
centers = np.load(root_path / "costmap-cell-centers-xy-m.npy", mmap_mode="r", allow_pickle=False)
centers = np.load(
root_path / "costmap-cell-centers-xy-m.npy", mmap_mode="r", allow_pickle=False
)
states_all = np.load(root_path / "costmap-states.npy", mmap_mode="r", allow_pickle=False)
z_all = np.load(root_path / "costmap-z-bounds-m.npy", mmap_mode="r", allow_pickle=False)
if (
@@ -258,6 +331,46 @@ def _chunk_json_cached(
).encode("utf-8")
def _artifact_descriptor(
result: M49TgsFullShadowResult,
name: str,
) -> dict[str, object]:
for artifact in result.manifest["artifacts"]:
if artifact.get("path") == name:
return artifact
raise ValueError(f"full-shadow artifact is not declared: {name}")
def _playback_manifest(result: M49TgsFullShadowResult) -> dict[str, object]:
tracks: list[dict[str, object]] = []
total_bytes = 0
for track_id, (name, media_type, dtype, shape) in PLAYBACK_TRACKS.items():
artifact = _artifact_descriptor(result, name)
byte_length = int(artifact["byte_length"])
total_bytes += byte_length
tracks.append(
{
"id": track_id,
"url": (f"{ENDPOINT_ROOT}/{result.result_id}/playback/tracks/{track_id}"),
"media_type": media_type,
"dtype": dtype,
"shape": shape,
"byte_length": byte_length,
"sha256": artifact["sha256"],
}
)
return {
"schema_version": "missioncore.m49-tgs-full-shadow-playback/v1",
"result_id": result.result_id,
"coordinate_frame": "map-gravity-local",
"frame_count": 4489,
"cell_count": 2244,
"total_byte_length": total_bytes,
"tracks": tracks,
"access": "read-only",
}
def _project(result: M49TgsFullShadowResult) -> dict[str, object]:
return {
**copy.deepcopy(result.report),
@@ -269,7 +382,9 @@ def _project(result: M49TgsFullShadowResult) -> dict[str, object]:
}
def _catalog(items: list[dict[str, object]], *, configured: bool, invalid_total: int) -> dict[str, object]:
def _catalog(
items: list[dict[str, object]], *, configured: bool, invalid_total: int
) -> dict[str, object]:
return {
"schema_version": "missioncore.m49-tgs-full-shadow-catalog/v1",
"configured": configured,
src/k1link/web/m4_threat_replay_api.py
+83 -3
View File
@@ -3,6 +3,7 @@
from __future__ import annotations
import copy
import hashlib
import json
import re
from collections.abc import Callable, Iterator
@@ -11,6 +12,7 @@ from pathlib import Path
from typing import Final
from fastapi import APIRouter, HTTPException, Query, Response
from fastapi.responses import StreamingResponse
from k1link.perception.threat_replay import (
THREAT_REPLAY_FRAME_SCHEMA,
@@ -166,7 +168,7 @@ def build_m4_threat_replay_router(
def get_timeline(result_id: str) -> dict[str, object]:
return copy.deepcopy(timeline(result_id).metadata())
@router.get("/results/{result_id}/timeline/chunk")
@router.get("/results/{result_id}/timeline/chunk", response_model=None)
def get_timeline_chunk(
result_id: str,
start: int = Query(default=0, ge=0),
@@ -175,14 +177,87 @@ def build_m4_threat_replay_router(
ge=1,
le=RECORDED_SPATIAL_MAX_CHUNK_FRAMES,
),
) -> dict[str, object]:
include_points: bool = Query(default=True),
) -> dict[str, object] | Response:
try:
return timeline(result_id).chunk(start_sequence=start, frame_count=count)
payload = timeline(result_id).chunk(
start_sequence=start,
frame_count=count,
include_points=include_points,
)
except RecordedThreatTimelineError:
raise HTTPException(
status_code=404,
detail="M4.6 timeline chunk не найден",
) from None
if include_points:
return payload
return Response(
content=json.dumps(
payload,
ensure_ascii=False,
separators=(",", ":"),
allow_nan=False,
),
media_type="application/json",
headers={
"Cache-Control": "private, max-age=31536000, immutable",
"Content-Encoding": "identity",
"X-Content-Type-Options": "nosniff",
},
)
@router.get("/results/{result_id}/timeline/playback")
def get_timeline_playback(result_id: str) -> dict[str, object]:
projected = timeline(result_id)
points = projected.store.playback_points_map()
offsets = projected.store.playback_point_offsets()
content_sha256 = hashlib.sha256(memoryview(points).cast("B")).hexdigest()
return {
"schema_version": "missioncore.recorded-spatial-playback/v1",
"result_id": result_id,
"frame_count": len(offsets) - 1,
"point_count": int(points.shape[0]),
"point_offsets": list(offsets),
"track": {
"id": "points-map-f32",
"url": (
f"/api/v1/laboratory/m4-threat/results/{result_id}"
"/timeline/playback/tracks/points-map-f32"
),
"media_type": "application/octet-stream",
"dtype": "<f4",
"shape": [int(points.shape[0]), 3],
"bytes": int(points.nbytes),
"sha256": content_sha256,
},
"source_pack_sha256": projected.profile.source_pack_sha256,
"coordinate_frame": "map",
"ground_truth": False,
"authority": "replay-simulated",
"access": "read-only-sealed-binary-playback",
}
@router.get(
"/results/{result_id}/timeline/playback/tracks/points-map-f32",
response_class=StreamingResponse,
)
def get_timeline_playback_points(result_id: str) -> StreamingResponse:
projected = timeline(result_id)
points = projected.store.playback_points_map()
source_digest = projected.profile.source_pack_sha256
return StreamingResponse(
_binary_chunks(memoryview(points).cast("B")),
media_type="application/octet-stream",
headers={
"Cache-Control": "private, max-age=31536000, immutable",
"Content-Encoding": "identity",
"Content-Length": str(points.nbytes),
"ETag": f'"{source_digest}-points-map-f32"',
"X-Content-Type-Options": "nosniff",
"X-Uncompressed-Content-Length": str(points.nbytes),
},
)
@router.get("/results/{result_id}/timeline/frames/{sequence}/camera")
def get_timeline_camera(result_id: str, sequence: int) -> Response:
@@ -196,6 +271,11 @@ def build_m4_threat_replay_router(
return router
def _binary_chunks(view: memoryview, chunk_size: int = 1024 * 1024) -> Iterator[bytes]:
for start in range(0, view.nbytes, chunk_size):
yield bytes(view[start : start + chunk_size])
@lru_cache(maxsize=4)
def _read_threat_result_cached(
root_value: str,