Files
orisys-app/apps/studio/tools/ui_check.py
T
ykcheungab 148573934f 采集端统一由 SDK 滤波、显示端不再二次滤波,session.json 记录生效值 - What: sdk_worker/finger_worker 删除硬编码的 sensor.configure(lowpass_cutoff_hz=LOWPASS_CUTOFF_HZ, lowpass_sample_rate_hz=...) —— 它会让界面上的曲线与 CSV 走两条不同的滤波路径;改为在会话打开设备后调用 FilterSettings.apply_to(sensor),并把两件事分别记下来:self._filter_state = effective_filters(sensor)(设备报的**实际**生效值,供界面与记录使用)与 self._filters_unsupported(该 SDK 应用不了的键,老 SDK 必须被看见而不是被吞掉);新增 worker 的 set_filter_settings()/filter_settings()/filter_state()/filters_unsupported() 访问器,start() 时清空上一会话的 filter_state(与 _calibration_version 同样的理由:新会话不能继承旧会话的说法)。finger_worker 保留原有的"runtime 可能拒绝 configure"的容错,但改成记日志并把全部键标为未应用。app_window 新增 _filter_settings()(从 preferences 的 plot_* 键读,空/坏值即"什么都不滤")与 _filter_metadata()(向**正在跑的** worker 要设备实际状态,写成 session.json 的 filters 块:measurement/readout/describe/source,SDK 缺键时附 unsupported_by_sdk);_start_session 在 start() 之前把设置交给 worker。ui/force_plots.py 把 MetricHistory 的 lowpass_cutoff_hz 从 LOWPASS_CUTOFF_HZ 改成 0.0:SDK 已经滤过一遍,且这个 buffer 就是进记录的那个,显示端再滤会让曲线画出 CSV 里根本没有的数据(原来用的还是零相位 filtfilt,非因果)。acquisition/constants.py 删掉 LOWPASS_CUTOFF_HZ("app 在 10 Hz 滤波"这件事不存在了),FORCE_SAMPLE_RATE_HZ 保留。tests/test_force_filters.py 增 3 项:worker 未开设备时 filter_state() 为 {}("没问过"不等于"没在滤")、effective_filters() 对没有 filters_state() 的 SDK 返回 {}、以及对着真实 SDK 的端到端断言(external-frame 构造 box Sensor → apply_to → filters_state 回报 flowEMA(0.5)+median(3)+lowpass(10Hz,order2) → force_triplets 两个 buffer 都在 → 全关也可表达)。apps/studio/tools/ui_check.py 的 _StuckWorker 补上新合约方法(set_filter_settings/filter_settings/filter_state/filters_unsupported),否则窗口一启动就 AttributeError。 - Why: 用户的核心要求是"显示的和记录的必须一致,包括所有滤波",并决定整条链留在 SDK、app 只通过 Sensor.configure 驱动。此前 app 硬编码 10 Hz 低通(SDK 侧)**加上**显示端 filtfilt(app 侧),同一次采集的两份数据经过两条不同的滤波路径;session.json 里的 filter 块也只是照抄常量,说的不是设备在做的事。 - Verification: shared 全量 156 passed(含新增的真实 SDK 端到端断言);studio 85 项测试通过(修好 _StuckWorker 合约后);ui_check 554 项检查、4 个场景 0 失败;编译通过。 - Not in this change: 曲线表头的齿轮按钮与滤波设置对话框(因此现在还只能改 preferences.json 的 plot_* 键来换滤波)、CSV 的 17 列与未滤波第二 buffer、回放界面显示滤波前曲线、发布说明里"默认不再滤波"的措辞。
2026-09-14 11:01:55 +08:00

3134 lines
117 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/usr/bin/env python3
"""Dependency-free smoke run for the studio UI.
``pytest`` is the canonical runner (``apps/studio/tests``), but this script
needs nothing beyond the app's own runtime dependencies, so it can be used in
minimal environments:
python tools/ui_check.py
It builds the real main window offscreen and asserts the layout, theming and
status-feedback invariants that the UI work depends on.
"""
from __future__ import annotations
import os
import sys
import threading
import time
from pathlib import Path
APP_ROOT = Path(__file__).resolve().parents[1]
REPO_ROOT = APP_ROOT.parents[1]
for _path in (APP_ROOT / "src", REPO_ROOT / "shared" / "src"):
if _path.is_dir() and str(_path) not in sys.path:
sys.path.insert(0, str(_path))
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
if sys.platform == "win32" and "QT_QPA_FONTDIR" not in os.environ:
_fonts = Path(os.environ.get("WINDIR", r"C:\Windows")) / "Fonts"
if _fonts.is_dir():
os.environ["QT_QPA_FONTDIR"] = str(_fonts)
os.chdir(APP_ROOT)
from orisys_app_shared.acquisition.sinks import RecordSinkMixin # noqa: E402
from orisys_app_shared.force.filters import FilterSettings # noqa: E402
from orisys_app_shared.record import FORCES_HEADER # noqa: E402
FAILURES: list[str] = []
CHECKS = 0
def check(condition: bool, message: str) -> None:
global CHECKS
CHECKS += 1
if not condition:
FAILURES.append(message)
def _inside_scroll_area(widget) -> bool:
"""True when ``widget`` lives inside a QScrollArea's content."""
from PyQt5.QtWidgets import QScrollArea
node = widget
while node is not None:
if isinstance(node, QScrollArea):
return True
node = node.parentWidget() if hasattr(node, "parentWidget") else None
return False
def rail_squeezed(window) -> list[str]:
"""Status-rail widgets showing less than their own text.
The rail shares one line with the session state, so every item that states a
fact (``摄像机 0``, ``开始记录``) must keep its full text; only the labels
that carry a tooltip are allowed to shrink. An elided button is how a
readable instrument turns back into a jumble.
"""
layout = window.toolbar.layout().itemAt(0).layout()
squeezed: list[str] = []
for index in range(layout.count()):
widget = layout.itemAt(index).widget()
if widget is None or not widget.isVisible():
continue
if getattr(widget, "minimumSizeHint", None) is None:
continue
if widget.minimumSizeHint().width() <= 0:
continue
if widget.width() < widget.sizeHint().width() - 1:
squeezed.append(
f"{type(widget).__name__}"
f"({getattr(widget, 'objectName', lambda: '')()}"
f" {widget.width()}<{widget.sizeHint().width()})"
)
return squeezed
def section(name: str) -> None:
print(f"\n== {name} ==")
#: Sequences that only appear when UTF-8 text was decoded with a single-byte
#: codepage and re-encoded. A PowerShell ``Get-Content -Raw | Set-Content``
#: round trip does that and silently mangles every non-ASCII string, which is
#: how the Chinese UI once ended up unreadable.
#:
#: Written as escapes, with no literal examples in the comments, so that this
#: guard cannot itself be corrupted by the same mistake and so that a scan of
#: the tree for these sequences stays clean.
MOJIBAKE_MARKERS = (
"\u00c3",
"\u00c2",
"\u00e2\u20ac",
"\u00e3\u20ac",
"\u00ef\u00bf\u00bd",
)
def audit_locale_encoding() -> list[str]:
"""Locale files must be BOM-free UTF-8 and not mojibake."""
import json
problems: list[str] = []
locales = APP_ROOT / "config" / "locales"
for name in ("zh_CN", "en_US"):
path = locales / f"{name}.json"
if not path.is_file():
problems.append(f"{name}.json is missing")
continue
raw = path.read_bytes()
if raw.startswith(b"\xef\xbb\xbf"):
problems.append(f"{name}.json starts with a UTF-8 BOM")
try:
text = raw.decode("utf-8")
except UnicodeDecodeError as exc:
problems.append(f"{name}.json is not valid UTF-8: {exc}")
continue
for marker in MOJIBAKE_MARKERS:
if marker in text:
problems.append(
f"{name}.json contains {marker!r} — written by a tool that "
"decoded UTF-8 with the wrong codepage"
)
break
try:
json.loads(text)
except json.JSONDecodeError as exc:
problems.append(f"{name}.json is not valid JSON: {exc}")
# A corrupted zh_CN.json still parses, so check that it holds Chinese text.
try:
import json as _json
zh = _json.loads((locales / "zh_CN.json").read_text(encoding="utf-8"))
sample = zh["menu"]["theme_dark"]
if not any("\u4e00" <= char <= "\u9fff" for char in sample):
problems.append(f"zh_CN.menu.theme_dark has no Chinese text: {sample!r}")
except Exception as exc: # noqa: BLE001
problems.append(f"could not verify the Chinese locale: {exc}")
return problems
def _calibration_sample(index: int, *, fn_raw: float, fx_raw: float = 0.4, fy_raw: float = -0.2):
"""A sample shaped for the calibration recorder."""
from orisys_app_shared.force.session import SessionSample
return SessionSample(
frame_id=index,
t_rel_s=index / 100.0,
fn_raw=fn_raw,
fx_raw=fx_raw,
fy_raw=fy_raw,
fn_n=fn_raw / 100.0,
fx_n=fx_raw / 100.0,
fy_n=fy_raw / 100.0,
valid=True,
)
def _fake_packet(
frame_id: int,
*,
valid: bool = True,
depth=None,
reason: str = "",
calibrated_axes=None,
timestamp: float | None = None,
mono_t: float | None = None,
shear_n: float | None = None,
frame=None,
):
"""A packet shaped like the real worker's, for the record-channel checks.
``frame`` replaces the blank 8x8 image, so a synthetic recording can carry
something visible (see ``ui_snapshot --scene replay``).
"""
import numpy as np
from orisys_app_shared.acquisition.sdk_worker import SdkPacket
nan = float("nan")
blank = np.zeros((8, 8, 3), np.uint8) if frame is None else frame
# A single tall sample makes it easy to prove decimation keeps peaks.
fn_raw = 99.0 if frame_id == 700 else (9.0 + frame_id * 0.01)
return SdkPacket(
raw_frame=blank,
processed_frame=blank,
timestamp=1000.0 + frame_id / 30.0 if timestamp is None else timestamp,
mono_t=mono_t,
fps=30.0,
process_ms=1.0,
normal_force=(2.0 + frame_id * 0.01) if valid else nan,
shear_x=(0.5 if shear_n is None else shear_n) if valid else nan,
shear_y=-0.5 if valid else nan,
normal_force_raw=(9.0 + frame_id * 0.1) if valid else nan,
shear_x_raw=0.4 if valid else nan,
shear_y_raw=-0.4 if valid else nan,
force_unit="N",
force_calibrated=True,
calibrated_axes=tuple(calibrated_axes or ()),
session_id=1,
frame_id=frame_id,
valid=valid,
invalid_reason=reason,
raw_available=True,
depth=depth,
)
class _StuckWorker(RecordSinkMixin):
"""A worker whose device open never returns.
Mirrors the real worker contract — including the record-channel sink API —
so the window can be exercised against the failure mode the Stop button
exists to survive.
"""
def __init__(self) -> None:
self._init_sinks()
self._thread: threading.Thread | None = None
self._end_reason: str | None = None
self._latest = None
self.thread_still_owned = False
self.stop_blocked_for = 0.0
#: Released by ``release()`` so daemon threads do not outlive the run.
self._release = threading.Event()
#: Calibration requests this worker was asked to apply.
self.calibration_requests: list[str] = []
#: Zeroing requests, and the confirmation callback the window installs.
self.reset_requests = 0
self.on_reset_applied = None
#: Filters handed over before a session, and what the fake device reports
#: as in effect (this stub has no ``configure``, so it reports nothing).
self.filter_settings_value = FilterSettings()
self.filter_state_value: dict = {}
def set_filter_settings(self, settings) -> None:
self.filter_settings_value = settings
def filter_settings(self):
return self.filter_settings_value
def filter_state(self) -> dict:
return dict(self.filter_state_value)
def filters_unsupported(self) -> tuple[str, ...]:
return ()
def set_pending_calibration(self, path) -> None:
self.calibration_requests.append(str(path))
def reset(self) -> None:
"""Request a zeroing; a real device confirms it from its own thread.
This stub confirms immediately (a fast device), so the window's event path
— request, confirmation, recorded event — is exercised end to end.
"""
self.reset_requests += 1
if callable(self.on_reset_applied):
packet = self._latest
frame_id = int(getattr(packet, "frame_id", -1))
mono_t = float(getattr(packet, "mono_t", 0.0) or 0.0)
self.on_reset_applied(frame_id, mono_t, 1)
def start(self, _source, *, config_ref=None) -> None:
self._release.clear()
self._thread = threading.Thread(target=self._hang, daemon=True)
self._thread.start()
def _hang(self) -> None:
self._release.wait(timeout=600.0)
def release(self) -> None:
"""Let the stuck thread finish so the process can exit cleanly."""
self._release.set()
if self._thread is not None:
self._thread.join(timeout=5.0)
def emit(self, packet) -> None:
"""Publish a packet exactly like the acquisition loop would."""
self._latest = packet
self._emit(packet)
@property
def is_running(self) -> bool:
return self._thread is not None and self._thread.is_alive()
@property
def phase(self) -> str:
return "connecting"
def latest(self):
return self._latest
def consume_end_reason(self):
reason, self._end_reason = self._end_reason, None
return reason
def stop(self, *, wait: bool = False) -> None:
self._end_reason = self._end_reason or "stopped"
started = time.perf_counter()
if wait and self._thread is not None and self._thread.is_alive():
self._thread.join(timeout=5.0)
if self._thread is None or not self._thread.is_alive():
self._thread = None
self.thread_still_owned = self._thread is not None
self.stop_blocked_for = time.perf_counter() - started
# --------------------------------------------------------------------------
# Palette rules
#
# These are the WCAG / colour-separation invariants the theme tokens must
# satisfy. ``tests/test_ui_smoke.py`` imports :func:`audit_palette` from here so
# both runners share one source of truth.
# --------------------------------------------------------------------------
TEXT_MIN_RATIO = 4.5
GRAPHIC_MIN_RATIO = 3.0
VIZ_MIN_HUE_GAP = 90.0
VIZ_MIN_LUMINANCE_RATIO = 1.2
RAMP_BLEND_STOPS = 2 # the low stops may merge into the canvas
TEXT_PAIRS: tuple[tuple[str, tuple[str, ...]], ...] = (
("text_primary", (
"window_bg", "panel_bg", "card_bg", "toolbar_bg", "input_bg",
"table_header_bg", "table_row_alt", "menu_bg", "button_bg",
"table_selected", "scrollbar_bg",
)),
("text_secondary", (
"window_bg", "panel_bg", "card_bg", "toolbar_bg",
"table_header_bg", "button_bg", "input_bg", "menu_bg",
)),
("text_muted", ("window_bg", "panel_bg", "card_bg", "toolbar_bg", "input_bg")),
("hint_text", ("card_bg", "panel_bg", "window_bg", "input_bg")),
("accent_text", ("accent", "accent_hover", "danger_fill", "danger_fill_hover")),
("danger_text", ("danger_fill", "danger_fill_hover")),
# The status rail lives on the toolbar background, so every semantic colour
# it can show is audited there too — not only on the panel/card backgrounds.
("danger", ("card_bg", "panel_bg", "toolbar_bg")),
("danger_hover", ("card_bg", "panel_bg", "toolbar_bg")),
("warning", ("card_bg", "panel_bg", "toolbar_bg")),
("info", ("card_bg", "panel_bg", "toolbar_bg")),
("success", ("card_bg", "panel_bg", "toolbar_bg")),
("status_idle", ("panel_bg", "toolbar_bg")),
("status_connecting", ("panel_bg", "toolbar_bg")),
("status_running", ("panel_bg", "toolbar_bg")),
("status_error", ("panel_bg", "toolbar_bg")),
("brand_bar_fg", ("brand_bar_bg",)),
("brand_bar_fg_muted", ("brand_bar_bg",)),
)
GRAPHIC_PAIRS: tuple[tuple[str, tuple[str, ...]], ...] = (
("plot_fn", ("plot_bg",)),
("plot_fx", ("plot_bg",)),
("plot_fy", ("plot_bg",)),
("scrollbar_handle", ("scrollbar_bg",)),
("scrollbar_handle_hover", ("scrollbar_bg",)),
("accent", ("panel_bg", "window_bg", "toolbar_bg")),
("accent_hover", ("panel_bg", "toolbar_bg")),
("focus_ring", ("window_bg", "input_bg", "panel_bg", "card_bg")),
("input_border", ("input_bg", "panel_bg", "card_bg", "toolbar_bg")),
)
VIZ_KEYS = ("plot_fn", "plot_fx", "plot_fy")
# Deliberate, documented exemptions (see apps/studio/docs/ui-palette.md):
# decorative 1px container outlines, and disabled states which WCAG 1.4.3
# explicitly exempts.
DECORATIVE_PAIRS: tuple[tuple[str, tuple[str, ...]], ...] = (
("card_border", ("card_bg",)),
("panel_border", ("panel_bg", "window_bg")),
("plot_grid", ("plot_bg",)),
("separator", ("toolbar_bg", "panel_bg")),
)
def _srgb_channel(value: int) -> float:
channel = value / 255.0
if channel <= 0.03928:
return channel / 12.92
return ((channel + 0.055) / 1.055) ** 2.4
def _rgb(hex_color: str) -> tuple[int, int, int]:
text = str(hex_color).strip().lstrip("#")
return tuple(int(text[i : i + 2], 16) for i in (0, 2, 4)) # type: ignore[return-value]
def relative_luminance(hex_color: str) -> float:
red, green, blue = _rgb(hex_color)
return (
0.2126 * _srgb_channel(red)
+ 0.7152 * _srgb_channel(green)
+ 0.0722 * _srgb_channel(blue)
)
def contrast_ratio(first: str, second: str) -> float:
high = relative_luminance(first)
low = relative_luminance(second)
if high < low:
high, low = low, high
return (high + 0.05) / (low + 0.05)
def hue_degrees(hex_color: str) -> float:
red, green, blue = (value / 255.0 for value in _rgb(hex_color))
high, low = max(red, green, blue), min(red, green, blue)
delta = high - low
if delta == 0:
return 0.0
if high == red:
return (60 * ((green - blue) / delta) + 360) % 360
if high == green:
return 60 * ((blue - red) / delta) + 120
return 60 * ((red - green) / delta) + 240
def _hue_gap(first: str, second: str) -> float:
delta = abs(hue_degrees(first) - hue_degrees(second))
return min(delta, 360 - delta)
def audit_palette(verbose: bool = False) -> list[str]:
"""Return the list of palette-rule violations (empty when compliant)."""
from orisys_app_shared.theme import load_colors
failures: list[str] = []
themes = ("dark", "light")
palettes = {theme: load_colors(theme) for theme in themes}
dark_keys, light_keys = set(palettes["dark"]), set(palettes["light"])
if dark_keys != light_keys:
failures.append(f"token keys differ between themes: {sorted(dark_keys ^ light_keys)}")
for theme, palette in palettes.items():
if verbose:
print(f" -- {theme} --")
for key, value in sorted(palette.items()):
if key == "flow_ramp":
stops = [stop.strip() for stop in value.split(",") if stop.strip()]
if not stops:
failures.append(f"{theme}.flow_ramp is empty")
for stop in stops:
if not (stop.startswith("#") and len(stop) == 7):
failures.append(f"{theme}.flow_ramp stop {stop!r} is not #RRGGBB")
elif not (str(value).startswith("#") and len(str(value)) == 7):
failures.append(f"{theme}.{key} = {value!r} is not #RRGGBB")
for key, backgrounds in TEXT_PAIRS:
for background in backgrounds:
ratio = contrast_ratio(palette[key], palette[background])
if verbose:
print(f" text {key:19s} on {background:17s} {ratio:5.2f}")
if ratio < TEXT_MIN_RATIO:
failures.append(
f"{theme}: text {key} on {background} = {ratio:.2f} "
f"(need {TEXT_MIN_RATIO})"
)
for key, backgrounds in GRAPHIC_PAIRS:
for background in backgrounds:
ratio = contrast_ratio(palette[key], palette[background])
if verbose:
print(f" graph {key:19s} on {background:17s} {ratio:5.2f}")
if ratio < GRAPHIC_MIN_RATIO:
failures.append(
f"{theme}: graphic {key} on {background} = {ratio:.2f} "
f"(need {GRAPHIC_MIN_RATIO})"
)
for index, first in enumerate(VIZ_KEYS):
for second in VIZ_KEYS[index + 1 :]:
gap = _hue_gap(palette[first], palette[second])
high = max(relative_luminance(palette[first]), relative_luminance(palette[second]))
low = min(relative_luminance(palette[first]), relative_luminance(palette[second]))
ratio = (high + 0.05) / (low + 0.05)
if verbose:
print(
f" viz {first:19s} vs {second:17s} "
f"dHue={gap:5.1f} LrelRatio={ratio:4.2f}"
)
if gap < VIZ_MIN_HUE_GAP and ratio < VIZ_MIN_LUMINANCE_RATIO:
failures.append(
f"{theme}: {first}/{second} are too similar "
f"(hue gap {gap:.0f} deg, luminance ratio {ratio:.2f})"
)
stops = [stop.strip() for stop in palette["flow_ramp"].split(",") if stop.strip()]
if len(stops) >= 2:
luminances = [relative_luminance(stop) for stop in stops]
if verbose:
print(f" ramp luminances {[round(value, 4) for value in luminances]}")
if not all(
luminances[i] < luminances[i + 1] for i in range(len(luminances) - 1)
):
failures.append(f"{theme}: flow_ramp is not monotone in luminance")
for stop in stops[RAMP_BLEND_STOPS:]:
ratio = contrast_ratio(stop, palette["arrow_canvas"])
if ratio < GRAPHIC_MIN_RATIO:
failures.append(
f"{theme}: flow_ramp stop {stop} on arrow_canvas = "
f"{ratio:.2f} (need {GRAPHIC_MIN_RATIO})"
)
for theme, palette in palettes.items():
if palette["brand_bar_bg"] == palette["window_bg"]:
failures.append(f"{theme}: brand_bar_bg collides with window_bg")
if verbose:
print(f" -- {theme} documented exemptions --")
for key, backgrounds in DECORATIVE_PAIRS:
for background in backgrounds:
print(
f" exempt {key:19s} on {background:17s} "
f"{contrast_ratio(palette[key], palette[background]):5.2f}"
)
return failures
def main() -> int:
from PyQt5.QtWidgets import QApplication
from orisys_app_shared.force.calibration_io import CalSample
from orisys_app_shared.i18n import set_locale, t
from orisys_app_shared.theme import build_qss, load_colors
from orisys_studio.app_window import VIEW_SPLITTER_SIZES, StudioMainWindow
app = QApplication.instance() or QApplication(sys.argv)
set_locale("zh_CN")
section("theme tokens")
dark = set(load_colors("dark"))
light = set(load_colors("light"))
check(dark == light, f"token keys differ: {sorted(dark ^ light)}")
for theme in ("dark", "light"):
qss = build_qss(theme)
check(len(qss) > 1000, f"{theme} qss suspiciously short")
check(
load_colors(theme)["arrow_canvas"] != load_colors(theme)["accent"],
f"{theme}: arrow canvas collides with accent",
)
check("QProgressBar::chunk" in qss, f"{theme}: progress bar not styled")
check("QHeaderView::section" in qss, f"{theme}: table header not styled")
check("QRadioButton::indicator" in qss, f"{theme}: radio not styled")
check("QScrollBar::handle" in qss, f"{theme}: scrollbar not styled")
check("QFrame#brandBar" in qss, f"{theme}: brand bar not styled")
check("brand_bar_bg" not in qss, f"{theme}: qss leaked a token name")
section("palette rules")
palette_failures = audit_palette(verbose=True)
for failure in palette_failures:
check(False, failure)
if not palette_failures:
check(True, "palette rules hold for both themes")
section("locale encoding")
encoding_failures = audit_locale_encoding()
for failure in encoding_failures:
check(False, failure)
if not encoding_failures:
check(True, "locale files are BOM-free UTF-8 and not mojibake")
scene_count = 0
for theme in ("dark", "light"):
for sensor in ("box", "finger"):
section(f"window {theme}/{sensor}")
scene_count += 1
app.setStyleSheet(build_qss(theme))
window = StudioMainWindow(
prefs={
"language": "zh_CN",
"theme": theme,
"sensor_type": sensor,
"show_cal_panel": True,
}
)
window.resize(1440, 880)
window.show()
app.processEvents()
check(window.windowTitle() != "", "window title empty")
check(
window.status_chip.text() != "",
"status chip has no phase text",
)
check(
window.toolbar.btn_run.text().strip() not in ("", "▶"),
f"run button is unlabelled: {window.toolbar.btn_run.text()!r}",
)
brand_bar = window.brand_bar
check(brand_bar.height() >= 28, "brand bar is too short to read")
check(brand_bar.name.text() != "", "brand bar wordmark empty")
check(brand_bar.tagline.text() != "", "brand bar tagline empty")
check(brand_bar.product.text() == t("app.title"), "brand bar product name wrong")
logo = brand_bar.logo.pixmap()
if logo is not None and not logo.isNull():
image = logo.toImage()
pixels = [
image.pixelColor(x, y).alpha()
for y in range(image.height())
for x in range(image.width())
]
check(any(a < 32 for a in pixels), "brand logo lost its transparency")
check(any(a >= 200 for a in pixels), "brand logo renders empty")
else:
check(not brand_bar.logo.isVisible(), "hidden logo still shown")
panels = [
window.mesh3d_panel,
window.arrow_panel,
window.force_plots,
window.cal_panel,
]
for panel, width in zip(panels, window.content_splitter.sizes()):
if panel.isVisible():
check(
width >= panel.minimumWidth(),
f"{type(panel).__name__} width {width} < min {panel.minimumWidth()}",
)
check(
window.force_plots.width() >= window.force_plots.minimumWidth(),
"force plots narrower than its minimum",
)
# value cards must keep their text visible
for card, label in zip(
window.force_plots.value_cards, window.force_plots.value_labels
):
check(
label.height() >= label.fontMetrics().height(),
"value card text is clipped",
)
check(label.text() != "", "value card is empty")
# demo data flows through the plots and the calibration table
for i in range(90):
window.force_plots.push(i / 30.0, float(i % 7), -1.0, 1.0, unit="N")
app.processEvents()
check(
window.force_plots.history.sample_count > 0,
"plots did not accept samples",
)
window.cal_panel.set_running(True)
check(window.cal_panel.btn_record.isEnabled(), "record button stays disabled")
check(
window.cal_panel.table.columnCount() == 6,
"the calibration table is no longer the six-column diagnostic view",
)
# Drive a real point through the recorder: the table is a view over
# the recorded points, not over a separate sample list.
window.cal_recorder.start(
channel="fn", reference_load_n=1.0, duration_s=1.0
)
for index in range(40):
window.cal_recorder.feed(
_calibration_sample(index, fn_raw=100.0 + index * 0.01)
)
point = window.cal_recorder.finish_now()
window.cal_panel.on_record_finished(point)
app.processEvents()
check(window.cal_panel.table.rowCount() == 1, "calibration row missing")
check(
not window.cal_panel.empty_hint.isVisible(),
"empty hint still visible with rows",
)
check(
window.cal_panel.fit_label.text() != "",
"the fit summary is empty",
)
# Four points, one of them off-trend (raw 400 where the others imply
# ~200). The exported model is piecewise-linear, so it passes through
# every point it is built from: the training residual is zero by
# construction and the off-trend point can only show up as a
# *validation* error. The summary must describe that model, not a
# private least-squares line.
for load, raw in ((2.0, 400.0), (3.0, 300.6), (5.0, 500.9)):
window.cal_recorder.start(
channel="fn", reference_load_n=load, duration_s=1.0
)
for index in range(40):
window.cal_recorder.feed(_calibration_sample(index, fn_raw=raw))
window.cal_panel.on_record_finished(window.cal_recorder.finish_now())
app.processEvents()
check(window.cal_panel.table.rowCount() == 4, "calibration rows missing")
fit = window.cal_recorder.fit("fn")
check(
fit.method == "piecewise_linear",
f"the summary is not about the exported model: {fit.method!r}",
)
check(
fit.breakpoints == 4,
f"the exported model has {fit.breakpoints} breakpoints, not 4",
)
check(
fit.max_residual_n <= 1e-9,
"the exported model does not reproduce its own calibration points: "
f"max residual {fit.max_residual_n}",
)
# Hold the off-trend point out: the exported model is then built
# without it and the misfit becomes visible as a validation error.
window.cal_panel.table.setCurrentCell(1, 0)
window.cal_panel._on_toggle_validation()
app.processEvents()
check(
t("cal.fit_validation").split("{")[0].strip()
in window.cal_panel.fit_label.text(),
"no validation summary after marking a validation point",
)
validation = window.cal_recorder.fit("fn").validation_error_n
check(
validation > 0.05,
f"the off-trend point produced no validation error: {validation}",
)
window.cal_panel._on_toggle_validation()
# Excluding a point removes it from the build, so the exported model
# no longer passes through it.
window.cal_panel._on_toggle_excluded()
app.processEvents()
check(
window.cal_panel.table.item(1, 5).text() == t("cal.state_excluded"),
"the excluded row does not show its state",
)
from orisys_app_shared.force.calibration_io import (
build_export_model,
build_force_calibration_v2 as _build_v2,
predict_channel,
)
import orisys.calibration as _sdk_calibration
store = window.cal_panel._store
model = build_export_model(
{key: store.channel(key) for key in ("fn", "fx", "fy")}
)
predicted = predict_channel(model, "fn", 400.0)
check(
abs(predicted - 2.0) > 0.05,
f"excluding a point left the exported model unchanged: {predicted}",
)
# One preparation: the store the panel shows is the prepared model's
# input, and the export serialises that same object.
prepared = window.cal_panel._prepared
check(
prepared is not None and prepared.ok,
"the panel has no prepared calibration to show or export",
)
check(
list(store.channel("fn")) == list(prepared.samples["fn"]),
"the panel store is not the prepared model's input",
)
doc, _warnings = _build_v2(prepared)
export_dir = APP_ROOT / "outputs" / "_cal_export_check"
if export_dir.exists():
for stale in export_dir.iterdir():
stale.unlink()
export_dir.mkdir(parents=True, exist_ok=True)
written = export_dir / "exported_calibration.json"
import json as _cal_json
written.write_text(_cal_json.dumps(doc), encoding="utf-8")
loaded = _sdk_calibration.load(written, strict=True)
for point in window.cal_recorder.points:
if point.excluded:
continue
raw = float(point.loaded_stats().mean)
check(
abs(
float(loaded.get("fn").convert(raw))
- prepared.predict("fn", raw)
)
<= 1e-9,
"the exported file and the panel disagree about the model",
)
check(
len(prepared.samples["fn"]) == 3,
f"the prepared build has {len(prepared.samples['fn'])} points, not 3",
)
window.cal_recorder.reset_session()
window.cal_panel._refresh_table()
window._set_phase("running")
check("status_running" not in window.status_chip.text(), "chip not localized")
window._set_theme("light" if theme == "dark" else "dark")
app.processEvents()
window._persist_prefs()
check(
len(window.prefs.get("splitter_sizes", [])) == 4,
"splitter sizes not persisted",
)
window.close()
app.processEvents()
section("small window")
app.setStyleSheet(build_qss("dark"))
window = StudioMainWindow(
prefs={
"language": "en_US",
"theme": "dark",
"sensor_type": "finger",
"show_cal_panel": True,
"show_finger_3d": True,
}
)
window.resize(1280, 720)
window.show()
app.processEvents()
for panel, width in zip(
[
window.mesh3d_panel,
window.arrow_panel,
window.force_plots,
window.cal_panel,
],
window.content_splitter.sizes(),
):
if panel.isVisible():
check(width >= panel.minimumWidth(), f"1280px: {type(panel).__name__} squeezed")
oversize = []
for child in window.findChildren(object):
if not hasattr(child, "isVisible") or not child.isVisible():
continue
if not hasattr(child, "geometry"):
continue
parent = child.parentWidget()
if parent is None or not parent.isVisible():
continue
if _inside_scroll_area(child):
# A scroll area's content is *meant* to exceed its viewport (the
# toolbar scrolls horizontally on narrow windows).
continue
geometry = child.geometry()
if geometry.width() > parent.width() + 1 or geometry.height() > parent.height() + 1:
oversize.append(
f"{type(child).__name__}({getattr(child, 'objectName', lambda: '')()})"
f" {geometry.width()}x{geometry.height()}"
f" in {type(parent).__name__} {parent.width()}x{parent.height()}"
)
check(not oversize, f"widgets overflow their parent: {oversize}")
window.close()
app.processEvents()
section("top area")
import orisys_studio.ui.record_status as _record_states
from orisys_app_shared.record.experiment import RecordStats as _RecordStats
#: Brand rail + status rail, collapsed settings, must fit in one instrument
#: line: the previous layout spent this height on a 40 px brand bar and a
#: tall sensor card stacked above the charts.
top_area_limit = 78.0
for theme in ("dark", "light"):
app.setStyleSheet(build_qss(theme))
window = StudioMainWindow(
prefs={"language": "zh_CN", "theme": theme, "sensor_type": "box"}
)
window.resize(1440, 880)
window.show()
for _ in range(3):
app.processEvents()
rail_height = window.brand_bar.height() + window.toolbar.height()
check(
rail_height <= top_area_limit,
f"{theme}: the collapsed top area is {rail_height}px (limit {top_area_limit})",
)
check(
not window.toolbar.settings_expanded(),
f"{theme}: the settings area starts expanded",
)
check(window.toolbar.status_chip.text().strip(), f"{theme}: the phase chip is empty")
# The sensor is one summary button on the rail; the pickers (and the
# calibration they belong to) live in the dialog it opens.
check(
window.toolbar.btn_sensor.text() == t(
"sensor.summary", type="BOX", config="sensor.json"
),
f"{theme}: the sensor summary is wrong: "
f"{window.toolbar.btn_sensor.text()!r}",
)
check(
window.toolbar.btn_sensor.isVisible(),
f"{theme}: the sensor button is not on the status rail",
)
check(
window.toolbar.camera_group.isVisible(),
f"{theme}: the live view does not offer the camera picker",
)
check(
not window.toolbar.video_group.isVisible(),
f"{theme}: the live camera view also shows the video-file group",
)
check(
window.toolbar.calibration_label.text().strip(),
f"{theme}: the rail has no calibration state",
)
check(
window.toolbar.record_chip.text() == t("record.state_idle"),
f"{theme}: the idle record state is "
f"{window.toolbar.record_chip.text()!r}",
)
check(
t("status.phase_running") != t("record.state_idle"),
"acquiring and recording do not have distinct labels",
)
check(
window.toolbar.record_chip.toolTip() != "",
"the record chip does not explain its state",
)
check(
not rail_squeezed(window),
f"{theme}: the status rail squeezes {rail_squeezed(window)}",
)
check(
not hasattr(window.toolbar, "source_combo"),
f"{theme}: the source mode is still a combobox",
)
# The status bar is a notice line, not a second state line: the config
# file and the phase are already on the rail, and a message must expire.
check(
not hasattr(window, "device_label"),
f"{theme}: the status bar still carries a permanent device line",
)
window.statusBar().showMessage("notice")
check(
window.statusBar().currentMessage() == "notice",
f"{theme}: the status bar does not show a message at all",
)
# The sensor dialog owns the pickers and the calibration facts.
window._open_sensor_dialog()
for _ in range(3):
app.processEvents()
dialog = window.sensor_dialog
check(dialog is not None and dialog.isVisible(), f"{theme}: the sensor dialog did not open")
check(
window.sensor_picker.isVisible(),
f"{theme}: the sensor pickers are not reachable in the dialog",
)
check(
window.sensor_picker.type_label.text() == "传感器"
and window.sensor_picker.config_label.text() == "传感器配置",
f"{theme}: the sensor picker labels are not the documented ones: "
f"{window.sensor_picker.type_label.text()!r} / "
f"{window.sensor_picker.config_label.text()!r}",
)
check(
dialog.version_label.text() != "" and dialog.axes_label.text() != "",
f"{theme}: the dialog states no calibration",
)
check(
t("sensor.export_none") in dialog.export_label.text()
or dialog.btn_apply.isEnabled(),
f"{theme}: the export row states neither a file nor its absence: "
f"{dialog.export_label.text()!r}",
)
# Changing the product in the dialog reaches the window.
window.sensor_picker.type_combo.setCurrentIndex(
window.sensor_picker.type_combo.findData("finger")
)
for _ in range(3):
app.processEvents()
check(
window.prefs.get("sensor_type") == "finger",
f"{theme}: the dialog's product change did not reach the window",
)
check(
"FINGER" in window.toolbar.btn_sensor.text(),
f"{theme}: the rail summary did not follow the product change: "
f"{window.toolbar.btn_sensor.text()!r}",
)
window.sensor_picker.type_combo.setCurrentIndex(
window.sensor_picker.type_combo.findData("box")
)
for _ in range(3):
app.processEvents()
dialog.close()
app.processEvents()
# The settings area is one click away, and remembers that it was opened.
window.toolbar.btn_settings.setChecked(True)
for _ in range(3):
app.processEvents()
expanded_height = window.brand_bar.height() + window.toolbar.height()
check(
expanded_height > rail_height,
f"{theme}: opening the settings area changed nothing "
f"({expanded_height} vs {rail_height})",
)
check(
window.toolbar.units_combo.isVisible() and window.toolbar.btn_preview.isVisible(),
f"{theme}: the unit and preview controls are unreachable",
)
check(
window.prefs.get("settings_expanded") is True,
f"{theme}: the settings state was not persisted",
)
check(
window.act_settings.isChecked(),
f"{theme}: the View menu does not follow the settings toggle",
)
window.toolbar.btn_settings.setChecked(False)
for _ in range(3):
app.processEvents()
check(
not window.toolbar.settings_expanded(),
f"{theme}: the settings area could not be collapsed again",
)
window.close()
app.processEvents()
# English labels are the widest case ("Start recording", "Sensor config"),
# and the tightest window is 1280: the rail must still fit both.
set_locale("en_US")
app.setStyleSheet(build_qss("dark"))
english_window = StudioMainWindow(
prefs={"language": "en_US", "theme": "dark", "sensor_type": "finger"}
)
english_window.resize(1280, 720)
english_window.show()
for _ in range(3):
app.processEvents()
check(
not rail_squeezed(english_window),
f"en_US at 1280x720: the status rail squeezes {rail_squeezed(english_window)}",
)
check(
t("sensor.summary", type="FINGER", config="sensor.json")
in english_window.toolbar.btn_sensor.text(),
f"the sensor summary is not stated in full: "
f"{english_window.toolbar.btn_sensor.text()!r}",
)
english_window.close()
app.processEvents()
set_locale("zh_CN")
# Exactly one *filled* action at a time: a red Stop dominating an otherwise
# idle top area was the original complaint.
app.setStyleSheet(build_qss("dark"))
window = StudioMainWindow(
prefs={"language": "zh_CN", "theme": "dark", "sensor_type": "box"}
)
window.resize(1440, 880)
window.show()
app.processEvents()
check(
window.toolbar.btn_run.objectName() == "primaryButton",
f"idle Run is not the filled primary: {window.toolbar.btn_run.objectName()!r}",
)
check(
not window.toolbar.btn_record.isEnabled(),
"recording is offered without a session",
)
stuck = _StuckWorker()
window._ensure_workers = lambda: None
window._box_worker = stuck
window._finger_worker = stuck
window.toolbar.btn_run.setChecked(True)
app.processEvents()
check(
window.toolbar.btn_run.objectName() == "dangerButton",
"Cancel is not the filled action while a connect hangs",
)
stuck.emit(_fake_packet(1))
window._on_run_tick()
app.processEvents()
check(window._session_phase == "running", "the session never reached running")
check(
window.toolbar.btn_run.objectName() == "dangerOutline",
f"Stop competes with the record action: "
f"{window.toolbar.btn_run.objectName()!r}",
)
check(
window.toolbar.btn_record.objectName() == "primaryButton",
"the record action is not the filled primary while acquiring",
)
check(
window._calibration_summary() == t("status.calibrated_all"),
f"a fully calibrated packet is not reported as such: "
f"{window._calibration_summary()!r}",
)
window.toolbar.set_record_checked(True)
app.processEvents()
check(
window.toolbar.btn_record.objectName() == "dangerButton",
"stopping the recording is not the filled danger",
)
check(
window.toolbar.btn_run.objectName() == "dangerOutline",
"two filled actions share the rail while recording",
)
real_stats = window._record_stats
window._record_stats = lambda: (
_record_states.RECORDING,
_RecordStats(duration_s=12.4, frames=372, segments=1, dropped_data=3),
"20240101T000000Z_session0001_deadbeef",
)
window._refresh_record_state()
app.processEvents()
chip = window.toolbar.record_chip
check(
"12.4" in chip.text(),
f"the record strip does not show the duration: {chip.text()!r}",
)
check(
t("record.dropped", count=3) in chip.text(),
f"dropped frames are not surfaced: {chip.text()!r}",
)
check(
"3" in chip.toolTip() and "372" in chip.toolTip(),
f"the record tooltip has no detail: {chip.toolTip()!r}",
)
window._record_stats = lambda: (
_record_states.ERROR,
_RecordStats(
frames=10, segments=1, failed=True, write_errors=2, last_error="disk full"
),
"",
)
window._refresh_record_state()
check(
chip.text() == t("record.state_error", count=2),
f"a write failure is not reported: {chip.text()!r}",
)
check(
"disk full" in chip.toolTip(),
f"the record tooltip hides the failure reason: {chip.toolTip()!r}",
)
window._record_stats = real_stats
window.toolbar.set_record_checked(False)
window._stop_run(reason="user_stop")
stuck.release()
window.close()
app.processEvents()
section("stopping a hung connect")
stuck = _StuckWorker()
window = StudioMainWindow(
prefs={
"language": "zh_CN",
"theme": "dark",
"sensor_type": "box",
"source_mode": "camera",
}
)
window.resize(1440, 880)
window.show()
window._ensure_workers = lambda: None
window._box_worker = stuck
window._finger_worker = stuck
app.processEvents()
check(not window.act_stop.isEnabled(), "Stop shortcut is live before a session")
window.toolbar.btn_run.setChecked(True)
app.processEvents()
check(window._session_phase == "connecting", "run did not enter the connecting phase")
check(
window.toolbar.btn_run.isEnabled(),
"Stop is disabled while connecting — a hung connect could not be cancelled",
)
check(window.toolbar.btn_run.isChecked(), "run button did not latch into the stop state")
check(
window.toolbar.btn_run.text() == t("toolbar.cancel_label"),
f"connecting label is not the cancel label: {window.toolbar.btn_run.text()!r}",
)
check(
window.toolbar.btn_run.toolTip() == t("toolbar.cancel_tip"),
"connecting tooltip does not explain how to cancel",
)
check(window.act_stop.isEnabled(), "Stop shortcut is disabled while connecting")
started = time.perf_counter()
window.toolbar.btn_run.setChecked(False)
elapsed = time.perf_counter() - started
check(elapsed < 0.5, f"Stop blocked the GUI for {elapsed:.2f}s on a hung connect")
check(not window._running, "session stayed active after Stop")
check(window._session_phase == "idle", "phase did not return to idle after Stop")
check(
stuck.thread_still_owned,
"Stop abandoned the still-running worker thread",
)
# The slow-connect hint must point at the escape hatch.
window.toolbar.btn_run.setChecked(True)
app.processEvents()
window._connect_started_at = time.perf_counter() - 3600.0
window._on_run_tick()
check(
window.statusBar().currentMessage() == t("status.connect_slow"),
f"no slow-connect hint: {window.statusBar().currentMessage()!r}",
)
escaped = time.perf_counter()
window._on_stop_shortcut()
check(
time.perf_counter() - escaped < 0.5,
"Esc did not cancel the hung connect immediately",
)
check(not window._running, "Esc left the session running")
window.close()
app.processEvents()
# Let the deliberately-stuck thread finish before the interpreter tears
# down Qt, otherwise it would outlive the widgets it can reach.
stuck.release()
section("record channel")
import numpy as np
stuck = _StuckWorker()
window = StudioMainWindow(
prefs={
"language": "zh_CN",
"theme": "dark",
"sensor_type": "box",
"source_mode": "camera",
}
)
window.resize(1440, 880)
window.show()
window._ensure_workers = lambda: None
window._box_worker = stuck
window._finger_worker = stuck
window.toolbar.btn_run.setChecked(True)
app.processEvents()
check(
window._on_packet_recorded in stuck.sinks(),
"the record channel is not attached to the running worker",
)
# The record channel must not depend on the UI refresh timer at all.
for index in range(90):
stuck.emit(_fake_packet(index))
check(
window.session_store.count == 90,
f"record channel dropped samples without a UI tick: {window.session_store.count}",
)
# Invalid readouts must never be averaged in as zeros.
for index in range(10):
stuck.emit(_fake_packet(200 + index, valid=False, reason="read failed"))
slice_ = window.session_store.data()
check(int(slice_.valid.sum()) == 90, "invalid samples were not marked invalid")
check(
not bool(slice_.valid[-1]) and slice_.raw[-1][0] != slice_.raw[-1][0],
"a failed readout was stored as a number instead of NaN",
)
# Calibration sampling must survive the panel being hidden.
window._show_cal = False
window._apply_panel_visibility()
check(not window.cal_panel.isVisible(), "the calibration panel is still visible")
window.cal_recorder.start(channel="fn", reference_load_n=2.0, duration_s=1.0)
for index in range(40):
stuck.emit(_fake_packet(300 + index))
point = window.cal_recorder.finish_now()
check(point is not None and point.n_valid == 40, "hidden panel stopped the sampling")
check(
point is not None and point.status == "ok",
f"calibration point was not usable: {getattr(point, 'status', None)}",
)
# While recording, the record button must stay enabled — it is the cancel.
window.cal_panel.set_running(True)
window.cal_panel._on_record_clicked()
app.processEvents()
check(window.cal_panel._recording, "recording did not start")
check(
window.cal_panel.btn_record.isEnabled(),
"the record button is disabled while recording, so it cannot cancel",
)
check(
window.cal_panel.btn_record.text() == t("cal.recording_stop"),
f"record button does not show the stop label: {window.cal_panel.btn_record.text()!r}",
)
window.cal_panel._on_record_clicked()
app.processEvents()
check(not window.cal_panel._recording, "the recording could not be cancelled")
# The depth view must not fall back to the processed frame.
window.arrow_panel.set_view_mode("depth")
window._last_rendered_view_mode = None
stuck.emit(_fake_packet(400, depth=None))
window._on_run_tick()
app.processEvents()
check(
window.arrow_panel.image_label.text() == t("panels.depth_unavailable"),
f"missing depth was not reported: {window.arrow_panel.image_label.text()!r}",
)
window._last_rendered_view_mode = None
stuck.emit(_fake_packet(401, depth=np.linspace(0.0, 1.0, 64).reshape(8, 8)))
window._on_run_tick()
first_scale = window.arrow_panel.depth_scale_label.text()
window._last_rendered_view_mode = None
stuck.emit(_fake_packet(402, depth=np.linspace(0.0, 9.0, 64).reshape(8, 8)))
window._on_run_tick()
check(
window.arrow_panel.depth_scale_label.text() == first_scale and first_scale != "",
"the depth colour scale re-fitted per frame instead of staying fixed",
)
window.close()
app.processEvents()
stuck.release()
section("instrument panels")
window = StudioMainWindow(
prefs={"language": "zh_CN", "theme": "dark", "sensor_type": "box"}
)
window.resize(1440, 880)
window.show()
app.processEvents()
plots_panel = window.force_plots
check(
plots_panel.value_group.y() < plots_panel.plot_frame.y(),
"the axis readouts are not above the charts",
)
check(
all(label.text() for label in plots_panel.value_labels),
"an axis readout is empty before any sample",
)
for label in plots_panel.value_labels:
check(
label.height() >= label.fontMetrics().height(),
"an axis readout is clipped",
)
check(
label.fontMetrics().height() >= 20,
f"the axis readouts are not large digits: "
f"{label.font().family()} {label.fontMetrics().height()}px",
)
check(
[plot.getAxis("bottom").style.get("showValues") for plot in plots_panel.plots]
== [False, False, True],
"the three charts do not share one bottom time axis",
)
arrow = window.arrow_panel
check(
arrow.scale_label.isVisible() and arrow.scale_label.text(),
"the arrow view does not show its display magnification",
)
check(
"2" in arrow.scale_label.text(),
f"the box magnification is not the documented 2x: {arrow.scale_label.text()!r}",
)
check(
arrow.pipeline_label.isVisible() and arrow.pipeline_label.text(),
"the arrow view does not name its pipeline",
)
check(
arrow.area_label.isVisible() and arrow.area_label.text(),
"the arrow view does not report the valid area",
)
check(
not arrow.colorbar.isVisible(),
"the depth colour bar is shown in the arrow view",
)
arrow.set_view_mode("depth")
arrow.set_depth_scale("深度色标 0.000 – 1.000")
for _ in range(2):
app.processEvents()
check(arrow.colorbar.isVisible(), "the depth colour bar is hidden in the depth view")
check(
not arrow.scale_label.isVisible(),
"the arrow magnification is still shown in the depth view",
)
check(
arrow.depth_scale_label.isVisible() and arrow.depth_scale_label.text(),
"the depth range is not shown next to its colour bar",
)
arrow.set_view_mode("arrows_pure")
for _ in range(2):
app.processEvents()
check(
not arrow.colorbar.isVisible() and arrow.scale_label.isVisible(),
"leaving the depth view did not restore the arrow legend",
)
# The overview strip is the timeline control: dragging it pans the window.
stuck = _StuckWorker()
window._ensure_workers = lambda: None
window._box_worker = stuck
window._finger_worker = stuck
window.toolbar.btn_run.setChecked(True)
app.processEvents()
for index in range(120):
stuck.emit(_fake_packet(index + 100))
window._on_run_tick()
for _ in range(3):
app.processEvents()
check(
window.session_store.count == 120,
f"the session store missed samples: {window.session_store.count}",
)
first, last = window.session_store.extent()
check(last > first, "the session store has no extent to browse")
check(
plots_panel.window_label.text() != "",
"the overview strip has no window readout",
)
plots_panel.show_all()
app.processEvents()
check(plots_panel.mode() == "browse", "the overview button did not enter browsing")
target = (first + (last - first) * 0.25, first + (last - first) * 0.5)
plots_panel.overview_region.setRegion(target)
for _ in range(3):
app.processEvents()
dragged = plots_panel.window()
check(
abs(dragged[0] - target[0]) < 0.05 and abs(dragged[1] - target[1]) < 0.05,
f"dragging the overview did not move the window: {dragged} vs {target}",
)
check(
plots_panel.mode() == "browse",
"dragging the overview left the charts in live mode",
)
window.brand_bar.btn_live.click()
for _ in range(2):
app.processEvents()
check(
plots_panel.mode() == "follow",
"the brand rail's live item did not return to live mode",
)
check(
window.brand_bar.btn_live.property("navActive") == "true",
"the live nav item is not marked as the current view",
)
window.act_cal_panel.setChecked(False)
app.processEvents()
window.brand_bar.btn_calibration.click()
app.processEvents()
check(
window.cal_panel.isVisible(),
"the calibration nav item did not reveal the panel",
)
window.brand_bar.btn_calibration.click()
for _ in range(3):
app.processEvents()
from PyQt5.QtWidgets import QHeaderView as _QHeaderView
sizes = window.content_splitter.sizes()
check(
window._view_preset == "calibration",
f"the calibration nav item did not switch the view: {window._view_preset!r}",
)
check(
window.cal_panel.isVisible() and not window.arrow_panel.isVisible(),
"the calibration view does not trade the 2D panel for the calibration panel",
)
check(
sizes[3] >= 500,
f"the calibration panel is too narrow for its table: {sizes[3]}px",
)
check(
window.cal_panel.table.horizontalHeader().sectionResizeMode(0)
== _QHeaderView.ResizeToContents,
"the calibration columns can still be elided instead of scrolled",
)
check(
window.cal_panel.scroll.isVisible(),
"the calibration panel cannot scroll to its own controls",
)
for button in (
window.cal_panel.btn_apply,
window.cal_panel.btn_export,
window.cal_panel.btn_toggle_validation,
):
check(
button.height() >= button.fontMetrics().height(),
f"a calibration button is cut off: {button.text()!r}",
)
check(
button.width() >= button.sizeHint().width(),
f"a calibration button is clipped: {button.text()!r} "
f"({button.width()} < {button.sizeHint().width()})",
)
window.brand_bar.btn_playback.click()
for _ in range(3):
app.processEvents()
check(
window._view_preset == "replay",
f"the replay nav item did not switch the view: {window._view_preset!r}",
)
check(
not window.cal_panel.isVisible() and window.arrow_panel.isVisible(),
"the replay view is not the 2D + charts arrangement",
)
# No recording is open yet: the charts keep following the live session
# instead of pretending to replay something, and the layout is the live one.
check(not window.replay.is_loaded(), "a recording is open before one was chosen")
check(
window.force_plots.mode() == "follow",
"the replay view browses before a recording is open",
)
# The replay view is the live arrangement, measured rather than assumed: the
# splitter hands out its handles' width, so compare like with like.
window.brand_bar.btn_live.click()
for _ in range(3):
app.processEvents()
live_sizes = tuple(window.content_splitter.sizes())
window.brand_bar.btn_playback.click()
for _ in range(3):
app.processEvents()
replay_sizes = tuple(window.content_splitter.sizes())
check(
replay_sizes == live_sizes,
f"the replay view is not the live arrangement: {replay_sizes} vs {live_sizes}",
)
check(
not hasattr(window, "replay_bar"),
"the replay view still carries a separate bar",
)
check(
window.toolbar.btn_replay_video.isVisible(),
"the replay view does not offer a video to replay",
)
# The one difference between the two views: the source slot.
window.brand_bar.btn_live.click()
for _ in range(3):
app.processEvents()
check(
not window.toolbar.btn_replay_video.isVisible()
and window.toolbar.source_slot.isVisible(),
"the live view still shows the replay's video picker",
)
window.brand_bar.btn_live.click()
for _ in range(3):
app.processEvents()
sizes = window.content_splitter.sizes()
check(
window._view_preset == "live",
f"the live nav item did not switch the view: {window._view_preset!r}",
)
check(
sizes[2] > sizes[1],
f"the charts are not given more room than the 2D panel: {sizes}",
)
check(
window.force_plots.mode() == "follow",
"the live view did not resume following the latest samples",
)
check(
window.brand_bar.btn_live.property("navActive") == "true",
"the live nav item is not marked as the current view",
)
window._stop_run(reason="user_stop")
stuck.release()
for _ in range(3):
app.processEvents()
# The 2D area keeps the shape of the image it shows, instead of stretching a
# 16:9 frame into whatever the panel happens to be.
finger_window = StudioMainWindow(
prefs={
"language": "zh_CN",
"theme": "dark",
"sensor_type": "finger",
"show_finger_3d": False,
}
)
finger_window.resize(1440, 880)
finger_window.show()
for _ in range(3):
app.processEvents()
image = finger_window.arrow_panel.image_label
check(
abs(image.aspect() - 16.0 / 9.0) < 0.01,
f"the 2D area is not 16:9 for a finger sensor: {image.aspect()}",
)
check(
abs(image.height() - image.width() / image.aspect()) <= 2,
"the 2D image area does not follow the output image's aspect ratio: "
f"{image.width()}x{image.height()}",
)
finger_window.close()
app.processEvents()
check(
window.toolbar.record_chip.text() == t("record.state_idle"),
f"the record strip did not fall back to idle: "
f"{window.toolbar.record_chip.text()!r}",
)
window.close()
app.processEvents()
section("chart header and unit chip")
from orisys_studio.ui.force_plots import unit_display
header_window = StudioMainWindow(
prefs={"language": "zh_CN", "theme": "dark", "sensor_type": "box"}
)
header_window.resize(1440, 880)
header_window.show()
for _ in range(3):
app.processEvents()
header_plots = header_window.force_plots
for button, name in (
(header_plots.btn_overview, "zoom to fit"),
(header_plots.btn_box_zoom, "box zoom"),
(header_plots.btn_latest, "back to latest"),
):
check(button.text() == "", f"the {name} action still shows text")
check(not button.icon().isNull(), f"the {name} action has no icon")
check(button.toolTip() != "", f"the {name} action has no tooltip")
check(button.accessibleName() != "", f"the {name} action has no accessible name")
check(
isinstance(header_plots.unit_badge, type(header_plots.btn_overview)),
"the unit chip is not a button",
)
check(
header_plots.unit_badge.toolTip() != "",
"the unit chip does not say what it does",
)
# The chip switches the same unit everywhere: the charts and the toolbar.
before_unit = header_plots.unit()
header_plots.unit_badge.click()
app.processEvents()
check(
header_plots.unit() != before_unit,
f"the unit chip did not switch the charts: {header_plots.unit()}",
)
check(
("newton" if header_plots.unit() == "N" else "raw")
== header_window.toolbar.units(),
"the unit chip and the toolbar disagree about the unit",
)
check(
header_plots.unit_badge.text() == unit_display(header_plots.unit()),
f"the chip does not state the active unit: {header_plots.unit_badge.text()!r}",
)
header_plots.unit_badge.click()
app.processEvents()
check(
header_plots.unit() == before_unit,
"the unit chip does not switch back",
)
# The arrow legend says what the factor multiplies.
check(
"×" in header_window.arrow_panel.scale_label.text(),
"the arrow footer lost its magnification",
)
check(
"位移" in header_window.arrow_panel.scale_label.text()
or "displacement" in header_window.arrow_panel.scale_label.text(),
f"the arrow footer does not say what is magnified: "
f"{header_window.arrow_panel.scale_label.text()!r}",
)
# The header's mode label shares its row with three icon buttons and the unit
# chip: the long form of the mode sentence was cut mid-word in English.
for lang in ("zh_CN", "en_US"):
header_window._set_language(lang)
for _ in range(3):
app.processEvents()
for mode, name in ((False, "follow"), (True, "browse")):
if mode:
header_plots.set_window(0.2, 1.0)
app.processEvents()
label = header_plots.mode_label
check(
label.width() >= label.sizeHint().width(),
f"{lang}/{name}: the chart mode label is clipped: "
f"{label.width()} < {label.sizeHint().width()} ({label.text()!r})",
)
check(
label.toolTip() != "",
f"{lang}/{name}: the mode label does not explain itself",
)
# The locale is process-wide: leave it where the other sections expect it.
header_window._set_language("zh_CN")
app.processEvents()
header_window.close()
app.processEvents()
section("acquisition correctness")
import csv as _csv
import shutil as _shutil
import numpy as np
# A device that only confirms its calibration a moment after the first
# frames reports raw counts for those frames. They must never be drawn on the
# newton axis: that was the tall transient at the start of every session.
stuck = _StuckWorker()
window = StudioMainWindow(
prefs={
"language": "zh_CN",
"theme": "dark",
"sensor_type": "box",
"source_mode": "camera",
}
)
window.resize(1440, 880)
window.show()
window._ensure_workers = lambda: None
window._box_worker = stuck
window._finger_worker = stuck
for _ in range(3):
app.processEvents()
window._start_run()
window._run_timer.stop()
window._set_phase("running")
plots = window.force_plots
for index in range(3):
packet = _fake_packet(index, calibrated_axes=())
packet.force_calibrated = False
stuck.emit(packet)
window._on_run_tick()
check(
plots.unit() == "raw",
f"an uncalibrated device is not shown in raw units: {plots.unit()}",
)
for index in range(3, 40):
stuck.emit(_fake_packet(index))
window._on_run_tick()
state = plots.history.state()
check(plots.unit() == "N", f"the unit did not follow the calibration: {plots.unit()}")
check(
plots.history.sample_count == 37,
f"the raw frames were kept in the newton buffer: "
f"{plots.history.sample_count} samples",
)
peak = (
float(np.nanmax(np.abs(state.values[:, 0]))) if state is not None else float("inf")
)
check(
state is not None and peak < 5.0,
f"a raw count is still drawn on the newton axis (peak {peak})",
)
# A new session must not inherit the previous one's samples.
window._stop_run(reason="user_stop")
window._start_run()
window._run_timer.stop()
window._set_phase("running")
check(
plots.history.sample_count == 0 and plots.mode() == "follow",
f"a new session started with {plots.history.sample_count} old samples",
)
# While recording, the reported duration is the span captured so far: the
# record strip used to sit at 00:00.0 for the whole recording.
app_module = sys.modules["orisys_studio.app_window"]
scratch = APP_ROOT / "outputs" / "_correctness_check"
_shutil.rmtree(scratch, ignore_errors=True)
scratch.mkdir(parents=True, exist_ok=True)
original_root = app_module.recordings_dir
app_module.recordings_dir = lambda: scratch
try:
window.toolbar.btn_record.setChecked(True)
app.processEvents()
check(window.recorder.is_recording, "the recording did not start")
for index in range(96):
# 48 frames per second of acquisition time: above the 30 fps the
# sensor config declares, which is what the container must follow.
packet = _fake_packet(index, mono_t=index / 48.0)
packet.fps = 48.0
stuck.emit(packet)
window._on_run_tick()
live_stats = window.recorder.stats()
check(
live_stats.duration_s > 1.5,
f"the live recording duration stayed at {live_stats.duration_s}",
)
check(
"00:0" not in window.toolbar.record_chip.text()
or window.toolbar.record_chip.text()
!= t("record.state_recording", duration="00:00.0"),
f"the record strip does not count up: "
f"{window.toolbar.record_chip.text()!r}",
)
check(
abs(window._video_fps() - 30.0) < 0.01,
f"the video rate is not the sensor config's: {window._video_fps()}",
)
window.toolbar.btn_record.setChecked(False)
app.processEvents()
report = window._record_last_report
check(report is not None, "no record report was published")
if report is not None:
check(
report.stats.resampled_video > 0,
"the faster-than-configured frames were not reported",
)
with (report.directory / "frames.csv").open(encoding="utf-8") as handle:
rows = list(_csv.DictReader(handle))
check(
len(rows) == 96,
f"the frame rows were thinned by the video pacing: {len(rows)}",
)
check(
any("faster than" in problem for problem in report.problems),
f"the report does not explain the resampled frames: {report.problems}",
)
import cv2 as _cv2
segments = sorted(report.directory.glob("raw_*.mp4"))
check(bool(segments), "the recording wrote no video segment")
if segments:
capture = _cv2.VideoCapture(str(segments[0]))
try:
written = int(capture.get(_cv2.CAP_PROP_FRAME_COUNT))
container_fps = float(capture.get(_cv2.CAP_PROP_FPS))
finally:
capture.release()
check(
abs(container_fps - 30.0) < 1.0,
f"the video declares {container_fps} fps, not the sensor's 30",
)
check(
55 <= written <= 65,
f"the video holds {written} frames for 2 s at 30 fps",
)
finally:
app_module.recordings_dir = original_root
_shutil.rmtree(scratch, ignore_errors=True)
# The rail's FPS is the worker's *delivered* frame rate. It used to be a count
# of loop iterations (a 30 fps device read 63), so the number the rail prints
# must be the packet's, verbatim.
window.act_show_fps.setChecked(True)
app.processEvents()
packet = _fake_packet(0, mono_t=0.0)
packet.fps = 30.4
stuck.emit(packet)
window._on_run_tick()
app.processEvents()
check(
"30.4 FPS" in window.toolbar.source_summary.text(),
f"the rail does not show the delivered frame rate: "
f"{window.toolbar.source_summary.text()!r}",
)
window.act_show_fps.setChecked(False)
app.processEvents()
window._stop_run(reason="user_stop")
stuck.release()
app.processEvents()
section("replay owns the 2D pane")
# A replay running next to a live session must not be overwritten by the
# camera: the pane belongs to whichever source the view is showing.
replay_root = APP_ROOT / "outputs" / "_pane_check"
_shutil.rmtree(replay_root, ignore_errors=True)
replay_root.mkdir(parents=True, exist_ok=True)
from orisys_app_shared.record import ExperimentRecorder, RecordConfig
recorder = ExperimentRecorder(RecordConfig(video=True, fps=30.0))
recording = recorder.start(root=replay_root, session_id=1, metadata={})
for index in range(20):
# A raw-only recording: no calibration curve, so the calibrated columns
# stay empty. Showing that in newton mode would draw three blank charts.
packet = _fake_packet(index)
packet.force_calibrated = False
recorder.on_packet(packet)
recorder.stop()
stuck = _StuckWorker()
pane_window = StudioMainWindow(
prefs={"language": "zh_CN", "theme": "dark", "sensor_type": "box"}
)
pane_window.resize(1440, 880)
pane_window.show()
pane_window._ensure_workers = lambda: None
pane_window._box_worker = stuck
pane_window._finger_worker = stuck
pane_window._start_run()
pane_window._run_timer.stop()
pane_window._set_phase("running")
pane_window.toolbar.set_units("newton")
app.processEvents()
pane_window.brand_bar.btn_playback.click()
app.processEvents()
check(pane_window._load_recording(recording), "the replay could not open the recording")
app.processEvents()
check(pane_window._replay_owns_image(), "the replay does not own the 2D pane")
# The recording has no calibrated axis, so newton would be an empty chart.
check(
pane_window.force_plots.unit() == "raw",
f"a raw-only recording was shown in {pane_window.force_plots.unit()}",
)
check(
pane_window.toolbar.units() == "raw",
"the toolbar and the charts disagree about a recording's unit",
)
check(
pane_window.force_plots.unit_badge.text() == t("units.raw_short"),
"the chart's unit chip disagrees with the charts",
)
drawn = [int(curve.getData()[0].size) for curve in pane_window.force_plots.curves]
check(all(count > 0 for count in drawn), f"the recording drew nothing: {drawn}")
# The cards read the point under the playhead, not the live acquisition.
duration = pane_window.replay.duration_s()
pane_window.replay.set_position(duration)
app.processEvents()
cursor_card = pane_window.force_plots.value_labels[0].text()
check(
cursor_card not in ("", t("chart.value_unavailable")),
f"the replay readout is empty: {cursor_card!r}",
)
check(
float(cursor_card) > 9.0,
f"the readout is not the sample at the cursor: {cursor_card}",
)
before_key = pane_window.arrow_panel.image_label.pixmap().cacheKey()
before_text = pane_window.arrow_panel.image_label.text()
for index in range(3):
packet = _fake_packet(index, mono_t=index / 30.0)
packet.raw_frame = np.full((24, 32, 3), 200, np.uint8)
stuck.emit(packet)
pane_window._on_run_tick()
app.processEvents()
check(
pane_window.arrow_panel.image_label.pixmap().cacheKey() == before_key
and pane_window.arrow_panel.image_label.text() == before_text,
"a live frame overwrote the replay's picture",
)
check(
pane_window.force_plots.unit() == "raw",
"a live sample switched the unit of the recording being replayed",
)
check(
pane_window.force_plots.value_labels[0].text() == cursor_card,
"a live sample overwrote the replay's readout",
)
pane_window.brand_bar.btn_live.click()
app.processEvents()
check(
not pane_window._replay_owns_image(),
"the live view still hands the 2D pane to the replay",
)
pane_window._on_run_tick()
app.processEvents()
pane_window.replay.close()
pane_window._stop_run(reason="user_stop")
stuck.release()
pane_window.close()
app.processEvents()
_shutil.rmtree(replay_root, ignore_errors=True)
section("clean shutdown")
# A camera SDK leaves native threads behind; the entry point must still come
# back to the console. This runs the real ``main()`` in a child process and
# closes the window from a timer, so a hang shows up as a timeout.
import subprocess as _subprocess
script = (
"import os, sys\n"
"os.environ.setdefault('QT_QPA_PLATFORM', 'offscreen')\n"
"os.environ.setdefault('QT_QPA_FONTDIR', r'C:\\Windows\\Fonts')\n"
"from PyQt5.QtCore import QTimer\n"
"import orisys_studio.main as studio_main\n"
"show = studio_main.StudioMainWindow.show\n"
"def show_and_close(self):\n"
" show(self)\n"
" QTimer.singleShot(400, self.close)\n"
"studio_main.StudioMainWindow.show = show_and_close\n"
"raise SystemExit(studio_main.main())\n"
)
env = dict(os.environ)
env["PYTHONPATH"] = os.pathsep.join(
[str(REPO_ROOT / "shared" / "src"), str(APP_ROOT / "src"), env.get("PYTHONPATH", "")]
)
started = time.perf_counter()
try:
completed = _subprocess.run(
[sys.executable, "-c", script],
cwd=str(APP_ROOT),
env=env,
capture_output=True,
timeout=60,
)
elapsed = time.perf_counter() - started
print(f" window closed and the process returned in {elapsed:.1f}s")
check(
completed.returncode == 0,
f"the app exited with {completed.returncode}: "
f"{completed.stderr[-400:]!r}",
)
check(
elapsed < 45.0,
f"closing the window did not return to the console ({elapsed:.1f}s)",
)
except _subprocess.TimeoutExpired:
check(False, "the app never exited after its window was closed")
section("experiment record")
import shutil as _shutil
scratch = APP_ROOT / "outputs" / "_record_check"
_shutil.rmtree(scratch, ignore_errors=True)
scratch.mkdir(parents=True, exist_ok=True)
import orisys_studio.app_window as _app_module
_app_module.recordings_dir = lambda: scratch
stuck = _StuckWorker()
window = StudioMainWindow(
prefs={
"language": "zh_CN",
"theme": "dark",
"sensor_type": "box",
"source_mode": "camera",
}
)
window.resize(1440, 880)
window.show()
window._ensure_workers = lambda: None
window._box_worker = stuck
window._finger_worker = stuck
window._start_run()
window._run_timer.stop()
window._set_phase("running")
app.processEvents()
check(
window.toolbar.btn_record.isEnabled(),
"the record control is disabled while a session is running",
)
check(
window.toolbar.btn_record.text() == t("record.start_label"),
f"record button label is wrong: {window.toolbar.btn_record.text()!r}",
)
check(
window.toolbar.record_mode() == "video_data",
"the default recording mode is not video+data",
)
window.toolbar.btn_record.setChecked(True)
app.processEvents()
check(window.recorder.is_recording, "the recording did not start")
check(
window.toolbar.btn_record.isEnabled()
and window.toolbar.btn_record.text() == t("record.stop_label"),
"the record button is not a usable stop control",
)
check(
not window.toolbar.record_mode_combo.isEnabled(),
"the recording mode can be changed mid-recording",
)
for index in range(30):
stuck.emit(_fake_packet(index))
window._on_run_tick()
stuck.emit(_fake_packet(90, valid=False, reason="read failed"))
window._on_run_tick()
# The zeroing is a *request*; the event is recorded only once the acquisition
# thread confirms it took effect (the stub confirms immediately).
stuck.on_reset_applied = window._on_reset_applied
window._on_reset()
window._on_run_tick()
app.processEvents()
window.toolbar.btn_record.setChecked(False)
app.processEvents()
report = window._record_last_report
check(not window.recorder.is_recording, "the recording did not stop")
check(report is not None and report.ok, f"recording reported problems: {getattr(report, 'problems', None)}")
check(
not window.toolbar.btn_record.isChecked(),
"the record button still looks active after the recording stopped",
)
check(
window._record_report_pending is False,
"a finished recording is still waiting for a report",
)
import csv as _csv
import json as _json
directory = report.directory
names = sorted(path.name for path in directory.iterdir())
check(
names == ["events.csv", "forces.csv", "frames.csv", "raw_0001.mp4", "session.json"],
f"unexpected experiment layout: {names}",
)
with (directory / "forces.csv").open(encoding="utf-8") as handle:
rows = list(_csv.DictReader(handle))
check(tuple(rows[0].keys()) == FORCES_HEADER, "forces.csv header drifted from the spec")
check(
rows[-1]["valid"] == "0" and rows[-1]["FN_raw"] == "",
"a failed readout was written as a number instead of being left empty",
)
with (directory / "frames.csv").open(encoding="utf-8") as handle:
frames = list(_csv.DictReader(handle))
check(len(frames) == len(rows), "forces.csv and frames.csv disagree on the frame count")
check(
all(f["frame_id"] == r["frame_id"] for f, r in zip(frames, rows)),
"the frame index does not share frame ids with the data",
)
meta = _json.loads((directory / "session.json").read_text(encoding="utf-8"))
check(meta.get("ended_at"), "session.json was not finalised")
check(meta["format_version"], "session.json has no format version")
check(
[event["kind"] for event in meta["events"]] == ["record_started", "zeroing", "record_stopped"],
f"events not recorded: {[e['kind'] for e in meta['events']]}",
)
check(meta["stats"]["frames"] == 31, f"unexpected frame count: {meta['stats']['frames']}")
window.close()
app.processEvents()
stuck.release()
_shutil.rmtree(scratch, ignore_errors=True)
section("record failure states")
from unittest.mock import patch as _patch
from orisys_app_shared.record.experiment import _SessionWriter as _Writer
scratch_fail = APP_ROOT / "outputs" / "_record_fail"
_shutil.rmtree(scratch_fail, ignore_errors=True)
scratch_fail.mkdir(parents=True, exist_ok=True)
stuck_fail = _StuckWorker()
fail_window = StudioMainWindow(
prefs={
"language": "zh_CN",
"theme": "dark",
"sensor_type": "box",
"source_mode": "camera",
}
)
fail_window.resize(1440, 880)
fail_window.show()
fail_window._ensure_workers = lambda: None
fail_window._box_worker = stuck_fail
fail_window._finger_worker = stuck_fail
fail_window._start_run()
fail_window._run_timer.stop()
fail_window._set_phase("running")
# A directory that cannot be created is a terminal, retryable failure.
blocked = scratch_fail / "blocked"
blocked.write_text("", encoding="utf-8")
_app_module.recordings_dir = lambda: blocked
fail_window.toolbar.btn_record.setChecked(True)
app.processEvents()
check(
not fail_window.recorder.is_recording,
"a recording started even though its directory could not be created",
)
check(
not fail_window.toolbar.btn_record.isChecked(),
"the record button stayed active after the start failed",
)
check(
fail_window.recorder.state == "failed",
f"a failed start left the recorder in {fail_window.recorder.state!r}",
)
# ...and the operator can simply try again.
_app_module.recordings_dir = lambda: scratch_fail
fail_window.toolbar.btn_record.setChecked(True)
app.processEvents()
check(
fail_window.recorder.is_recording,
"the recorder refused to retry after a failed directory",
)
# A writer that keeps failing must fail the recording, and the window must
# consume that report even though ``is_recording`` is already False.
with _patch.object(_Writer, "write_sample", side_effect=OSError("disk full")):
for index in range(8):
stuck_fail.emit(_fake_packet(index))
fail_window._on_run_tick()
deadline = time.monotonic() + 3.0
while (
fail_window.recorder.stats().write_errors < 5
and time.monotonic() < deadline
):
app.processEvents()
time.sleep(0.01)
check(
fail_window.recorder.stats().failed,
"repeated write failures never failed the recording",
)
check(
not fail_window.recorder.is_recording,
"a dead writer still claimed to be recording",
)
fail_window._stop_recording()
app.processEvents()
failed_report = fail_window._record_last_report
check(
failed_report is not None and failed_report.stats.failed,
"the window did not consume the failure report",
)
check(
not fail_window.toolbar.btn_record.isChecked(),
"the record button stayed active after the recording failed",
)
check(
fail_window._session_phase == "error",
f"the session phase is {fail_window._session_phase!r} after a failed recording",
)
fail_window.close()
app.processEvents()
stuck_fail.release()
_app_module.recordings_dir = lambda: scratch
_shutil.rmtree(scratch_fail, ignore_errors=True)
section("calibration record identity")
import json as _cal_json
from unittest.mock import patch as _patch_mismatch
import orisys_studio.ui.cal_panel as _cal_panel_module
import orisys_studio.calibration_service as _cal_service_module
from PyQt5.QtWidgets import QMessageBox as _QMessageBox
from orisys_app_shared.force.calibration_io import (
build_force_calibration_v2 as _build_v2b,
write_force_calibration as _write_cal,
)
from orisys_app_shared.force.calibration_session import (
build_session_document as _build_session_doc,
)
record_dir = APP_ROOT / "outputs" / "_cal_record_check"
if record_dir.exists():
for stale in record_dir.iterdir():
stale.unlink()
record_dir.mkdir(parents=True, exist_ok=True)
identity_window = StudioMainWindow(
prefs={
"language": "zh_CN",
"theme": "dark",
"sensor_type": "box",
"source_mode": "camera",
}
)
identity_window.resize(1400, 880)
identity_window.show()
identity_window.cal_panel.set_running(True)
taken_points = []
for load, raw in ((1.0, 100.0), (3.0, 300.0)):
identity_window.cal_recorder.start(
channel="fn", reference_load_n=load, duration_s=1.0
)
for index in range(20):
identity_window.cal_recorder.feed(
_calibration_sample(index, fn_raw=raw)
)
taken_points.append(identity_window.cal_recorder.finish_now())
device_a = record_dir / "device_a_session.json"
device_a.write_text(
_cal_json.dumps(
_build_session_doc(
taken_points,
meta={
"sensor_type": "box",
"sn": "SN-A",
"config_ref": "D:/lab/deviceA/sensor.json",
},
)
),
encoding="utf-8",
)
identity_window.cal_recorder.clear_points()
# The live device is B; opening A's record must not restamp it.
panel = identity_window.cal_panel
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(device_a))
app.processEvents()
check(
panel.sn_edit.text() == "SN-A",
f"the reopened record's SN was overwritten: {panel.sn_edit.text()!r}",
)
meta = panel._export_meta()
check(meta["sensor_type"] == "box", "the record's sensor type was overwritten")
check(meta["sn"] == "SN-A", "the record's SN was not kept")
check(
meta["config_ref"] == "D:/lab/deviceA/sensor.json",
f"the record's config was overwritten: {meta['config_ref']!r}",
)
check(
str(meta.get("source_record", "")).endswith(device_a.name),
"the export does not reference the source record",
)
saved = sorted(record_dir.glob("calibration_session_box_*.json"))
with _patch_mismatch.object(
_cal_panel_module, "calibration_export_dir", lambda: record_dir
):
with _patch_mismatch.object(
_cal_service_module, "calibration_export_dir", lambda: record_dir
):
with _patch_mismatch.object(
_cal_panel_module.QMessageBox, "information", lambda *a, **k: None
):
panel._on_save_record()
saved = sorted(record_dir.glob("calibration_session_box_*.json"))
check(bool(saved), "saving a reopened record produced no file")
if saved:
saved_meta = _cal_json.loads(saved[-1].read_text(encoding="utf-8"))["meta"]
check(
saved_meta.get("sn") == "SN-A" and saved_meta.get("sensor_type") == "box",
f"the saved record was restamped: {saved_meta.get('sn')!r}",
)
doc, _warnings = _build_v2b(panel._prepared)
check(
doc["meta"]["sn"] == "SN-A" and doc["meta"]["sensor_type"] == "box",
"the exported coefficients carry the wrong device identity",
)
# A reopened point with only its samples can still be inspected.
reopened = identity_window.cal_recorder.points[0]
for entry in _build_session_doc([reopened])["points"]:
entry["trace"] = {}
dialog = panel.point_dialog(reopened)
check(dialog is not None, "a reopened point could not be inspected")
if dialog is not None:
check(
len(dialog.plot_widget.ci.items) == 3,
"the rebuilt diagnostic is not one trace per axis",
)
dialog.close()
# Applying a calibration that names another device is flagged and defaults
# to "no".
stuck_identity = _StuckWorker()
identity_window._ensure_workers = lambda: None
identity_window._box_worker = stuck_identity
identity_window._finger_worker = stuck_identity
coefficients = record_dir / "device_a_coefficients.json"
_write_cal(doc, coefficients)
check(
bool(
identity_window.calibration_target_mismatch(
str(coefficients),
sensor_type="finger",
config_ref="D:/lab/B/sensor.json",
)
),
"a calibration for another device was not reported as a mismatch",
)
check(
not identity_window.calibration_target_mismatch(
str(coefficients),
sensor_type="box",
config_ref="D:/lab/deviceA/sensor.json",
),
"matching identities were reported as a mismatch",
)
with _patch_mismatch.object(
_QMessageBox, "question", staticmethod(lambda *a, **k: _QMessageBox.No)
):
identity_window._on_calibration_apply(str(coefficients))
check(
not stuck_identity.calibration_requests,
"a mismatched calibration was applied after the user declined",
)
check(
identity_window.statusBar().currentMessage() == t("cal.apply_mismatch_skipped"),
f"the refusal was not reported: {identity_window.statusBar().currentMessage()!r}",
)
# A matching calibration still goes through the normal confirmation.
identity_window.sensor_picker.set_sensor_type("box")
with _patch_mismatch.object(
_QMessageBox, "question", staticmethod(lambda *a, **k: _QMessageBox.Yes)
):
identity_window._on_calibration_apply(str(coefficients))
check(
stuck_identity.calibration_requests == [str(coefficients)],
f"a matching calibration was not requested: {stuck_identity.calibration_requests}",
)
identity_window.close()
app.processEvents()
_shutil.rmtree(record_dir, ignore_errors=True)
section("display units and time base")
import math as _math
stuck_units = _StuckWorker()
units_window = StudioMainWindow(
prefs={
"language": "zh_CN",
"theme": "dark",
"sensor_type": "box",
"source_mode": "camera",
}
)
units_window.resize(1440, 880)
units_window.show()
units_window._ensure_workers = lambda: None
units_window._box_worker = stuck_units
units_window._finger_worker = stuck_units
units_window._start_run()
units_window._run_timer.stop()
units_window._set_phase("running")
units_plots = units_window.force_plots
units_window.toolbar.units_combo.setCurrentIndex(0) # newton
units_plots.reset()
wall = 1_700_000_000.0
for index in range(90):
stuck_units.emit(
_fake_packet(index, timestamp=wall + index / 30.0, mono_t=index / 30.0)
)
units_window._on_run_tick()
before = units_plots.history.state()
end_before = float(before.times[-1]) if before is not None else -1.0
check(
before is not None and end_before < 60.0,
f"the live X axis is not session-relative (ends at {end_before})",
)
for units_index in (1, 0, 1, 0):
units_window.toolbar.units_combo.setCurrentIndex(units_index)
app.processEvents()
after = units_plots.history.state()
check(
after is not None and abs(float(after.times[-1]) - end_before) < 1.0,
"the X axis jumped across a raw/N switch: "
f"{end_before} -> {None if after is None else after.times[-1]}",
)
check(
units_plots.history.sample_count > 0,
"the rebuilt history lost every sample",
)
units_window.close()
units_window._box_worker = None
stuck_units.release()
app.processEvents()
# A device with only its normal-force axis calibrated: FN in newtons, and the
# other two axes explicitly unavailable instead of showing raw counts.
stuck_one = _StuckWorker()
one_window = StudioMainWindow(
prefs={
"language": "zh_CN",
"theme": "dark",
"sensor_type": "box",
"source_mode": "camera",
}
)
one_window.resize(1440, 880)
one_window.show()
one_window._ensure_workers = lambda: None
one_window._box_worker = stuck_one
one_window._finger_worker = stuck_one
one_window._start_run()
one_window._run_timer.stop()
one_window._set_phase("running")
one_window.toolbar.units_combo.setCurrentIndex(0) # newton
one_window.force_plots.reset()
for index in range(10):
stuck_one.emit(
_fake_packet(index, calibrated_axes=("fn",), shear_n=float("nan"))
)
one_window._on_run_tick()
app.processEvents()
one_plots = one_window.force_plots
check(
one_plots._available == (True, False, False),
f"per-axis availability is wrong: {one_plots._available}",
)
check(
one_plots.value_labels[0].text() != t("chart.value_unavailable"),
"the calibrated axis was reported as unavailable",
)
check(
all(
one_plots.value_labels[index].text() == t("chart.value_unavailable")
for index in (1, 2)
),
"an axis without a curve showed a number in newton mode",
)
one_state = one_plots.history.state()
check(
one_state is not None
and _math.isfinite(float(one_state.values[:, 0].max())),
"the calibrated axis was not plotted",
)
check(
one_state is not None
and not any(_math.isfinite(v) for v in one_state.values[:, 1]),
"FX carried a value without a calibration curve",
)
one_window.close()
one_window._box_worker = None
stuck_one.release()
app.processEvents()
section("session browsing")
import numpy as np
stuck = _StuckWorker()
window = StudioMainWindow(
prefs={
"language": "zh_CN",
"theme": "dark",
"sensor_type": "box",
"source_mode": "camera",
}
)
window.resize(1440, 880)
window.show()
window._ensure_workers = lambda: None
window._box_worker = stuck
window._finger_worker = stuck
window._start_run()
window._run_timer.stop()
window._set_phase("running")
for index in range(1500):
stuck.emit(_fake_packet(index))
app.processEvents()
plots = window.force_plots
check(plots.mode() == "follow", "the charts do not start in live mode")
check(
plots.stats_label.text() == "",
"the selection readout is not blank while following live",
)
# Zoom and pan are available while following too: a chart that ignored the
# mouse could never be dragged into browsing in the first place.
check(
all(plot.getViewBox().state["mouseEnabled"][0] for plot in plots.plots),
"panning and zooming are disabled while following live",
)
check(
not any(cursor.movable for cursor in plots.cursors),
"the cursor is draggable while following live",
)
first, last = window.session_store.extent()
check(last > first, "the session store has no extent after recording")
plots.show_all()
check(plots.mode() == "browse", "the overview did not enter browse mode")
check(
all(plot.getViewBox().state["mouseEnabled"][0] for plot in plots.plots),
"browsing does not enable panning",
)
# "Whole session" is the default fit, not a selection: the cards still state
# the sample under the cursor, and the strip carries no selection summary.
check(
plots._selection is None,
"fitted to the whole session, the cards still hold selection statistics",
)
check(
plots.stats_label.text() == "",
f"a fitted session shows a selection summary: {plots.stats_label.text()!r}",
)
# Decimation must keep the spike while bounding the point count.
drawn = max(int(curve.getData()[0].size) for curve in plots.curves)
check(drawn <= 2 * plots._max_draw_points, f"overview drew {drawn} points")
spike = float(np.nanmax(plots.curves[0].getData()[1]))
check(spike > 90.0, f"the overview lost the spike: {spike}")
plots.set_window(10.0, 20.0)
check(plots.window() == (10.0, 20.0), f"set_window did not apply: {plots.window()}")
check(
plots.overview_curve.getData()[0].size > 0,
"the whole-session overview strip is empty",
)
# A window narrower than the session is a selection: its peak becomes the big
# number on each card and its mean/σ the line under it, while the strip keeps
# only the size of the selection (the long per-channel line did not fit).
check(plots._selection is not None, "zooming in did not record a selection")
summary = plots.stats_label.text()
check(summary != "", "the strip does not state the size of the selection")
check(
"峰" not in summary and "均" not in summary,
f"the per-channel statistics are still in the strip: {summary!r}",
)
check(
"选区" in summary or "Selection" in summary,
f"the strip does not say that the numbers describe a selection: {summary!r}",
)
expected_peak = float(plots._selection.peaks[0])
peak_text = (
f"{expected_peak:.3f}" if plots.unit() == "N" else f"{expected_peak:.1f}"
)
check(
plots.value_labels[0].text() == peak_text,
f"the FN card is not the selection's peak: {plots.value_labels[0].text()!r}",
)
check(
plots.value_details[0].text() != ""
and ("均" in plots.value_details[0].text() or "mean" in plots.value_details[0].text()),
f"the card does not state the selection's mean: "
f"{plots.value_details[0].text()!r}",
)
for card, label, detail in zip(
plots.value_cards, plots.value_labels, plots.value_details
):
check(
detail.width() >= detail.sizeHint().width(),
f"the selection detail line is clipped: "
f"{detail.width()} < {detail.sizeHint().width()}",
)
# Leaving the selection hands the cards back to the cursor readout.
plots.follow_latest()
check(plots._selection is None, "the selection survived 'back to latest'")
check(
plots.value_details[0].text() == "",
f"the mean line survived leaving the selection: "
f"{plots.value_details[0].text()!r}",
)
# A user gesture (the signal pyqtgraph emits for a wheel/drag) enters browse;
# "back to latest" must actually leave it.
plots._syncing = False
view_box = plots.plots[0].getViewBox()
view_box.setXRange(5.0, 8.0, padding=0.0)
view_box.sigRangeChangedManually.emit([True, False])
app.processEvents()
check(plots.mode() == "browse", "a user pan did not enter browse mode")
check(
plots.window() == (5.0, 8.0),
f"the dragged window was not applied: {plots.window()}",
)
check(
all(cursor.movable for cursor in plots.cursors),
"the cursor is not draggable while browsing",
)
plots.follow_latest()
check(plots.mode() == "follow", "back-to-latest did not restore live mode")
check(
not any(cursor.movable for cursor in plots.cursors),
"the cursor stayed draggable after leaving browsing",
)
# A programmatic (or X-link propagated) range change is not a gesture.
plots._syncing = False
view_box.setXRange(1.0, 2.0, padding=0.0)
app.processEvents()
check(
plots.mode() == "follow",
"a programmatic range change switched the view into browsing",
)
plots.set_cursor(3.5, emit=True)
check(
[round(cursor.value(), 3) for cursor in plots.cursors] == [3.5, 3.5, 3.5],
"the time cursor is not shared across the three channels",
)
# Dragging the shared cursor moves the playhead and notifies the listeners.
plots.show_all()
app.processEvents()
dragged_cursor: list[float] = []
plots.cursor_moved.connect(dragged_cursor.append)
plots.cursors[0].setValue(2.5)
app.processEvents()
check(
[round(cursor.value(), 3) for cursor in plots.cursors] == [2.5, 2.5, 2.5],
"dragging one cursor did not move the shared cursor",
)
check(
dragged_cursor and dragged_cursor[-1] == 2.5,
f"the cursor drag did not notify the other views: {dragged_cursor}",
)
# History browsing is a display pause: capture and recording keep going.
plots.show_all()
before = window.session_store.count
for index in range(1500, 1560):
stuck.emit(_fake_packet(index))
check(
window.session_store.count > before,
"browsing history stopped the record channel",
)
# Unit switches rebuild from the attached source without leaving live mode.
plots.follow_latest()
plots._unit = "raw"
window._on_units_changed("newton")
app.processEvents()
check(
plots.mode() == "follow",
f"switching units left the charts in browsing: {plots.mode()}",
)
check(plots._source_ready, "switching units detached the curve source")
check(plots._unit == "N", f"the unit switch did not reach the charts: {plots._unit}")
window.close()
app.processEvents()
stuck.release()
section("views and their source pickers")
stuck = _StuckWorker()
window = StudioMainWindow(
prefs={
"language": "zh_CN",
"theme": "dark",
"sensor_type": "finger",
"source_mode": "camera",
"show_finger_3d": True,
}
)
window.resize(1600, 900)
window.show()
window._ensure_workers = lambda: None
window._box_worker = stuck
window._finger_worker = stuck
app.processEvents()
# The 3D panel needs a finger sensor and its own toggle, and no layout preset
# is involved any more.
check(window.mesh3d_panel.isVisible(), "the finger 3D panel is hidden in the live view")
check(window.act_finger_3d.isEnabled(), "the 3D toggle is disabled for a finger sensor")
check(
not hasattr(window, "_layout_preset"),
"the removed layout preset is still part of the window",
)
check(
not hasattr(window, "act_layout_2d"),
"the removed layout menu entry is still part of the window",
)
# Live: the camera picker; replay: the recording picker and no camera.
check(
window.toolbar.camera_group.isVisible(),
"the live view does not show the camera picker",
)
window.brand_bar.btn_playback.click()
for _ in range(3):
app.processEvents()
check(
not window.toolbar.camera_group.isVisible()
and not window.toolbar.video_group.isVisible(),
"the replay view still shows the live capture pickers",
)
window.brand_bar.btn_live.click()
for _ in range(3):
app.processEvents()
check(
window.toolbar.camera_group.isVisible(),
"returning to the live view did not restore the camera picker",
)
# A video file is a *replay* source: the live view offers the way back to the
# camera, and the video is named only in the replay view.
window.toolbar.set_video_path("C:/tmp/demo.mp4")
window._set_source_mode("video")
for _ in range(3):
app.processEvents()
check(
window.toolbar.video_group.isVisible() and not window.toolbar.camera_group.isVisible(),
"video mode still offers the camera group",
)
check(
"demo.mp4" not in window.toolbar.source_summary.text(),
f"the live view names a replay's video: "
f"{window.toolbar.source_summary.text()!r}",
)
window.brand_bar.btn_playback.click()
for _ in range(3):
app.processEvents()
check(
"demo.mp4" in window.toolbar.source_summary.text(),
f"the rail does not name the video file: {window.toolbar.source_summary.text()!r}",
)
window.brand_bar.btn_live.click()
for _ in range(3):
app.processEvents()
check(
"demo.mp4" not in window.toolbar.source_summary.text(),
"the video name stayed on the rail after leaving the replay",
)
window.toolbar.use_camera_clicked.emit()
for _ in range(3):
app.processEvents()
check(
window.toolbar.source_mode() == "camera" and window.toolbar.video_path() == "",
"the way back to the camera did not take effect",
)
window.close()
app.processEvents()
stuck.release()
section("replay")
replay_root = APP_ROOT / "outputs" / "_replay_check"
_shutil.rmtree(replay_root, ignore_errors=True)
replay_root.mkdir(parents=True, exist_ok=True)
from orisys_app_shared.record import ExperimentRecorder, RecordConfig
recorder = ExperimentRecorder(
RecordConfig(video=True, fps=30.0, segment_seconds=30.0)
)
recording = recorder.start(
root=replay_root,
session_id=1,
metadata={
"device": {
"sensor_type": "box",
"config_ref": str(replay_root / "sensor.json"),
"config_sha256": "recorded",
},
"versions": {"sdk": "0.5.1"},
"depth": {"stored": False, "replay": "recompute"},
},
)
(replay_root / "sensor.json").write_text('{"camera": {}}\n', encoding="utf-8")
for index in range(40):
recorder.on_packet(_fake_packet(index))
recorder.stop()
from orisys_studio.replay import ReplayController
# The replay lives in the main window: one engine, one progress bar, and the
# charts read the recording itself.
replay_window = StudioMainWindow(
prefs={
"language": "zh_CN",
"theme": "dark",
"sensor_type": "box",
"source_mode": "camera",
}
)
replay_window.resize(1440, 880)
replay_window.show()
app.processEvents()
replay_window.brand_bar.btn_playback.click()
app.processEvents()
check(
replay_window.toolbar.btn_replay_video.isVisible(),
"the replay view does not offer a recorded video",
)
check(
not replay_window.toolbar.camera_group.isVisible(),
"the replay view still offers the camera picker",
)
check(
replay_window._load_recording(recording),
"the replay view could not open the recording",
)
app.processEvents()
replay_window.toolbar.set_video_path(str(replay_root / "raw_0001.mp4"))
replay_window._refresh_status_rail()
check(
"raw_0001.mp4" in replay_window.toolbar.source_summary.text(),
f"the replay does not name its video: "
f"{replay_window.toolbar.source_summary.text()!r}",
)
replay = replay_window.replay
replay_plots = replay_window.force_plots
check(
replay.position() == 0.0,
f"replay did not start at the beginning: {replay.position()}",
)
check(
all(int(curve.getData()[0].size) == 40 for curve in replay_plots.curves),
"replay did not draw every force sample",
)
check(
replay_plots.is_recording_source() and replay_plots.mode() == "browse",
"the charts do not browse the recording",
)
check(
replay_plots.mode_label.text() == t("chart.mode_replay"),
f"the charts do not say they are replaying: {replay_plots.mode_label.text()!r}",
)
check(
not replay_plots.btn_latest.isEnabled(),
"a recording offers 'back to latest'",
)
# The chart cursor is the playhead, and the transport moves it at 1x.
check(
replay_window.toolbar.run_role() == "playback",
"the rail's primary button is not the replay's transport",
)
check(replay_window.toolbar.btn_run.isEnabled(), "the replay has no transport")
check(
not replay_window.toolbar.btn_run.isChecked(),
"the replay starts while playing",
)
check(
replay_window.toolbar.btn_run.text() == t("toolbar.playback"),
f"the button does not say it plays: {replay_window.toolbar.btn_run.text()!r}",
)
replay_window.toolbar.btn_run.setChecked(True)
app.processEvents()
check(
replay_window.toolbar.btn_run.text() == t("toolbar.playback_pause"),
f"the button does not offer pause: {replay_window.toolbar.btn_run.text()!r}",
)
started = time.perf_counter()
for _ in range(12):
app.processEvents()
time.sleep(0.03)
elapsed = time.perf_counter() - started
replay_window.toolbar.btn_run.setChecked(False)
app.processEvents()
check(
replay.position() > 0.05,
f"the transport did not advance the playhead: {replay.position()}",
)
check(
replay.position() <= elapsed + 0.2,
f"the playhead ran ahead of the clock: {replay.position()} > {elapsed}",
)
check(
abs(float(replay_plots.cursors[0].value()) - replay.position()) < 0.05,
"the chart cursor did not follow the transport",
)
check(
not replay_window.toolbar.btn_run.isChecked(),
"the transport stayed checked after pause",
)
# A manual gesture on the timeline takes over: playback pauses.
replay_window.toolbar.btn_run.setChecked(True)
app.processEvents()
replay_plots.cursors[0].setValue(0.25)
app.processEvents()
check(
not replay_window.toolbar.btn_run.isChecked(),
"dragging the cursor did not pause the playback",
)
check(
abs(replay.position() - 0.25) < 0.05,
f"dragging the chart cursor did not scrub the replay: {replay.position()}",
)
check(
replay.video_image() is not None,
"no video frame was rendered at the cursor position",
)
check(
replay_window.arrow_panel.view_modes() == ("recorded",),
"the 2D panel still offers the live-only views while replaying",
)
# Long-history browsing: an interval read is bounded by the interval, and the
# whole-session envelope is a band rather than every row.
from orisys_app_shared.force.source import RecordingCurveSource
from orisys_app_shared.record.index import RecordingIndex
from orisys_app_shared.record.playback import RecordingReader as _RecordingReader
reader = _RecordingReader(recording, max_rows=10)
source = RecordingCurveSource(reader)
check(source.kind == "disk", "the replay source is not the recording")
index = reader.index()
check(index.rows == 40, f"the index found {index.rows} rows, not 40")
start, stop = index.row_range(0.5, 1.0)
check(
0 <= start < stop <= index.rows and (stop - start) <= 20,
f"an interval read spans {stop - start} rows",
)
interval = source.window(0.5, 1.0, unit="N")
check(0 < len(interval) <= 20, f"the interval read returned {len(interval)} rows")
envelope = source.envelope(buckets=8, unit="N")
check(0 < len(envelope) <= 8, f"the envelope has {len(envelope)} points")
check(
reader.info().total_rows == 40,
f"the recording reports {reader.info().total_rows} rows",
)
check(
t("playback.warn_config_changed").split("{")[0].strip()[:4]
in replay.warnings_text(),
f"the replay notes are missing: {replay.warnings_text()!r}",
)
check(
t("playback.warn_config_changed").split("{")[0].strip()[:4]
in replay_window.toolbar.source_summary.toolTip(),
f"the replay notes are not reachable from the rail: "
f"{replay_window.toolbar.source_summary.toolTip()!r}",
)
moved: list[float] = []
replay.position_changed.connect(moved.append)
replay.set_position(1.0)
app.processEvents()
check(
[round(line.value(), 2) for line in replay_plots.cursors] == [1.0, 1.0, 1.0],
"the replay cursor is not shared across the three charts",
)
check(moved == [1.0], f"the cursor signal did not fire once: {moved}")
check(
abs(replay_plots.cursor_time() - 1.0) < 0.05,
f"the chart cursor did not follow a programmatic position: "
f"{replay_plots.cursor_time()}",
)
# The chart's unit chip rebuilds the recording's curves, not the live ones.
replay_plots.unit_badge.click()
app.processEvents()
check(
replay.unit() == "N" and replay_plots._unit == "N",
f"the unit chip did not reach the replay: {replay.unit()}",
)
# Playing means decoding a frame per tick. Reopening and seeking the file for
# every frame (10 ms for a small clip, hundreds for 1080p H.264) is what made
# the video stand still; the reader keeps one decoder open and steps it.
span = max(0.1, replay.duration_s() - 0.05)
seeks: list[float] = []
for fraction in (0.0, 0.25, 0.5, 0.75, 1.0):
moment = fraction * span
started_at = time.perf_counter()
frame_image, _frame = replay.reader().video_frame_at(moment)
seeks.append((time.perf_counter() - started_at) * 1000.0)
check(frame_image is not None, f"no frame at {moment:.2f}s")
check(
max(seeks) < 250.0,
f"a seek costs {max(seeks):.0f} ms, so playback would stall",
)
steps: list[float] = []
for index in range(1, 8):
started_at = time.perf_counter()
replay.reader().video_frame_at(index / 30.0)
steps.append((time.perf_counter() - started_at) * 1000.0)
check(
max(steps) < 60.0,
f"stepping one frame costs {max(steps):.0f} ms, so 30 fps cannot hold",
)
# There is no depth recomputation any more: the engine was deleted with
# the replay bar (depth is not stored with a recording, and the replay
# view shows the recorded frames). The arrow panel in the replay view
# therefore offers the recorded image and nothing else.
check(
replay_window.arrow_panel.view_modes() == ("recorded",),
f"the replay still offers a depth pane: "
f"{replay_window.arrow_panel.view_modes()}",
)
# The recompute mode: off (default) the charts read the recorded CSV; on, the
# recorded video is run through the pipeline instead. The run itself is not
# started here (that needs the SDK), only the wiring around it.
started: list = []
replay_window._start_run = lambda: started.append(True)
replay_window.act_recompute.setChecked(True)
for _ in range(3):
app.processEvents()
check(
started == [True],
"the recompute mode did not hand the video to the pipeline",
)
check(
not replay_plots.is_recording_source(),
"the charts still read the CSV while recomputing",
)
check(
replay_window.toolbar.source_mode() == "video"
and replay_window.toolbar.video_path().endswith(".mp4"),
f"the recompute mode did not select the video: "
f"{replay_window.toolbar.source_mode()!r} / "
f"{replay_window.toolbar.video_path()!r}",
)
replay_window._recompute_running = True
replay_window.act_recompute.setChecked(False)
for _ in range(3):
app.processEvents()
check(
replay_plots.is_recording_source(),
"turning the recompute off did not restore the recorded curves",
)
check(
replay_window.toolbar.source_mode() == "camera" or not replay_window._running,
"the recompute pass was left running",
)
# Leaving the replay view returns the charts to the live session and hides
# the bar: the replay stays available, it just stops being on screen.
check(
replay_window._load_recording(recording),
"the replay view could not be re-opened",
)
app.processEvents()
replay_window.brand_bar.btn_live.click()
app.processEvents()
check(
not replay_plots.is_recording_source() and replay_plots.mode() == "follow",
"leaving the replay view did not restore the live charts",
)
check(
replay_window.toolbar.run_role() != "playback",
"the rail kept the replay's transport in the live view",
)
check(
not replay_window.toolbar.source_summary.text().endswith(".mp4")
and "raw_0001" not in replay_window.toolbar.source_summary.text(),
f"the replay's video is still named in the live view: "
f"{replay_window.toolbar.source_summary.text()!r}",
)
check(
replay_window.arrow_panel.view_modes() == ("arrows_pure", "arrows_stitched", "depth"),
"the 2D panel did not return to its live views",
)
check(
replay_window.replay.is_loaded(),
"leaving the view dropped the open recording",
)
# The replay engine is inside the shell now: there is no second window.
check(
not hasattr(replay_window, "_playback"),
"the window still keeps a replay dialog reference",
)
check(
not hasattr(replay_window, "_open_playback"),
"the window still opens a separate replay window",
)
replay_window.replay.close()
replay_window.close()
app.processEvents()
_shutil.rmtree(replay_root, ignore_errors=True)
print(f"\nscenes: {scene_count} checks: {CHECKS} failures: {len(FAILURES)}")
for failure in FAILURES:
print(f" FAIL: {failure}")
return 1 if FAILURES else 0
if __name__ == "__main__":
raise SystemExit(main())