"""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 #: The block a session recorded without video writes (interfaces.md 2.2). VIDEO_OFF = {"video": False, "codec": "mp4v", "lossless_frames": False} def test_a_session_recorded_without_video_is_still_eligible(tmp_path): """The operator chose data only: the force rows are the evidence, not a video.""" root = multipoint_session( tmp_path / "rec", sites=("P1",), repeats=1, levels=_levels(), video=False, recording_block=VIDEO_OFF, capture_status="complete", ) 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"]) assert summary["sensitivity"] assert summary["missing_reasons"] == [] def test_a_session_without_a_recording_block_still_expects_video(tmp_path): """Old sessions have no block, so the video stays required exactly as before.""" root = multipoint_session( tmp_path / "rec", sites=("P1",), repeats=1, levels=_levels(), video=False, # data rows only, and no block saying that was intended ) from orisys_autocal.recording_options import RecordingOptions from orisys_autocal.session_review import review_session review = review_session(root) assert review["recording"] == RecordingOptions().as_block() assert RecordingOptions.from_block(review.get("recording")).video is True summary = build_summary(root) 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 REASON_NO_FRAME_SOURCE in summary["missing_reasons"]