ykcheungab
327 lines
15 KiB
Python
327 lines
15 KiB
Python
"""Saved-session review uses frame identity and excludes ramps and invalid data."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import csv
|
|
import json
|
|
|
|
import pytest
|
|
from orisys_autocal.calibration_session import write_json
|
|
from orisys_autocal.session_review import review_session, save_review
|
|
|
|
|
|
def _csv(path, rows):
|
|
with path.open("w", newline="", encoding="utf-8") as handle:
|
|
writer = csv.DictWriter(handle, fieldnames=list(rows[0]))
|
|
writer.writeheader()
|
|
writer.writerows(rows)
|
|
|
|
|
|
@pytest.fixture
|
|
def session(tmp_path):
|
|
manifest = {
|
|
"format_version": "autocal-session-1.0", "session_id": "test-session",
|
|
"status": "incomplete", "uv_definition": {"mapping_sha256": "test-map"},
|
|
"sites": [{"site_id": "site-001", "uv_xy_px": [1, -2]},
|
|
{"site_id": "site-002", "uv_xy_px": [-3, 4]}],
|
|
"trials": [],
|
|
}
|
|
for i, sensitivity in enumerate((2.0, 4.0), 1):
|
|
trial_id, site_id = f"trial-{i:03d}", f"site-{i:03d}"
|
|
entry = {"trial_id": trial_id, "site_id": site_id,
|
|
"status": "complete", "directory": f"trials/{trial_id}"}
|
|
manifest["trials"].append(entry)
|
|
path = tmp_path / entry["directory"]
|
|
path.mkdir(parents=True)
|
|
write_json(path / "session.json", {
|
|
"trial_id": trial_id, "site_id": site_id, "session_id": 7,
|
|
"calibration_session_id": "test-session", "trial_status": "complete",
|
|
"reference_alignment_possible": False, "uv_definition": manifest["uv_definition"],
|
|
})
|
|
frames, forces, references = [], [], []
|
|
for step, (branch, load) in enumerate((
|
|
("zero", 0), ("load", 5), ("load", 10), ("unload", 5), ("unload", 0))):
|
|
start = (step * 20 + 1) * 1_000_000_000
|
|
end = start + 10_000_000_000
|
|
references.append({"step_index": step, "branch": branch,
|
|
"t_start_host_mono_ns": start, "t_end_host_mono_ns": end,
|
|
"F_ref_normal_N": load, "hold_complete": 1, "stability_verified": 0})
|
|
raw = 1 + sensitivity * load + (1 if branch == "unload" and load else 0)
|
|
for offset, value, valid, dropped in (
|
|
(0, raw, 1, 0), (1_000_000_000, raw, 1, 0),
|
|
(2_000_000_000, 999, 0, 0), (3_000_000_000, 999, 1, 1),
|
|
(10_000_000_000, 999, 1, 0)):
|
|
frame_id = len(frames) + 1
|
|
frames.append({"frame_id": frame_id, "t_host_mono_ns": start + offset,
|
|
"dropped": dropped})
|
|
forces.append({"session_id": 7, "frame_id": frame_id, "FN_raw": value, "valid": valid})
|
|
_csv(path / "frames.csv", frames)
|
|
_csv(path / "forces.csv", list(reversed(forces)))
|
|
_csv(path / "reference.csv", references)
|
|
manifest["trials"].append({"trial_id": "trial-003", "status": "incomplete"})
|
|
write_json(tmp_path / "session.json", manifest)
|
|
return tmp_path
|
|
|
|
|
|
def test_review_fits_each_site_without_ramps_invalid_samples_or_unloading_bias(session):
|
|
result = review_session(session)
|
|
assert result["status"] == "exploratory"
|
|
assert result["issues"] == [{"trial_id": "trial-003", "reason": "incomplete_trial"}]
|
|
first, second = result["sites"]
|
|
assert first["fit"]["gain_n_per_raw"] == pytest.approx(.5)
|
|
assert first["fit"]["offset_n"] == pytest.approx(-.5)
|
|
assert second["fit"]["sensitivity_raw_per_n"] == pytest.approx(4)
|
|
assert all(w["samples"] == 2 and w["raw_std"] == 0 for s in result["sites"] for w in s["windows"])
|
|
assert first["windows"][3]["raw_mean"] == 12
|
|
assert not first["windows"][0]["reference_alignment_possible"]
|
|
assert not first["windows"][0]["stability_verified"]
|
|
before = (session / "session.json").read_bytes()
|
|
a, b = save_review(result), save_review(result)
|
|
assert a != b and a.is_file() and b.is_file()
|
|
assert (session / "session.json").read_bytes() == before
|
|
|
|
|
|
def test_no_samples_does_not_produce_a_coefficient(session):
|
|
path = session / "trials/trial-001/forces.csv"
|
|
_csv(path, [{"session_id": 7, "frame_id": 1, "FN_raw": "", "valid": 0}])
|
|
first = review_session(session)["sites"][0]
|
|
assert first["fit"] is None
|
|
assert all(w["raw_mean"] is None and w["status"] == "insufficient_samples" for w in first["windows"])
|
|
|
|
|
|
def test_review_refuses_mismatched_identity_and_paths_outside_session(session):
|
|
path = session / "trials/trial-001/session.json"
|
|
document = json.loads(path.read_text(encoding="utf-8"))
|
|
document["site_id"] = "wrong-site"
|
|
write_json(path, document)
|
|
with pytest.raises(ValueError, match="mismatch"):
|
|
review_session(session)
|
|
manifest_path = session / "session.json"
|
|
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
|
|
manifest["trials"][0]["directory"] = "../outside"
|
|
write_json(manifest_path, manifest)
|
|
with pytest.raises(ValueError, match="outside"):
|
|
review_session(session)
|
|
|
|
|
|
def test_review_dialog_shows_saved_windows_charts_and_new_export(qt_app, session, tmp_path):
|
|
from orisys_autocal.ui.session_review_dialog import SessionReviewDialog
|
|
|
|
dialog = SessionReviewDialog(session)
|
|
assert dialog.sites.count() == 2
|
|
assert dialog.table.rowCount() == 5
|
|
assert dialog.pages.currentIndex() == 0 and dialog.pages.count() == 2
|
|
assert len(dialog.trace_chart.axes()) == 2
|
|
assert all(len(series.attachedAxes()) == 2 for series in dialog.trace_chart.series())
|
|
assert len(dialog.chart.series()) == 4
|
|
assert len(dialog.uv_chart.series()) == 2
|
|
dialog.sites.setCurrentIndex(1)
|
|
assert "4" in dialog.summary.text()
|
|
dialog.uv_chart.series()[0].clicked.emit(dialog.uv_chart.series()[0].at(0))
|
|
assert dialog.sites.currentIndex() == 0
|
|
dialog.save_button.click()
|
|
assert len(list(session.glob("derived/*/review.json"))) == 1
|
|
dialog.show()
|
|
qt_app.processEvents()
|
|
assert dialog.grab().save(str(tmp_path / "review.png"))
|
|
dialog.close()
|
|
|
|
|
|
def test_reviewed_windows_roundtrip_and_reject_changed_inputs(session):
|
|
from orisys_autocal.session_review import load_review
|
|
|
|
selections = {f"trial-001:{index}": {
|
|
"t_start_host_mono_ns": (index * 20 + 1) * 1_000_000_000,
|
|
"t_end_host_mono_ns": (index * 20 + 3) * 1_000_000_000,
|
|
"stability_verified": True,
|
|
} for index in (0, 1, 2)}
|
|
before = (session / "trials/trial-001/reference.csv").read_bytes()
|
|
result = review_session(session, selections=selections)
|
|
first = result["sites"][0]
|
|
assert first["confirmed_fit"]["gain_n_per_raw"] == pytest.approx(.5)
|
|
assert all(w["duration_s"] == 2 for w in first["windows"][:3])
|
|
saved = save_review(result)
|
|
assert load_review(session, saved)["sites"] == result["sites"]
|
|
document = json.loads(saved.read_text(encoding="utf-8"))
|
|
assert document["baselines"][0]["status"] == "unavailable"
|
|
assert document["baselines"][0]["tracker_reference"] is False
|
|
assert document["analysis_versions"]["numpy"]
|
|
assert (session / "trials/trial-001/reference.csv").read_bytes() == before
|
|
with (session / "trials/trial-001/forces.csv").open("a", encoding="utf-8") as handle:
|
|
handle.write("\n")
|
|
with pytest.raises(ValueError, match="match"):
|
|
load_review(session, saved)
|
|
with pytest.raises(ValueError, match="changed"):
|
|
save_review(result)
|
|
|
|
|
|
def test_invalid_review_selection_does_not_become_verified(session):
|
|
for start, end in ((0, 2), (1, 1), (1, 12), (3, 4)):
|
|
with pytest.raises(ValueError):
|
|
review_session(session, selections={"trial-001:0": {
|
|
"t_start_host_mono_ns": start * 1_000_000_000,
|
|
"t_end_host_mono_ns": end * 1_000_000_000,
|
|
"stability_verified": True,
|
|
}})
|
|
|
|
|
|
def test_operator_confirms_windows_in_dialog_and_reloads_them(qt_app, session, monkeypatch):
|
|
from orisys_autocal.ui import session_review_dialog
|
|
|
|
dialog = session_review_dialog.SessionReviewDialog(session)
|
|
for row in (0, 1):
|
|
dialog.table.selectRow(row)
|
|
dialog.end_s.setValue(2)
|
|
dialog.verified.setChecked(True)
|
|
dialog.apply_button.click()
|
|
assert dialog.review["sites"][0]["windows"][row]["stability_verified"]
|
|
assert dialog.review["sites"][0]["confirmed_fit"] is not None
|
|
dialog.save_button.click()
|
|
saved = next(session.glob("derived/*/review.json"))
|
|
fresh = session_review_dialog.SessionReviewDialog(session)
|
|
monkeypatch.setattr(session_review_dialog.QFileDialog, "getOpenFileName",
|
|
lambda *args: (str(saved), ""))
|
|
fresh.load_button.click()
|
|
assert fresh.review["sites"] == dialog.review["sites"]
|
|
dialog.close()
|
|
fresh.close()
|
|
|
|
|
|
def test_baseline_image_matches_first_reviewed_zero_frame(session):
|
|
import cv2
|
|
import numpy as np
|
|
|
|
path = session / "trials/trial-001"
|
|
frames = list(csv.DictReader((path / "frames.csv").open(encoding="utf-8")))
|
|
writer = cv2.VideoWriter(str(path / "raw_0001.mp4"), cv2.VideoWriter_fourcc(*"mp4v"),
|
|
30, (32, 24))
|
|
assert writer.isOpened()
|
|
try:
|
|
for index, row in enumerate(frames):
|
|
writer.write(np.full((24, 32, 3), 20 if index == 0 else 220, np.uint8))
|
|
row.update(segment=1, segment_frame=index + 1, in_video=1)
|
|
finally:
|
|
writer.release()
|
|
_csv(path / "frames.csv", frames)
|
|
result = review_session(session, selections={"trial-001:0": {
|
|
"t_start_host_mono_ns": 1_000_000_000,
|
|
"t_end_host_mono_ns": 3_000_000_000, "stability_verified": True,
|
|
}})
|
|
saved = save_review(result)
|
|
document = json.loads(saved.read_text(encoding="utf-8"))
|
|
baseline = document["baselines"][0]
|
|
assert baseline["status"] == "saved" and baseline["segment_frame"] == 1
|
|
image = cv2.imread(str(saved.parent / baseline["image"]))
|
|
assert 10 < image.mean() < 30
|
|
|
|
|
|
@pytest.fixture
|
|
def repeated_session(session):
|
|
import shutil
|
|
|
|
source = session / "trials/trial-001"
|
|
destination = session / "trials/trial-004"
|
|
shutil.copytree(source, destination)
|
|
metadata = json.loads((destination / "session.json").read_text(encoding="utf-8"))
|
|
metadata["trial_id"] = "trial-004"
|
|
write_json(destination / "session.json", metadata)
|
|
with (destination / "forces.csv").open(encoding="utf-8") as handle:
|
|
forces = list(csv.DictReader(handle))
|
|
for row in forces:
|
|
row["FN_raw"] = float(row["FN_raw"]) + 2
|
|
_csv(destination / "forces.csv", forces)
|
|
manifest = json.loads((session / "session.json").read_text(encoding="utf-8"))
|
|
manifest["trials"].append({"trial_id": "trial-004", "site_id": "site-001",
|
|
"status": "complete", "directory": "trials/trial-004"})
|
|
write_json(session / "session.json", manifest)
|
|
result = review_session(session)
|
|
selections = {f"{w['trial_id']}:{w['step_index']}": {
|
|
"t_start_host_mono_ns": w["t_start_host_mono_ns"],
|
|
"t_end_host_mono_ns": w["t_end_host_mono_ns"], "stability_verified": True,
|
|
} for site in result["sites"] for w in site["windows"]}
|
|
return session, selections
|
|
|
|
|
|
def test_quality_uses_distinct_trials_and_never_trains_on_held_out_data(repeated_session):
|
|
import math
|
|
from orisys_autocal.session_review import load_review
|
|
|
|
directory, selections = repeated_session
|
|
result = review_session(directory, selections=selections)
|
|
quality = result["sites"][0]["quality"]
|
|
assert quality["reference_alignment_possible"] is False
|
|
assert all(row["trials"] == 2 and row["raw_range"] == 2
|
|
and row["raw_std_between_trials"] == pytest.approx(math.sqrt(2))
|
|
for row in quality["repeatability"])
|
|
assert len(quality["hysteresis"]) == 2
|
|
assert all(row["unload_minus_load_raw"] == 1 for row in quality["hysteresis"])
|
|
assert all(row["final_minus_initial_raw"] == 0 for row in quality["zero_return"])
|
|
validation = quality["validation"]
|
|
assert validation["evaluated_windows"] == 10
|
|
assert validation["rmse_n"] == pytest.approx(math.sqrt(1.05))
|
|
assert validation["mae_n"] == pytest.approx(1)
|
|
assert validation["max_abs_error_n"] == pytest.approx(1.5)
|
|
assert validation["outside_training_raw_range"] == 3
|
|
for fold in validation["folds"]:
|
|
assert fold["held_out_trial"] not in fold["training_trials"]
|
|
assert len(fold["training_trials"]) == 1
|
|
assert validation["folds"][0]["fit"]["offset_n"] == pytest.approx(-1.5)
|
|
assert validation["folds"][1]["fit"]["offset_n"] == pytest.approx(-.5)
|
|
second = result["sites"][1]["quality"]
|
|
assert second["validation"]["rmse_n"] is None
|
|
assert all(row["raw_range"] is None for row in second["repeatability"])
|
|
saved = save_review(result)
|
|
assert load_review(directory, saved)["sites"][0]["quality"] == quality
|
|
|
|
|
|
def test_quality_ignores_unconfirmed_data_and_unmatched_loads(repeated_session):
|
|
directory, selections = repeated_session
|
|
result = review_session(directory)
|
|
assert result["sites"][0]["quality"]["repeatability"] == []
|
|
assert result["sites"][0]["quality"]["validation"]["rmse_n"] is None
|
|
# Keep only one confirmed loading level in the other trial: no usable fit.
|
|
for key in list(selections):
|
|
if key.startswith("trial-004:") and key != "trial-004:1":
|
|
selections[key]["stability_verified"] = False
|
|
result = review_session(directory, selections=selections)
|
|
quality = result["sites"][0]["quality"]
|
|
assert quality["validation"]["folds"][0]["status"] == "insufficient_training"
|
|
assert len(quality["hysteresis"]) == 1
|
|
path = directory / "trials/trial-001/reference.csv"
|
|
with path.open(encoding="utf-8") as handle:
|
|
rows = list(csv.DictReader(handle))
|
|
rows[3]["F_ref_normal_N"] = "5.01"
|
|
_csv(path, rows)
|
|
quality = review_session(directory, selections=selections)["sites"][0]["quality"]
|
|
assert quality["hysteresis"] == []
|
|
|
|
|
|
def test_quality_dialog_is_reachable_and_shows_validation_tables(qt_app, repeated_session, monkeypatch, tmp_path):
|
|
from PySide6.QtWidgets import QTableWidget
|
|
from orisys_autocal.ui.session_review_dialog import SessionReviewDialog
|
|
from orisys_autocal.ui.quality_review_dialog import QualityReviewDialog
|
|
|
|
directory, selections = repeated_session
|
|
review = SessionReviewDialog(directory)
|
|
review.review = review_session(directory, selections=selections)
|
|
review._show_site()
|
|
opened = []
|
|
monkeypatch.setattr(QualityReviewDialog, "exec", lambda dialog: opened.append(dialog))
|
|
review.quality_button.click()
|
|
assert len(opened) == 1
|
|
dialog = opened[0]
|
|
assert dialog.tabs.count() == 4
|
|
repeat = dialog.findChild(QTableWidget, "quality_repeatability")
|
|
validation = dialog.findChild(QTableWidget, "quality_validation")
|
|
assert repeat.rowCount() == 5 and validation.rowCount() == 10
|
|
assert validation.item(0, 0).text() == "trial-001"
|
|
assert validation.item(0, 1).text() == "trial-004"
|
|
dialog.show()
|
|
qt_app.processEvents()
|
|
assert dialog.grab().save(str(tmp_path / "repeatability.png"))
|
|
dialog.tabs.setCurrentIndex(3)
|
|
qt_app.processEvents()
|
|
assert dialog.grab().save(str(tmp_path / "validation.png"))
|
|
dialog.close()
|
|
review.close()
|