ykcheungab
73592f085f
What
- record/index.py: RecordingIndex streams forces.csv once and keeps only the
small things — one time, one validity flag, one frame id per row plus a byte
offset every 4096 rows (~6.5 MB per 500k rows). row_range() bisects the times,
read_rows() seeks to the nearest checkpoint and scans forward, and overview()
folds every row into its bucket during the same pass to build a per-axis
min/max envelope.
- read_rows(max_points, unit) decimates **while streaming**: each bucket keeps
the extreme row of every axis in the requested unit, so a long interval never
has to exist in memory in full and a shear spike survives a flat normal channel.
- RecordingReader: info() decimates above max_rows (default 200k) and marks
itself decimated, with total_rows and a duration taken from the frame index
rather than from the view; window()/window_all()/overview()/statistics() read
the interval being asked about. list/window semantics unchanged for small
recordings.
- force/source.py: StoreCurveSource (live memory) and RecordingCurveSource (disk
interval + cached envelope) implement one interface — extent, window,
envelope, statistics, duration_s — so ForcePlotsPanel asks a *source* instead
of knowing where the data lives.
- force_plots: reads through the source (window, envelope, statistics, extent),
which also replaces its last direct store accesses.
- playback_dialog: draws the whole-session view through window_all() in the unit
being shown (bounded by CHART_MAX_POINTS), and says so when the view is
decimated ("显示 N / 共 M 行") instead of letting the operator count points.
- experiment._run: a failure that lands while the files are opening can no longer
be overwritten by the writer's own transition to "recording" (found by a flaky
test; the state check now covers STATE_FAILED).
Why
The store releases old blocks above ~4.6 h and the replay loaded every CSV row
into Python lists before drawing anything, so neither the live browser nor the
replay could show an earlier interval of a long recording. The index makes the
disk the data source and memory a cache of what is on screen.
Verification
- shared/tests/test_history_source.py (new, 8 cases) over a 12 000-row recording:
the index reports its rows and reads a bounded interval; a large recording is
marked decimated with its true duration and total rows; a small one is not; an
interval read sees the exact rows including a single peak; a decimated window
keeps the peak; the envelope fits the whole session in ≤200 points, keeps the
spike and is cached; the disk and memory sources agree on extent, values and
statistics; the whole session can be read without holding it all. Shared
suites: 146 tests, 0 failures (three consecutive runs).
- apps/studio/tools/ui_check.py: the replay section now checks the index row
count, that an interval read spans only the interval, the envelope size and
total_rows. 324 checks, 0 failures.
- apps/studio/tests/test_ui_smoke.py: 43 tests, 36 pass; the 7 failures are the
unchanged sandbox/harness artifacts.
- D:/orisys_ws/_review_phase12/reproduce.py: all ten lines unchanged;
compileall, locale and prefs parity clean.
- apps/studio/docs/record-channel.md §8 documents the index, the reader API and
the unified source, and lists what is still hardware work.
Not in this change
- Maximum RSS / backlog / loss measurements at real resolution on the target ARM
board (needs the hardware), and range CSV export.
1149 lines
42 KiB
Python
1149 lines
42 KiB
Python
"""Offscreen UI smoke tests: construction, layout adaptation, theme tokens."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
|
|
import pytest
|
|
|
|
import ui_check
|
|
from tests.conftest import visible_descendants
|
|
|
|
RESOLUTIONS = ((1280, 720), (1600, 900), (1920, 1080))
|
|
|
|
|
|
def _show(window, size):
|
|
window.resize(*size)
|
|
window.show()
|
|
from PyQt5.QtWidgets import QApplication
|
|
|
|
QApplication.processEvents()
|
|
return window
|
|
|
|
|
|
@pytest.mark.parametrize(("theme", "sensor_type"), [
|
|
("dark", "box"),
|
|
("light", "box"),
|
|
("dark", "finger"),
|
|
("light", "finger"),
|
|
])
|
|
def test_window_builds_in_every_theme(make_window, theme, sensor_type):
|
|
window = _show(make_window(theme=theme, sensor_type=sensor_type), (1400, 880))
|
|
assert window.windowTitle()
|
|
assert window.content_splitter.count() == 4
|
|
|
|
|
|
@pytest.mark.parametrize("size", RESOLUTIONS)
|
|
def test_panels_fit_within_window(make_window, size):
|
|
window = _show(
|
|
make_window(sensor_type="finger", show_cal_panel=True),
|
|
size,
|
|
)
|
|
sizes = window.content_splitter.sizes()
|
|
panels = [
|
|
window.mesh3d_panel,
|
|
window.arrow_panel,
|
|
window.force_plots,
|
|
window.cal_panel,
|
|
]
|
|
for panel, width in zip(panels, sizes):
|
|
assert panel.isVisible()
|
|
assert width >= panel.minimumWidth(), (
|
|
f"{panel.objectName() or type(panel).__name__} too narrow: {width}"
|
|
)
|
|
assert window.force_plots.width() >= window.force_plots.minimumWidth()
|
|
assert window.arrow_panel.width() >= window.arrow_panel.minimumWidth()
|
|
|
|
|
|
@pytest.mark.parametrize("size", RESOLUTIONS)
|
|
def test_children_stay_inside_their_parent(make_window, size):
|
|
window = _show(make_window(sensor_type="finger", show_cal_panel=True), size)
|
|
oversize = []
|
|
for child in visible_descendants(window):
|
|
parent = child.parentWidget()
|
|
if parent is None or not parent.isVisible():
|
|
continue
|
|
geometry = child.geometry()
|
|
if geometry.width() > parent.width() + 1 or geometry.height() > parent.height() + 1:
|
|
oversize.append(
|
|
f"{type(child).__name__} {geometry.width()}x{geometry.height()}"
|
|
f" > parent {parent.width()}x{parent.height()}"
|
|
)
|
|
assert not oversize, oversize
|
|
|
|
|
|
def test_theme_tokens_have_identical_keys():
|
|
from orisys_app_shared.theme import build_qss, load_colors
|
|
|
|
dark = set(load_colors("dark"))
|
|
light = set(load_colors("light"))
|
|
assert dark == light, {"dark_only": sorted(dark - light), "light_only": sorted(light - dark)}
|
|
assert len(build_qss("dark")) > 1000
|
|
assert len(build_qss("light")) > 1000
|
|
|
|
|
|
def test_arrow_canvas_differs_from_accent():
|
|
"""The pure-color arrow canvas must not collide with the accent pen color."""
|
|
from orisys_app_shared.theme import load_colors
|
|
|
|
for theme in ("dark", "light"):
|
|
colors = load_colors(theme)
|
|
assert colors["arrow_canvas"] != colors["accent"], theme
|
|
|
|
|
|
def test_palette_rules_hold():
|
|
"""WCAG contrast and colour-separation rules from ``tools/ui_check.py``."""
|
|
import ui_check
|
|
|
|
assert ui_check.audit_palette() == []
|
|
|
|
|
|
def test_flow_colormap_follows_the_theme_ramp():
|
|
from orisys_app_shared.acquisition.arrows import (
|
|
_lut_from_ramp,
|
|
flow_colormap,
|
|
hex_to_bgr,
|
|
)
|
|
from orisys_app_shared.theme import load_colors
|
|
|
|
for theme in ("dark", "light"):
|
|
stops = [s.strip() for s in load_colors(theme)["flow_ramp"].split(",")]
|
|
lut = flow_colormap(theme)
|
|
assert lut.shape == (256, 1, 3)
|
|
assert tuple(int(v) for v in lut[0, 0]) == hex_to_bgr(stops[0])
|
|
assert tuple(int(v) for v in lut[-1, 0]) == hex_to_bgr(stops[-1])
|
|
|
|
# Malformed ramps must never raise; they fall back to the built-in ramp.
|
|
fallback = _lut_from_ramp("not,a,ramp")
|
|
assert fallback.shape == (256, 1, 3)
|
|
|
|
|
|
def test_brand_bar_renders_and_localizes(make_window):
|
|
from orisys_app_shared.i18n import t
|
|
|
|
window = _show(make_window(language="zh_CN"), (1440, 880))
|
|
bar = window.brand_bar
|
|
assert bar.height() >= 28
|
|
assert bar.name.text()
|
|
assert bar.product.text() == t("app.title")
|
|
|
|
zh_tagline = bar.tagline.text()
|
|
window._set_language("en_US")
|
|
assert bar.tagline.text() != zh_tagline
|
|
assert bar.name.text() == "ORISYS"
|
|
|
|
logo = bar.logo.pixmap()
|
|
if logo is not None and not logo.isNull():
|
|
image = logo.toImage()
|
|
alphas = [
|
|
image.pixelColor(x, y).alpha()
|
|
for y in range(image.height())
|
|
for x in range(image.width())
|
|
]
|
|
assert any(a < 32 for a in alphas)
|
|
assert any(a >= 200 for a in alphas)
|
|
|
|
|
|
def test_brand_bar_background_comes_from_tokens(make_window):
|
|
from orisys_app_shared.theme import load_colors
|
|
|
|
window = _show(make_window(), (1440, 880))
|
|
for theme in ("dark", "light"):
|
|
window._set_theme(theme)
|
|
colors = load_colors(theme)
|
|
assert colors["brand_bar_bg"] != colors["window_bg"], theme
|
|
assert window.brand_bar._theme == theme
|
|
|
|
|
|
def test_disabled_buttons_use_muted_text_not_hover_background():
|
|
from orisys_app_shared.theme import build_qss, load_colors
|
|
|
|
for theme in ("dark", "light"):
|
|
colors = load_colors(theme)
|
|
qss = build_qss(theme)
|
|
assert f"color: {colors['text_muted']}" in qss, theme
|
|
# The old rule painted disabled buttons with the brighter hover color.
|
|
assert f"background-color: {colors['button_hover']};\n border-color: {colors['panel_border']}" not in qss
|
|
|
|
|
|
def test_splitter_sizes_round_trip_through_prefs(make_window):
|
|
window = _show(make_window(), (1440, 880))
|
|
window.content_splitter.setSizes([300, 420, 520, 0])
|
|
window._persist_prefs()
|
|
stored = window.prefs["splitter_sizes"]
|
|
assert len(stored) == 4
|
|
assert stored[2] > stored[0]
|
|
|
|
|
|
def test_status_chip_tracks_phase(make_window):
|
|
window = _show(make_window(), (1400, 880))
|
|
assert window._session_phase == "idle"
|
|
window._set_phase("running")
|
|
assert window.status_chip.text()
|
|
assert "status_running" not in window.status_chip.text()
|
|
window._set_phase("nonexistent-phase")
|
|
assert window._session_phase == "idle"
|
|
|
|
|
|
def test_run_tooltip_explains_disabled_reason(make_window):
|
|
window = _show(make_window(source_mode="video", video_path=""), (1400, 880))
|
|
assert not window.toolbar.btn_run.isEnabled()
|
|
assert window.toolbar.btn_run.toolTip()
|
|
|
|
window.toolbar.set_video_path("C:/tmp/demo.mp4")
|
|
window._update_run_enabled()
|
|
assert window.toolbar.btn_run.isEnabled()
|
|
|
|
|
|
def test_toolbar_run_button_is_labelled(make_window):
|
|
window = _show(make_window(), (1400, 880))
|
|
label = window.toolbar.btn_run.text()
|
|
assert label.strip()
|
|
assert label != "â–¶"
|
|
window.toolbar.set_run_checked(True)
|
|
assert window.toolbar.btn_run.text() != label
|
|
|
|
|
|
def test_stop_cancels_a_hung_connect(make_window):
|
|
"""Stop must stay live and take effect instantly while connecting."""
|
|
import time
|
|
|
|
import ui_check
|
|
from orisys_app_shared.i18n import t
|
|
|
|
window = _show(make_window(sensor_type="box", source_mode="camera"), (1440, 880))
|
|
stuck = ui_check._StuckWorker()
|
|
window._ensure_workers = lambda: None
|
|
window._box_worker = stuck
|
|
window._finger_worker = stuck
|
|
|
|
window.toolbar.btn_run.setChecked(True)
|
|
assert window._session_phase == "connecting"
|
|
assert window.toolbar.btn_run.isEnabled(), "Stop is disabled while connecting"
|
|
assert window.toolbar.btn_run.text() == t("toolbar.cancel_label")
|
|
assert window.act_stop.isEnabled()
|
|
|
|
started = time.perf_counter()
|
|
window.toolbar.btn_run.setChecked(False)
|
|
elapsed = time.perf_counter() - started
|
|
assert elapsed < 0.5, f"Stop blocked the GUI for {elapsed:.2f}s"
|
|
assert not window._running
|
|
assert stuck.thread_still_owned, "Stop abandoned the still-running worker thread"
|
|
|
|
|
|
def test_escape_cancels_a_hung_connect(make_window):
|
|
import time
|
|
|
|
import ui_check
|
|
|
|
window = _show(make_window(sensor_type="box", source_mode="camera"), (1440, 880))
|
|
stuck = ui_check._StuckWorker()
|
|
window._ensure_workers = lambda: None
|
|
window._box_worker = stuck
|
|
window._finger_worker = stuck
|
|
|
|
window.toolbar.btn_run.setChecked(True)
|
|
assert window.act_stop.shortcut().toString() == "Esc"
|
|
started = time.perf_counter()
|
|
window._on_stop_shortcut()
|
|
assert time.perf_counter() - started < 0.5
|
|
assert not window._running
|
|
|
|
|
|
def test_theme_switch_updates_panels(make_window):
|
|
window = _show(make_window(), (1400, 880))
|
|
window._set_theme("light")
|
|
assert window._theme == "light"
|
|
window._set_theme("dark")
|
|
assert window._theme == "dark"
|
|
|
|
|
|
def test_cal_panel_records_and_lists_samples(make_window):
|
|
window = _show(make_window(), (1400, 880))
|
|
cal = window.cal_panel
|
|
cal.set_running(True)
|
|
assert cal.btn_record.isEnabled()
|
|
assert cal.btn_record.text()
|
|
assert cal.table.columnCount() == 6, "the diagnostic columns changed"
|
|
|
|
_add_calibration_point(window, load=1.0, raw=100.2)
|
|
_add_calibration_point(window, load=2.0, raw=400.0) # off-trend
|
|
_add_calibration_point(window, load=3.0, raw=300.6)
|
|
assert cal.table.rowCount() == 3
|
|
assert not cal.empty_hint.isVisible()
|
|
|
|
headers = [
|
|
cal.table.horizontalHeaderItem(i).text() for i in range(cal.table.columnCount())
|
|
]
|
|
from orisys_app_shared.i18n import t
|
|
|
|
assert headers == [
|
|
t("cal.col_weight"),
|
|
t("cal.col_fn"),
|
|
t("cal.col_fx"),
|
|
t("cal.col_fy"),
|
|
t("cal.col_stability"),
|
|
t("cal.col_state"),
|
|
]
|
|
|
|
|
|
def _add_calibration_point(window, *, load: float, raw: float, samples: int = 40):
|
|
"""Drive one point through the recorder and into the panel's table."""
|
|
from orisys_app_shared.force.session import SessionSample
|
|
|
|
window.cal_recorder.start(channel="fn", reference_load_n=load, duration_s=1.0)
|
|
for index in range(samples):
|
|
window.cal_recorder.feed(
|
|
SessionSample(
|
|
frame_id=index,
|
|
t_rel_s=index / 100.0,
|
|
fn_raw=raw,
|
|
fx_raw=0.4,
|
|
fy_raw=-0.2,
|
|
valid=True,
|
|
)
|
|
)
|
|
point = window.cal_recorder.finish_now()
|
|
window.cal_panel.on_record_finished(point)
|
|
return point
|
|
|
|
|
|
def test_calibration_diagnostics_drive_the_fit(make_window):
|
|
"""The diagnostics must describe the model that would be exported.
|
|
|
|
The export builds a piecewise-linear curve, which passes through every point
|
|
it is built from — so the training residual is zero by construction and an
|
|
off-trend point can only show up as a *validation* error. A private
|
|
least-squares residual would have described a model nobody applies.
|
|
"""
|
|
window = _show(make_window(), (1400, 880))
|
|
window.cal_panel.set_running(True)
|
|
_add_calibration_point(window, load=1.0, raw=100.2)
|
|
_add_calibration_point(window, load=2.0, raw=400.0)
|
|
_add_calibration_point(window, load=3.0, raw=300.6)
|
|
_add_calibration_point(window, load=5.0, raw=500.9)
|
|
|
|
recorder = window.cal_recorder
|
|
fit = recorder.fit("fn")
|
|
assert fit.usable and fit.n_points == 4
|
|
assert fit.method == "piecewise_linear", fit.method
|
|
assert fit.breakpoints == 4, fit.breakpoints
|
|
assert fit.max_residual_n <= 1e-9, "the exported model misses its own points"
|
|
assert window.cal_panel.fit_label.text()
|
|
|
|
# Hold the off-trend point (load 2.0 at raw 400) out: the exported model is
|
|
# then built without it and the misfit is measurable.
|
|
window.cal_panel.table.setCurrentCell(1, 0)
|
|
window.cal_panel._on_toggle_validation()
|
|
held_out = recorder.fit("fn")
|
|
assert held_out.validation_points == 1
|
|
assert held_out.validation_error_n > 0.05, held_out.validation_error_n
|
|
assert recorder.points[1].is_validation
|
|
assert np.isnan(recorder.points[1].residual_n), "validation point entered the fit"
|
|
|
|
window.cal_panel._on_toggle_validation()
|
|
window.cal_panel._on_toggle_excluded()
|
|
assert recorder.points[1].excluded
|
|
# An excluded point leaves the export store but stays on record.
|
|
assert len(window.cal_panel._store.channel("fn")) == 3
|
|
|
|
# The exported model no longer passes through the excluded point.
|
|
from orisys_app_shared.force.calibration_io import (
|
|
build_export_model,
|
|
predict_channel,
|
|
)
|
|
|
|
store = window.cal_panel._store
|
|
model = build_export_model({key: store.channel(key) for key in ("fn", "fx", "fy")})
|
|
assert abs(predict_channel(model, "fn", 400.0) - 2.0) > 0.05
|
|
|
|
# The store, the label and the export are the same preparation: the panel
|
|
# must not derive the model twice and show numbers for one of them.
|
|
prepared = window.cal_panel._prepared
|
|
assert prepared is not None and prepared.ok
|
|
assert list(store.channel("fn")) == list(prepared.samples["fn"])
|
|
assert prepared.points_in_fit("fn") == 3
|
|
|
|
|
|
def test_the_panel_shows_the_numbers_of_the_file_it_exports(make_window, record_root):
|
|
"""Displayed diagnostics must match the written coefficients.
|
|
|
|
A repeated reference load makes this sharp: the two samples of one load must
|
|
be merged before the build, so the exported curve and the panel's residual
|
|
and validation figures all come from that single prepared model.
|
|
"""
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import orisys.calibration as sdk_calibration
|
|
|
|
from orisys_app_shared.force.calibration_io import build_force_calibration_v2
|
|
|
|
window = _show(make_window(), (1400, 880))
|
|
window.cal_panel.set_running(True)
|
|
window.cal_panel.set_meta_context(sensor_type="box", config_ref=None)
|
|
_add_calibration_point(window, load=1.0, raw=100.0)
|
|
_add_calibration_point(window, load=2.0, raw=200.0)
|
|
_add_calibration_point(window, load=2.0, raw=240.0) # a deliberate repeat
|
|
_add_calibration_point(window, load=4.0, raw=400.0)
|
|
_add_calibration_point(window, load=3.0, raw=290.0)
|
|
|
|
panel = window.cal_panel
|
|
panel.table.setCurrentCell(4, 0)
|
|
panel._on_toggle_validation()
|
|
panel._rebuild_store()
|
|
panel._refresh_fit_label()
|
|
prepared = panel._prepared
|
|
assert prepared is not None and prepared.ok
|
|
assert len(prepared.samples["fn"]) == 3, "the repeated load was not merged"
|
|
|
|
# Write the document the export would write, load it back with the SDK and
|
|
# check that its predictions are the prepared ones.
|
|
document, _warnings = build_force_calibration_v2(prepared)
|
|
path = Path(record_root) / "force_calibration.json"
|
|
path.write_text(json.dumps(document), encoding="utf-8")
|
|
loaded = sdk_calibration.load(path, strict=True)
|
|
|
|
for point in window.cal_recorder.points:
|
|
raw = float(point.loaded_stats().mean)
|
|
predicted = float(loaded.get("fn").convert(raw))
|
|
assert predicted == pytest.approx(prepared.predict("fn", raw), abs=1e-9)
|
|
|
|
# ...and the label describes that same model.
|
|
label = panel.fit_label.text()
|
|
assert "分段线性" in label or "piecewise" in label, label
|
|
assert "3" in label
|
|
assert prepared.max_validation_error("fn") > 0.0
|
|
|
|
|
|
def test_calibration_point_dialog_shows_three_axes(make_window):
|
|
window = _show(make_window(), (1400, 880))
|
|
_add_calibration_point(window, load=1.0, raw=100.2)
|
|
point = window.cal_recorder.points[0]
|
|
assert set(point.trace) == {"fn", "fx", "fy"}
|
|
assert point.trace_dt_s > 0
|
|
|
|
from orisys_studio.ui.cal_point_dialog import CalibrationPointDialog
|
|
|
|
dialog = CalibrationPointDialog(point, theme="dark")
|
|
assert dialog.windowTitle()
|
|
assert len(dialog.plot_widget.ci.items) == 3, "not one trace per axis"
|
|
dialog.close()
|
|
|
|
|
|
def _write_session_file(root, *, sensor_type, sn, config_ref, points):
|
|
import json
|
|
|
|
from orisys_app_shared.force.calibration_session import build_session_document
|
|
|
|
document = build_session_document(
|
|
points,
|
|
meta={"sensor_type": sensor_type, "sn": sn, "config_ref": config_ref},
|
|
)
|
|
path = Path(root) / f"calibration_session_{sensor_type}.json"
|
|
path.write_text(json.dumps(document), encoding="utf-8")
|
|
return path
|
|
|
|
|
|
def test_opening_device_a_record_keeps_device_a_identity(make_window, record_root):
|
|
"""Device A's record must not be restamped with device B's identity."""
|
|
import json
|
|
from unittest.mock import patch as _patch
|
|
|
|
import orisys_studio.ui.cal_panel as cal_panel_module
|
|
from orisys_app_shared.force.calibration_io import build_force_calibration_v2
|
|
|
|
window = _show(make_window(), (1400, 880))
|
|
panel = window.cal_panel
|
|
# A record taken on device A, with a full audit series.
|
|
session_points = [
|
|
_add_calibration_point(window, load=1.0, raw=100.0),
|
|
_add_calibration_point(window, load=3.0, raw=300.0),
|
|
]
|
|
source = _write_session_file(
|
|
record_root,
|
|
sensor_type="box",
|
|
sn="SN-A",
|
|
config_ref="D:/lab/deviceA/sensor.json",
|
|
points=session_points,
|
|
)
|
|
window.cal_recorder.clear_points()
|
|
|
|
# ...and the live context is device B.
|
|
panel.set_meta_context(sensor_type="finger", config_ref="D:/lab/deviceB/sensor.json")
|
|
panel.sn_edit.setText("SN-B")
|
|
|
|
panel._on_open_record(str(source))
|
|
assert panel.sn_edit.text() == "SN-A", "the record's SN was overwritten"
|
|
meta = panel._export_meta()
|
|
assert meta["sensor_type"] == "box"
|
|
assert meta["sn"] == "SN-A"
|
|
assert meta["config_ref"] == "D:/lab/deviceA/sensor.json"
|
|
assert meta["source_record"].endswith(source.name)
|
|
|
|
# Saving the record again still says A.
|
|
import orisys_studio.calibration_service as cal_service_module
|
|
|
|
with _patch.object(cal_panel_module, "calibration_export_dir", lambda: Path(record_root)):
|
|
with _patch.object(
|
|
cal_service_module, "calibration_export_dir", lambda: Path(record_root)
|
|
):
|
|
with _patch.object(cal_panel_module.QMessageBox, "information", lambda *a, **k: None):
|
|
panel._on_save_record()
|
|
saved = sorted(Path(record_root).glob("calibration_session_box_*.json"))
|
|
assert saved, "the record was not saved under device A's type"
|
|
saved_meta = json.loads(saved[-1].read_text(encoding="utf-8"))["meta"]
|
|
assert saved_meta["sn"] == "SN-A"
|
|
assert saved_meta["sensor_type"] == "box"
|
|
assert saved_meta["config_ref"] == "D:/lab/deviceA/sensor.json"
|
|
|
|
# ...and so does the exported coefficient document.
|
|
document, _warnings = build_force_calibration_v2(panel._prepared)
|
|
assert document["meta"]["sensor_type"] == "box"
|
|
assert document["meta"]["sn"] == "SN-A"
|
|
assert document["meta"]["config_ref"] == "D:/lab/deviceA/sensor.json"
|
|
|
|
|
|
def test_a_reopened_point_can_be_inspected_from_its_samples(make_window, record_root):
|
|
"""A record with samples but no cached trace still opens its diagnostics."""
|
|
window = _show(make_window(), (1400, 880))
|
|
panel = window.cal_panel
|
|
points = [
|
|
_add_calibration_point(window, load=1.0, raw=100.0),
|
|
_add_calibration_point(window, load=2.0, raw=200.0),
|
|
]
|
|
import json
|
|
|
|
from orisys_app_shared.force.calibration_session import build_session_document
|
|
|
|
document = build_session_document(points, meta={"sensor_type": "box"})
|
|
for entry in document["points"]:
|
|
entry["trace"] = {} # only the audit series survived
|
|
entry["trace_dt_s"] = 0.0
|
|
source = Path(record_root) / "no_trace.json"
|
|
source.write_text(json.dumps(document), encoding="utf-8")
|
|
|
|
panel._on_open_record(str(source))
|
|
reopened = window.cal_recorder.points[0]
|
|
assert reopened.samples, "the reopened point lost its samples"
|
|
dialog = panel.point_dialog(reopened)
|
|
assert dialog is not None, "the reopened point could not be inspected"
|
|
assert len(dialog.plot_widget.ci.items) == 3
|
|
dialog.close()
|
|
|
|
|
|
def test_applying_another_devices_calibration_is_flagged(make_window, record_root):
|
|
"""The apply path states a sensor/config mismatch instead of silently applying."""
|
|
import json
|
|
|
|
from orisys_app_shared.force.calibration_io import build_export_model, write_force_calibration
|
|
from orisys_app_shared.force.calibration_io import CalSample, CalSessionStore
|
|
|
|
window, stuck = _running_window(make_window)
|
|
store = CalSessionStore()
|
|
store.aggregate("fn", CalSample(weight_n=0.0, raw_mean=0.0, n=10))
|
|
store.aggregate("fn", CalSample(weight_n=5.0, raw_mean=500.0, n=10))
|
|
model = build_export_model({key: store.channel(key) for key in ("fn", "fx", "fy")})
|
|
document = model.to_dict()
|
|
document["identity"] = {"sensor_type": "box", "sn": "SN-A"}
|
|
document["pipeline"] = {"config_ref": "D:/lab/deviceA/sensor.json"}
|
|
path = Path(record_root) / "deviceA_calibration.json"
|
|
path.write_text(json.dumps(document), encoding="utf-8")
|
|
|
|
# The window is connected to a finger device.
|
|
mismatch = window.calibration_target_mismatch(
|
|
str(path), sensor_type="finger", config_ref="D:/lab/deviceB/sensor.json"
|
|
)
|
|
assert any("sensor_type" in reason for reason in mismatch), mismatch
|
|
assert any("config_ref" in reason for reason in mismatch), mismatch
|
|
assert not window.calibration_target_mismatch(
|
|
str(path), sensor_type="box", config_ref="D:/lab/deviceA/sensor.json"
|
|
)
|
|
_finish(window, stuck)
|
|
|
|
|
|
def test_force_plots_render_demo_data(make_window):
|
|
window = _show(make_window(), (1440, 880))
|
|
for i in range(60):
|
|
window.force_plots.push(i / 30.0, float(i), -1.0, 1.0, unit="N")
|
|
from PyQt5.QtWidgets import QApplication
|
|
|
|
QApplication.processEvents()
|
|
assert window.force_plots.history.sample_count > 0
|
|
assert all(label.text() for label in window.force_plots.value_labels)
|
|
|
|
|
|
# --------------------------------------------------------------- record channel
|
|
|
|
|
|
def _running_window(make_window):
|
|
"""A window in the ``running`` phase whose worker never opens a device.
|
|
|
|
The Qt timer is stopped deliberately: these tests drive the display path by
|
|
calling ``_on_run_tick`` themselves, and a concurrent 16 ms timer would race
|
|
the assertions (and outlive the window on teardown).
|
|
"""
|
|
window = _show(make_window(sensor_type="box", source_mode="camera"), (1440, 880))
|
|
stuck = ui_check._StuckWorker()
|
|
window._ensure_workers = lambda: None
|
|
window._box_worker = stuck
|
|
window._finger_worker = stuck
|
|
window._start_run()
|
|
window._run_timer.stop()
|
|
window._set_phase("running")
|
|
return window, stuck
|
|
|
|
|
|
def _finish(window, stuck):
|
|
window._stop_run(reason="user_stop")
|
|
stuck.release()
|
|
|
|
|
|
def test_record_channel_does_not_depend_on_the_ui_timer(make_window):
|
|
window, stuck = _running_window(make_window)
|
|
for index in range(90):
|
|
stuck.emit(ui_check._fake_packet(index))
|
|
|
|
assert window.session_store.count == 90, "samples were lost without a UI tick"
|
|
_finish(window, stuck)
|
|
|
|
|
|
def test_invalid_samples_are_excluded_not_zeroed(make_window):
|
|
import math
|
|
|
|
window, stuck = _running_window(make_window)
|
|
for index in range(9):
|
|
stuck.emit(ui_check._fake_packet(index))
|
|
stuck.emit(ui_check._fake_packet(99, valid=False, reason="read failed"))
|
|
|
|
slice_ = window.session_store.data()
|
|
assert slice_.valid.tolist()[-1] is False
|
|
assert math.isnan(slice_.raw[-1][0]), "a failed readout became a number"
|
|
assert sum(1 for _ in window.session_store.iter_valid("raw")) == 9
|
|
_finish(window, stuck)
|
|
|
|
|
|
def test_calibration_sampling_survives_a_hidden_panel(make_window):
|
|
window, stuck = _running_window(make_window)
|
|
window._show_cal = False
|
|
window._apply_panel_visibility()
|
|
assert not window.cal_panel.isVisible()
|
|
|
|
window.cal_recorder.start(channel="fn", reference_load_n=2.0, duration_s=1.0)
|
|
for index in range(40):
|
|
stuck.emit(ui_check._fake_packet(index))
|
|
point = window.cal_recorder.finish_now()
|
|
|
|
assert point is not None
|
|
assert point.n_valid == 40, "hiding the panel stopped the sampling"
|
|
assert point.status == "ok"
|
|
_finish(window, stuck)
|
|
|
|
|
|
def test_record_button_stays_a_cancel_affordance(make_window):
|
|
from orisys_app_shared.i18n import t
|
|
|
|
window, stuck = _running_window(make_window)
|
|
window.cal_panel.set_running(True)
|
|
window.cal_panel._on_record_clicked()
|
|
|
|
assert window.cal_panel._recording
|
|
assert window.cal_panel.btn_record.isEnabled(), "the recording cannot be stopped"
|
|
assert window.cal_panel.btn_record.text() == t("cal.recording_stop")
|
|
|
|
window.cal_panel._on_record_clicked()
|
|
assert not window.cal_panel._recording
|
|
_finish(window, stuck)
|
|
|
|
|
|
def test_unit_switch_rebuilds_from_the_session_store(make_window):
|
|
import math
|
|
|
|
window, stuck = _running_window(make_window)
|
|
window.toolbar.units_combo.setCurrentIndex(1) # raw
|
|
window.force_plots.reset()
|
|
for index in range(60):
|
|
stuck.emit(ui_check._fake_packet(index))
|
|
window._on_run_tick()
|
|
|
|
state = window.force_plots.history.state()
|
|
assert float(state.values[:, 0].max()) > 5.0, "raw values were not plotted"
|
|
|
|
window.toolbar.units_combo.setCurrentIndex(0) # newton
|
|
assert window.force_plots._unit == "N"
|
|
newtons = window.force_plots.history.state().values[:, 0]
|
|
finite = [v for v in newtons if math.isfinite(v)]
|
|
assert finite and max(finite) < 5.0, "the buffer still held raw values in N mode"
|
|
_finish(window, stuck)
|
|
|
|
|
|
def test_history_keeps_negative_fn(make_window):
|
|
window, stuck = _running_window(make_window)
|
|
window.force_plots.set_fixed_newton(True)
|
|
for index in range(20):
|
|
window.force_plots.push(index / 30.0, -2.5, 0.0, 0.0, unit="N")
|
|
window.force_plots.push(1.0, -3.0, 0.0, 0.0, unit="N")
|
|
|
|
state = window.force_plots.history.state()
|
|
assert float(state.values[:, 0].min()) < 0.0, (
|
|
"negative FN was clamped before entering the buffer"
|
|
)
|
|
_finish(window, stuck)
|
|
|
|
|
|
def test_the_live_axis_is_session_relative_and_survives_a_unit_switch(make_window):
|
|
"""The live buffer, the store and a rebuilt history share one time base.
|
|
|
|
Feeding the wall clock into the plot put the live axis ~1.7e9 s away from the
|
|
session store's, so a unit switch (which rebuilds from the store) jumped the
|
|
X axis and the chart looked like it had lost its history.
|
|
"""
|
|
wall = 1_700_000_000.0
|
|
window, stuck = _running_window(make_window)
|
|
window.toolbar.units_combo.setCurrentIndex(0) # newton
|
|
window.force_plots.reset()
|
|
for index in range(90):
|
|
stuck.emit(
|
|
ui_check._fake_packet(
|
|
index, timestamp=wall + index / 30.0, mono_t=index / 30.0
|
|
)
|
|
)
|
|
window._on_run_tick()
|
|
|
|
before = window.force_plots.history.state()
|
|
assert before is not None
|
|
end_before = float(before.times[-1])
|
|
assert end_before < 60.0, (
|
|
f"the live X axis is not session-relative: it ends at {end_before}"
|
|
)
|
|
assert abs(end_before - 89 / 30.0) < 0.5, end_before
|
|
|
|
for units_index in (1, 0, 1, 0): # raw -> N -> raw -> N
|
|
window.toolbar.units_combo.setCurrentIndex(units_index)
|
|
|
|
after = window.force_plots.history.state()
|
|
assert after is not None, "the unit switch emptied the rebuilt history"
|
|
assert abs(float(after.times[-1]) - end_before) < 1.0, (
|
|
f"the X axis jumped across a unit switch: {end_before} -> {after.times[-1]}"
|
|
)
|
|
_finish(window, stuck)
|
|
|
|
|
|
def test_a_single_calibrated_axis_is_shown_and_the_others_are_unavailable(make_window):
|
|
"""Only FN calibrated: FN in newtons, FX/FY explicitly unavailable."""
|
|
import math
|
|
|
|
from orisys_app_shared.i18n import t
|
|
|
|
window, stuck = _running_window(make_window)
|
|
window.toolbar.units_combo.setCurrentIndex(0) # newton
|
|
window.force_plots.reset()
|
|
for index in range(10):
|
|
stuck.emit(
|
|
ui_check._fake_packet(
|
|
index,
|
|
calibrated_axes=("fn",),
|
|
shear_n=float("nan"),
|
|
)
|
|
)
|
|
window._on_run_tick()
|
|
|
|
plots = window.force_plots
|
|
assert plots._available == (True, False, False), plots._available
|
|
assert plots.value_labels[0].text() != t("chart.value_unavailable")
|
|
for index in (1, 2):
|
|
assert plots.value_labels[index].text() == t("chart.value_unavailable"), (
|
|
"an axis with no curve must not show a number"
|
|
)
|
|
state = plots.history.state()
|
|
assert math.isfinite(float(state.values[:, 0].max()))
|
|
assert not any(math.isfinite(v) for v in state.values[:, 1]), (
|
|
"FX carried a value in newton mode without a curve"
|
|
)
|
|
_finish(window, stuck)
|
|
|
|
|
|
def test_a_slow_close_is_not_reported_as_a_finished_recording(make_window):
|
|
"""Stopping must not present an unclosed file as a finished experiment."""
|
|
import threading as _threading
|
|
from unittest.mock import patch as _patch
|
|
|
|
from orisys_app_shared.i18n import t
|
|
from orisys_app_shared.record.experiment import _SessionWriter
|
|
|
|
window, stuck = _running_window(make_window)
|
|
window.toolbar.btn_record.setChecked(True)
|
|
stuck.emit(ui_check._fake_packet(0))
|
|
window._on_run_tick()
|
|
|
|
started = _threading.Event()
|
|
release = _threading.Event()
|
|
original = _SessionWriter.write_sample
|
|
|
|
def slow(self, *args):
|
|
started.set()
|
|
release.wait(10.0)
|
|
return original(self, *args)
|
|
|
|
with _patch.object(_SessionWriter, "write_sample", slow):
|
|
stuck.emit(ui_check._fake_packet(1))
|
|
window._on_run_tick()
|
|
assert started.wait(5.0), "the writer never picked up the frame"
|
|
|
|
window._stop_recording(wait=False)
|
|
assert window.recorder.report is None, "the stop claimed a finished recording"
|
|
assert window._record_report_pending, "the pending report was not tracked"
|
|
assert window._record_last_report is None
|
|
assert not window.toolbar.btn_record.isChecked(), (
|
|
"the record button still looks active while the file closes"
|
|
)
|
|
assert window.statusBar().currentMessage() == t("record.closing")
|
|
|
|
release.set()
|
|
deadline = __import__("time").monotonic() + 10.0
|
|
while window.recorder.report is None and __import__("time").monotonic() < deadline:
|
|
__import__("time").sleep(0.01)
|
|
window._on_run_tick()
|
|
|
|
report = window._record_last_report
|
|
assert report is not None and report.completed, "the terminal report was never published"
|
|
assert not window._record_report_pending
|
|
_finish(window, stuck)
|
|
|
|
|
|
def test_a_failed_recording_is_consumed_even_when_it_is_not_running(make_window):
|
|
"""A recording that died on its own still has a report to publish."""
|
|
import time as _time
|
|
from unittest.mock import patch as _patch
|
|
|
|
from orisys_app_shared.record.experiment import _SessionWriter
|
|
|
|
window, stuck = _running_window(make_window)
|
|
window.toolbar.btn_record.setChecked(True)
|
|
assert window.recorder.is_recording
|
|
|
|
with _patch.object(_SessionWriter, "write_sample", side_effect=OSError("disk full")):
|
|
for index in range(8):
|
|
stuck.emit(ui_check._fake_packet(index))
|
|
window._on_run_tick()
|
|
deadline = _time.monotonic() + 3.0
|
|
while window.recorder.stats().write_errors < 5 and _time.monotonic() < deadline:
|
|
_time.sleep(0.01)
|
|
assert window.recorder.stats().failed
|
|
assert not window.recorder.is_recording
|
|
|
|
# ``is_recording`` is already False: the stop must still consume the report.
|
|
window._stop_recording()
|
|
report = window._record_last_report
|
|
assert report is not None and report.stats.failed
|
|
assert not window.toolbar.btn_record.isChecked()
|
|
assert window._session_phase == "error"
|
|
_finish(window, stuck)
|
|
|
|
|
|
def test_the_depth_request_belongs_to_the_recording_it_started_in(make_window, record_root):
|
|
"""Opening another recording must invalidate the request still running."""
|
|
import threading as _threading
|
|
import time as _time
|
|
|
|
from PyQt5.QtWidgets import QApplication
|
|
|
|
from orisys_app_shared.record.depth import DepthResult
|
|
from orisys_studio.ui.playback_dialog import PlaybackDialog
|
|
|
|
window = _show(make_window(), (1200, 700))
|
|
dialog = PlaybackDialog(theme="dark", directory=Path(record_root))
|
|
first_serial = dialog._record_serial
|
|
|
|
started = _threading.Event()
|
|
release = _threading.Event()
|
|
closed: list[str] = []
|
|
|
|
class _Blocking:
|
|
def preflight(self) -> str:
|
|
return ""
|
|
|
|
def depth_at(self, _position, **_kwargs):
|
|
started.set()
|
|
release.wait(10.0)
|
|
return DepthResult(
|
|
depth=np.ones((2, 2)), frame_id=0, t_rel_s=0.0, record_id="old"
|
|
)
|
|
|
|
def close(self) -> None:
|
|
closed.append("closed")
|
|
|
|
blocking = _Blocking()
|
|
dialog._recomputer = blocking
|
|
dialog._on_recompute_depth()
|
|
assert started.wait(5.0), "the depth request never started"
|
|
|
|
# Another recording is opened while the request is in flight.
|
|
dialog.open_recording(Path(record_root))
|
|
assert dialog._record_serial == first_serial + 1
|
|
assert dialog._recomputer is not blocking, "the processor was not replaced"
|
|
assert closed == [], "the UI thread closed a processor a worker was reading"
|
|
|
|
release.set()
|
|
deadline = _time.monotonic() + 10.0
|
|
while not closed and _time.monotonic() < deadline:
|
|
QApplication.processEvents()
|
|
_time.sleep(0.01)
|
|
assert closed == ["closed"], "the retired processor was never closed"
|
|
assert dialog._depth_thread is not None and not dialog._depth_thread.is_alive()
|
|
|
|
# The answer belongs to the previous recording, so it is dropped, not drawn.
|
|
dialog._poll_depth()
|
|
QApplication.processEvents()
|
|
assert dialog._depth_ok is False
|
|
pixmap = dialog.depth_label.pixmap()
|
|
assert pixmap is None or pixmap.isNull(), "an answer for another recording was drawn"
|
|
dialog.close()
|
|
window.close()
|
|
|
|
|
|
def test_a_busy_depth_processor_is_not_closed_by_the_ui_thread(make_window, record_root):
|
|
import threading as _threading
|
|
import time as _time
|
|
|
|
from PyQt5.QtWidgets import QApplication
|
|
|
|
from orisys_app_shared.record.depth import DepthResult
|
|
from orisys_studio.ui.playback_dialog import PlaybackDialog
|
|
|
|
window = _show(make_window(), (1200, 700))
|
|
dialog = PlaybackDialog(theme="dark", directory=Path(record_root))
|
|
started = _threading.Event()
|
|
release = _threading.Event()
|
|
closed: list[str] = []
|
|
|
|
class _Blocking:
|
|
def preflight(self) -> str:
|
|
return ""
|
|
|
|
def depth_at(self, _position, **_kwargs):
|
|
started.set()
|
|
release.wait(10.0)
|
|
return DepthResult(depth=np.ones((2, 2)), frame_id=0, t_rel_s=0.0)
|
|
|
|
def close(self) -> None:
|
|
closed.append("closed")
|
|
|
|
dialog._recomputer = _Blocking()
|
|
dialog._on_recompute_depth()
|
|
assert started.wait(5.0)
|
|
dialog.close() # closing the window must not free a processor in use
|
|
QApplication.processEvents()
|
|
assert closed == [], "the window closed a processor that was still reading"
|
|
release.set()
|
|
deadline = _time.monotonic() + 10.0
|
|
while not closed and _time.monotonic() < deadline:
|
|
QApplication.processEvents()
|
|
_time.sleep(0.01)
|
|
assert closed == ["closed"], "the owning thread never closed the processor"
|
|
window.close()
|
|
|
|
|
|
def test_a_confirmed_calibration_is_snapshotted_into_the_recording(make_window, record_root):
|
|
"""The recording keeps the coefficients themselves, not only their hash."""
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import orisys_studio.app_window as app_module
|
|
from orisys_app_shared.i18n import t
|
|
|
|
window, stuck = _running_window(make_window)
|
|
app_module.recordings_dir = lambda: record_root
|
|
window.toolbar.btn_record.setChecked(True)
|
|
assert window.recorder.is_recording
|
|
|
|
coefficients = Path(record_root) / "force_calibration_box.json"
|
|
coefficients.write_text('{"unit": "N", "channels": {}}', encoding="utf-8")
|
|
version = "deadbeefcafe1234"
|
|
window._on_calibration_applied(
|
|
str(coefficients), version, 21, 0.7, int(window.session_store.session_id)
|
|
)
|
|
window._drain_calibration_events()
|
|
|
|
directory = window.recorder.directory
|
|
assert directory is not None
|
|
snapshots = sorted(path.name for path in (directory / "calibration").iterdir())
|
|
assert snapshots == [f"force_calibration_{version[:12]}.json"], snapshots
|
|
assert (directory / "calibration" / snapshots[0]).read_text(
|
|
encoding="utf-8"
|
|
) == '{"unit": "N", "channels": {}}'
|
|
assert window._active_calibration_id == version
|
|
assert window.session_store.calibration_id == version
|
|
|
|
window._stop_recording()
|
|
meta = json.loads((directory / "session.json").read_text(encoding="utf-8"))
|
|
entries = meta["calibration_snapshots"]
|
|
assert entries and entries[0]["version"] == version
|
|
assert entries[0]["file"] == f"calibration/{snapshots[0]}"
|
|
assert entries[0]["frame_id"] == 21
|
|
event = next(e for e in meta["events"] if e["kind"] == "calibration_applied")
|
|
assert event["frame_id"] == 21 and event["mono_t"] == pytest.approx(0.7)
|
|
assert snapshots[0] in event["detail"]
|
|
_finish(window, stuck)
|
|
|
|
|
|
def test_a_confirmed_absence_of_calibration_does_not_keep_the_old_version(make_window):
|
|
"""An empty version is the honest answer; it must not fall back to the path."""
|
|
from orisys_app_shared.i18n import t
|
|
|
|
window = _show(make_window(), (1400, 880))
|
|
window._active_calibration_id = "previous-hash"
|
|
window._on_calibration_applied("/tmp/whatever.json", "", 5, 0.2, 1)
|
|
window._drain_calibration_events()
|
|
assert window._active_calibration_id == "", (
|
|
"an uncalibrated device kept claiming the previous coefficients"
|
|
)
|
|
assert window.statusBar().currentMessage() == t("cal.applied_none")
|
|
|
|
|
|
def test_a_zeroing_is_recorded_only_after_the_worker_confirms_it(make_window, record_root):
|
|
import json
|
|
|
|
import orisys_studio.app_window as app_module
|
|
|
|
window, stuck = _running_window(make_window)
|
|
app_module.recordings_dir = lambda: record_root
|
|
window.toolbar.btn_record.setChecked(True)
|
|
stuck.emit(ui_check._fake_packet(3))
|
|
window._on_run_tick()
|
|
|
|
window._on_reset()
|
|
assert stuck.reset_requests == 1, "the zeroing was not requested"
|
|
# The stub confirms from the acquisition side, as a real device does.
|
|
stuck.on_reset_applied = window._on_reset_applied
|
|
window._on_reset()
|
|
window._on_run_tick()
|
|
window._stop_recording()
|
|
|
|
directory = window.recorder.directory
|
|
meta = json.loads((directory / "session.json").read_text(encoding="utf-8"))
|
|
zeroing = [e for e in meta["events"] if e["kind"] == "zeroing"]
|
|
assert len(zeroing) == 1, f"unexpected zeroing events: {zeroing}"
|
|
assert zeroing[0]["frame_id"] == 3, zeroing[0]
|
|
assert zeroing[0]["mono_t"] is not None
|
|
_finish(window, stuck)
|
|
|
|
|
|
def test_depth_view_reports_missing_data_and_fixes_its_scale(make_window):
|
|
import numpy as np
|
|
|
|
window, stuck = _running_window(make_window)
|
|
window.arrow_panel.set_view_mode("depth")
|
|
|
|
window._last_rendered_view_mode = None
|
|
stuck.emit(ui_check._fake_packet(0, depth=None))
|
|
window._on_run_tick()
|
|
from orisys_app_shared.i18n import t
|
|
|
|
assert window.arrow_panel.image_label.text() == t("panels.depth_unavailable"), (
|
|
"a missing depth map silently fell back to the processed frame"
|
|
)
|
|
|
|
window._last_rendered_view_mode = None
|
|
stuck.emit(ui_check._fake_packet(1, depth=np.linspace(0.0, 1.0, 64).reshape(8, 8)))
|
|
window._on_run_tick()
|
|
latched = window.arrow_panel.depth_scale_label.text()
|
|
assert latched
|
|
|
|
window._last_rendered_view_mode = None
|
|
stuck.emit(ui_check._fake_packet(2, depth=np.linspace(0.0, 9.0, 64).reshape(8, 8)))
|
|
window._on_run_tick()
|
|
assert window.arrow_panel.depth_scale_label.text() == latched, (
|
|
"the depth colour scale re-fitted on every frame"
|
|
)
|
|
_finish(window, stuck)
|
|
|
|
|
|
# ------------------------------------------------------------- experiment record
|
|
|
|
#: Recording scratch goes under the app's gitignored ``outputs`` tree rather than
|
|
#: the system temp directory, so the tests also run under a file sandbox that
|
|
#: only permits writes inside the workspace.
|
|
RECORD_SCRATCH = Path(__file__).resolve().parents[1] / "outputs" / "_test_record"
|
|
|
|
|
|
@pytest.fixture
|
|
def record_root():
|
|
import shutil
|
|
|
|
shutil.rmtree(RECORD_SCRATCH, ignore_errors=True)
|
|
RECORD_SCRATCH.mkdir(parents=True, exist_ok=True)
|
|
yield RECORD_SCRATCH
|
|
shutil.rmtree(RECORD_SCRATCH, ignore_errors=True)
|
|
|
|
|
|
def test_recording_writes_the_documented_experiment_layout(make_window, record_root):
|
|
import csv
|
|
import json
|
|
|
|
import orisys_studio.app_window as app_module
|
|
from orisys_app_shared.i18n import t
|
|
|
|
window, stuck = _running_window(make_window)
|
|
app_module.recordings_dir = lambda: record_root
|
|
assert window.toolbar.btn_record.isEnabled()
|
|
assert window.toolbar.btn_record.text() == t("record.start_label")
|
|
|
|
window.toolbar.btn_record.setChecked(True)
|
|
assert window.recorder.is_recording, "the recording did not start"
|
|
assert window.toolbar.btn_record.text() == t("record.stop_label")
|
|
assert not window.toolbar.record_mode_combo.isEnabled()
|
|
|
|
for index in range(20):
|
|
stuck.emit(ui_check._fake_packet(index))
|
|
window._on_run_tick()
|
|
window.toolbar.btn_record.setChecked(False)
|
|
|
|
report = window._record_last_report
|
|
assert report is not None and report.ok, getattr(report, "problems", None)
|
|
names = sorted(path.name for path in report.directory.iterdir())
|
|
assert names == ["events.csv", "forces.csv", "frames.csv", "raw_0001.mp4", "session.json"]
|
|
|
|
with (report.directory / "forces.csv").open(encoding="utf-8") as handle:
|
|
forces = list(csv.DictReader(handle))
|
|
with (report.directory / "frames.csv").open(encoding="utf-8") as handle:
|
|
frames = list(csv.DictReader(handle))
|
|
assert len(forces) == len(frames) == 20
|
|
assert [f["frame_id"] for f in frames] == [r["frame_id"] for r in forces]
|
|
assert [f["t_rel_s"] for f in frames] == [r["t_rel_s"] for r in forces]
|
|
|
|
meta = json.loads((report.directory / "session.json").read_text(encoding="utf-8"))
|
|
assert meta["ended_at"]
|
|
assert meta["format_version"]
|
|
assert meta["stats"]["frames"] == 20
|
|
_finish(window, stuck)
|
|
|
|
|
|
def test_recording_leaves_unavailable_values_empty(make_window, record_root):
|
|
import csv
|
|
|
|
import orisys_studio.app_window as app_module
|
|
|
|
window, stuck = _running_window(make_window)
|
|
app_module.recordings_dir = lambda: record_root
|
|
window.toolbar.btn_record.setChecked(True)
|
|
stuck.emit(ui_check._fake_packet(0))
|
|
window._on_run_tick()
|
|
stuck.emit(ui_check._fake_packet(1, valid=False, reason="read failed"))
|
|
window._on_run_tick()
|
|
window.toolbar.btn_record.setChecked(False)
|
|
|
|
with (window._record_last_report.directory / "forces.csv").open(encoding="utf-8") as handle:
|
|
rows = list(csv.DictReader(handle))
|
|
failed = rows[-1]
|
|
assert failed["valid"] == "0"
|
|
assert failed["FN_raw"] == "" and failed["FN_N"] == "", (
|
|
"a failed readout was persisted as a number"
|
|
)
|
|
_finish(window, stuck)
|
|
|
|
|
|
|