"""Saved-session review uses frame identity and excludes ramps and invalid data.""" from __future__ import annotations import csv import json import pytest from review_test_helpers import wait_review 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() wait_review(dialog) 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() wait_review(dialog) 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() wait_review(fresh) 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()