修复回放标定选择与曲线显示数据保留

让标定选择进入处理和缓存，避免显示设置清空曲线。102 项针对性检查通过，全量门禁仍有停止时序、模型尺寸与原生崩溃问题。
2026-10-19 16:30:36 +08:00
parent 768f560820
commit 9d4563e929
12 changed files with 478 additions and 97 deletions
@@ -0,0 +1,114 @@
+"""Display controls retain measurements and redraw the same source timeline."""
+
+import numpy as np
+import pytest
+from orisys_app_shared.force.session import SessionSample, SessionStore
+from orisys_app_shared.force.source import StoreCurveSource
+
+from orisys_studio.ui.force_plots import ForcePlotsPanel
+
+
+def _store(*, partial=False):
+    store = SessionStore()
+    for index in range(21):
+        raw = float(index + 100)
+        shear = float("nan") if partial else raw / 100
+        store.append(SessionSample(
+            frame_id=index, t_rel_s=float(index),
+            fn_raw=raw, fx_raw=-raw, fy_raw=raw,
+            fn_n=raw / 100, fx_n=shear, fy_n=shear,
+            fn_raw_processed=raw, fx_raw_processed=-raw, fy_raw_processed=raw,
+            fn_n_processed=raw / 100, fx_n_processed=shear, fy_n_processed=shear,
+        ))
+    return store
+
+
+def _curve(panel, axis=0):
+    x, y = panel.curves[axis].getData()
+    if x is None:
+        return np.empty(0), np.empty(0)
+    return np.asarray(x, dtype=float), np.asarray(y, dtype=float)
+
+
+@pytest.mark.parametrize("browse", [False, True])
+def test_y_limit_settings_keep_trace_window_cursor_and_store(make_window, browse):
+    window = make_window(units="raw", plot_window_sec=4.0)
+    store = _store()
+    window.force_plots.set_store(store)
+    if browse:
+        window.force_plots.set_window(5.0, 12.0)
+    window.force_plots.set_cursor(8.0)
+    before = tuple(array.copy() for array in _curve(window.force_plots))
+    x_range = window.force_plots.plots[0].viewRange()[0]
+    # No acquisition tick follows: an idle or paused view must redraw immediately.
+    window.display_ctl.apply_plot_settings({"fn_raw_max": 2000.0})
+    for actual, expected in zip(_curve(window.force_plots), before):
+        np.testing.assert_array_equal(actual, expected)
+    assert window.force_plots.plots[0].viewRange()[0] == pytest.approx(x_range)
+    assert window.force_plots.cursor_time() == 8.0
+    assert len(store.visible(0.0, 20.0).times) == 21
+
+
+@pytest.mark.parametrize("kind", ["memory", "disk", "analysis"])
+def test_window_and_unit_changes_preserve_partial_axes_and_session_time(qt_app, kind):
+    store = _store(partial=True)
+    panel = ForcePlotsPanel()
+    try:
+        source = StoreCurveSource(store)
+        source.kind = kind
+        panel.set_curve_source(source)
+        for unit in ("N", "raw", "N"):
+            panel.set_units(unit)
+            for seconds in (4.0, 8.0, 8.0):
+                panel.set_window_sec(seconds)
+                x, y = _curve(panel)
+                assert len(x) > 1
+                assert x[-1] == 20.0
+                assert x[0] == 20.0 - seconds
+                assert y[-1] == pytest.approx(1.2 if unit == "N" else 120.0)
+                assert panel.plots[0].viewRange()[0][1] == pytest.approx(20.0)
+                if unit == "N":
+                    assert np.isnan(_curve(panel, 1)[1]).all()
+                    assert np.isnan(_curve(panel, 2)[1]).all()
+        assert len(store.visible(0.0, 20.0).times) == 21
+    finally:
+        panel.close()
+        panel.deleteLater()
+
+
+def test_back_to_latest_reloads_the_live_tail_without_duplicate_samples(qt_app):
+    panel = ForcePlotsPanel()
+    try:
+        panel.set_store(_store())
+        for _ in range(2):
+            panel.set_window(2.0, 7.0)
+            panel.follow_latest()
+            x, y = _curve(panel)
+            np.testing.assert_array_equal(x, np.arange(16.0, 21.0))
+            np.testing.assert_array_equal(y, np.arange(116.0, 121.0))
+        panel.reset()
+        assert _curve(panel)[0].size == 0
+    finally:
+        panel.close()
+        panel.deleteLater()
+
+
+def test_auto_y_range_keeps_replayed_data_visible_after_unit_switch(qt_app):
+    panel = ForcePlotsPanel()
+    try:
+        panel.set_store(_store())
+        panel.show_all()
+        panel.set_axis_limits(
+            fn_min=0, fn_max=20, shear_max=20,
+            fn_raw_min=0, fn_raw_max=0, shear_raw_max=0,
+        )
+        panel.set_units("N")
+        panel.set_units("raw")
+        qt_app.processEvents()
+        x, y = _curve(panel)
+        assert x.size == 21
+        low, high = panel.plots[0].viewRange()[1]
+        assert low <= y.min() <= y.max() <= high
+    finally:
+        panel.close()
+        panel.deleteLater()
@@ -0,0 +1,173 @@
+"""A recording's analysis choices reach processing, cache identity and curves."""
+
+import hashlib
+import json
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+from orisys_app_shared.analysis.bare_video import BareVideoSetup
+from orisys_app_shared.analysis.service import AnalysisService, ChainSpec
+from orisys_app_shared.i18n import t
+from PySide6.QtWidgets import QDialog
+
+from orisys_studio import app_window, replay_controller
+
+
+def _calibration(tmp_path):
+    from orisys.calibration import load
+
+    path = tmp_path / "chosen.json"
+    path.write_text(json.dumps(load(sensor_type="box").to_dict()), encoding="utf-8")
+    return path
+
+
+@pytest.mark.parametrize("calibrated", [True, False])
+def test_recording_dialog_keeps_all_choices_for_its_own_recording(
+    make_window, replay_recording, tmp_path, monkeypatch, calibrated,
+):
+    calibration = _calibration(tmp_path) if calibrated else None
+    setup = BareVideoSetup(
+        profile_ref="shipped/box", calibration=calibration, reference_frame=12,
+        device_id="confirmed" if calibrated else None,
+        unloaded_reference_frames=30 if calibrated else 0,
+    )
+    monkeypatch.setattr(app_window.VideoAnalysisDialog, "exec", lambda self: QDialog.Accepted)
+    monkeypatch.setattr(app_window.VideoAnalysisDialog, "choice", lambda self: setup)
+    window = make_window()
+    before = (Path(replay_recording) / "session.json").read_bytes()
+    assert window._recording_analysis_overrides(replay_recording)
+    service = window._analysis_service(replay_recording)
+    assert service.chain.calibration == calibration
+    assert service.chain.device_id == setup.device_id
+    assert service.chain.calibration_disabled is (not calibrated)
+    assert service.reference_frame == 12
+    assert service.unloaded_reference_frames == setup.unloaded_reference_frames
+    assert service.store_force
+    # Another recording must not inherit this one's reference or model.
+    other = tmp_path / "other"
+    other.mkdir()
+    (other / "session.json").write_bytes(before)
+    untouched = window._analysis_service(other)
+    assert untouched.chain.calibration is None
+    assert untouched.reference_frame is None
+    assert not untouched.store_force
+    assert (Path(replay_recording) / "session.json").read_bytes() == before
+
+
+def test_chosen_calibration_content_and_force_channel_change_cache_key(replay_recording, tmp_path):
+    calibration = _calibration(tmp_path)
+    chain = ChainSpec(sensor_type="box", calibration=calibration)
+    service = AnalysisService(replay_recording, chain=chain, store_force=True)
+    original = service.dependency_report()["key"].digest
+    data = json.loads(calibration.read_text(encoding="utf-8"))
+    data["meta"] = {**data.get("meta", {}), "note": "changed coefficients file"}
+    calibration.write_text(json.dumps(data), encoding="utf-8")
+    assert service.dependency_report()["key"].digest != original
+    recorded = AnalysisService(replay_recording, chain=chain, store_force=False)
+    assert recorded.dependency_report()["key"].digest != service.dependency_report()["key"].digest
+
+
+def test_reference_selector_uses_recorded_ids_and_times(qt_app, tmp_path):
+    from orisys_app_shared.analysis.bare_video import VideoFacts
+    from orisys_app_shared.sensors import scan_sensor_catalog
+
+    dialog = app_window.VideoAnalysisDialog(
+        video=tmp_path / "clip.mp4",
+        facts=VideoFacts(tmp_path / "clip.mp4", 33, 30.0, 640, 480),
+        catalog=scan_sensor_catalog(),
+    )
+    try:
+        assert dialog.select_profile("shipped/box")
+        dialog.select_calibration(None)
+        dialog.set_reference_times({20: 0.0, 36: .9, 52: 1.9})
+        assert dialog.reference_spin.minimum() == 20
+        assert dialog.reference_spin.maximum() == 52
+        assert dialog.choice().reference_frame is None
+        dialog.reference_spin.setValue(36)
+        assert dialog.choice().reference_frame == 36
+        assert dialog.reference_time_label.text() == t("video_analysis.reference_time", seconds="0.900")
+        dialog.reference_spin.setValue(35)
+        assert dialog.choice() is None
+        assert dialog.note_label.text() == t("video_analysis.reference_missing", frame=35)
+    finally:
+        dialog.close()
+        dialog.deleteLater()
+
+
+def test_recalibrated_recording_uses_recomputed_trace_through_display_controls(
+    make_window, replay_recording, monkeypatch,
+):
+    from orisys_app_shared.force.session import SessionSample, SessionStore
+    from orisys_app_shared.force.source import StoreCurveSource
+
+    window = make_window(units="raw")
+    assert window.replay.open_recording(replay_recording)
+    session = SimpleNamespace(reader=SimpleNamespace(
+        provenance=SimpleNamespace(sources={"forces": "recomputed"}), meta={},
+    ))
+    store = SessionStore()
+    for index in range(3):
+        store.append(SessionSample(
+            frame_id=index, t_rel_s=float(index), fn_raw=100.0 + index,
+            fn_raw_processed=100.0 + index,
+        ))
+    source = StoreCurveSource(store)
+    source.kind = "analysis"
+    source.reader = session.reader
+    source.unit_available = lambda unit: True
+    monkeypatch.setattr(replay_controller, "AnalysisCurveSource", lambda reader: source)
+    monkeypatch.setattr(window.replay_ctl, "available_result", lambda: session)
+    files = [Path(replay_recording) / name for name in ("forces.csv", "frames.csv", "session.json")]
+    before = [hashlib.sha256(path.read_bytes()).hexdigest() for path in files]
+    assert window.replay_ctl._serve_curves(session) == "recomputed"
+    assert window.force_plots.curve_kind() == "analysis"
+    assert window.force_plots.mode_label.text() == t("chart.mode_recomputed")
+    window.display_ctl.apply_plot_settings({"fn_raw_max": 2000.0})
+    window.replay_ctl.apply_source(True)
+    assert window.force_plots.curve_kind() == "analysis"
+    window._reset_display()
+    assert window.force_plots.curve_kind() == "analysis"
+    assert [hashlib.sha256(path.read_bytes()).hexdigest() for path in files] == before
+
+
+def test_accepting_setup_then_cancelling_keeps_the_last_complete_result(
+    make_window, replay_recording, monkeypatch,
+):
+    window = make_window()
+    assert window.replay.open_recording(replay_recording)
+    previous = SimpleNamespace(recording=Path(replay_recording), close=lambda: None)
+    window.replay_ctl.results.open_session(previous)
+    monkeypatch.setattr(window, "_recording_analysis_overrides", lambda recording: True)
+    monkeypatch.setattr(window.replay_ctl, "refresh_state", lambda: None)
+    monkeypatch.setattr(window.replay_ctl, "render_frame", lambda: None)
+    requested = []
+    monkeypatch.setattr(window.replay_ctl, "on_analysis_requested", lambda: requested.append(True))
+    window.replay_ctl.on_analysis_setup()
+    assert requested == [True]
+    assert window.replay_ctl.available_result() is previous
+    window.replay_ctl.on_analysis_finished(SimpleNamespace(cancelled=True, frames_written=0))
+    assert window.replay_ctl.available_result() is previous
+
+
+@pytest.mark.parametrize("family", ["box", "finger"])
+def test_explicit_raw_choice_disables_default_calibration_before_processing(monkeypatch, tmp_path, family):
+    from orisys_app_shared.analysis import service as service_module
+    import orisys.finger
+
+    applied = []
+    sensor = SimpleNamespace(
+        cap=None, _read_frame=lambda: (False, None), get_img=lambda **kwargs: None,
+        set_calibration=applied.append,
+    )
+    runtime = SimpleNamespace(
+        tactile=sensor, capture=None, read_source=lambda: (False, None),
+        process_frame=lambda frame: None, set_calibration=applied.append,
+    )
+    monkeypatch.setattr(service_module, "_import_sdk", lambda: SimpleNamespace(Sensor=lambda *args, **kwargs: sensor))
+    monkeypatch.setattr(orisys.finger, "FingerRuntime", SimpleNamespace(open=lambda **kwargs: runtime))
+    video = tmp_path / "clip.mp4"
+    video.touch()
+    chain = ChainSpec(sensor_type=family, calibration_disabled=True)
+    service_module._OpenChain(chain, video, tmp_path).open()
+    assert applied == [None]