ykcheungab
280 lines
10 KiB
Python
280 lines
10 KiB
Python
"""Calibration identity and operator events as the acquisition thread sees them.
|
|
|
|
The review's P1-7: the application event was written before the device had loaded
|
|
anything, the recorded ``calibration_id`` never changed mid-session, and an
|
|
uncalibrated device could keep claiming the coefficients of an earlier session.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import shutil
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
import pytest
|
|
from orisys_app_shared.acquisition.sdk_worker import SdkAcquisitionWorker
|
|
from orisys_app_shared.record import ExperimentRecorder, RecordConfig
|
|
from orisys_app_shared.record.calibration_snapshot import (
|
|
list_calibration_snapshots,
|
|
snapshot_name,
|
|
store_calibration_snapshot,
|
|
)
|
|
|
|
SCRATCH = Path(__file__).resolve().parent / "_scratch"
|
|
|
|
|
|
def _root() -> Path:
|
|
shutil.rmtree(SCRATCH, ignore_errors=True)
|
|
SCRATCH.mkdir(parents=True, exist_ok=True)
|
|
return SCRATCH
|
|
|
|
|
|
class _Sensor:
|
|
"""A device with a swappable calibration, like the SDK's Sensor."""
|
|
|
|
def __init__(self, calibration=None, path=None):
|
|
self.calibration = calibration
|
|
self.calibration_path = path
|
|
self.applied: list[str] = []
|
|
self.resets = 0
|
|
|
|
def set_calibration(self, path):
|
|
if str(path) == "reject-me":
|
|
raise ValueError("bad document")
|
|
self.calibration = object()
|
|
self.calibration_path = Path(path)
|
|
self.applied.append(str(path))
|
|
|
|
def reset(self):
|
|
self.resets += 1
|
|
|
|
|
|
def _calibration_file(root: Path, name: str, body: str = '{"unit": "N"}') -> Path:
|
|
path = root / name
|
|
path.write_text(body, encoding="utf-8")
|
|
return path
|
|
|
|
|
|
def _worker(**kwargs) -> SdkAcquisitionWorker:
|
|
for key in ("on_calibration_applied", "on_reset_applied"):
|
|
kwargs.setdefault(key, None)
|
|
return SdkAcquisitionWorker(**kwargs)
|
|
|
|
|
|
# ------------------------------------------------------------------ worker state
|
|
|
|
|
|
def test_a_new_session_does_not_inherit_the_previous_version():
|
|
"""An uncalibrated device must report no calibration, not the last one."""
|
|
root = _root()
|
|
worker = _worker()
|
|
worker._calibration_version = "previous-session-hash"
|
|
worker._stop.set()
|
|
worker.start("clip.mp4", config_ref=str(_calibration_file(root, "sensor.json")))
|
|
assert worker._calibration_version == "", (
|
|
"the new session kept the previous session's calibration version"
|
|
)
|
|
worker.stop(wait=True)
|
|
|
|
|
|
def test_the_applied_version_is_the_content_hash_and_carries_the_frame():
|
|
root = _root()
|
|
applied: list[tuple] = []
|
|
worker = _worker(on_calibration_applied=lambda *args: applied.append(args))
|
|
path = _calibration_file(root, "force_calibration.json")
|
|
sensor = _Sensor()
|
|
worker._session_id = 4
|
|
worker.set_pending_calibration(path)
|
|
worker._consume_pending_calibration(sensor, frame_id=17, mono_t=0.5)
|
|
|
|
assert len(applied) == 1
|
|
got_path, version, frame_id, mono_t, session = applied[0]
|
|
assert got_path == str(path)
|
|
assert version and version == worker._calibration_version
|
|
assert version != str(path), "the path was used as the version"
|
|
assert (frame_id, mono_t, session) == (17, 0.5, 4)
|
|
assert sensor.applied == [str(path)]
|
|
|
|
|
|
def test_a_rejected_calibration_keeps_the_version_the_device_still_has():
|
|
root = _root()
|
|
applied: list[tuple] = []
|
|
worker = _worker(on_calibration_applied=lambda *args: applied.append(args))
|
|
sensor = _Sensor()
|
|
worker._session_id = 1
|
|
worker._calibration_version = "loaded-earlier"
|
|
worker.set_pending_calibration("reject-me")
|
|
with pytest.raises(ValueError):
|
|
worker._consume_pending_calibration(sensor, frame_id=3, mono_t=0.1)
|
|
assert not applied, "a rejected document was reported as applied"
|
|
assert worker._calibration_version == "loaded-earlier", (
|
|
"the device still has those coefficients, so the version is still true"
|
|
)
|
|
|
|
|
|
def test_a_zeroing_is_applied_by_the_worker_and_reported_with_its_frame():
|
|
root = _root()
|
|
events: list[tuple] = []
|
|
worker = _worker(on_reset_applied=lambda *args: events.append(args))
|
|
sensor = _Sensor()
|
|
worker._session_id = 2
|
|
worker.reset() # a request, not the zeroing itself
|
|
assert sensor.resets == 0, "the zeroing ran on the caller's thread"
|
|
worker._consume_pending_reset(sensor, frame_id=42, mono_t=1.4)
|
|
assert sensor.resets == 1
|
|
assert events == [(42, 1.4, 2)]
|
|
# The request is consumed once.
|
|
worker._consume_pending_reset(sensor, frame_id=43, mono_t=1.5)
|
|
assert events == [(42, 1.4, 2)]
|
|
|
|
|
|
def test_a_started_session_drops_a_stale_zeroing_request():
|
|
root = _root()
|
|
events: list[tuple] = []
|
|
worker = _worker(on_reset_applied=lambda *args: events.append(args))
|
|
worker._stop.set()
|
|
worker.start("clip.mp4", config_ref=str(_calibration_file(root, "sensor.json")))
|
|
sensor = _Sensor()
|
|
worker._consume_pending_reset(sensor, frame_id=1, mono_t=0.0)
|
|
assert sensor.resets == 0, "a reset requested before this session fired into it"
|
|
assert not events
|
|
worker.stop(wait=True)
|
|
|
|
|
|
# ------------------------------------------------------------------- snapshots
|
|
|
|
|
|
def test_a_snapshot_is_content_addressed_and_listed():
|
|
root = _root()
|
|
directory = root / "session"
|
|
source = _calibration_file(root, "force_calibration_box.json", '{"a": 1}')
|
|
name = store_calibration_snapshot(directory, source, "abcdef1234567890")
|
|
assert name == "calibration/force_calibration_abcdef123456.json"
|
|
assert (directory / name).is_file()
|
|
assert (directory / name).read_text(encoding="utf-8") == '{"a": 1}'
|
|
# Applying the same coefficients twice stores one file.
|
|
again = store_calibration_snapshot(directory, source, "abcdef1234567890")
|
|
assert again == name
|
|
assert list_calibration_snapshots(directory) == [name]
|
|
# A different calibration cannot collide with it.
|
|
other = _calibration_file(root, "force_calibration_box.json", '{"a": 2}')
|
|
second = store_calibration_snapshot(directory, other, "ffffff000000")
|
|
assert second != name
|
|
assert len(list_calibration_snapshots(directory)) == 2
|
|
assert snapshot_name("", source).startswith("force_calibration_")
|
|
|
|
|
|
def test_the_recording_metadata_names_the_coefficients_it_ran_with():
|
|
root = _root()
|
|
recorder = ExperimentRecorder(RecordConfig(video=False))
|
|
directory = recorder.start(root=root, session_id=1, metadata={})
|
|
source = _calibration_file(root, "force_calibration_box.json", '{"a": 1}')
|
|
name = store_calibration_snapshot(directory, source, "abcdef1234567890")
|
|
recorder.note_calibration_snapshot(
|
|
version="abcdef1234567890", name=name, frame_id=12, mono_t=0.4
|
|
)
|
|
recorder.log_event(
|
|
"calibration_applied", name, mono_t=0.4, frame_id=12
|
|
)
|
|
recorder.stop()
|
|
|
|
meta = json.loads((Path(directory) / "session.json").read_text(encoding="utf-8"))
|
|
snapshots = meta["calibration_snapshots"]
|
|
assert snapshots and snapshots[0]["version"] == "abcdef1234567890"
|
|
assert snapshots[0]["file"] == name
|
|
assert snapshots[0]["frame_id"] == 12
|
|
assert snapshots[0]["mono_t"] == pytest.approx(0.4)
|
|
event = next(e for e in meta["events"] if e["kind"] == "calibration_applied")
|
|
assert event["frame_id"] == 12
|
|
assert event["mono_t"] == pytest.approx(0.4), "the event lost its acquisition time"
|
|
assert event["at"], "the event lost its wall clock"
|
|
|
|
# The CSV carries the same acquisition time in the session's relative base.
|
|
# An event that predates the first frame has no base yet, so its relative
|
|
# time is empty — the absolute (session-relative) time is in the metadata.
|
|
with (Path(directory) / "events.csv").open(encoding="utf-8", newline="") as handle:
|
|
import csv as _csv
|
|
|
|
rows = list(_csv.DictReader(handle))
|
|
applied = next(row for row in rows if row["kind"] == "calibration_applied")
|
|
assert applied["frame_id"] == "12"
|
|
assert applied["t_rel_s"] == "", "an event before the first frame invented a base"
|
|
|
|
|
|
def test_an_event_during_the_recording_gets_the_session_relative_time():
|
|
import csv as _csv
|
|
|
|
from orisys_app_shared.acquisition.sdk_worker import SdkPacket
|
|
|
|
root = _root()
|
|
recorder = ExperimentRecorder(RecordConfig(video=False))
|
|
directory = recorder.start(root=root, session_id=1, metadata={})
|
|
frame = np.zeros((4, 4, 3), np.uint8)
|
|
recorder.on_packet(
|
|
SdkPacket(
|
|
raw_frame=frame,
|
|
processed_frame=frame,
|
|
timestamp=1000.0,
|
|
mono_t=0.5,
|
|
fps=30.0,
|
|
process_ms=1.0,
|
|
frame_id=7,
|
|
)
|
|
)
|
|
recorder.log_event("zeroing", "tracker reset", mono_t=0.9, frame_id=9)
|
|
recorder.stop()
|
|
|
|
with (Path(directory) / "events.csv").open(encoding="utf-8", newline="") as handle:
|
|
rows = list(_csv.DictReader(handle))
|
|
zeroing = next(row for row in rows if row["kind"] == "zeroing")
|
|
assert float(zeroing["t_rel_s"]) == pytest.approx(0.4, abs=1e-6)
|
|
assert zeroing["frame_id"] == "9"
|
|
|
|
|
|
def test_an_event_without_a_mono_time_still_gets_a_wall_clock():
|
|
root = _root()
|
|
recorder = ExperimentRecorder(RecordConfig(video=False))
|
|
directory = recorder.start(root=root, session_id=1, metadata={})
|
|
recorder.log_event("operator_note", "no frame attached")
|
|
recorder.stop()
|
|
meta = json.loads((Path(directory) / "session.json").read_text(encoding="utf-8"))
|
|
event = next(e for e in meta["events"] if e["kind"] == "operator_note")
|
|
assert event["at"], "the event lost its wall clock"
|
|
assert event["frame_id"] == -1
|
|
|
|
|
|
def test_the_recorded_row_uses_the_packet_version_not_the_session_default():
|
|
from orisys_app_shared.record.sample import to_record_sample
|
|
|
|
class Packet:
|
|
frame_id = 3
|
|
mono_t = 0.2
|
|
wall_t = 0.2
|
|
valid = True
|
|
calibration_version = ""
|
|
calibrated_axes = ("fn",)
|
|
force_calibrated = True
|
|
raw_available = True
|
|
|
|
# An empty packet version is the device saying "no calibration": the session
|
|
# default must not be pasted in behind its back.
|
|
assert to_record_sample(Packet(), calibration_version="session-default").calibration_version == ""
|
|
|
|
class LegacyPacket(Packet):
|
|
calibration_version = None # an older producer without the field
|
|
|
|
# ...but a packet that has no version field at all still gets the default.
|
|
assert (
|
|
to_record_sample(LegacyPacket(), calibration_version="session-default").calibration_version
|
|
== "session-default"
|
|
)
|
|
|
|
class StampedPacket(Packet):
|
|
calibration_version = "packet-hash"
|
|
|
|
assert (
|
|
to_record_sample(StampedPacket(), calibration_version="session-default").calibration_version
|
|
== "packet-hash"
|
|
)
|