328 lines
11 KiB
Python
328 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import sys
|
|
import threading
|
|
import time
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
from k1link.sessions.camera_frame import RecordedCameraFrame, RecordedCameraFrameService
|
|
from k1link.sessions.media import (
|
|
RecordedMediaEpoch,
|
|
RecordedMediaManifest,
|
|
RecordedMediaSegment,
|
|
)
|
|
from k1link.sessions.models import RecordedMediaArtifact, SessionIntegrityError
|
|
|
|
_JPEG = b"\xff\xd8\xff\xc0\x00\x07\x08\x00\x01\x00\x01\xff\xd9"
|
|
|
|
|
|
class _Store:
|
|
def __init__(self, *, block_prepare: bool = False) -> None:
|
|
self.prepare_calls = 0
|
|
self.list_calls = 0
|
|
self.prepare_started = threading.Event()
|
|
self.release_prepare = threading.Event()
|
|
if not block_prepare:
|
|
self.release_prepare.set()
|
|
self.artifact = RecordedMediaArtifact(
|
|
session_id="recorded-session",
|
|
public_source_id="camera",
|
|
artifact_id="camera-artifact",
|
|
source_path=Path("/sealed/sensor.camera.right"),
|
|
byte_length=123,
|
|
)
|
|
|
|
def prepare_replay(self, session_id: str, **_: Any) -> SimpleNamespace:
|
|
self.prepare_calls += 1
|
|
self.prepare_started.set()
|
|
assert self.release_prepare.wait(timeout=2.0)
|
|
return SimpleNamespace(session_id=session_id)
|
|
|
|
def list_recorded_media(self, session_id: str) -> tuple[RecordedMediaArtifact, ...]:
|
|
self.list_calls += 1
|
|
return (self.artifact,)
|
|
|
|
|
|
class _Inspector:
|
|
def __init__(self, manifest: RecordedMediaManifest) -> None:
|
|
self.manifest = manifest
|
|
self.inspect_calls = 0
|
|
|
|
def inspect(self, artifact: RecordedMediaArtifact, replay: object) -> RecordedMediaManifest:
|
|
assert artifact.artifact_id == "camera-artifact"
|
|
assert replay is not None
|
|
self.inspect_calls += 1
|
|
return self.manifest
|
|
|
|
|
|
def _manifest(tmp_path: Path) -> RecordedMediaManifest:
|
|
epoch_path = tmp_path / "epoch-1"
|
|
segments = tuple(
|
|
RecordedMediaSegment(
|
|
sequence=sequence,
|
|
path=epoch_path / "segments" / f"{sequence:08d}.m4s",
|
|
byte_length=10,
|
|
sha256=hashlib.sha256(str(sequence).encode()).hexdigest(),
|
|
random_access=True,
|
|
end_time_seconds=float(sequence),
|
|
)
|
|
for sequence in range(1, 4)
|
|
)
|
|
return RecordedMediaManifest(
|
|
session_id="recorded-session",
|
|
public_source_id="camera",
|
|
artifact_id="camera-artifact",
|
|
synchronization="recorded",
|
|
generation_sha256="a" * 64,
|
|
timeline_start_seconds=0.0,
|
|
timeline_end_seconds=3.0,
|
|
byte_length=123,
|
|
epochs=(
|
|
RecordedMediaEpoch(
|
|
ordinal=1,
|
|
path=epoch_path,
|
|
init_path=epoch_path / "init.mp4",
|
|
init_byte_length=10,
|
|
init_sha256="b" * 64,
|
|
media_type="video/mp4",
|
|
timeline_start_seconds=0.0,
|
|
timeline_end_seconds=3.0,
|
|
segments=segments,
|
|
),
|
|
),
|
|
)
|
|
|
|
|
|
def _service(
|
|
tmp_path: Path,
|
|
store: _Store,
|
|
inspector: _Inspector,
|
|
*,
|
|
max_decode_lanes: int = 32,
|
|
max_source_manifests: int = 32,
|
|
) -> RecordedCameraFrameService:
|
|
return RecordedCameraFrameService(
|
|
store, # type: ignore[arg-type]
|
|
inspector, # type: ignore[arg-type]
|
|
ffmpeg_path=Path(sys.executable),
|
|
cache_root=tmp_path / "cache",
|
|
max_decode_lanes=max_decode_lanes,
|
|
max_source_manifests=max_source_manifests,
|
|
)
|
|
|
|
|
|
def _wait_for_latest(service: RecordedCameraFrameService, ticket: int) -> None:
|
|
deadline = time.monotonic() + 2.0
|
|
source_key = ("recorded-session", "sensor.camera.right")
|
|
while time.monotonic() < deadline:
|
|
with service._coordination: # noqa: SLF001
|
|
lane = service._lanes.get(source_key) # noqa: SLF001
|
|
if lane is not None and lane.latest_ticket == ticket:
|
|
return
|
|
time.sleep(0.005)
|
|
raise AssertionError(f"camera request ticket {ticket} was not registered")
|
|
|
|
|
|
def _frame(epoch: RecordedMediaEpoch, sequence: int) -> RecordedCameraFrame:
|
|
return RecordedCameraFrame(
|
|
payload=_JPEG,
|
|
media_type="image/jpeg",
|
|
width=1,
|
|
height=1,
|
|
sha256=hashlib.sha256(_JPEG).hexdigest(),
|
|
source_fragment_sha256=epoch.segments[sequence - 1].sha256,
|
|
)
|
|
|
|
|
|
def test_camera_frame_burst_scans_manifest_once_and_decodes_only_latest(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
store = _Store(block_prepare=True)
|
|
inspector = _Inspector(_manifest(tmp_path))
|
|
service = _service(tmp_path, store, inspector)
|
|
decoded: list[int] = []
|
|
|
|
def decode(
|
|
manifest: RecordedMediaManifest,
|
|
epoch: RecordedMediaEpoch,
|
|
sequence: int,
|
|
) -> RecordedCameraFrame:
|
|
assert manifest is inspector.manifest
|
|
decoded.append(sequence)
|
|
return _frame(epoch, sequence)
|
|
|
|
monkeypatch.setattr(service, "_decode", decode)
|
|
with ThreadPoolExecutor(max_workers=3) as pool:
|
|
first = pool.submit(service.extract, "recorded-session", 0)
|
|
assert store.prepare_started.wait(timeout=2.0)
|
|
second = pool.submit(service.extract, "recorded-session", 1)
|
|
_wait_for_latest(service, 2)
|
|
third = pool.submit(service.extract, "recorded-session", 2)
|
|
_wait_for_latest(service, 3)
|
|
store.release_prepare.set()
|
|
|
|
with pytest.raises(SessionIntegrityError, match="superseded"):
|
|
first.result(timeout=2.0)
|
|
with pytest.raises(SessionIntegrityError, match="superseded"):
|
|
second.result(timeout=2.0)
|
|
assert third.result(timeout=2.0).source_fragment_sha256 == (
|
|
inspector.manifest.epochs[0].segments[2].sha256
|
|
)
|
|
|
|
assert store.prepare_calls == 1
|
|
assert store.list_calls == 1
|
|
assert inspector.inspect_calls == 1
|
|
assert decoded == [3]
|
|
|
|
|
|
def test_camera_frame_lane_keeps_only_latest_waiter_behind_active_decode(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
store = _Store()
|
|
inspector = _Inspector(_manifest(tmp_path))
|
|
service = _service(tmp_path, store, inspector)
|
|
decode_started = threading.Event()
|
|
release_decode = threading.Event()
|
|
decoded: list[int] = []
|
|
|
|
def decode(
|
|
manifest: RecordedMediaManifest,
|
|
epoch: RecordedMediaEpoch,
|
|
sequence: int,
|
|
) -> RecordedCameraFrame:
|
|
assert manifest is inspector.manifest
|
|
decoded.append(sequence)
|
|
if sequence == 1:
|
|
decode_started.set()
|
|
assert release_decode.wait(timeout=2.0)
|
|
return _frame(epoch, sequence)
|
|
|
|
monkeypatch.setattr(service, "_decode", decode)
|
|
with ThreadPoolExecutor(max_workers=3) as pool:
|
|
first = pool.submit(service.extract, "recorded-session", 0)
|
|
assert decode_started.wait(timeout=2.0)
|
|
second = pool.submit(service.extract, "recorded-session", 1)
|
|
_wait_for_latest(service, 2)
|
|
third = pool.submit(service.extract, "recorded-session", 2)
|
|
_wait_for_latest(service, 3)
|
|
release_decode.set()
|
|
|
|
with pytest.raises(SessionIntegrityError, match="superseded"):
|
|
first.result(timeout=2.0)
|
|
with pytest.raises(SessionIntegrityError, match="superseded"):
|
|
second.result(timeout=2.0)
|
|
assert third.result(timeout=2.0).width == 1
|
|
|
|
assert decoded == [1, 3]
|
|
assert store.prepare_calls == 1
|
|
assert inspector.inspect_calls == 1
|
|
|
|
|
|
def test_camera_frame_memory_caches_are_lru_bounded(tmp_path: Path) -> None:
|
|
store = _Store()
|
|
inspector = _Inspector(_manifest(tmp_path))
|
|
service = _service(
|
|
tmp_path,
|
|
store,
|
|
inspector,
|
|
max_decode_lanes=2,
|
|
max_source_manifests=2,
|
|
)
|
|
|
|
for ordinal in range(4):
|
|
source_key = (f"session-{ordinal}", "sensor.camera.right")
|
|
lane, _ = service._acquire_lane(source_key) # noqa: SLF001
|
|
service._release_lane(source_key, lane) # noqa: SLF001
|
|
assert tuple(service._lanes) == ( # noqa: SLF001
|
|
("session-2", "sensor.camera.right"),
|
|
("session-3", "sensor.camera.right"),
|
|
)
|
|
|
|
for ordinal in range(3):
|
|
source_key = (f"session-{ordinal}", "sensor.camera.right")
|
|
service._source_manifest( # noqa: SLF001
|
|
source_key,
|
|
session_id=source_key[0],
|
|
expected_source_name=source_key[1],
|
|
)
|
|
assert tuple(service._source_manifests) == ( # noqa: SLF001
|
|
("session-1", "sensor.camera.right"),
|
|
("session-2", "sensor.camera.right"),
|
|
)
|
|
|
|
service._source_manifest( # noqa: SLF001
|
|
("session-1", "sensor.camera.right"),
|
|
session_id="session-1",
|
|
expected_source_name="sensor.camera.right",
|
|
)
|
|
service._source_manifest( # noqa: SLF001
|
|
("session-3", "sensor.camera.right"),
|
|
session_id="session-3",
|
|
expected_source_name="sensor.camera.right",
|
|
)
|
|
assert tuple(service._source_manifests) == ( # noqa: SLF001
|
|
("session-1", "sensor.camera.right"),
|
|
("session-3", "sensor.camera.right"),
|
|
)
|
|
assert store.prepare_calls == 4
|
|
assert inspector.inspect_calls == 4
|
|
|
|
|
|
def test_camera_playback_source_reuses_manifest_and_exposes_exact_segment_clock(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
store = _Store()
|
|
manifest = _manifest(tmp_path)
|
|
manifest = RecordedMediaManifest(
|
|
session_id=manifest.session_id,
|
|
public_source_id=manifest.public_source_id,
|
|
artifact_id=manifest.artifact_id,
|
|
synchronization="host-arrival-best-effort",
|
|
generation_sha256=manifest.generation_sha256,
|
|
timeline_start_seconds=manifest.timeline_start_seconds,
|
|
timeline_end_seconds=manifest.timeline_end_seconds,
|
|
byte_length=manifest.byte_length,
|
|
epochs=(
|
|
RecordedMediaEpoch(
|
|
ordinal=1,
|
|
path=manifest.epochs[0].path,
|
|
init_path=manifest.epochs[0].init_path,
|
|
init_byte_length=manifest.epochs[0].init_byte_length,
|
|
init_sha256=manifest.epochs[0].init_sha256,
|
|
media_type='video/mp4; codecs="avc1.640028"',
|
|
timeline_start_seconds=0.0,
|
|
timeline_end_seconds=3.0,
|
|
segments=manifest.epochs[0].segments,
|
|
),
|
|
),
|
|
)
|
|
inspector = _Inspector(manifest)
|
|
service = _service(tmp_path, store, inspector)
|
|
|
|
first = service.playback_source("recorded-session")
|
|
second = service.playback_source("recorded-session")
|
|
|
|
assert first == second
|
|
assert first.segment_count == 3
|
|
assert first.segment_sha256s == tuple(
|
|
segment.sha256 for segment in manifest.epochs[0].segments
|
|
)
|
|
assert first.segment_start_times_ns == (0, 1_000_000_000, 2_000_000_000)
|
|
assert store.prepare_calls == 1
|
|
assert inspector.inspect_calls == 1
|
|
|
|
|
|
def test_camera_frame_rejects_unbounded_memory_cache_configuration(tmp_path: Path) -> None:
|
|
store = _Store()
|
|
inspector = _Inspector(_manifest(tmp_path))
|
|
with pytest.raises(SessionIntegrityError, match="cache bounds"):
|
|
_service(tmp_path, store, inspector, max_decode_lanes=0)
|