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

fix(lab): enforce canonical replay runtime

Browse Source
This commit is contained in:
DCCONSTRUCTIONS
2026-08-29 22:42:30 +03:00
parent 525ab74168
commit bd2892140f
17 changed files with 1765 additions and 507 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/k1link/sessions/canonical_lab_spatial.py
+259
View File
@@ -0,0 +1,259 @@
"""Canonical recorded-LAB spatial adapter for sealed Rerun recordings.
The LAB viewer must not run an independent Rerun transport beside the camera
transport. This adapter reads the immutable recording once, indexes the
recorded source cloud, sensor pose and SLAM trajectory, and returns the latest
source-paced spatial sample in the current body frame. Camera, spatial layers
and the common timeline can therefore be driven by one host clock.
"""
from __future__ import annotations
from bisect import bisect_right
from dataclasses import dataclass
from functools import lru_cache
from pathlib import Path
from threading import Lock
from typing import Any, Final
import numpy as np
import rerun_bindings as rr_bindings
_POINT_ENTITY: Final = "/world/points"
_POSE_ENTITY: Final = "/world/sensor_pose"
_TRAJECTORY_ENTITY: Final = "/world/trajectory"
_POINT_COMPONENT: Final = "Points3D:positions"
_POSE_TRANSLATION_COMPONENT: Final = "Transform3D:translation"
_POSE_QUATERNION_COMPONENT: Final = "Transform3D:quaternion"
_TRAJECTORY_COMPONENT: Final = "LineStrips3D:strips"
_INDEX_LOCK: Final = Lock()
@dataclass(frozen=True)
class _TimedPoints:
times_ns: tuple[int, ...]
values: tuple[np.ndarray, ...]
@dataclass(frozen=True)
class _TimedPoses:
times_ns: tuple[int, ...]
translations: tuple[np.ndarray, ...]
quaternions_xyzw: tuple[np.ndarray, ...]
@dataclass(frozen=True)
class _CanonicalSpatialIndex:
points: _TimedPoints
poses: _TimedPoses
trajectories: _TimedPoints
def _session_times(batch: Any) -> Any | None:
if "session_time" not in batch.schema.names:
return None
return batch.column("session_time")
def _point_rows(chunks: list[Any], entity: str, component: str, *, nested: bool = False) -> _TimedPoints:
rows: list[tuple[int, np.ndarray]] = []
for chunk in chunks:
if chunk.entity_path != entity:
continue
batch = chunk.to_record_batch()
times = _session_times(batch)
if times is None or component not in batch.schema.names:
continue
column = batch.column(component)
for row_index in range(batch.num_rows):
timestamp = int(times[row_index].value)
payload = column[row_index].as_py()
if nested:
payload = payload[0] if payload else []
values = np.asarray(payload, dtype=np.float32)
if values.ndim != 2 or values.shape[1] != 3 or not np.isfinite(values).all():
continue
values.setflags(write=False)
rows.append((timestamp, values))
rows.sort(key=lambda item: item[0])
return _TimedPoints(
times_ns=tuple(timestamp for timestamp, _ in rows),
values=tuple(values for _, values in rows),
)
def _pose_rows(chunks: list[Any]) -> _TimedPoses:
rows: list[tuple[int, np.ndarray, np.ndarray]] = []
for chunk in chunks:
if chunk.entity_path != _POSE_ENTITY:
continue
batch = chunk.to_record_batch()
times = _session_times(batch)
if (
times is None
or _POSE_TRANSLATION_COMPONENT not in batch.schema.names
or _POSE_QUATERNION_COMPONENT not in batch.schema.names
):
continue
translations = batch.column(_POSE_TRANSLATION_COMPONENT)
quaternions = batch.column(_POSE_QUATERNION_COMPONENT)
for row_index in range(batch.num_rows):
translation_values = translations[row_index].as_py()
quaternion_values = quaternions[row_index].as_py()
if len(translation_values) != 1 or len(quaternion_values) != 1:
continue
translation = np.asarray(translation_values[0], dtype=np.float64)
quaternion = np.asarray(quaternion_values[0], dtype=np.float64)
if (
translation.shape != (3,)
or quaternion.shape != (4,)
or not np.isfinite(translation).all()
or not np.isfinite(quaternion).all()
):
continue
norm = float(np.linalg.norm(quaternion))
if norm <= 1e-9:
continue
translation.setflags(write=False)
normalized = quaternion / norm
normalized.setflags(write=False)
rows.append((int(times[row_index].value), translation, normalized))
rows.sort(key=lambda item: item[0])
return _TimedPoses(
times_ns=tuple(timestamp for timestamp, _, _ in rows),
translations=tuple(translation for _, translation, _ in rows),
quaternions_xyzw=tuple(quaternion for _, _, quaternion in rows),
)
@lru_cache(maxsize=4)
def _load_index_cached(
path_text: str,
byte_length: int,
modified_ns: int,
generation_sha256: str,
) -> _CanonicalSpatialIndex:
path = Path(path_text)
stat = path.stat()
if stat.st_size != byte_length or stat.st_mtime_ns != modified_ns:
raise ValueError("Recorded LAB source changed during spatial indexing")
if len(generation_sha256) != 64:
raise ValueError("Recorded LAB generation is invalid")
# Decode only the three canonical entities in one pass. Building a lazy
# store first decodes the complete RRD (including unrelated payloads), and
# then scanning that store once per layer made first-open take more than a
# minute on RAVNOVES004TREE.
chunks = (
rr_bindings.RrdReaderInternal(str(path))
.stream()
.filter(content=[_POINT_ENTITY, _POSE_ENTITY, _TRAJECTORY_ENTITY])
.to_chunks()
)
points = _point_rows(chunks, _POINT_ENTITY, _POINT_COMPONENT)
poses = _pose_rows(chunks)
trajectories = _point_rows(
chunks,
_TRAJECTORY_ENTITY,
_TRAJECTORY_COMPONENT,
nested=True,
)
if not points.times_ns or not poses.times_ns or not trajectories.times_ns:
raise ValueError("Recorded LAB source has no canonical spatial layers")
return _CanonicalSpatialIndex(points=points, poses=poses, trajectories=trajectories)
def _load_index(
path_text: str,
byte_length: int,
modified_ns: int,
generation_sha256: str,
) -> _CanonicalSpatialIndex:
# functools.lru_cache is coherent but intentionally releases its lock
# during a miss. Serialize cold RRD indexing so simultaneous camera/TGS
# admission cannot parse the same 80 MiB recording twice.
with _INDEX_LOCK:
return _load_index_cached(
path_text,
byte_length,
modified_ns,
generation_sha256,
)
def _latest_index(times_ns: tuple[int, ...], target_ns: int) -> int:
return max(0, min(len(times_ns) - 1, bisect_right(times_ns, target_ns) - 1))
def _rotation_map_from_body(quaternion_xyzw: np.ndarray) -> np.ndarray:
x, y, z, w = (float(value) for value in quaternion_xyzw)
return np.asarray(
[
[1 - 2 * (y * y + z * z), 2 * (x * y - z * w), 2 * (x * z + y * w)],
[2 * (x * y + z * w), 1 - 2 * (x * x + z * z), 2 * (y * z - x * w)],
[2 * (x * z - y * w), 2 * (y * z + x * w), 1 - 2 * (x * x + y * y)],
],
dtype=np.float64,
)
def _map_points_to_body(
points_map: np.ndarray,
translation_map: np.ndarray,
quaternion_xyzw: np.ndarray,
) -> np.ndarray:
rotation = _rotation_map_from_body(quaternion_xyzw)
# Row vectors: inverse(map_from_body) == right-multiply by map_from_body.
body = (points_map.astype(np.float64) - translation_map) @ rotation
return body.astype(np.float32)
def canonical_lab_spatial_frame(
recording_path: Path,
generation_sha256: str,
target_time_ns: int,
) -> dict[str, object]:
"""Return the latest sealed source cloud and SLAM route on one host time."""
if target_time_ns < 0:
raise ValueError("Recorded LAB target time is invalid")
stat = recording_path.stat()
index = _load_index(
str(recording_path),
stat.st_size,
stat.st_mtime_ns,
generation_sha256,
)
point_index = _latest_index(index.points.times_ns, target_time_ns)
pose_index = _latest_index(index.poses.times_ns, index.points.times_ns[point_index])
trajectory_index = _latest_index(index.trajectories.times_ns, target_time_ns)
translation = index.poses.translations[pose_index]
quaternion = index.poses.quaternions_xyzw[pose_index]
basis_map_from_body = _rotation_map_from_body(quaternion)
points_body = _map_points_to_body(index.points.values[point_index], translation, quaternion)
trajectory_body = _map_points_to_body(
index.trajectories.values[trajectory_index],
translation,
quaternion,
)
# The canonical local-SLAM layer is bounded around the vehicle. It must
# never turn into the full world-route "blob" seen in the raw Rerun view.
local_mask = (
(np.abs(trajectory_body[:, 0]) <= 30.0)
& (np.abs(trajectory_body[:, 1]) <= 30.0)
& (np.abs(trajectory_body[:, 2]) <= 6.0)
)
local_trajectory = trajectory_body[local_mask]
return {
"schema_version": "missioncore.canonical-recorded-lab-spatial-frame/v1",
"target_time_ns": target_time_ns,
"source_time_ns": index.points.times_ns[point_index],
"pose_time_ns": index.poses.times_ns[pose_index],
"trajectory_time_ns": index.trajectories.times_ns[trajectory_index],
"body_frame": {
"origin_map_xyz_m": translation.tolist(),
"basis_map_from_body": basis_map_from_body.tolist(),
},
"source_point_count": int(points_body.shape[0]),
"source_points_body_xyz_m": points_body.tolist(),
"local_slam_body_xyz_m": local_trajectory.tolist(),
}
src/k1link/web/session_api.py
+71
View File
@@ -37,6 +37,7 @@ from k1link.sessions import (
SessionStore,
validate_recorded_media_timeline,
)
from k1link.sessions.canonical_lab_spatial import canonical_lab_spatial_frame
from k1link.sessions.plugin_contract import RecordedPointColorRenderer
from k1link.viewer.recorded import (
APPLICATION_ID as RECORDED_APPLICATION_ID,
@@ -824,6 +825,76 @@ def build_session_router(
**response_kwargs,
)
@router.get(
"/api/v1/observation-sessions/{session_id}/canonical-lab/spatial-frame"
)
async def get_observation_session_canonical_lab_spatial_frame(
session_id: str,
generation: Annotated[str, Query(min_length=64, max_length=64)],
time_ns: Annotated[int, Query(ge=0, le=MAX_SAFE_INTEGER)],
) -> JSONResponse:
"""Serve one body-frame sample for the canonical recorded-LAB clock.
The camera timeline owns playback. Spatial evidence is sampled from
the same immutable recording instead of starting a second Rerun clock.
"""
if SAFE_SHA256.fullmatch(generation) is None:
raise HTTPException(
status_code=412,
detail="Поколение spatial-записи не совпадает.",
)
if recording_preparation_manager is None:
raise HTTPException(
status_code=503,
detail="Сервис canonical LAB spatial playback не настроен.",
)
snapshot = recording_preparation_manager.status(session_id)
if snapshot is None or snapshot.state != "ready" or snapshot.recording is None:
raise HTTPException(
status_code=409,
detail="Запись canonical LAB ещё не подготовлена.",
)
_require_matching_recording_generation(snapshot.recording.sha256, generation)
pinned = recording_preparation_manager.pin_ready(
session_id,
preparation_id=snapshot.preparation_id,
)
if pinned is None:
raise HTTPException(
status_code=412,
detail="Подготовленная spatial-запись была заменена.",
)
pinned_snapshot, release_recording = pinned
try:
recording = pinned_snapshot.recording
if recording is None:
raise HTTPException(
status_code=500,
detail="Подготовленная spatial-запись недоступна.",
)
payload = await run_in_threadpool(
canonical_lab_spatial_frame,
recording.path,
generation,
time_ns,
)
except (OSError, ValueError) as exc:
raise HTTPException(
status_code=503,
detail="Canonical LAB spatial frame не прошёл проверку.",
) from exc
finally:
release_recording()
return JSONResponse(
payload,
headers={
"Cache-Control": "private, max-age=31536000, immutable",
"ETag": f'"{generation}:{payload["source_time_ns"]}"',
"X-Content-Type-Options": "nosniff",
},
)
@router.post("/api/v1/observation-sessions/{session_id}/blueprint.rrd")
async def get_observation_session_blueprint(
session_id: str,
src/k1link/web/vegetation_shadow_lab_api.py
+106 -1
View File
@@ -11,8 +11,9 @@ from functools import lru_cache
from pathlib import Path, PurePosixPath
from typing import Any, Final
import numpy as np
from fastapi import APIRouter, HTTPException
from fastapi.responses import FileResponse, Response
from fastapi.responses import FileResponse, JSONResponse, Response
from k1link.laboratory.evidence_registry import LaboratoryEvidenceDefinition
from k1link.laboratory.evidence_report import (
@@ -262,9 +263,113 @@ def _build_vegetation_lab_router(
},
)
@router.get("/{result_id}/route-tgs-anchor/{source_sequence}")
def get_route_tgs_anchor(result_id: str, source_sequence: int) -> JSONResponse:
candidate = _resolve_candidate(root_provider, definition, result_id)
manifest = _read_verified(candidate, definition)
review = manifest.get("route_review")
cases = review.get("cases") if isinstance(review, dict) else None
if (
not isinstance(cases, list)
or review.get("source_id") != "RAVNOVES004TREE"
or review.get("session_id") != "20260828T130511Z_viewer_live"
or not any(
isinstance(item, dict) and item.get("source_sequence") == source_sequence
for item in cases
)
):
raise HTTPException(status_code=404, detail="Route TGS anchor not found")
artifacts = manifest.get("artifacts")
descriptor = next(
(
item
for item in artifacts if isinstance(item, dict)
and item.get("role") == "mixed-route-tgs-evidence"
and item.get("path") == "proofs/tgs-evidence.npz"
and item.get("media_type") == "application/x-npz"
),
None,
) if isinstance(artifacts, list) else None
if descriptor is None:
raise HTTPException(status_code=404, detail="Route TGS anchor not found")
try:
payload = _route_tgs_anchor_payload(
candidate / "proofs" / "tgs-evidence.npz",
source_sequence,
)
except (KeyError, OSError, ValueError):
raise HTTPException(
status_code=503,
detail="Route TGS anchor failed verification",
) from None
return JSONResponse(
payload,
headers={
"Cache-Control": "private, max-age=31536000, immutable",
"ETag": f'"{descriptor.get("sha256", "")}"',
"X-Content-Type-Options": "nosniff",
},
)
return router
def _route_tgs_anchor_payload(path: Path, source_sequence: int) -> dict[str, object]:
before = path.stat()
with np.load(path, allow_pickle=False) as archive:
source_indices = archive["source_frame_indices"]
offsets = archive["current_increment_point_offsets"]
points = archive["current_increment_points_xyz_m"]
centers = archive["costmap_cell_centers_xy_m"]
states = archive["causal_rolling_1s_costmap_states"]
z_bounds = archive["causal_rolling_1s_costmap_z_bounds_m"]
if (
source_indices.shape != (10,)
or offsets.shape != (11,)
or points.ndim != 2
or points.shape[1] != 3
or centers.shape != (2244, 2)
or states.shape != (10, 2244)
or z_bounds.shape != (10, 2244, 2)
):
raise ValueError("Route TGS evidence shape changed")
matches = np.flatnonzero(source_indices == source_sequence - 1)
if matches.shape != (1,):
raise ValueError("Route TGS source sequence changed")
slot = int(matches[0])
start = int(offsets[slot])
end = int(offsets[slot + 1])
if not 0 <= start <= end <= points.shape[0]:
raise ValueError("Route TGS point offsets changed")
selected_points = np.ascontiguousarray(points[start:end], dtype=np.float32)
selected_states = np.ascontiguousarray(states[slot], dtype=np.uint8)
selected_z_bounds = np.ascontiguousarray(z_bounds[slot], dtype=np.float32)
if not np.isfinite(selected_points).all() or not np.isin(selected_states, [0, 1, 2, 3]).all():
raise ValueError("Route TGS payload changed")
result = {
"schema_version": "missioncore.lab-v1-route-tgs-anchor/v1",
"source_sequence": source_sequence,
"slot": slot,
"current_points_xyz_m": selected_points.astype(float).tolist(),
"costmap": {
"cell_size_m": 0.45,
"centers_xy_m": centers.astype(float).tolist(),
"state_codes": selected_states.astype(int).tolist(),
"z_bounds_m": [
[
float(row[0]) if np.isfinite(row[0]) else None,
float(row[1]) if np.isfinite(row[1]) else None,
]
for row in selected_z_bounds
],
},
}
after = path.stat()
if before.st_size != after.st_size or before.st_mtime_ns != after.st_mtime_ns:
raise ValueError("Route TGS evidence changed during read")
return result
def _zip_mask_response(archive_path: Path, sequence: int) -> Response:
member = f"masks/frame-{sequence + 1:06d}.png"
try: