Files
orisys-app/apps/studio/tools/ui_check.py
T
ykcheungab 7d1fca4c93 fix(studio-3d): 字段级来源与失效状态 —— 旧光流/算法失败不再冒充实时测量
- What:
  1. **采集侧声明每个派生数组的归属**(`acquisition/models.py`):`SdkPacket` 新增
     `flow_valid/flow_frame_id/depth_valid/depth_frame_id/forces_valid`(`None` = 未声明)。两个真实
     生产者都填写:`finger_worker`/`sdk_worker` 逐帧记录"该数组实际算出来的 frame_id",链失败时
     `*_valid=False` 并把上一次成功帧号带出来。光流与形变链分开判定(力读取失败不等于光流失败),
     但算法链失败时本帧的一切派生量都不可声明。
  2. **adapter 按声明收数据**(`scene3d/adapters.py`):`_field_validity()` 三态——未声明 → 拒绝并报
     `field_unverified`(没声明就不算数);声明为假 → `field_stale`(把 `invalid_reason` 带进诊断);
    声明帧号不是本帧 → `field_stale`(两个帧号都写进 details)。`_intersect_device_mask` 换成
     `_device_mask`:设备 mask 形状不符**拒绝该帧**(`field_domain_mismatch`,两个形状都写进诊断),
     既不拉伸也不静默忽略。发布的帧带 `computed_frame_id`,身份一路带到显示侧。
  3. **契约**(`scene3d/contracts.py`):`SceneState` 加 `"stale"`;诊断码加 `field_stale`/`field_unverified`
     (warning);`SurfaceFieldFrame.computed_frame_id` 与 `RenderFrame.computed_frame_id`(由 composer
     原样透传);`validate_surface_frame` 校验帧号非负且不晚于所属帧。新增 locale:
     `scene3d.state_stale`(中英各一条)。
  4. **协调器不再"边失败边报 live"**(`scene3d/coordinator.py`):新增 `_stale_since` 宽限状态机与
     `note_field_failure()`——字段被拒、帧被判 invalid、compose 抛错、或合成结果的 `computed_frame_id`
     落后于本帧,都立即把状态改为 `stale` 并保留最后一帧;宽限(`stale_grace_s`,默认 1.0 s)到期后
     回参考模型并转 `no_data`;收到好帧即回 `live` 并清掉宽限。顺带修掉"等待首帧时每个 tick 都把参考
     模型重新 present 一次"(`_rest_presented`,16 ms 一次的无谓上传)。
  5. **诊断真正到达界面**(`scene3d/view.py`):`submit_packet()` 在 adapter 拒绝时把
     `adapter.last_diagnostic` 交给 `note_field_failure()`;`tick()` 把协调器诊断的变化经已有
     `diagnosticRaised` 冒泡(协调器自身仍然零 Qt 依赖)。新增 `_snapshot_diagnostic()` 与 `_adapter`
     引用(后续"按能力禁用控件"也要问它)。
- Why: 审查报告 P1/R2:`FingerFrameAdapter.snapshot()` 从不读 `valid`/`stale_forces`/`invalid_reason`,
  只看到有限的 `flow` 就当作新数据上色(复现:`valid=False, stale_forces=True` + 有限 flow → 仍有有效
  着色点、diagnostic=None);compose 抛错只写 `last_diagnostic`,画面与 `state` 保持不变;
  `snapshot()` 返回 None 时诊断无人读取;mask 域不符时静默忽略会扩大可信范围。力读数失败不该机械地
  判死三维,但算法失败也不能因为旧数组是有限值就当成新数据——所以按**字段级来源**处理,并让失效有
  明确的、可见的状态而不是继续保持 live。
- Verification:
  · `test_scene3d_adapters.py` 23 通过(+6 新增):stale flow 被拒且带原因、未声明 flow → `field_unverified`、
  帧号不符 → `field_stale`(details 两个帧号)、新帧记录 `computed_frame_id`、box 侧"形变链失败仍用光流"、
  box 全部字段不可声明时拒绝;mask 域不符由"被忽略"改为"拒绝该帧"(原断言已改)。
  · `test_scene3d_coordinator.py` 30 通过(+6 新增):落后帧号的合成结果永不上屏且状态 stale、
  字段被拒 → 宽限内保留画面/宽限后回参考模型/好帧回 live、compose 抛错不再保持 live、invalid 帧也计入
  宽限。
  · `test_scene3d_view.py`(+3 新增,offscreen 可跑):adapter 拒绝 → 协调器 stale 且诊断经
  `diagnosticRaised` 回到面板、`scene3d.state_stale` 不是裸键;未声明字段 → `field_unverified`;
  新鲜字段照常上屏回到 live。
  · 全量:apps/studio 228 项 0 失败 14 跳过;shared 301 项 0 失败 1 跳过(`--junitxml` 计数)。
- Not in this change: `docs/architecture/studio-3d-contracts.pyi` 与新契约的同步放在阶段 6 的文档提交里
  一起做(本阶段契约改动较多,避免每提交改一次文档);设备几何身份(R3)、资产加载请求号(R7)、
  二维后端统一、限速与真实统计均未动;真机上"链失败时 VRAW 是否也过期"这一条仍需实测(若证明链失败
  不影响 VRAW,`flow_valid` 放宽为只看读取结果是一行改动,已写进计划的风险项)。
2026-09-14 21:03:07 +08:00

3402 lines
128 KiB
Python
Raw Blame History

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