Files
orisys-app/apps/studio/tools/ui_check.py
T
ykcheungab 9b3c5fe46c 回放与实时统一为一套界面;修复 30 fps 设备被以 16.7 fps 消费的计时器缺陷 - What: 1) 删除 ui/replay_bar.py(进度条、录制目录选择、重算深度按钮、提示行)与回放条的全部接线:历史回放现在与实时监测同布局(分割比例同为 0,360,1080,0)、同一条状态轨、同一套曲线手势,唯一差别是来源槽——实时是摄像机,回放是「选择录制视频…」(toolbar.btn_replay_video,与文件菜单「打开视频…」同一条 _load_replay_video 路径:视频所在目录即录制目录,曲线自动配对同目录 forces.csv)。时间轴改为曲线游标本身:表头新增播放/暂停图标按钮(icons.play/pause,空格键同效)以 1× 推进(按墙钟计时,漏拍不变慢),到末尾自动停;拖动游标、框选、滚轮平移或拖总览条一律自动暂停(ForcePlotsPanel.pause_playback)。回放的降采样/跨版本提示移入来源摘要 tooltip,打开录制时另发一条 6 秒通知。修复视频文件名残留在实时视图的问题:_refresh_status_rail 只在 _view_preset == "replay" 时写视频名。2) 录制格式不变:raw_*.mp4 与 forces.csv/frames.csv/session.json 继续一起写。按「两者都要」新增 视图 → 由视频重算力值(act_recompute):默认曲线读录制的 CSV,打开后把录制视频重新跑一遍视觉管线(进 SessionStore、2D 面板交给管线绘制、只能顺序前进,播放按钮即开始/暂停,关闭即挂回 CSV),启动路径唯一(set_playing → _on_chart_play_toggled → _start_run),并让文件源的时间轴改为视频自身时间(mono_t = 帧号 × 1/fps),否则重算结果会落在与 CSV 不同的横轴上。回放 2D 面板不再提供深度视图(REPLAY_VIEW_MODES=("recorded",)),离线深度重算引擎保留但无界面入口。3) 采集只有 16.67 fps 的根因是上一轮引入的等待:Windows 默认计时器分辨率 15.6 ms,time.sleep(0.002) 实测睡 15.5 ms,重复帧等两拍即跨过 30 fps 的 33 ms 周期,30 fps 设备被以 ~16.7 fps 消费并隔帧丢弃(配置本身正确:mini/sensor.json 写 camera.fps=30;离线实测整套管线 15.2 ms/帧≈66 fps,余量充足)。新增 _high_resolution_timer() 在会话期间把分辨率提到 1 ms(结束还原),并把重复帧等待改为 min(距下一帧时间, 半个周期)、到期只 sleep(0)。 - Why: 用户评审 3 点——回放界面与实时不一致且拖拽条效果一般、不想为回放录制深度信息、想同时保留视频与 CSV 记录、以及采集帧率只有 16.67 fps(应为 30)。 - Verification: ui_check.py 562 项 0 失败(新增两视图同布局/无 bar、来源槽随视图切换且无残留、播放暂停与手动手势暂停、重算模式切换与视频源、回放提示在 tooltip);test_ui_smoke.py 73 项 0 失败(新增视图统一、传输控制、重算模式);drive_shared.py 150 项 0 失败(新增 shared/tests/test_worker_frame_rate.py::test_a_file_source_runs_on_the_video_s_own_timeline,并把 30 fps 用例改成必须落在 27–31.5 fps;同文件实测无 1 ms 计时器 21.6 fps、有则 30.0 fps,同一帧返回 3 次时 11317:39);locale_studio.py 266 键齐全、zh==en;reproduce.py 退出 0;ui_snapshot.py 三张复核图(统一后的回放、实时、选区中英)。 - Not in this change: 录制文件格式与 CSV 列序不变;离线深度重算引擎(ReplayController.request_depth / DepthRecomputer)保留并有独立测试,但界面已无入口,可下一步删除;重算模式用有损 mp4,数值与原 CSV 会有差异(这正是它要显式呈现的)。
2026-09-13 22:04:01 +08:00

3214 lines
120 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.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
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_plots.btn_play.isEnabled(), "the replay has no transport")
check(not replay_plots.btn_play.isChecked(), "the replay starts while playing")
replay_plots.btn_play.setChecked(True)
started = time.perf_counter()
for _ in range(12):
app.processEvents()
time.sleep(0.03)
elapsed = time.perf_counter() - started
replay_plots.btn_play.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_plots.btn_play.isChecked(), "the transport stayed checked after pause")
# A manual gesture on the timeline takes over: playback pauses.
replay_plots.btn_play.setChecked(True)
app.processEvents()
replay_plots.cursors[0].setValue(0.25)
app.processEvents()
check(
not replay_plots.btn_play.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_depth_recomputed").split("{")[0].strip()[:4]
in replay.warnings_text(),
f"the replay notes are missing: {replay.warnings_text()!r}",
)
check(
t("playback.warn_depth_recomputed").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()}",
)
# The depth recomputation must never run on the UI thread. A stub that blocks
# proves the call returns immediately instead of freezing the window.
import threading as _threading
import time as _time
class _Blocking:
released = _threading.Event()
def preflight(self) -> str:
return ""
def depth_at(self, _t):
_Blocking.released.wait(timeout=5.0)
return type("R", (), {"ok": False, "error": "stub", "depth": None})()
def close(self) -> None:
return None
replay._recomputer = _Blocking() # noqa: SLF001 - exercising the contract
started = _time.perf_counter()
replay.request_depth()
elapsed = _time.perf_counter() - started
check(elapsed < 0.5, f"the depth recompute blocked the UI for {elapsed:.2f}s")
check(
replay._depth_thread is not None and replay._depth_thread.is_alive(),
"the depth recompute did not run on a worker thread",
)
check(
replay._depth_thread.daemon,
"the depth worker is not a daemon, so a blocked SDK call would keep the "
"process alive at exit",
)
replay.request_depth() # a second request is refused, not queued
check(
replay.depth_message() == t("playback.depth_busy"),
f"a concurrent recompute was not refused: {replay.depth_message()!r}",
)
_Blocking.released.set()
for _ in range(40):
app.processEvents()
_time.sleep(0.05)
if not replay._depth_busy(): # noqa: SLF001
break
check(not replay._depth_busy(), "the depth worker never finished")
# Deliver the worker result deterministically instead of waiting for the
# 120 ms poll timer.
replay._poll_depth()
app.processEvents()
check(
t("playback.depth_failed").split("{")[0].strip()[:4] in replay.depth_message(),
f"a failed recomputation was not reported: {replay.depth_message()!r}",
)
check(
replay.depth_image() is None,
"a failed recomputation left an image in the depth pane",
)
# The replay view no longer offers a depth pane: depth is not stored with a
# recording, and re-running the pipeline over the video is the recompute
# mode's job (see the `recompute` section).
check(
replay_window.arrow_panel.view_modes() == ("recorded",),
f"the replay still offers a depth pane: "
f"{replay_window.arrow_panel.view_modes()}",
)
# The plan must state where to start, which reset it is anchored to and
# whether the state is really recoverable; and a request must belong to the
# recording it started in.
from orisys_app_shared.record.depth import (
STATE_APPROXIMATE,
STATE_PREROLL,
DepthRecomputer,
)
from orisys_app_shared.record.playback import RecordingReader
plan_reader = RecordingReader(recording)
planner = DepthRecomputer(plan_reader)
frames = plan_reader.info().frames
check(bool(frames), "the replay recording has no frame index")
plan = planner.plan_for(frames[10])
check(
plan.state == STATE_APPROXIMATE and plan.anchor == "segment_start",
f"a recording with no reset/pre-roll is not reported as approximate: {plan.state}",
)
check(
plan.steps and plan.steps[0].source == plan_reader.segment_path(frames[10].segment),
"the plan does not name the video to read",
)
check(
"state_approximate" in plan.warnings,
f"the approximate replay is not reported: {plan.warnings}",
)
# A pre-roll makes the start of a recording reproducible.
session_path = recording / "session.json"
session_doc = _json.loads(session_path.read_text(encoding="utf-8"))
prerecord = recording / "preroll_0001.mp4"
prerecord.write_bytes(b"")
session_doc["preroll_frames"] = 30
session_doc["preroll_segment"] = prerecord.name
session_path.write_text(_json.dumps(session_doc), encoding="utf-8")
planner.close()
planner = DepthRecomputer(RecordingReader(recording))
plan = planner.plan_for(planner.reader.info().frames[5])
check(
plan.state == STATE_PREROLL and plan.preroll == prerecord,
f"the pre-roll was not used for the start of the recording: {plan.state}",
)
check(plan.frames_to_read == 30 + 6, f"unexpected replay cost: {plan.frames_to_read}")
planner.close()
# A reset boundary is re-applied, and a request is bound to its recording.
_Blocking.released.clear()
second_controller = ReplayController()
second_controller.open_recording(recording)
serial_before = second_controller._record_serial
blocking = _Blocking()
second_controller._recomputer = blocking
second_controller.request_depth()
check(
second_controller._depth_pending_record == serial_before,
"the request has no recording id",
)
check(
second_controller._depth_pending_seq is not None,
"the request carries no sequence number",
)
# Opening another recording invalidates the running request...
second_controller.open_recording(recording)
check(
second_controller._record_serial == serial_before + 1,
"opening a recording did not invalidate the running request",
)
check(
second_controller._recomputer is not blocking,
"the invalidated processor was not replaced",
)
_Blocking.released.set()
for _ in range(40):
app.processEvents()
_time.sleep(0.05)
if not second_controller._depth_busy():
break
second_controller._poll_depth()
app.processEvents()
check(
second_controller.depth_image() is None,
"an answer for a previous recording was drawn into the new one",
)
second_controller.close()
app.processEvents()
# 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(
not replay_plots.btn_play.isEnabled(),
"the live view still offers the replay's transport",
)
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())