orisys-tech/orisys-app
3
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases Wiki Activity

Fix repeat ownership and close out capture evidence

Browse Source 
Start each repeat after its own manifest entry, refuse packets from a restarted
or stopped camera, keep a simulated repeat from reading as recorded evidence,
and decide completeness from the writer report. Add the operator summary (four
files plus an atomic pointer) and read-only video review of a hold window.

Verified with the full Autocal suite (567 passed), the shared recorder/playback
files (57 passed), the lint gate and the documentation gate; the two pre-existing
panel_check layout failures also fail on a clean HEAD worktree.
This commit is contained in:
ykcheungab
2026-10-17 10:44:30 +08:00
parent d7540f52ec
commit 19426c3c4a
32 changed files with 5461 additions and 191 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/autocal/tests/evidence_helpers.py
+563
View File
@@ -0,0 +1,563 @@
"""Synthetic recording evidence for the acquisition, summary and review tests.
Two rules govern this module:
* a fixture writes the **shared recorder's real headers**, so a reader is tested
against the files it will meet rather than against a convenient subset;
* a video fixture is a **real decodable clip**. A file of fake MP4 bytes proves
nothing about seeking, decoding or frame identity.
Nothing here touches the app's preference file, its ``outputs/`` directory or a
real camera. Fixtures are built inside the test's ``tmp_path``; the run's
``--basetemp`` points at the task scratch directory.
"""
from __future__ import annotations
import csv
import json
import time
from dataclasses import dataclass
from pathlib import Path
from orisys_app_shared.acquisition.models import SdkPacket
from orisys_app_shared.record import FORCES_HEADER, FRAMES_HEADER
#: One working day in nanoseconds: holds inside a fixture are this far apart.
HOLD_STRIDE_NS = 10_000_000_000
#: A fixed, arbitrary session-relative origin for synthetic host timestamps.
BASE_NS = 1_700_000_000_000_000_000
FRAMES_FIELDS = list(FRAMES_HEADER)
FORCES_FIELDS = list(FORCES_HEADER)
def write_csv(path: Path, fields: list[str], rows: list[dict]) -> Path:
path.parent.mkdir(parents=True, exist_ok=True)
with path.open("w", newline="", encoding="utf-8") as handle:
writer = csv.DictWriter(handle, fieldnames=fields, extrasaction="ignore")
writer.writeheader()
writer.writerows(rows)
return path
def read_csv(path: Path) -> list[dict]:
with path.open(encoding="utf-8-sig", newline="") as handle:
return list(csv.DictReader(handle))
def frame_row(
frame_id: int,
t_host_mono_ns: int,
*,
segment: int = 1,
segment_frame: int | None = None,
in_video: int = 1,
dropped: int = 0,
time_basis: str = "host_camera",
t_rel_s: float = 0.0,
) -> dict:
return {
"frame_id": frame_id,
"source_frame_id": frame_id,
"t_rel_s": repr(float(t_rel_s)),
"t_host_mono_ns": int(t_host_mono_ns),
"t_host_utc_ns": int(t_host_mono_ns),
"acquired_at": "2026-10-17T00:00:00.000+00:00",
"time_basis": time_basis,
"media_time_ms": "",
"segment": segment if in_video else 0,
"segment_frame": (segment_frame if segment_frame is not None else frame_id)
if in_video
else 0,
"in_video": in_video,
"dropped": dropped,
"capture_dropped": 0,
}
def force_row(
frame_id: int,
*,
fn: float | None,
fx: float | None = 0.0,
fy: float | None = 0.0,
valid: int = 1,
session_id: int = 7,
t_rel_s: float = 0.0,
) -> dict:
def cell(value):
return "" if value is None else repr(float(value))
return {
"session_id": session_id,
"frame_id": frame_id,
"t_rel_s": repr(float(t_rel_s)),
"FN_raw": cell(fn),
"FX_raw": cell(fx),
"FY_raw": cell(fy),
"FN_N": "",
"FX_N": "",
"FY_N": "",
"valid": valid,
"calibration_id": "",
"FN_output_raw": "",
"FX_output_raw": "",
"FY_output_raw": "",
"FN_output_N": "",
"FX_output_N": "",
"FY_output_N": "",
"Fmag_raw": "",
}
def write_video(path: Path, *, frames: int = 6, fps: float = 10.0) -> Path:
"""A real, short, decodable clip whose frame number is visible in the picture.
Frame ``i`` carries a white 8x8 marker at column ``4 + (i % 6) * 8``. A test
can therefore ask the decoder which frame it got instead of trusting a seek.
"""
import cv2
import numpy as np
path.parent.mkdir(parents=True, exist_ok=True)
size = (64, 48) # width, height
writer = cv2.VideoWriter(str(path), cv2.VideoWriter_fourcc(*"mp4v"), float(fps), size)
if not writer.isOpened():
raise RuntimeError("the MP4 writer is unavailable in this environment")
try:
for index in range(frames):
image = np.zeros((size[1], size[0], 3), dtype=np.uint8)
column = 4 + (index % 6) * 8
image[4:12, column : column + 8] = 255
writer.write(image)
finally:
writer.release()
return path
def decoded_frame_number(path: Path, *, segment_frame: int | None = None) -> int:
"""Decode one frame of a :func:`write_video` clip and read its marker column."""
import cv2
capture = cv2.VideoCapture(str(path))
try:
if segment_frame is not None:
capture.set(cv2.CAP_PROP_POS_FRAMES, max(0, int(segment_frame) - 1))
ok, image = capture.read()
if not ok:
raise RuntimeError(f"could not decode {path.name}")
column = int(image[4:12].sum(axis=(0, 2)).argmax())
return (column - 4) // 8
finally:
capture.release()
def write_capture(
directory: Path,
frames: list[dict],
forces: list[dict],
*,
fields: list[str] | None = None,
video_segments: int = 1,
video_frames: int = 6,
) -> Path:
"""Write ``frames.csv``, ``forces.csv`` and one real MP4 per video segment."""
directory.mkdir(parents=True, exist_ok=True)
write_csv(directory / "frames.csv", fields or FRAMES_FIELDS, frames)
write_csv(directory / "forces.csv", FORCES_FIELDS, forces)
for segment in range(1, video_segments + 1):
write_video(directory / f"raw_{segment:04d}.mp4", frames=video_frames)
write_json_file(
directory / "session.json",
{
"format_version": "1.4",
"session_id": forces[0]["session_id"] if forces else 7,
"files": {"frames": "frames.csv", "forces": "forces.csv", "video": "raw_%04d.mp4"},
},
)
return directory
def write_json_file(path: Path, payload: dict) -> Path:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
return path
@dataclass(frozen=True)
class Hold:
"""One declared reference window as the campaign writes it."""
step_index: int
branch: str
target_load_n: float
reference_n: float
start_ns: int
end_ns: int
complete: int = 1
@property
def stability_verified(self) -> int:
return 0
def reference_row(hold: Hold, *, site_id: str, repeat_index: int, repeat_count: int) -> dict:
row = {name: "" for name in (
"step_index", "branch", "target_load_n", "t_start_host_mono_ns",
"t_end_host_mono_ns", "F_ref_normal_N", "site_id", "x", "y", "z",
"coordinate_frame", "repeat_index", "repeat_count", "hold_complete",
"reference_source", "fx_N", "fy_N", "fz_N", "force_to_tool_rotation",
"nx", "ny", "nz", "pose_x_mm", "pose_y_mm", "pose_yaw_deg",
"pose_pitch_deg", "pose_z_mm", "normal_aligned_to",
)}
row.update({
"step_index": hold.step_index,
"branch": hold.branch,
"target_load_n": hold.target_load_n,
"t_start_host_mono_ns": hold.start_ns,
"t_end_host_mono_ns": hold.end_ns,
"F_ref_normal_N": hold.reference_n,
"site_id": site_id,
"x": 0.0, "y": 0.0, "z": 0.0,
"coordinate_frame": "sensor",
"repeat_index": repeat_index,
"repeat_count": repeat_count,
"hold_complete": hold.complete,
"reference_source": "weight",
})
return row
REFERENCE_FIELDS = list(reference_row(
Hold(0, "baseline", 0.0, 0.0, 0, 1), site_id="", repeat_index=1, repeat_count=1,
).keys())
def holds_for_levels(
levels: list[float], *, first_start_ns: int = BASE_NS, hold_ns: int = 4_000_000_000,
) -> list[Hold]:
"""baseline, one load per level, then unload back to zero."""
steps: list[tuple[str, float]] = [("baseline", 0.0)]
steps += [("load", float(level)) for level in levels]
steps += [("unload", value) for value in [*reversed(levels[:-1]), 0.0]]
holds = []
for index, (branch, target) in enumerate(steps):
start = first_start_ns + index * HOLD_STRIDE_NS
holds.append(Hold(index, branch, target, target, start, start + hold_ns))
return holds
def hold_samples(
hold: Hold,
*,
count: int = 3,
first_frame_id: int,
raw_base: float,
raw_per_n: float,
raw_per_axis: dict[str, float] | None = None,
) -> tuple[list[dict], list[dict]]:
"""Evenly spaced samples inside a hold, raw linear in the declared force."""
frames: list[dict] = []
forces: list[dict] = []
span = (hold.end_ns - hold.start_ns) // (count + 1)
fn = raw_base + raw_per_n * hold.reference_n
for offset in range(count):
frame_id = first_frame_id + offset
stamp = hold.start_ns + span * (offset + 1)
frames.append(frame_row(frame_id, stamp, t_rel_s=(stamp - BASE_NS) / 1e9))
values = {"fn": fn}
if raw_per_axis:
values.update({
"fx": raw_per_axis["FX"] * hold.reference_n,
"fy": raw_per_axis["FY"] * hold.reference_n,
})
forces.append(force_row(frame_id, **values))
return frames, forces
def _write_multipoint_trial(
path: Path,
*,
session_id: str,
site_id: str,
repeat_index: int,
repeat_count: int,
holds: list[Hold],
with_capture: bool,
video: bool,
raw_base: float,
raw_per_n: float,
samples_per_hold: int,
frame_start: int,
) -> list[dict]:
write_csv(
path / "reference.csv",
REFERENCE_FIELDS,
[reference_row(hold, site_id=site_id, repeat_index=repeat_index,
repeat_count=repeat_count) for hold in holds],
)
write_csv(
path / "events.csv",
["t_host_mono_ns", "t_host_utc_ns", "kind", "detail"],
[],
)
write_json_file(path / "session.json", {
"trial_id": f"{site_id}-repeat-{repeat_index:03d}",
"site_id": site_id,
"status": "complete",
"uv_px": {"u": 128, "v": 64},
"images": {"captured": with_capture,
"reason": "" if with_capture else "camera not attached yet"},
})
if not with_capture:
return []
frames: list[dict] = []
forces: list[dict] = []
for hold in holds:
hold_frames, hold_forces = hold_samples(
hold,
count=samples_per_hold,
first_frame_id=frame_start + len(frames),
raw_base=raw_base,
raw_per_n=raw_per_n,
)
frames.extend(hold_frames)
forces.extend(hold_forces)
write_capture(path / "capture", frames, forces, fields=None if video else FRAMES_FIELDS,
video_segments=1)
return []
def multipoint_session(
root: Path,
*,
sites: tuple[str, ...] = ("P1",),
repeats: int = 1,
levels: tuple[float, ...] = (5.0,),
with_capture: bool = True,
video: bool = True,
raw_base: float = 100.0,
raw_per_n: float = 40.0,
samples_per_hold: int = 3,
simulate: bool = False,
trial_status: str = "complete",
extra_holds: list[Hold] | None = None,
) -> Path:
"""A complete multi-point session directory through the review contract."""
session_id = f"finger-SN1_{root.name}_test0001"
trials: list[dict] = []
completion: list[dict] = []
site_records = []
for site_id in sites:
site_records.append({
"site_id": site_id,
"position": {"x": 1.0, "y": 2.0, "z": 3.0},
"position_frame": "sensor",
"uv_px": {"u": 128, "v": 64},
"levels_n": list(levels),
"hold_s": 4.0,
"repeats": repeats,
})
complete = 0
for repeat in range(1, repeats + 1):
site_dir = root / "sites" / site_id / f"repeat-{repeat:03d}"
holds = holds_for_levels(list(levels))
if extra_holds:
holds = [*holds, *extra_holds]
directory = f"sites/{site_id}/repeat-{repeat:03d}"
_write_multipoint_trial(
site_dir,
session_id=session_id,
site_id=site_id,
repeat_index=repeat,
repeat_count=repeats,
holds=holds,
with_capture=with_capture,
video=video,
raw_base=raw_base,
raw_per_n=raw_per_n,
samples_per_hold=samples_per_hold,
frame_start=1,
)
trials.append({
"trial_id": f"{site_id}-repeat-{repeat:03d}",
"site_id": site_id,
"repeat_index": repeat,
"directory": directory,
"status": trial_status,
"reason": "" if trial_status == "complete" else "stopped by the operator",
})
if trial_status == "complete":
complete += 1
completion.append({"site_id": site_id, "repeats_planned": repeats,
"repeats_complete": complete,
"complete": complete >= repeats})
write_json_file(root / "session.json", {
"format_version": "autocal-multipoint-1.0",
"session_id": session_id,
"sensor_model": "finger",
"sensor_sn": "SN1",
"coordinate_frame": "sensor",
"max_force_n": 20.0,
"min_force_n": 0.0,
"simulate": simulate,
"status": "complete" if trial_status == "complete" else "incomplete",
"sites": site_records,
"trials": trials,
"completion": completion,
"started_at": "2026-10-17T00:00:00+00:00",
"sync": {"association": "A frame belongs to a hold when its t_host_mono_ns is in "
"[t_start_host_mono_ns, t_end_host_mono_ns). Rows are not paired."},
"platform_motion": False,
"automatic_force_control": False,
})
return root
def single_session(
root: Path,
*,
sites: tuple[str, ...] = ("site-001",),
trials_per_site: int = 1,
levels: tuple[float, ...] = (5.0,),
with_capture: bool = True,
raw_base: float = 100.0,
raw_per_n: float = 40.0,
samples_per_hold: int = 3,
trial_status: str = "complete",
reference_alignment_possible: bool = True,
) -> Path:
"""A complete single-trial calibration session (``autocal-session-1.0``)."""
session_id = f"finger-SN1_{root.name}_test0001"
uv_definition = {
"unit": "px", "u_positive": "right", "v_positive": "up",
"origin_raster_xy_px": [0, 0], "raster_size_hw": [64, 64],
"mapping_sha256": "test-mapping",
}
site_records = [
{"site_id": site_id, "uv_xy_px": [10 + index, 20 + index],
"position_sensor_mm": [1.0, 2.0, 3.0]}
for index, site_id in enumerate(sites)
]
trials: list[dict] = []
for site_index, site_id in enumerate(sites):
for repeat in range(1, trials_per_site + 1):
trial_id = f"trial-{len(trials) + 1:03d}"
directory = f"trials/{trial_id}"
path = root / directory
holds = holds_for_levels(list(levels))
write_csv(path / "reference.csv", REFERENCE_FIELDS, [
reference_row(hold, site_id=site_id, repeat_index=repeat, repeat_count=1)
for hold in holds
])
frames: list[dict] = []
forces: list[dict] = []
if with_capture:
for hold in holds:
hold_frames, hold_forces = hold_samples(
hold, count=samples_per_hold,
first_frame_id=1 + len(frames),
raw_base=raw_base, raw_per_n=raw_per_n,
)
frames.extend(hold_frames)
forces.extend(hold_forces)
write_csv(path / "frames.csv", FRAMES_FIELDS, frames)
write_csv(path / "forces.csv", FORCES_FIELDS, forces)
write_json_file(path / "session.json", {
"trial_id": trial_id,
"site_id": site_id,
"session_id": 7,
"calibration_session_id": session_id,
"trial_status": trial_status,
"trial_reason": "" if trial_status == "complete" else "application closed",
"reference_alignment_possible": reference_alignment_possible,
"uv_definition": uv_definition,
"steps": [{"step_index": hold.step_index, "branch": hold.branch,
"target_load_n": hold.target_load_n} for hold in holds],
})
trials.append({
"trial_id": trial_id, "site_id": site_id, "repeat_index": repeat,
"directory": directory, "status": trial_status,
"reason": "" if trial_status == "complete" else "application closed",
})
write_json_file(root / "session.json", {
"format_version": "autocal-session-1.0",
"session_id": session_id,
"status": "complete" if trial_status == "complete" else "incomplete",
"sites": site_records,
"trials": trials,
"uv_definition": uv_definition,
"sensor_model": "finger",
"sensor_sn": "SN1",
"started_at": "2026-10-17T00:00:00+00:00",
})
return root
class FakeWorker:
"""A fingerprint worker stand-in whose session can be restarted mid-recording."""
def __init__(self, *, session_id: int = 1, frame_id: int = 1, time_basis: str = "host_camera",
clock=None):
self.session_id = int(session_id)
self.is_running = True
self.sinks: list = []
self.time_basis = time_basis
self._clock = clock
self._frame_id = int(frame_id)
self._started = self._now_ns()
self.packet = self._packet()
def _now_ns(self) -> int:
return time.perf_counter_ns() if self._clock is None else int(self._clock())
def _packet(self) -> SdkPacket:
import numpy as np
now = self._now_ns()
return SdkPacket(
raw_frame=np.zeros((16, 16, 3), dtype=np.uint8),
processed_frame=np.zeros((16, 16, 3), dtype=np.uint8),
timestamp=time.time(),
fps=30,
process_ms=1,
session_id=self.session_id,
frame_id=self._frame_id,
mono_t=(now - self._started) / 1e9,
capture_mono_ns=now,
capture_utc_ns=time.time_ns(),
source_frame_id=self._frame_id,
time_basis=self.time_basis,
normal_force_raw=1.0,
)
def advance(self) -> SdkPacket:
self._frame_id += 1
self.packet = self._packet()
return self.packet
def latest(self):
return self.packet
def add_sink(self, sink) -> None:
if sink not in self.sinks:
self.sinks.append(sink)
def remove_sink(self, sink) -> None:
if sink in self.sinks:
self.sinks.remove(sink)
def emit(self) -> None:
for sink in list(self.sinks):
sink(self.packet)
def restart(self) -> None:
"""A new acquisition session with fresh frame numbers, sinks still attached."""
self.session_id += 1
self._frame_id = 1
self._started = self._now_ns()
self.packet = self._packet()
def stop(self, *, wait: bool = True) -> None:
self.is_running = False
apps/autocal/tests/test_binding_purity.py
+3
View File
@@ -62,6 +62,9 @@ _SCOPED_ENUMS = {
"TextFlag",
"AspectRatioMode",
"WindowType",
# ``Qt.WidgetAttribute.WA_DeleteOnClose`` is the scoped spelling too; the
# class simply was not needed before the video review dialog.
"WidgetAttribute",
),
"QTableWidget": ("SelectionMode", "EditTrigger", "SelectionBehavior"),
"QSizePolicy": ("Policy",),
apps/autocal/tests/test_frames_index.py
+170
View File
@@ -0,0 +1,170 @@
"""A force row belongs to a frame by identity, never by its line number."""
from __future__ import annotations
import pytest
from evidence_helpers import (
BASE_NS,
force_row,
frame_row,
write_capture,
)
from orisys_autocal.frames_index import (
DUPLICATE_REASON,
joined_samples,
read_frame_index,
window_samples,
)
def _capture(tmp_path, frames, forces, **kwargs):
return write_capture(tmp_path / "capture", frames, forces, video_segments=0, **kwargs)
def test_samples_are_paired_by_frame_id_not_by_row_order(tmp_path):
frames = [
frame_row(1, BASE_NS + 10),
frame_row(2, BASE_NS + 20),
frame_row(3, BASE_NS + 30),
]
# The force rows are reversed on disk, as an out-of-order write would leave them.
forces = [force_row(3, fn=30.0), force_row(2, fn=20.0), force_row(1, fn=10.0)]
_capture(tmp_path, frames, forces)
samples, report = joined_samples(tmp_path / "capture")
assert [sample.frame_id for sample in samples] == [1, 2, 3]
assert [sample.t_host_mono_ns for sample in samples] == [
BASE_NS + 10, BASE_NS + 20, BASE_NS + 30,
]
assert [sample.value("FN") for sample in samples] == [10.0, 20.0, 30.0]
assert report.unjoined_force_rows == 0
window = window_samples(samples, BASE_NS + 15, BASE_NS + 30)
assert [sample.frame_id for sample in window] == [2]
def test_a_repeated_frame_id_excludes_the_samples_it_would_pair(tmp_path):
"""A camera that restarted reuses frame numbers; neither copy is the measurement."""
frames = [
frame_row(1, BASE_NS + 10),
frame_row(2, BASE_NS + 20),
frame_row(1, BASE_NS + 1_000),
frame_row(2, BASE_NS + 1_010),
]
forces = [
force_row(1, fn=10.0),
force_row(2, fn=20.0),
force_row(1, fn=11.0),
force_row(2, fn=21.0),
]
_capture(tmp_path, frames, forces)
samples, report = joined_samples(tmp_path / "capture")
assert samples == []
assert report.duplicate_frame_ids == 2
assert DUPLICATE_REASON in report.reasons
def test_frames_outside_the_video_are_kept_only_when_asked_for(tmp_path):
frames = [
frame_row(1, BASE_NS + 10, in_video=1),
frame_row(2, BASE_NS + 20, in_video=0),
frame_row(3, BASE_NS + 30, in_video=1),
]
forces = [force_row(1, fn=1.0), force_row(2, fn=2.0), force_row(3, fn=3.0)]
_capture(tmp_path, frames, forces)
with_video, report = joined_samples(tmp_path / "capture", require_video=True)
everything, _ = joined_samples(tmp_path / "capture")
assert [sample.frame_id for sample in with_video] == [1, 3]
assert [sample.frame_id for sample in everything] == [1, 2, 3]
assert report.frames_not_in_video == 1
def test_a_dropped_frame_row_has_no_time_to_pair_with(tmp_path):
frames = [
frame_row(1, BASE_NS + 10),
frame_row(2, 0, dropped=1),
frame_row(3, BASE_NS + 30),
]
forces = [force_row(1, fn=1.0), force_row(2, fn=2.0), force_row(3, fn=3.0)]
_capture(tmp_path, frames, forces)
samples, report = joined_samples(tmp_path / "capture")
assert [sample.frame_id for sample in samples] == [1, 3]
assert report.unjoined_force_rows == 1
assert report.frames_dropped == 1
def test_a_recording_without_an_in_video_column_still_joins(tmp_path):
"""Older recordings have no membership column; unknown is not "outside"."""
fields = [
name for name in (
"frame_id", "source_frame_id", "t_rel_s", "t_host_mono_ns", "t_host_utc_ns",
"acquired_at", "time_basis", "media_time_ms", "segment", "segment_frame",
"dropped", "capture_dropped",
)
]
frames = [
{"frame_id": 1, "t_host_mono_ns": BASE_NS + 10, "dropped": 0, "t_rel_s": "0.0"},
{"frame_id": 2, "t_host_mono_ns": BASE_NS + 20, "dropped": 0, "t_rel_s": "0.0"},
]
_capture(tmp_path, frames, [force_row(1, fn=1.0), force_row(2, fn=2.0)], fields=fields)
samples, report = joined_samples(tmp_path / "capture", require_video=True)
assert [sample.frame_id for sample in samples] == [1, 2]
assert report.frames_not_in_video == 0
def test_a_missing_axis_is_empty_not_zero(tmp_path):
frames = [frame_row(1, BASE_NS + 10), frame_row(2, BASE_NS + 20)]
forces = [
force_row(1, fn=10.0, fx=0.5, fy=-0.25),
force_row(2, fn=None, fx=None, fy=None, valid=0),
]
_capture(tmp_path, frames, forces)
samples, _ = joined_samples(tmp_path / "capture")
assert samples[0].values == (10.0, 0.5, -0.25)
assert samples[1].value("FN") is None
assert samples[1].valid is False
assert samples[1].usable() is False
assert samples[0].usable() is True
def test_the_frame_index_keeps_repeated_ids_and_unknown_video_membership(tmp_path):
frames = [
frame_row(1, BASE_NS + 10, in_video=1),
frame_row(1, BASE_NS + 20, in_video=0),
]
_capture(tmp_path, frames, [force_row(1, fn=1.0)])
index = read_frame_index(tmp_path / "capture")
assert [row.frame_id for row in index] == [1, 1]
assert [row.in_video for row in index] == [True, False]
def test_a_missing_capture_directory_reads_as_empty(tmp_path):
samples, report = joined_samples(tmp_path / "absent")
assert samples == []
assert report.frames_total == 0 and report.force_rows == 0
@pytest.mark.parametrize("frame_id", [0, 120])
def test_frame_ids_are_used_verbatim(tmp_path, frame_id):
"""Zero and large ids are still identities; nothing is renumbered."""
frames = [frame_row(frame_id, BASE_NS + 10)]
_capture(tmp_path, frames, [force_row(frame_id, fn=2.0)])
samples, _ = joined_samples(tmp_path / "capture")
assert samples[0].frame_id == frame_id
apps/autocal/tests/test_multipoint.py
+50 -6
View File
@@ -347,8 +347,13 @@ def test_an_outlier_in_the_stable_half_is_left_out_of_its_own_mean(tmp_path):
assert float(row["FY_mean"]) == pytest.approx(-0.375)
def test_a_reconnected_camera_keeps_each_hold_on_its_own_frames(tmp_path):
"""A camera that starts again reuses frame numbers. Each hold keeps its own rows."""
def test_reused_frame_numbers_are_left_out_of_the_means(tmp_path):
"""A camera that restarts reuses frame numbers; neither copy is a measurement.
Pairing by row order used to keep both copies, which silently moved a sample
into the wrong hold. A repeated identity now excludes the samples it would
pair and the row says so through ``samples``/``unit`` being empty.
"""
from orisys_autocal.multipoint import stable_component_rows
repeat = tmp_path / "repeat-001"
@@ -389,10 +394,49 @@ def test_a_reconnected_camera_keeps_each_hold_on_its_own_frames(tmp_path):
rows = stable_component_rows(repeat, site_id="P1", repeat_index=1)
assert float(rows[0]["FN_mean"]) == pytest.approx(20.0)
assert rows[0]["branch"] == "load"
assert float(rows[1]["FN_mean"]) == pytest.approx(5.0)
assert rows[1]["branch"] == "unload"
assert rows[0]["branch"] == "load" and rows[1]["branch"] == "unload"
assert all(row["samples"] == 0 and row["unit"] == "" for row in rows)
assert all(row["FN_mean"] == row["FX_mean"] == row["FY_mean"] == "" for row in rows)
def test_a_missing_force_row_does_not_shift_the_other_holds(tmp_path):
"""The defect the line-number join caused: one lost row moved every later value."""
from orisys_autocal.multipoint import stable_component_rows
repeat = tmp_path / "repeat-001"
(repeat / "capture").mkdir(parents=True)
first, second = (1_000, 1_100), (5_000, 5_100)
with (repeat / "reference.csv").open("w", newline="", encoding="utf-8") as handle:
writer = csv.DictWriter(handle, fieldnames=[
"step_index", "branch", "target_load_n", "t_start_host_mono_ns",
"t_end_host_mono_ns", "hold_complete", "reference_source",
])
writer.writeheader()
writer.writerow({"step_index": 1, "branch": "load", "target_load_n": 5,
"t_start_host_mono_ns": first[0], "t_end_host_mono_ns": first[1],
"hold_complete": 1, "reference_source": "weight"})
writer.writerow({"step_index": 2, "branch": "load", "target_load_n": 5,
"t_start_host_mono_ns": second[0], "t_end_host_mono_ns": second[1],
"hold_complete": 1, "reference_source": "weight"})
with (repeat / "capture" / "frames.csv").open("w", newline="", encoding="utf-8") as handle:
writer = csv.DictWriter(handle, fieldnames=["frame_id", "t_host_mono_ns", "dropped"])
writer.writeheader()
for frame_id, stamp in ((1, 1060), (2, 5070)):
writer.writerow({"frame_id": frame_id, "t_host_mono_ns": stamp, "dropped": 0})
# Frame 1's force row was lost; only the second hold has a sample.
with (repeat / "capture" / "forces.csv").open("w", newline="", encoding="utf-8") as handle:
writer = csv.DictWriter(handle, fieldnames=[
"frame_id", "valid", "FN_output_N", "FX_output_N", "FY_output_N",
])
writer.writeheader()
writer.writerow({"frame_id": 2, "valid": 1,
"FN_output_N": 42.0, "FX_output_N": 0.0, "FY_output_N": 0.0})
rows = stable_component_rows(repeat, site_id="P1", repeat_index=1)
assert rows[0]["FN_mean"] == "" and rows[0]["samples"] == 0
assert float(rows[1]["FN_mean"]) == pytest.approx(42.0)
assert rows[1]["samples"] == 1
def _hold_for(campaign: Campaign, force: float) -> None:
apps/autocal/tests/test_multipoint_recording.py
+362
View File
@@ -0,0 +1,362 @@
"""Multi-point capture: repeat ownership, one source session, one verdict.
These tests use the shared recorder with a fake worker and a real camera-sized
packet, so the video file, frame index and closing report are the production
ones. No controller and no real camera are involved.
"""
from __future__ import annotations
import json
import time
from types import SimpleNamespace
import pytest
from evidence_helpers import FakeWorker
from orisys_autocal.multipoint import Campaign
class _Clock:
"""One clock for both the campaign and the fake worker's frame stamps."""
def __init__(self, now: int) -> None:
self.now = int(now)
def advance(self, seconds: float) -> None:
self.now += round(seconds * 1e9)
@pytest.fixture
def clock(monkeypatch):
from orisys_autocal import multipoint as module
value = _Clock(time.perf_counter_ns())
monkeypatch.setattr(module, "time", SimpleNamespace(
perf_counter_ns=lambda: value.now,
perf_counter=lambda: value.now / 1e9,
time_ns=time.time_ns,
))
return value
def _campaign(*, repeats: int = 2) -> Campaign:
campaign = Campaign()
campaign.add_site("P1", 1, 2, 3, [5], 1, repeats, max_force_n=20)
return campaign
def _start(campaign: Campaign, root, *, simulate: bool = False) -> "object":
return campaign.start(
root, sensor_model="finger", sensor_sn="SN1",
coordinate_frame="fixture", max_force_n=20, simulate=simulate,
)
def _drain(campaign: Campaign, timeout: float = 20.0) -> None:
deadline = time.monotonic() + timeout
while campaign.capture_state == "closing" and time.monotonic() < deadline:
campaign.poll()
time.sleep(0.005)
assert campaign.capture_state != "closing", "the writer never closed"
def _record_step(campaign: Campaign, worker: FakeWorker, target: float, *, clock=None) -> None:
campaign.confirm_reference(target)
# Three frames spread across the hold: the summary averages the latter half
# and needs two usable samples there, which is the review's own rule.
for _ in range(3):
if clock is not None:
clock.advance(0.25)
worker.advance()
worker.emit()
campaign.complete_hold(
now_ns=campaign._hold_started_ns + int(campaign._hold_s * 1e9)
)
def _record_repeat(campaign: Campaign, worker: FakeWorker, *, clock=None) -> None:
for _branch, target in list(campaign._steps):
_record_step(campaign, worker, target, clock=clock)
_drain(campaign)
def _session(root) -> dict:
return json.loads((root / "session.json").read_text(encoding="utf-8"))
def test_each_repeat_owns_its_own_video_and_manifest(tmp_path, clock):
campaign = _campaign(repeats=2)
root = _start(campaign, tmp_path)
worker = FakeWorker(clock=lambda: clock.now)
campaign.attach_capture(worker, fps=10.0)
assert campaign.capture_ready
_record_repeat(campaign, worker, clock=clock)
assert campaign.phase == "load"
assert campaign._trial_entry["trial_id"] == "P1-repeat-002"
_record_repeat(campaign, worker, clock=clock)
assert campaign.phase == "done"
_drain(campaign)
session = _session(root)
assert session["status"] == "complete"
trials = {trial["trial_id"]: trial for trial in session["trials"]}
assert set(trials) == {"P1-repeat-001", "P1-repeat-002"}
for trial in trials.values():
assert trial["status"] == "complete", trial
assert trial["capture_status"] == "complete"
assert trial["images"]["captured"] is True
capture = root / trial["directory"] / "capture"
assert (capture / "raw_0001.mp4").is_file()
manifest = json.loads((root / trial["directory"] / "session.json").read_text(encoding="utf-8"))
assert manifest["status"] == "complete"
assert manifest["images"]["captured"] is True
assert manifest["capture_status"] == "complete"
assert manifest["capture_report"]["stats"]["video_frames"] == manifest["capture_report"]["stats"]["frames"] > 0
# The evidence the real recorder wrote is what the summary reads.
from orisys_autocal.summary import build_summary
summary = build_summary(root)
assert summary["windows"]
assert all(row["capture_status"] == "complete" for row in summary["windows"])
assert all(row["eligible"] for row in summary["windows"])
assert all(row["FN_mean"] is not None for row in summary["windows"])
def test_a_restarted_source_aborts_the_repeat_instead_of_mixing_identities(tmp_path):
campaign = _campaign(repeats=1)
root = _start(campaign, tmp_path)
worker = FakeWorker(session_id=4)
campaign.attach_capture(worker, fps=10.0)
_record_step(campaign, worker, campaign._steps[0][1])
# The camera is restarted under the attached sink: new session, frame 1 again.
worker.restart()
worker.emit()
assert campaign.poll()["phase"] in {"closing", "aborted"}
_drain(campaign)
session = _session(root)
trial = session["trials"][0]
assert trial["status"] == "incomplete"
assert trial["capture_status"] == "incomplete"
assert trial["capture_reason"] == "source_restarted"
assert trial["capture_report"]["guard"]["refused"] >= 1
assert session["status"] == "incomplete"
assert campaign.phase == "aborted"
def test_no_recorded_frames_keep_a_confirmed_repeat_incomplete(tmp_path):
"""The operator confirmed every hold, but the evidence is empty."""
campaign = _campaign(repeats=1)
root = _start(campaign, tmp_path)
worker = FakeWorker()
campaign.attach_capture(worker, fps=10.0)
# No packet is emitted, so the writer holds no frame at all.
for _branch, target in list(campaign._steps):
campaign.confirm_reference(target)
campaign.complete_hold(
now_ns=campaign._hold_started_ns + int(campaign._hold_s * 1e9)
)
_drain(campaign)
session = _session(root)
trial = session["trials"][0]
assert trial["status"] == "incomplete"
assert trial["capture_reason"] == "zero_frames"
assert trial["capture_report"]["stats"]["frames"] == 0
assert session["status"] == "incomplete"
def test_a_stopped_source_keeps_the_repeat_incomplete(tmp_path):
"""Frames that stop describing the end of the repeat are not complete evidence."""
campaign = _campaign(repeats=1)
root = _start(campaign, tmp_path)
worker = FakeWorker()
campaign.attach_capture(worker, fps=10.0)
_record_step(campaign, worker, campaign._steps[0][1])
# The camera goes away; the operator keeps confirming holds by hand.
worker.is_running = False
campaign.poll()
assert campaign.phase in {"closing", "aborted"}
_drain(campaign)
session = _session(root)
trial = session["trials"][0]
assert trial["capture_status"] == "incomplete"
assert trial["capture_reason"] == "source_stopped"
assert session["status"] == "incomplete"
def test_a_simulated_repeat_is_steps_complete_but_not_recorded_evidence(tmp_path):
campaign = _campaign(repeats=1)
root = campaign.start(
tmp_path, sensor_model="finger", sensor_sn="SN1",
coordinate_frame="fixture", max_force_n=20, simulate=True,
)
while campaign.phase in {"load", "hold"}:
if campaign.phase == "load":
campaign.confirm_reference(campaign._steps[campaign._step_index][1])
campaign.complete_hold(
now_ns=campaign._hold_started_ns + int(campaign._hold_s * 1e9)
)
session = _session(root)
trial = session["trials"][0]
assert campaign.phase == "done"
assert trial["status"] == "complete" # the operator finished the steps
assert trial["capture_status"] == "not_recorded" # but nothing was recorded
assert trial["capture_reason"] == "not_recorded"
assert session["status"] == "complete"
from orisys_autocal.summary import build_summary
summary = build_summary(root)
assert all(row["capture_status"] == "not_recorded" for row in summary["windows"])
assert all(row["eligible"] is False for row in summary["windows"])
def test_a_new_session_cannot_start_while_the_last_repeat_closes(tmp_path):
campaign = _campaign(repeats=1)
_start(campaign, tmp_path)
worker = FakeWorker()
campaign.attach_capture(worker, fps=10.0)
for _branch, target in list(campaign._steps):
_record_step(campaign, worker, target)
assert campaign.capture_state == "closing"
assert campaign.busy and not campaign.is_active
with pytest.raises(RuntimeError, match="already running"):
_start(campaign, tmp_path)
_drain(campaign)
# After the writer closed, a new session is allowed again.
second = _start(campaign, tmp_path)
assert second.is_dir()
def test_attaching_a_source_without_a_frame_raises_and_leaves_no_capture(tmp_path):
campaign = _campaign(repeats=1)
_start(campaign, tmp_path)
worker = FakeWorker()
worker.packet = None
with pytest.raises(RuntimeError, match="frame"):
campaign.attach_capture(worker, fps=10.0)
assert campaign.capture_state == "idle"
assert campaign._recorder is None
assert not (campaign.directory / "sites" / "P1" / "repeat-001" / "capture").exists()
def test_a_new_session_does_not_inherit_the_previous_source_or_rotation(tmp_path):
"""A live worker and a six-axis rotation from the last session must not leak."""
campaign = _campaign(repeats=1)
worker = FakeWorker()
campaign.start(
tmp_path, sensor_model="finger", sensor_sn="SN1",
coordinate_frame="fixture", max_force_n=20, simulate=False,
)
campaign.set_force_rotation([1, 0, 0, 0, 1, 0, 0, 0, 1])
campaign.attach_capture(worker, fps=10.0)
campaign.stop("operator stopped")
_drain(campaign)
second = campaign.start(
tmp_path, sensor_model="finger", sensor_sn="SN1",
coordinate_frame="fixture", max_force_n=20, simulate=True,
)
assert campaign._pending_worker is None
assert campaign._rotation is None
assert campaign.capture_state == "idle"
repeat = second / "sites" / "P1" / "repeat-001"
assert not (repeat / "capture").exists()
manifest = json.loads((repeat / "session.json").read_text(encoding="utf-8"))
assert manifest["images"]["captured"] is False
assert "force_to_tool_rotation" not in _session(second)
def test_a_repeat_directory_that_already_exists_is_never_reused(tmp_path):
campaign = _campaign(repeats=2)
root = _start(campaign, tmp_path)
# A leftover directory from an interrupted attempt must not be written into.
leftover = root / "sites" / "P1" / "repeat-002"
leftover.mkdir(parents=True)
with pytest.raises(RuntimeError, match="already exists"):
campaign._advance_after_repeat()
def test_the_confirm_button_waits_until_the_capture_is_recording(make_window, tmp_path):
"""A hold confirmed before the writer exists would be evidence with no video."""
from types import SimpleNamespace
from orisys_app_shared.i18n import t
window = make_window()
panel = window.calibration_panel
multipoint = panel.multipoint
worker = FakeWorker()
panel.source_panel.worker = worker
panel.source_panel.mode = SimpleNamespace(fps=10)
multipoint.campaign.add_site("P1", 0, 0, 1, [5.0], 1, 1, max_force_n=20)
panel.output_path.setText(str(tmp_path))
panel.sensor_sn.setText("SN1")
# A started campaign whose capture was never attached: not ready.
multipoint.campaign.start(
tmp_path, sensor_model="finger", sensor_sn="SN1",
coordinate_frame="sensor", max_force_n=20, simulate=False,
)
panel._sync()
assert multipoint.campaign.phase == "load"
assert not panel.confirm_button.isEnabled()
assert t("multipoint.capture_wait") in panel.target_label.text()
panel.reference.setText("0")
panel._confirm()
assert multipoint.campaign.phase == "load"
multipoint.campaign.attach_capture(worker, fps=10.0)
panel._sync()
assert panel.confirm_button.isEnabled()
assert multipoint.campaign.capture_ready
window.close()
def test_a_restarted_source_is_reported_on_the_panel(make_window, tmp_path):
from types import SimpleNamespace
from orisys_app_shared.i18n import t
window = make_window()
panel = window.calibration_panel
multipoint = panel.multipoint
worker = FakeWorker()
panel.source_panel.worker = worker
panel.source_panel.mode = SimpleNamespace(fps=10)
multipoint.campaign.add_site("P1", 0, 0, 1, [5.0], 1, 1, max_force_n=20)
multipoint.campaign.start(
tmp_path, sensor_model="finger", sensor_sn="SN1",
coordinate_frame="sensor", max_force_n=20, simulate=False,
)
multipoint.campaign.attach_capture(worker, fps=10.0)
multipoint.campaign.poll()
worker.restart()
worker.emit()
multipoint._poll()
_drain(multipoint.campaign)
multipoint._refresh()
assert multipoint.campaign.phase == "aborted"
assert t("multipoint.capture_incomplete") in multipoint.capture_status.text()
assert t("multipoint.capture_reason_restarted") in multipoint.capture_status.text()
manifest = json.loads(
(multipoint.campaign.directory / "sites" / "P1" / "repeat-001" / "session.json")
.read_text(encoding="utf-8")
)
assert manifest["capture_status"] == "incomplete"
assert manifest["capture_reason"] == "source_restarted"
window.close()
apps/autocal/tests/test_summary.py
+265
View File
@@ -0,0 +1,265 @@
"""One window table over both recording formats, with nothing invented."""
from __future__ import annotations
import pytest
from evidence_helpers import (
BASE_NS,
FORCES_FIELDS,
FRAMES_FIELDS,
Hold,
HOLD_STRIDE_NS,
multipoint_session,
read_csv,
single_session,
write_csv,
)
from orisys_autocal.summary import (
HOLD_WINDOW_METHOD,
LEGACY_WINDOW_METHOD,
REASON_INSUFFICIENT_SAMPLES,
REASON_LEGACY,
REASON_NO_FRAME_SOURCE,
REASON_REMEASURED,
SCHEMA_VERSION,
build_summary,
legacy_window_rows,
)
def _levels() -> tuple[float, ...]:
return (5.0, 10.0)
def _means(summary: dict) -> list[float | None]:
return [row["FN_mean"] for row in summary["windows"]]
def test_both_formats_produce_the_same_identity_and_means(tmp_path):
multipoint = build_summary(multipoint_session(
tmp_path / "multi", sites=("P1",), repeats=1, levels=_levels()))
single = build_summary(single_session(
tmp_path / "single", sites=("site-001",), trials_per_site=1, levels=_levels()))
for summary in (multipoint, single):
assert summary["schema_version"] == SCHEMA_VERSION
assert summary["formal_conclusion"] is False
assert summary["units"]["raw"] == "raw"
assert summary["methods"]["fit"] == "raw_on_reference_ols_v1"
assert len(summary["windows"]) == 5
assert [row["branch"] for row in summary["windows"]] == [
"baseline", "load", "load", "unload", "unload"]
assert [row["hold_id"] for row in summary["windows"]] == [1, 2, 3, 4, 5]
assert all(row["session_id"] for row in summary["windows"])
assert all(row["trial_id"] for row in summary["windows"])
assert all(row["source_file"].endswith("reference.csv") for row in summary["windows"])
assert all(row["unit"] == "raw" for row in summary["windows"])
assert all(row["window_method"] == HOLD_WINDOW_METHOD for row in summary["windows"])
assert all(row["stability_verified"] is False for row in summary["windows"])
assert _means(multipoint) == pytest.approx([100.0, 300.0, 500.0, 300.0, 100.0])
assert _means(single) == pytest.approx(_means(multipoint))
assert [row["samples_valid"] for row in multipoint["windows"]] == [2, 2, 2, 2, 2]
assert all(row["capture_status"] == "unknown" for row in multipoint["windows"])
def test_a_six_axis_site_keeps_raw_components_apart(tmp_path):
summary = build_summary(multipoint_session(
tmp_path / "rec", sites=("P1",), repeats=1, levels=_levels(),
raw_base=100.0, raw_per_n=40.0))
load = summary["windows"][2]
assert load["FN_mean"] == pytest.approx(500.0)
assert load["FX_mean"] == pytest.approx(0.0)
assert load["FY_mean"] == pytest.approx(0.0)
assert load["FN_std"] == pytest.approx(0.0)
# The newton columns never appear in this table.
assert "FN_mean" in load and "FN_N" not in load
sensitivity = {row["component"]: row for row in summary["sensitivity"]}
assert sensitivity["FN"]["slope_raw_per_N"] == pytest.approx(40.0)
assert sensitivity["FN"]["status"] == "available"
def test_a_remeasured_hold_is_listed_with_its_reason(tmp_path):
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,
)
summary = build_summary(multipoint_session(
tmp_path / "rec", sites=("P1",), repeats=1, levels=_levels(),
extra_holds=[replaced]))
assert len(summary["windows"]) == 6
replaced_rows = [row for row in summary["windows"] if row["excluded_reason"] == REASON_REMEASURED]
assert len(replaced_rows) == 1
assert replaced_rows[0]["hold_id"] == 6
assert replaced_rows[0]["eligible"] is False
assert replaced_rows[0]["FN_mean"] is None
assert replaced_rows[0]["samples_total"] == 0
def test_a_missing_frame_source_is_reported_not_filled_in(tmp_path):
summary = build_summary(multipoint_session(
tmp_path / "rec", sites=("P1",), repeats=1, levels=_levels(),
with_capture=False))
assert all(row["eligible"] is False for row in summary["windows"])
assert all(row["excluded_reason"] == REASON_NO_FRAME_SOURCE for row in summary["windows"])
assert all(row["FN_mean"] is None for row in summary["windows"])
assert all(row["capture_status"] == "missing" for row in summary["windows"])
assert REASON_NO_FRAME_SOURCE in summary["missing_reasons"]
assert summary["sensitivity"] == []
def test_a_confirmed_window_uses_the_reviewed_interval(tmp_path):
root = multipoint_session(tmp_path / "rec", sites=("P1",), repeats=1, levels=_levels())
from orisys_autocal.session_review import review_session
first = review_session(root)
window = first["sites"][0]["windows"][2]
selections = {f"{window['trial_id']}:{window['step_index']}": {
"t_start_host_mono_ns": window["hold_start_host_mono_ns"] + 1,
"t_end_host_mono_ns": window["hold_end_host_mono_ns"],
"stability_verified": True,
}}
summary = build_summary(root, selections=selections)
confirmed = [row for row in summary["windows"] if row["stability_verified"]]
assert len(confirmed) == 1
assert confirmed[0]["window_method"] == "review_selection_v1"
assert confirmed[0]["review_status"] == "confirmed"
assert confirmed[0]["t_start_host_mono_ns"] == window["hold_start_host_mono_ns"] + 1
assert confirmed[0]["samples_valid"] == 3
assert summary["selections_sha256"]
assert {row["review_status"] for row in summary["windows"] if not row["stability_verified"]} == {"candidate"}
def test_out_of_order_force_rows_do_not_change_the_means(tmp_path):
root = multipoint_session(tmp_path / "rec", sites=("P1",), repeats=1, levels=_levels())
before = build_summary(root)
capture = root / "sites" / "P1" / "repeat-001" / "capture"
rows = read_csv(capture / "forces.csv")
write_csv(capture / "forces.csv", FORCES_FIELDS, list(reversed(rows)))
after = build_summary(root)
assert _means(after) == pytest.approx(_means(before))
def test_a_repeated_frame_identity_excludes_the_window_and_is_reported(tmp_path):
root = multipoint_session(tmp_path / "rec", sites=("P1",), repeats=1, levels=_levels())
capture = root / "sites" / "P1" / "repeat-001" / "capture"
frames = read_csv(capture / "frames.csv")
forces = read_csv(capture / "forces.csv")
last_frames = frames[-3:]
last_forces = forces[-3:]
for frame, force in zip(last_frames, last_forces):
copy = dict(frame)
copy["t_host_mono_ns"] = int(frame["t_host_mono_ns"]) + 1
frames.append(copy)
copied = dict(force)
copied["FN_raw"] = repr(float(force["FN_raw"]) + 1.0)
forces.append(copied)
write_csv(capture / "frames.csv", FRAMES_FIELDS, frames)
write_csv(capture / "forces.csv", FORCES_FIELDS, forces)
summary = build_summary(root)
assert any(issue["reason"] == "duplicate_frame_identity" for issue in summary["issues"])
last = summary["windows"][-1]
assert last["samples_total"] == 0
assert last["excluded_reason"] == REASON_INSUFFICIENT_SAMPLES
assert last["eligible"] is False
# The earlier windows still have their own, unambiguous samples.
assert all(row["samples_valid"] == 2 for row in summary["windows"][:-1])
def test_two_sessions_are_never_merged(tmp_path):
first = build_summary(multipoint_session(
tmp_path / "a", sites=("P1",), repeats=1, levels=_levels()))
second = build_summary(multipoint_session(
tmp_path / "b", sites=("P2",), repeats=2, levels=(5.0,)))
assert {row["session_id"] for row in first["windows"]} == {first["session_id"]}
assert {row["session_id"] for row in second["windows"]} == {second["session_id"]}
assert first["session_id"] != second["session_id"]
assert len(second["windows"]) == 6 # baseline, load, unload per repeat
def test_an_old_stable_means_table_is_kept_but_never_fitted(tmp_path):
path = tmp_path / "stable_means.csv"
write_csv(path, [
"site_id", "U", "V", "负载类型", "repeat_index", "step_index", "branch",
"target_load_n", "samples", "FN_mean", "FX_mean", "FY_mean", "unit",
], [{
"site_id": "20261016point0", "U": "-145", "V": "604", "负载类型": "砝码",
"repeat_index": 1, "step_index": 1, "branch": "load", "target_load_n": 1.0,
"samples": 181, "FN_mean": 10013.7, "FX_mean": -6484.37, "FY_mean": 707.088,
"unit": "raw",
}])
rows = legacy_window_rows(path, summary_id="legacy-1")
assert len(rows) == 1
row = rows[0]
assert row["window_method"] == LEGACY_WINDOW_METHOD
assert row["eligible"] is False
assert row["excluded_reason"] == REASON_LEGACY
assert row["FN_mean"] == pytest.approx(10013.7)
assert row["t_start_host_mono_ns"] is None and row["FN_std"] is None
assert row["unit"] == "raw"
def test_the_table_states_what_it_did_not_conclude(tmp_path):
summary = build_summary(multipoint_session(
tmp_path / "rec", sites=("P1",), repeats=1, levels=_levels()))
assert summary["formal_conclusion"] is False
assert summary["methods"]["window_unverified"] == HOLD_WINDOW_METHOD
assert summary["input_sha256"]
assert summary["software"]["numpy"]
assert summary["generated_at"].startswith("20")
assert summary["summary_id"]
assert all(row["summary_id"] == summary["summary_id"] for row in summary["windows"])
assert summary["sensitivity_summary"]
assert summary["sensitivity_summary"][0]["trials"] == 1
assert summary["sensitivity_summary"][0]["slope_std_raw_per_N"] is None
def test_a_sensitivity_row_exists_per_trial_and_review_set(tmp_path):
summary = build_summary(multipoint_session(
tmp_path / "rec", sites=("P1",), repeats=2, levels=_levels()))
fn_rows = [row for row in summary["sensitivity"] if row["component"] == "FN"]
assert len(fn_rows) == 2
assert {row["review_status"] for row in fn_rows} == {"candidate"}
assert [row["window_count"] for row in fn_rows] == [3, 3]
assert all(row["status"] == "available" for row in fn_rows)
assert all(row["slope_raw_per_N"] == pytest.approx(40.0) for row in fn_rows)
assert all(row["endpoint_raw_per_N"] == pytest.approx(40.0) for row in fn_rows)
assert all(row["initial_step_raw"] == pytest.approx(200.0) for row in fn_rows)
def test_an_axis_without_samples_is_unavailable_without_blocking_the_others(tmp_path):
root = multipoint_session(tmp_path / "rec", sites=("P1",), repeats=1, levels=_levels())
capture = root / "sites" / "P1" / "repeat-001" / "capture"
forces = read_csv(capture / "forces.csv")
for row in forces:
row["FX_raw"] = ""
write_csv(capture / "forces.csv", FORCES_FIELDS, forces)
summary = build_summary(root)
by_axis = {row["component"]: row for row in summary["sensitivity"]}
assert by_axis["FN"]["status"] == "available"
assert by_axis["FX"]["status"] == "no_data"
# A constant component has a unitless slope of zero and no R2; that is a
# flat signal, not a gain.
assert by_axis["FY"]["status"] == "flat_signal"
assert by_axis["FY"]["R2"] is None
assert by_axis["FY"]["slope_raw_per_N"] == pytest.approx(0.0)
assert summary["windows"][0]["axis_samples"]["FX"] == 0
apps/autocal/tests/test_summary_export.py
+175
View File
@@ -0,0 +1,175 @@
"""Publishing a summary: one set of files, one pointer, never a partial result."""
from __future__ import annotations
import csv
import json
import pytest
from evidence_helpers import multipoint_session
from orisys_autocal.summary import build_summary
from orisys_autocal.summary_export import (
POINTER_FORMAT,
POINTER_NAME,
SummaryExportError,
load_summary,
publish_summary,
render_html,
)
@pytest.fixture
def session(tmp_path):
root = multipoint_session(
tmp_path / "rec", sites=("P1", "P2"), repeats=2, levels=(5.0, 10.0))
return root
def _pointer(root):
return json.loads((root / POINTER_NAME).read_text(encoding="utf-8"))
def _read_csv(path):
with path.open(encoding="utf-8-sig", newline="") as handle:
return list(csv.DictReader(handle))
def test_publishing_writes_all_four_files_and_moves_the_pointer(session):
summary = build_summary(session)
published = publish_summary(session, summary)
assert published.is_dir()
assert published.name == f"summary_{summary['summary_id']}"
for name in ("summary.json", "summary.html", "baseline_load_data.csv",
"sensitivity_values.csv"):
assert (published / name).is_file(), name
pointer = _pointer(session)
assert pointer["format_version"] == POINTER_FORMAT
assert pointer["summary_id"] == summary["summary_id"]
assert pointer["path"] == f"derived/{published.name}"
assert set(pointer["files"]) == {
"summary.json", "summary.html", "baseline_load_data.csv", "sensitivity_values.csv"}
loaded, pointer_again = load_summary(session)
assert loaded["summary_id"] == summary["summary_id"]
assert pointer_again["summary_id"] == summary["summary_id"]
# The export is a snapshot of the same object, not a second computation.
stored = json.loads((published / "summary.json").read_text(encoding="utf-8"))
assert stored["windows"] == summary["windows"]
assert stored["sensitivity"] == summary["sensitivity"]
def test_the_html_is_self_contained_and_escapes_operator_text(session):
summary = build_summary(session)
summary = dict(summary)
summary["sensor_sn"] = "<script>alert('x')</script>"
document = render_html(summary)
assert "<script>" not in document
assert "&lt;script&gt;" in document
assert "http://" not in document and "https://" not in document
assert "not a pass/fail certificate" in document
def test_a_write_failure_leaves_the_previous_pointer_in_place(session, monkeypatch):
import orisys_autocal.summary_export as module
first = build_summary(session)
publish_summary(session, first)
before = (session / POINTER_NAME).read_text(encoding="utf-8")
second = build_summary(session)
second["generated_at"] = "2999-01-01T00:00:00+00:00"
def explode(*args, **kwargs):
raise OSError("no space left on device")
monkeypatch.setattr(module, "_write_csv", explode)
with pytest.raises(SummaryExportError, match="could not publish"):
publish_summary(session, second)
assert (session / POINTER_NAME).read_text(encoding="utf-8") == before
assert not (session / "derived" / f"summary_{second['summary_id']}").exists()
leftovers = [path for path in (session / "derived").iterdir() if path.name.startswith(".tmp-")]
assert leftovers == []
def test_the_pointer_never_moves_back_to_an_older_summary(session):
first = build_summary(session)
first["generated_at"] = "2026-10-18T00:00:00+00:00"
publish_summary(session, first)
older = build_summary(session)
older["generated_at"] = "2026-10-17T00:00:00+00:00"
with pytest.raises(SummaryExportError, match="back"):
publish_summary(session, older)
assert _pointer(session)["summary_id"] == first["summary_id"]
def test_changed_inputs_refuse_to_publish(session):
summary = build_summary(session)
forces = session / "sites" / "P1" / "repeat-001" / "capture" / "forces.csv"
text = forces.read_text(encoding="utf-8-sig")
forces.write_text(text + "\n", encoding="utf-8")
with pytest.raises(SummaryExportError, match="changed"):
publish_summary(session, summary)
assert not (session / POINTER_NAME).exists()
def test_a_cancelled_export_writes_nothing(session):
summary = build_summary(session)
with pytest.raises(SummaryExportError, match="cancelled"):
publish_summary(session, summary, cancelled=lambda: True)
assert not (session / POINTER_NAME).exists()
assert not (session / "derived").exists()
def test_the_same_summary_is_not_published_twice(session):
summary = build_summary(session)
publish_summary(session, summary)
with pytest.raises(SummaryExportError, match="already published"):
publish_summary(session, summary)
def test_load_refuses_a_tampered_export(session):
summary = build_summary(session)
published = publish_summary(session, summary)
(published / "sensitivity_values.csv").write_text("changed", encoding="utf-8")
with pytest.raises(SummaryExportError, match="modified"):
load_summary(session)
def test_the_csv_keeps_full_precision_and_utf8_text(session):
summary = build_summary(session)
summary["windows"][1]["FN_mean"] = 40.1234567890123
summary["windows"][1]["site_id"] = "测点一"
published = publish_summary(session, summary)
rows = _read_csv(published / "baseline_load_data.csv")
assert rows[1]["FN_mean"] == repr(40.1234567890123)
assert float(rows[1]["FN_mean"]) == pytest.approx(40.1234567890123)
assert rows[1]["site_id"] == "测点一"
assert rows[1]["unit"] == "raw"
assert rows[1]["window_method"] == "hold_second_half_v1"
assert rows[1]["eligible"] == "1"
assert rows[1]["samples_FN"] == "2"
def test_the_sensitivity_csv_has_one_row_per_point_trial_axis(session):
summary = build_summary(session)
published = publish_summary(session, summary)
rows = _read_csv(published / "sensitivity_values.csv")
assert len(rows) == len(summary["sensitivity"]) == 2 * 2 * 3
assert {row["component"] for row in rows} == {"FN", "FX", "FY"}
assert {row["method"] for row in rows} == {"raw_on_reference_ols_v1"}
assert {row["review_status"] for row in rows} == {"candidate"}
fn = [row for row in rows if row["component"] == "FN"]
assert all(float(row["slope_raw_per_N"]) == pytest.approx(40.0) for row in fn)
apps/autocal/tests/test_summary_fit.py
+166
View File
@@ -0,0 +1,166 @@
"""The summary fit is a forward regression of raw on the declared reference.
Hand-computed lines, one nonlinear set, and the proof that its slope is not the
reciprocal of the review's inverse fit (``N = gain * raw + offset``).
"""
from __future__ import annotations
import math
import pytest
from orisys_autocal.summary import fit_axis
def _window(load: float, fn: float | None, *, branch: str = "load", verified: bool = False,
eligible: bool = True, site: str = "P1", trial: str = "t1") -> dict:
return {
"session_id": "session-1", "site_id": site, "trial_id": trial,
"repeat_index": 1, "step_index": 0, "branch": branch,
"reference_load_n": load, "eligible": eligible,
"stability_verified": verified,
"FN_mean": fn, "FX_mean": 0.0, "FY_mean": 0.0,
}
def test_a_straight_line_is_recovered_exactly():
fit = fit_axis([(0.0, 100.0), (2.0, 180.0), (4.0, 260.0)])
assert fit["status"] == "available"
assert fit["slope_raw_per_N"] == pytest.approx(40.0)
assert fit["intercept_raw"] == pytest.approx(100.0)
assert fit["R2"] == pytest.approx(1.0)
assert fit["rmse_raw"] == pytest.approx(0.0)
assert fit["reference_min_N"] == 0.0 and fit["reference_max_N"] == 4.0
assert fit["window_count"] == 3
def test_a_nonlinear_set_matches_the_hand_computation():
# x mean 1, y mean 166.667; slope 80; intercept 86.667; SSE 1066.667;
# SST 13866.667 -> R2 1 - 1066.667 / 13866.667 = 0.923077.
fit = fit_axis([(0.0, 100.0), (1.0, 140.0), (2.0, 260.0)])
assert fit["slope_raw_per_N"] == pytest.approx(80.0)
assert fit["intercept_raw"] == pytest.approx(86.6666666667, rel=1e-9)
assert fit["R2"] == pytest.approx(0.9230769231, rel=1e-9)
assert fit["rmse_raw"] == pytest.approx(math.sqrt(1066.6666667 / 3), rel=1e-9)
def test_a_constant_signal_has_no_coefficient_of_determination():
fit = fit_axis([(0.0, 5.0), (1.0, 5.0), (2.0, 5.0)])
assert fit["status"] == "flat_signal"
assert fit["R2"] is None
assert fit["slope_raw_per_N"] == pytest.approx(0.0)
assert fit["intercept_raw"] == pytest.approx(5.0)
def test_one_reference_level_cannot_be_fitted():
fit = fit_axis([(5.0, 300.0), (5.0, 310.0)])
assert fit["status"] == "insufficient_reference_levels"
assert "slope_raw_per_N" not in fit
def test_no_finite_sample_is_reported_as_missing_data():
fit = fit_axis([])
assert fit["status"] == "no_data"
def test_repeated_reference_levels_are_weighted_but_counted():
fit = fit_axis([(0.0, 100.0), (5.0, 300.0), (5.0, 320.0), (10.0, 520.0)])
assert fit["window_count"] == 4
assert fit["reference_min_N"] == 0.0 and fit["reference_max_N"] == 10.0
assert fit["slope_raw_per_N"] == pytest.approx(42.0)
def test_the_forward_slope_is_not_the_reciprocal_of_the_inverse_fit():
"""``N = gain*raw + offset`` and ``raw = a*N + b`` answer different questions."""
import numpy as np
from orisys_autocal.session_review import _fit
raw = [100.0, 300.0, 520.0]
loads = [0.0, 5.0, 10.0]
forward = fit_axis(list(zip(loads, raw)))
inverse = _fit([
{"raw_mean": value, "reference_n": load, "status": "candidate", "trial_id": "t1"}
for value, load in zip(raw, loads)
])
gain, _offset = np.polyfit(np.array(raw), np.array(loads), 1)
assert forward["slope_raw_per_N"] == pytest.approx(42.0)
assert inverse["sensitivity_raw_per_n"] == pytest.approx(1.0 / gain)
assert abs(forward["slope_raw_per_N"] - 1.0 / gain) > 1e-3
def test_unloading_is_shown_but_never_fitted():
from orisys_autocal.summary import build_sensitivity
windows = [
_window(0.0, 100.0, branch="baseline"),
_window(5.0, 300.0),
_window(5.0, 340.0, branch="unload"),
_window(0.0, 120.0, branch="unload"),
]
rows = build_sensitivity(windows)
assert {row["component"] for row in rows} == {"FN", "FX", "FY"}
fn = next(row for row in rows if row["component"] == "FN")
assert fn["window_count"] == 2
assert fn["slope_raw_per_N"] == pytest.approx(40.0)
# The unloading values are far higher; a fit that included them would not be 40.
assert fn["endpoint_raw_per_N"] == pytest.approx(40.0)
def test_endpoint_and_initial_step_use_the_declared_loads():
from orisys_autocal.summary import build_sensitivity
windows = [
_window(0.0, 100.0, branch="baseline"),
_window(2.0, 180.0),
_window(5.0, 300.0),
]
fn = next(row for row in build_sensitivity(windows) if row["component"] == "FN")
# endpoint: (raw at the highest load - raw at zero) / highest load.
assert fn["endpoint_raw_per_N"] == pytest.approx((300.0 - 100.0) / 5.0)
# initial step: the lowest positive load minus zero.
assert fn["initial_step_raw"] == pytest.approx(80.0)
def test_a_missing_zero_load_leaves_the_derived_numbers_empty():
from orisys_autocal.summary import build_sensitivity
fn = next(row for row in build_sensitivity([_window(5.0, 300.0), _window(10.0, 500.0)])
if row["component"] == "FN")
assert fn["slope_raw_per_N"] == pytest.approx(40.0)
assert fn["endpoint_raw_per_N"] is None
assert fn["initial_step_raw"] is None
def test_confirmed_and_candidate_windows_are_never_fitted_together():
from orisys_autocal.summary import build_sensitivity
windows = [
_window(0.0, 100.0, branch="baseline"),
_window(5.0, 300.0),
_window(0.0, 100.0, branch="baseline", verified=True),
_window(10.0, 500.0, verified=True),
]
rows = build_sensitivity(windows)
by_review = {}
for row in rows:
if row["component"] == "FN":
by_review[row["review_status"]] = row
assert set(by_review) == {"candidate", "confirmed"}
assert by_review["candidate"]["window_count"] == 2
assert by_review["candidate"]["slope_raw_per_N"] == pytest.approx(40.0)
assert by_review["confirmed"]["window_count"] == 2
assert by_review["confirmed"]["slope_raw_per_N"] == pytest.approx(40.0)
apps/autocal/tests/test_summary_lifecycle.py
+150
View File
@@ -0,0 +1,150 @@
"""The candidate export, the reviewed export, and what happens when it fails."""
from __future__ import annotations
import hashlib
import time
from PySide6.QtCore import QCoreApplication
from review_test_helpers import wait_review
from evidence_helpers import multipoint_session
from orisys_autocal.summary_export import POINTER_NAME, load_summary
def _pump(panel, timeout: float = 30.0) -> None:
deadline = time.monotonic() + timeout
while panel.summary_publishing and time.monotonic() < deadline:
QCoreApplication.processEvents()
time.sleep(0.005)
assert not panel.summary_publishing, "the summary export never finished"
def _finish_simulated_campaign(campaign) -> None:
while campaign.phase in {"load", "hold"}:
if campaign.phase == "load":
branch, target = campaign._steps[campaign._step_index]
campaign.confirm_reference(target)
campaign.complete_hold(
now_ns=campaign._hold_started_ns + int(campaign._hold_s * 1e9)
)
def _acquired_hashes(root) -> dict[str, str]:
"""Hashes of the recorded evidence only: derived results are not evidence."""
return {
str(path.relative_to(root)): hashlib.sha256(path.read_bytes()).hexdigest()
for path in root.rglob("*")
if path.is_file() and "derived" not in path.parts and path.name != POINTER_NAME
}
def test_closing_the_session_publishes_a_candidate_summary(make_window, tmp_path):
root = multipoint_session(
tmp_path / "rec", sites=("P1",), repeats=1, levels=(5.0, 10.0))
before = _acquired_hashes(root)
window = make_window()
panel = window.calibration_panel
assert panel.publish_summary_for(root)
_pump(panel)
assert (root / POINTER_NAME).is_file()
summary, pointer = load_summary(root)
assert summary["formal_conclusion"] is False
assert summary["windows"]
assert all(row["review_status"] == "candidate" for row in summary["windows"])
assert pointer["path"].startswith("derived/summary_")
# Raw evidence is never rewritten by an export.
assert _acquired_hashes(root) == before
window.close()
def test_a_failed_export_is_reported_and_can_be_retried(
make_window, tmp_path, monkeypatch
):
import orisys_autocal.ui.calibration_panel as panel_module
root = multipoint_session(tmp_path / "rec", sites=("P1",), repeats=1, levels=(5.0,))
window = make_window()
panel = window.calibration_panel
def explode(*args, **kwargs):
raise RuntimeError("disk full")
monkeypatch.setattr(panel_module, "publish_summary", explode)
assert panel.publish_summary_for(root)
_pump(panel)
assert not (root / POINTER_NAME).exists()
assert "disk full" in panel.result_label.text()
assert panel.trial.status == "idle" # the acquisition result is untouched
monkeypatch.undo()
assert panel.publish_summary_for(root)
_pump(panel)
assert (root / POINTER_NAME).is_file()
window.close()
def test_a_published_candidate_is_republished_when_the_review_is_saved(qt_app, tmp_path):
from orisys_autocal.session_review import review_session
from orisys_autocal.ui.record_overview_dialog import RecordOverviewDialog
root = multipoint_session(
tmp_path / "rec", sites=("P1",), repeats=1, levels=(5.0, 10.0))
review = review_session(root)
window = review["sites"][0]["windows"][2]
selections = {f"{window['trial_id']}:{window['step_index']}": {
"t_start_host_mono_ns": window["hold_start_host_mono_ns"] + 1,
"t_end_host_mono_ns": window["hold_end_host_mono_ns"],
"stability_verified": True,
}}
reviewed = review_session(root, selections=selections)
dialog = RecordOverviewDialog(root, review=reviewed)
try:
dialog._save()
wait_review(dialog)
assert (root / POINTER_NAME).is_file()
summary, pointer = load_summary(root)
confirmed = [row for row in summary["windows"] if row["stability_verified"]]
assert confirmed and confirmed[0]["review_status"] == "confirmed"
assert pointer["summary_id"] == summary["summary_id"]
assert summary["review_relative_path"].startswith("derived/review_")
assert summary["sensitivity"]
# A second explicit publish (the retry action) makes a new version.
first_id = pointer["summary_id"]
dialog._publish_summary()
wait_review(dialog)
second, pointer_again = load_summary(root)
assert pointer_again["summary_id"] != first_id
assert second["summary_id"] == pointer_again["summary_id"]
finally:
dialog.close()
def test_the_multipoint_end_of_campaign_starts_the_export(make_window, tmp_path):
window = make_window()
panel = window.calibration_panel
multipoint = panel.multipoint
multipoint.campaign.add_site("P1", 0, 0, 1, [5.0], 1, 1, max_force_n=20)
directory = multipoint.campaign.start(
tmp_path, sensor_model="finger", sensor_sn="SN1",
coordinate_frame="sensor", max_force_n=20, simulate=True,
)
_finish_simulated_campaign(multipoint.campaign)
assert multipoint.campaign.phase == "done"
multipoint._poll()
_pump(panel)
assert (directory / POINTER_NAME).is_file()
summary, _pointer = load_summary(directory)
assert summary["session_id"] == directory.name
# A simulated campaign produces no frames, so its windows must not read as
# measurements: they are excluded with the reason the review recorded.
assert all(row["eligible"] is False for row in summary["windows"])
assert all(row["excluded_reason"] == "no_frame_source" for row in summary["windows"])
assert any(issue["reason"] == "no_frame_source" for issue in summary["issues"])
window.close()
apps/autocal/tests/test_video_review.py
+246
View File
@@ -0,0 +1,246 @@
"""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_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
apps/autocal/tests/test_video_review_dialog.py
+149
View File
@@ -0,0 +1,149 @@
"""The review dialog that shows the recorded pictures beside the curve."""
from __future__ import annotations
import pytest
from review_test_helpers import wait_review
from evidence_helpers import multipoint_session
from orisys_autocal.session_review import review_session
@pytest.fixture
def reviewed(tmp_path):
root = multipoint_session(
tmp_path / "rec", sites=("P1",), repeats=1, levels=(5.0, 10.0))
return root, review_session(root)
def _window(review: dict) -> dict:
return review["sites"][0]["windows"][1]
def test_the_dialog_shows_the_first_frame_of_the_selected_window(qt_app, reviewed):
from orisys_autocal.ui.video_review_dialog import VideoReviewDialog
root, review = reviewed
window = _window(review)
dialog = VideoReviewDialog(
root, review, trial_id=window["trial_id"], step_index=int(window["step_index"]))
try:
wait_review(dialog)
assert dialog.video is not None
assert dialog.location.frame_id > 0
assert dialog.image.pixmap() is not None and not dialog.image.pixmap().isNull()
assert "frame" in dialog.position.text().lower() or "帧" in dialog.position.text()
assert dialog.slider.maximum() == dialog.video.frame_count - 1
finally:
dialog.close()
assert dialog.video is None
def test_stepping_moves_one_recorded_frame(qt_app, reviewed):
from orisys_autocal.ui.video_review_dialog import VideoReviewDialog
root, review = reviewed
window = _window(review)
dialog = VideoReviewDialog(
root, review, trial_id=window["trial_id"], step_index=int(window["step_index"]))
try:
wait_review(dialog)
start = dialog.index
dialog._step(1)
wait_review(dialog)
assert dialog.index == start + 1
dialog._step(-1)
wait_review(dialog)
assert dialog.index == start
finally:
dialog.close()
def test_a_stale_frame_request_is_dropped(qt_app, reviewed):
from orisys_autocal.ui.video_review_dialog import VideoReviewDialog
root, review = reviewed
window = _window(review)
dialog = VideoReviewDialog(
root, review, trial_id=window["trial_id"], step_index=int(window["step_index"]))
try:
wait_review(dialog)
dialog._request(0)
dialog._request(1)
wait_review(dialog)
assert dialog.index == 1
assert dialog.video.frame_count >= 2
finally:
dialog.close()
def test_a_recording_without_video_explains_itself(qt_app, tmp_path):
from orisys_autocal.ui.video_review_dialog import VideoReviewDialog
root = multipoint_session(
tmp_path / "rec", sites=("P1",), repeats=1, levels=(5.0,), with_capture=False)
review = review_session(root)
window = review["sites"][0]["windows"][1]
dialog = VideoReviewDialog(
root, review, trial_id=window["trial_id"], step_index=int(window["step_index"]))
try:
wait_review(dialog)
assert dialog.video is None
assert not dialog.play_button.isEnabled()
assert dialog.image.text()
assert "no" in dialog.image.text().lower() or "没有" in dialog.image.text()
finally:
dialog.close()
def test_the_hold_review_dialog_offers_the_video_for_the_selected_window(qt_app, reviewed):
from orisys_autocal.ui.session_review_dialog import SessionReviewDialog
from orisys_autocal.ui.video_review_dialog import VideoReviewDialog
root, review = reviewed
dialog = SessionReviewDialog(root, review=review)
try:
assert dialog.video_button.isEnabled()
opened = dialog._video()
assert isinstance(opened, VideoReviewDialog)
wait_review(opened)
opened.close()
finally:
dialog.close()
def test_the_record_overview_offers_the_video_action(qt_app, reviewed):
from orisys_autocal.ui.record_overview_dialog import RecordOverviewDialog
from orisys_autocal.ui.video_review_dialog import VideoReviewDialog
root, review = reviewed
dialog = RecordOverviewDialog(root, review=review)
try:
actions = [action.objectName() for action in dialog.tools.menu().actions()]
assert "overview_video" in actions
dialog.table.clearSelection()
dialog.table.setCurrentCell(-1, -1)
assert dialog._video() is None
assert dialog.notice.text()
dialog.table.selectRow(0)
opened = dialog._video()
assert isinstance(opened, VideoReviewDialog)
wait_review(opened)
opened.close()
finally:
dialog.close()
def test_an_unclosed_trial_does_not_offer_the_video(qt_app, tmp_path):
from orisys_autocal.ui.session_review_dialog import SessionReviewDialog
root = multipoint_session(
tmp_path / "rec", sites=("P1",), repeats=1, levels=(5.0,), trial_status="incomplete")
review = review_session(root)
dialog = SessionReviewDialog(root, review=review)
try:
assert dialog._video() is None
assert dialog.message.text()
finally:
dialog.close()