feat(perception): add dual evidence replay threat

This commit is contained in:
DCCONSTRUCTIONS
2026-08-05 19:18:01 +03:00
parent 6b810952a4
commit b0d0bc8d7f
22 changed files with 3122 additions and 14 deletions
+42
View File
@@ -245,6 +245,48 @@ class RecordedGeometryStore:
points.setflags(write=False)
return points
def pose_values_for_frame(
self,
frame_id: str,
) -> tuple[tuple[float, float, float], tuple[float, float, float, float]] | None:
"""Return one verified replay pose without exposing the source archive."""
prefix = "frame-"
if not frame_id.startswith(prefix) or not frame_id[len(prefix) :].isdigit():
raise GeometryProviderError("replay pose frame identity is invalid")
frame_index = int(frame_id[len(prefix) :])
if not 0 <= frame_index < self.profile.frame_count:
raise GeometryProviderError("replay pose frame is outside the source profile")
if not bool(self._source["sample_available"][frame_index]):
return None
position = np.asarray(
self._source["pose_positions_map"][frame_index],
dtype=np.float64,
)
orientation = np.asarray(
self._source["pose_quaternions_map_from_lidar"][frame_index],
dtype=np.float64,
)
if not np.isfinite(position).all() or not np.isfinite(orientation).all():
raise GeometryProviderError("available replay pose is not finite")
return (
(float(position[0]), float(position[1]), float(position[2])),
(
float(orientation[0]),
float(orientation[1]),
float(orientation[2]),
float(orientation[3]),
),
)
def available_frame_indices(self) -> tuple[int, ...]:
"""Expose the immutable availability partition for deterministic sampling."""
return tuple(
int(index)
for index in np.flatnonzero(self._source["sample_available"])
)
def _validate(self) -> None:
source_required = {
"frame_indices",