ykcheungab
280 lines
10 KiB
Python
280 lines
10 KiB
Python
"""Locating a reviewed hold's frames in the existing recording."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from evidence_helpers import (
|
|
BASE_NS,
|
|
FRAMES_FIELDS,
|
|
decoded_frame_number,
|
|
force_row,
|
|
frame_row,
|
|
multipoint_session,
|
|
write_capture,
|
|
write_csv,
|
|
write_json_file,
|
|
)
|
|
from orisys_autocal.session_review import review_session
|
|
from orisys_autocal.video_review import (
|
|
NO_FRAME_SOURCE,
|
|
NO_VIDEO_FRAME,
|
|
Location,
|
|
in_window_frames,
|
|
locate_window,
|
|
open_window_video,
|
|
)
|
|
|
|
|
|
def _review(tmp_path) -> dict:
|
|
root = multipoint_session(
|
|
tmp_path / "rec", sites=("P1",), repeats=1, levels=(5.0,))
|
|
return review_session(root)
|
|
|
|
|
|
def _capture(review: dict):
|
|
trial_id = review["trials"][0]["trial_id"]
|
|
step = review["sites"][0]["windows"][1]["step_index"]
|
|
return trial_id, step
|
|
|
|
|
|
def test_a_reviewed_window_locates_its_own_first_video_frame(tmp_path):
|
|
review = _review(tmp_path)
|
|
trial_id, step = _capture(review)
|
|
|
|
location = locate_window(review, trial_id=trial_id, step_index=step)
|
|
|
|
assert isinstance(location, Location)
|
|
assert location.frame_id > 0
|
|
assert location.segment == 1 and location.segment_frame == location.frame_id
|
|
# The located picture is the one the frame index names.
|
|
root = tmp_path / "rec"
|
|
video = root / location.capture_directory / "raw_0001.mp4"
|
|
assert decoded_frame_number(video, segment_frame=location.segment_frame) == location.segment_frame - 1
|
|
|
|
with open_window_video(review, trial_id=trial_id, step_index=step) as player:
|
|
assert player.frame_count >= 1
|
|
assert player.frame(0) is not None
|
|
assert player.location.frame_id == location.frame_id
|
|
assert player.frame(0).shape[:2] == (48, 64)
|
|
|
|
|
|
def test_a_frame_outside_the_video_is_skipped_for_the_next_one(tmp_path):
|
|
review = _review(tmp_path)
|
|
root = tmp_path / "rec"
|
|
trial_id = review["trials"][0]["trial_id"]
|
|
window = review["sites"][0]["windows"][1]
|
|
capture = root / "sites" / "P1" / "repeat-001" / "capture"
|
|
from evidence_helpers import read_csv
|
|
|
|
rows = read_csv(capture / "frames.csv")
|
|
# The second hold starts at frame 4; take its first frame out of the video.
|
|
first_of_hold = int(window["hold_start_host_mono_ns"])
|
|
row = next(item for item in rows if int(item["t_host_mono_ns"]) >= first_of_hold)
|
|
row["in_video"] = 0
|
|
row["segment"] = 0
|
|
row["segment_frame"] = 0
|
|
skipped = int(row["frame_id"])
|
|
write_csv(capture / "frames.csv", FRAMES_FIELDS, rows)
|
|
|
|
location = locate_window(review, trial_id=trial_id, step_index=window["step_index"])
|
|
|
|
assert isinstance(location, Location)
|
|
assert location.frame_id == skipped + 1
|
|
|
|
|
|
def test_no_capture_directory_is_a_reason_not_a_picture(tmp_path):
|
|
root = multipoint_session(tmp_path / "rec", sites=("P1",), repeats=1,
|
|
levels=(5.0,), with_capture=False)
|
|
review = review_session(root)
|
|
trial_id = review["trials"][0]["trial_id"]
|
|
step = review["sites"][0]["windows"][1]["step_index"]
|
|
|
|
location = locate_window(review, trial_id=trial_id, step_index=step)
|
|
|
|
assert location == {"reason": NO_FRAME_SOURCE}
|
|
with pytest.raises(ValueError, match=NO_FRAME_SOURCE):
|
|
open_window_video(review, trial_id=trial_id, step_index=step)
|
|
|
|
|
|
def test_a_hold_with_no_picture_in_the_video_has_no_frame(tmp_path):
|
|
review = _review(tmp_path)
|
|
root = tmp_path / "rec"
|
|
capture = root / "sites" / "P1" / "repeat-001" / "capture"
|
|
from evidence_helpers import read_csv
|
|
|
|
rows = read_csv(capture / "frames.csv")
|
|
for row in rows:
|
|
row["in_video"] = 0
|
|
row["segment"] = 0
|
|
row["segment_frame"] = 0
|
|
write_csv(capture / "frames.csv", FRAMES_FIELDS, rows)
|
|
trial_id, step = _capture(review)
|
|
|
|
assert locate_window(review, trial_id=trial_id, step_index=step) == {
|
|
"reason": NO_VIDEO_FRAME}
|
|
|
|
|
|
def test_a_missing_segment_file_is_reported(tmp_path):
|
|
review = _review(tmp_path)
|
|
root = tmp_path / "rec"
|
|
for video in root.glob("sites/*/*/capture/raw_0001.mp4"):
|
|
video.unlink()
|
|
trial_id, step = _capture(review)
|
|
|
|
assert locate_window(review, trial_id=trial_id, step_index=step) == {
|
|
"reason": NO_FRAME_SOURCE}
|
|
|
|
|
|
def test_a_hold_split_across_segments_opens_the_right_one(tmp_path):
|
|
"""Segment 2 restarts its own frame count; seeking must follow the index."""
|
|
root = tmp_path / "rec"
|
|
trial = root / "trial"
|
|
start = BASE_NS + 10
|
|
frames = [
|
|
frame_row(1, start, segment=1, segment_frame=1),
|
|
frame_row(2, start + 10, segment=1, segment_frame=2),
|
|
frame_row(3, start + 20, segment=2, segment_frame=1),
|
|
]
|
|
forces = [force_row(1, fn=1.0), force_row(2, fn=2.0), force_row(3, fn=3.0)]
|
|
write_capture(trial, frames, forces, video_segments=2)
|
|
write_json_file(root / "session.json", {"format_version": "autocal-session-1.0"})
|
|
review = {
|
|
"source_directory": str(root),
|
|
"session_id": "s1",
|
|
"trials": [{"trial_id": "t1", "directory": "trial"}],
|
|
"sites": [{
|
|
"site_id": "P1", "uv_xy_px": [1, 2],
|
|
"windows": [{
|
|
"trial_id": "t1", "step_index": 1,
|
|
"hold_start_host_mono_ns": start, "hold_end_host_mono_ns": start + 100,
|
|
"frame_source": "trial",
|
|
"trace": [(start, 1, 1.0), (start + 10, 2, 2.0), (start + 20, 3, 3.0)],
|
|
}],
|
|
}],
|
|
}
|
|
|
|
location = locate_window(review, trial_id="t1", step_index=1)
|
|
|
|
assert isinstance(location, Location) and location.segment == 1
|
|
with open_window_video(review, trial_id="t1", step_index=1) as player:
|
|
assert player.frame_count == 3
|
|
assert player.frame(0) is not None
|
|
# Frame 3 lives in the second container, at its own position 1.
|
|
assert player.frame(2) is not None
|
|
assert player.frames[2].segment == 2 and player.frames[2].segment_frame == 1
|
|
|
|
|
|
def test_the_decoder_can_be_closed_and_reopened(tmp_path):
|
|
review = _review(tmp_path)
|
|
trial_id, step = _capture(review)
|
|
|
|
first = open_window_video(review, trial_id=trial_id, step_index=step)
|
|
assert first.frame(0) is not None
|
|
first.close()
|
|
second = open_window_video(review, trial_id=trial_id, step_index=step)
|
|
try:
|
|
assert second.frame(0) is not None
|
|
finally:
|
|
second.release()
|
|
|
|
|
|
def test_an_unknown_trial_or_window_is_a_reason(tmp_path):
|
|
review = _review(tmp_path)
|
|
trial_id, step = _capture(review)
|
|
|
|
assert locate_window(review, trial_id="missing", step_index=step) == {"reason": "trial_not_found"}
|
|
assert locate_window(review, trial_id=trial_id, step_index=999) == {"reason": "window_not_found"}
|
|
|
|
|
|
def test_the_frame_list_is_bounded_by_the_hold(tmp_path):
|
|
review = _review(tmp_path)
|
|
trial_id, step = _capture(review)
|
|
window = next(
|
|
item for site in review["sites"] for item in site["windows"]
|
|
if item["trial_id"] == trial_id and item["step_index"] == step
|
|
)
|
|
|
|
frames = in_window_frames(review, trial_id=trial_id, step_index=step)
|
|
|
|
assert isinstance(frames, list) and frames
|
|
assert all(
|
|
window["hold_start_host_mono_ns"] <= frame.t_host_mono_ns < window["hold_end_host_mono_ns"]
|
|
for frame in frames
|
|
)
|
|
assert all(frame.in_video for frame in frames)
|
|
|
|
|
|
def test_confirmed_playback_uses_the_reviewers_selected_interval(tmp_path):
|
|
review = _review(tmp_path)
|
|
trial_id, step = _capture(review)
|
|
window = review["sites"][0]["windows"][1]
|
|
selected_start = window["trace"][1][0]
|
|
selected_end = window["hold_end_host_mono_ns"]
|
|
review = review_session(tmp_path / "rec", selections={f"{trial_id}:{step}": {
|
|
"t_start_host_mono_ns": selected_start,
|
|
"t_end_host_mono_ns": selected_end,
|
|
"stability_verified": True,
|
|
}})
|
|
with open_window_video(review, trial_id=trial_id, step_index=step) as player:
|
|
assert len(player.frames) == 2
|
|
assert all(selected_start <= frame.t_host_mono_ns < selected_end for frame in player.frames)
|
|
|
|
|
|
def test_playback_does_not_retain_previously_decoded_images(tmp_path):
|
|
import gc
|
|
import weakref
|
|
|
|
review = _review(tmp_path)
|
|
trial_id, step = _capture(review)
|
|
refs = []
|
|
with open_window_video(review, trial_id=trial_id, step_index=step) as player:
|
|
for index in range(player.frame_count):
|
|
image = player.frame(index)
|
|
assert image is not None
|
|
refs.append(weakref.ref(image))
|
|
del image
|
|
gc.collect()
|
|
assert sum(ref() is not None for ref in refs) <= 1
|
|
|
|
|
|
def test_decode_failure_is_reported_rather_than_showing_a_neighbour(tmp_path):
|
|
review = _review(tmp_path)
|
|
root = tmp_path / "rec"
|
|
trial_id, step = _capture(review)
|
|
location = locate_window(review, trial_id=trial_id, step_index=step)
|
|
assert isinstance(location, Location)
|
|
(root / location.capture_directory / "raw_0001.mp4").write_bytes(b"not a video")
|
|
window = review["sites"][0]["windows"][1]
|
|
player = open_window_video(review, trial_id=trial_id, step_index=window["step_index"])
|
|
try:
|
|
assert player.frame(0) is None
|
|
assert player.position == -1
|
|
finally:
|
|
player.close()
|
|
|
|
def test_a_discarded_hold_attempt_cannot_be_played(tmp_path):
|
|
"""A remeasured hold keeps its row id; its window must not be shown as evidence."""
|
|
from evidence_helpers import HOLD_STRIDE_NS, Hold, multipoint_session
|
|
|
|
replaced = Hold(
|
|
step_index=1, branch="load", target_load_n=5.0, reference_n=5.0,
|
|
start_ns=BASE_NS + 200 * HOLD_STRIDE_NS,
|
|
end_ns=BASE_NS + 200 * HOLD_STRIDE_NS + 4_000_000_000,
|
|
complete=0,
|
|
)
|
|
root = multipoint_session(
|
|
tmp_path / "rec", sites=("P1",), repeats=1, levels=(5.0,), extra_holds=[replaced])
|
|
review = review_session(root)
|
|
trial_id = review["trials"][0]["trial_id"]
|
|
step = next(item for site in review["sites"] for item in site["windows"]
|
|
if item["trial_id"] == trial_id)["step_index"]
|
|
|
|
assert locate_window(review, trial_id=trial_id, step_index=step, hold_id=4) == {
|
|
"reason": "hold_not_complete"}
|
|
assert locate_window(review, trial_id=trial_id, step_index=step, hold_id=99) == {
|
|
"reason": "hold_not_found"}
|
|
# The completed attempt still opens, and names the attempt it shows.
|
|
location = locate_window(review, trial_id=trial_id, step_index=step, hold_id=1)
|
|
assert isinstance(location, Location) and location.hold_id == 1
|