ykcheungab
06e06126ca
选点、对准、加载与录像分散在两个面板，状态靠隐藏相位判断。现改由不依赖 Qt 的采集控制器驱动，并加入录像选项、汇总平铺副本、保持到时自动结束、会话结束回到空闲，以及无设备时的模拟采集。已运行 Autocal 非 Qt 测试并通过范围内门禁；界面测试待 Windows 复核，未做实机验收。
308 lines
13 KiB
Python
308 lines
13 KiB
Python
"""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"]
|