#!/usr/bin/env python3 """Dependency-free smoke run for the studio UI. ``pytest`` is the canonical runner (``apps/studio/tests``), but this script needs nothing beyond the app's own runtime dependencies, so it can be used in minimal environments: python tools/ui_check.py It builds the real main window offscreen and asserts the layout, theming and status-feedback invariants that the UI work depends on. """ from __future__ import annotations import os import sys import threading import time from pathlib import Path APP_ROOT = Path(__file__).resolve().parents[1] REPO_ROOT = APP_ROOT.parents[1] for _path in (APP_ROOT / "src", REPO_ROOT / "shared" / "src"): if _path.is_dir() and str(_path) not in sys.path: sys.path.insert(0, str(_path)) os.environ.setdefault("QT_QPA_PLATFORM", "offscreen") if sys.platform == "win32" and "QT_QPA_FONTDIR" not in os.environ: _fonts = Path(os.environ.get("WINDIR", r"C:\Windows")) / "Fonts" if _fonts.is_dir(): os.environ["QT_QPA_FONTDIR"] = str(_fonts) os.chdir(APP_ROOT) from orisys_app_shared.acquisition.sinks import RecordSinkMixin # noqa: E402 from orisys_app_shared.record import FORCES_HEADER # noqa: E402 FAILURES: list[str] = [] CHECKS = 0 def check(condition: bool, message: str) -> None: global CHECKS CHECKS += 1 if not condition: FAILURES.append(message) def _inside_scroll_area(widget) -> bool: """True when ``widget`` lives inside a QScrollArea's content.""" from PyQt5.QtWidgets import QScrollArea node = widget while node is not None: if isinstance(node, QScrollArea): return True node = node.parentWidget() if hasattr(node, "parentWidget") else None return False def rail_squeezed(window) -> list[str]: """Status-rail widgets showing less than their own text. The rail shares one line with the session state, so every item that states a fact (``摄像机 0``, ``开始记录``) must keep its full text; only the labels that carry a tooltip are allowed to shrink. An elided button is how a readable instrument turns back into a jumble. """ layout = window.toolbar.layout().itemAt(0).layout() squeezed: list[str] = [] for index in range(layout.count()): widget = layout.itemAt(index).widget() if widget is None or not widget.isVisible(): continue if getattr(widget, "minimumSizeHint", None) is None: continue if widget.minimumSizeHint().width() <= 0: continue if widget.width() < widget.sizeHint().width() - 1: squeezed.append( f"{type(widget).__name__}" f"({getattr(widget, 'objectName', lambda: '')()}" f" {widget.width()}<{widget.sizeHint().width()})" ) return squeezed def section(name: str) -> None: print(f"\n== {name} ==") #: Sequences that only appear when UTF-8 text was decoded with a single-byte #: codepage and re-encoded. A PowerShell ``Get-Content -Raw | Set-Content`` #: round trip does that and silently mangles every non-ASCII string, which is #: how the Chinese UI once ended up unreadable. #: #: Written as escapes, with no literal examples in the comments, so that this #: guard cannot itself be corrupted by the same mistake and so that a scan of #: the tree for these sequences stays clean. MOJIBAKE_MARKERS = ( "\u00c3", "\u00c2", "\u00e2\u20ac", "\u00e3\u20ac", "\u00ef\u00bf\u00bd", ) def audit_locale_encoding() -> list[str]: """Locale files must be BOM-free UTF-8 and not mojibake.""" import json problems: list[str] = [] locales = APP_ROOT / "config" / "locales" for name in ("zh_CN", "en_US"): path = locales / f"{name}.json" if not path.is_file(): problems.append(f"{name}.json is missing") continue raw = path.read_bytes() if raw.startswith(b"\xef\xbb\xbf"): problems.append(f"{name}.json starts with a UTF-8 BOM") try: text = raw.decode("utf-8") except UnicodeDecodeError as exc: problems.append(f"{name}.json is not valid UTF-8: {exc}") continue for marker in MOJIBAKE_MARKERS: if marker in text: problems.append( f"{name}.json contains {marker!r} — written by a tool that " "decoded UTF-8 with the wrong codepage" ) break try: json.loads(text) except json.JSONDecodeError as exc: problems.append(f"{name}.json is not valid JSON: {exc}") # A corrupted zh_CN.json still parses, so check that it holds Chinese text. try: import json as _json zh = _json.loads((locales / "zh_CN.json").read_text(encoding="utf-8")) sample = zh["menu"]["theme_dark"] if not any("\u4e00" <= char <= "\u9fff" for char in sample): problems.append(f"zh_CN.menu.theme_dark has no Chinese text: {sample!r}") except Exception as exc: # noqa: BLE001 problems.append(f"could not verify the Chinese locale: {exc}") return problems def _calibration_sample(index: int, *, fn_raw: float, fx_raw: float = 0.4, fy_raw: float = -0.2): """A sample shaped for the calibration recorder.""" from orisys_app_shared.force.session import SessionSample return SessionSample( frame_id=index, t_rel_s=index / 100.0, fn_raw=fn_raw, fx_raw=fx_raw, fy_raw=fy_raw, fn_n=fn_raw / 100.0, fx_n=fx_raw / 100.0, fy_n=fy_raw / 100.0, valid=True, ) def _fake_packet( frame_id: int, *, valid: bool = True, depth=None, reason: str = "", calibrated_axes=None, timestamp: float | None = None, mono_t: float | None = None, shear_n: float | None = None, ): """A packet shaped like the real worker's, for the record-channel checks.""" import numpy as np from orisys_app_shared.acquisition.sdk_worker import SdkPacket nan = float("nan") blank = np.zeros((8, 8, 3), np.uint8) # A single tall sample makes it easy to prove decimation keeps peaks. fn_raw = 99.0 if frame_id == 700 else (9.0 + frame_id * 0.01) return SdkPacket( raw_frame=blank, processed_frame=blank, timestamp=1000.0 + frame_id / 30.0 if timestamp is None else timestamp, mono_t=mono_t, fps=30.0, process_ms=1.0, normal_force=(2.0 + frame_id * 0.01) if valid else nan, shear_x=(0.5 if shear_n is None else shear_n) if valid else nan, shear_y=-0.5 if valid else nan, normal_force_raw=(9.0 + frame_id * 0.1) if valid else nan, shear_x_raw=0.4 if valid else nan, shear_y_raw=-0.4 if valid else nan, force_unit="N", force_calibrated=True, calibrated_axes=tuple(calibrated_axes or ()), session_id=1, frame_id=frame_id, valid=valid, invalid_reason=reason, raw_available=True, depth=depth, ) class _StuckWorker(RecordSinkMixin): """A worker whose device open never returns. Mirrors the real worker contract — including the record-channel sink API — so the window can be exercised against the failure mode the Stop button exists to survive. """ def __init__(self) -> None: self._init_sinks() self._thread: threading.Thread | None = None self._end_reason: str | None = None self._latest = None self.thread_still_owned = False self.stop_blocked_for = 0.0 #: Released by ``release()`` so daemon threads do not outlive the run. self._release = threading.Event() #: Calibration requests this worker was asked to apply. self.calibration_requests: list[str] = [] #: Zeroing requests, and the confirmation callback the window installs. self.reset_requests = 0 self.on_reset_applied = None def set_pending_calibration(self, path) -> None: self.calibration_requests.append(str(path)) def reset(self) -> None: """Request a zeroing; a real device confirms it from its own thread. This stub confirms immediately (a fast device), so the window's event path — request, confirmation, recorded event — is exercised end to end. """ self.reset_requests += 1 if callable(self.on_reset_applied): packet = self._latest frame_id = int(getattr(packet, "frame_id", -1)) mono_t = float(getattr(packet, "mono_t", 0.0) or 0.0) self.on_reset_applied(frame_id, mono_t, 1) def start(self, _source, *, config_ref=None) -> None: self._release.clear() self._thread = threading.Thread(target=self._hang, daemon=True) self._thread.start() def _hang(self) -> None: self._release.wait(timeout=600.0) def release(self) -> None: """Let the stuck thread finish so the process can exit cleanly.""" self._release.set() if self._thread is not None: self._thread.join(timeout=5.0) def emit(self, packet) -> None: """Publish a packet exactly like the acquisition loop would.""" self._latest = packet self._emit(packet) @property def is_running(self) -> bool: return self._thread is not None and self._thread.is_alive() @property def phase(self) -> str: return "connecting" def latest(self): return self._latest def consume_end_reason(self): reason, self._end_reason = self._end_reason, None return reason def stop(self, *, wait: bool = False) -> None: self._end_reason = self._end_reason or "stopped" started = time.perf_counter() if wait and self._thread is not None and self._thread.is_alive(): self._thread.join(timeout=5.0) if self._thread is None or not self._thread.is_alive(): self._thread = None self.thread_still_owned = self._thread is not None self.stop_blocked_for = time.perf_counter() - started # -------------------------------------------------------------------------- # Palette rules # # These are the WCAG / colour-separation invariants the theme tokens must # satisfy. ``tests/test_ui_smoke.py`` imports :func:`audit_palette` from here so # both runners share one source of truth. # -------------------------------------------------------------------------- TEXT_MIN_RATIO = 4.5 GRAPHIC_MIN_RATIO = 3.0 VIZ_MIN_HUE_GAP = 90.0 VIZ_MIN_LUMINANCE_RATIO = 1.2 RAMP_BLEND_STOPS = 2 # the low stops may merge into the canvas TEXT_PAIRS: tuple[tuple[str, tuple[str, ...]], ...] = ( ("text_primary", ( "window_bg", "panel_bg", "card_bg", "toolbar_bg", "input_bg", "table_header_bg", "table_row_alt", "menu_bg", "button_bg", "table_selected", "scrollbar_bg", )), ("text_secondary", ( "window_bg", "panel_bg", "card_bg", "toolbar_bg", "table_header_bg", "button_bg", "input_bg", "menu_bg", )), ("text_muted", ("window_bg", "panel_bg", "card_bg", "toolbar_bg", "input_bg")), ("hint_text", ("card_bg", "panel_bg", "window_bg", "input_bg")), ("accent_text", ("accent", "accent_hover", "danger_fill", "danger_fill_hover")), ("danger_text", ("danger_fill", "danger_fill_hover")), # The status rail lives on the toolbar background, so every semantic colour # it can show is audited there too — not only on the panel/card backgrounds. ("danger", ("card_bg", "panel_bg", "toolbar_bg")), ("danger_hover", ("card_bg", "panel_bg", "toolbar_bg")), ("warning", ("card_bg", "panel_bg", "toolbar_bg")), ("info", ("card_bg", "panel_bg", "toolbar_bg")), ("success", ("card_bg", "panel_bg", "toolbar_bg")), ("status_idle", ("panel_bg", "toolbar_bg")), ("status_connecting", ("panel_bg", "toolbar_bg")), ("status_running", ("panel_bg", "toolbar_bg")), ("status_error", ("panel_bg", "toolbar_bg")), ("brand_bar_fg", ("brand_bar_bg",)), ("brand_bar_fg_muted", ("brand_bar_bg",)), ) GRAPHIC_PAIRS: tuple[tuple[str, tuple[str, ...]], ...] = ( ("plot_fn", ("plot_bg",)), ("plot_fx", ("plot_bg",)), ("plot_fy", ("plot_bg",)), ("scrollbar_handle", ("scrollbar_bg",)), ("scrollbar_handle_hover", ("scrollbar_bg",)), ("accent", ("panel_bg", "window_bg", "toolbar_bg")), ("accent_hover", ("panel_bg", "toolbar_bg")), ("focus_ring", ("window_bg", "input_bg", "panel_bg", "card_bg")), ("input_border", ("input_bg", "panel_bg", "card_bg", "toolbar_bg")), ) VIZ_KEYS = ("plot_fn", "plot_fx", "plot_fy") # Deliberate, documented exemptions (see apps/studio/docs/ui-palette.md): # decorative 1px container outlines, and disabled states which WCAG 1.4.3 # explicitly exempts. DECORATIVE_PAIRS: tuple[tuple[str, tuple[str, ...]], ...] = ( ("card_border", ("card_bg",)), ("panel_border", ("panel_bg", "window_bg")), ("plot_grid", ("plot_bg",)), ("separator", ("toolbar_bg", "panel_bg")), ) def _srgb_channel(value: int) -> float: channel = value / 255.0 if channel <= 0.03928: return channel / 12.92 return ((channel + 0.055) / 1.055) ** 2.4 def _rgb(hex_color: str) -> tuple[int, int, int]: text = str(hex_color).strip().lstrip("#") return tuple(int(text[i : i + 2], 16) for i in (0, 2, 4)) # type: ignore[return-value] def relative_luminance(hex_color: str) -> float: red, green, blue = _rgb(hex_color) return ( 0.2126 * _srgb_channel(red) + 0.7152 * _srgb_channel(green) + 0.0722 * _srgb_channel(blue) ) def contrast_ratio(first: str, second: str) -> float: high = relative_luminance(first) low = relative_luminance(second) if high < low: high, low = low, high return (high + 0.05) / (low + 0.05) def hue_degrees(hex_color: str) -> float: red, green, blue = (value / 255.0 for value in _rgb(hex_color)) high, low = max(red, green, blue), min(red, green, blue) delta = high - low if delta == 0: return 0.0 if high == red: return (60 * ((green - blue) / delta) + 360) % 360 if high == green: return 60 * ((blue - red) / delta) + 120 return 60 * ((red - green) / delta) + 240 def _hue_gap(first: str, second: str) -> float: delta = abs(hue_degrees(first) - hue_degrees(second)) return min(delta, 360 - delta) def audit_palette(verbose: bool = False) -> list[str]: """Return the list of palette-rule violations (empty when compliant).""" from orisys_app_shared.theme import load_colors failures: list[str] = [] themes = ("dark", "light") palettes = {theme: load_colors(theme) for theme in themes} dark_keys, light_keys = set(palettes["dark"]), set(palettes["light"]) if dark_keys != light_keys: failures.append(f"token keys differ between themes: {sorted(dark_keys ^ light_keys)}") for theme, palette in palettes.items(): if verbose: print(f" -- {theme} --") for key, value in sorted(palette.items()): if key == "flow_ramp": stops = [stop.strip() for stop in value.split(",") if stop.strip()] if not stops: failures.append(f"{theme}.flow_ramp is empty") for stop in stops: if not (stop.startswith("#") and len(stop) == 7): failures.append(f"{theme}.flow_ramp stop {stop!r} is not #RRGGBB") elif not (str(value).startswith("#") and len(str(value)) == 7): failures.append(f"{theme}.{key} = {value!r} is not #RRGGBB") for key, backgrounds in TEXT_PAIRS: for background in backgrounds: ratio = contrast_ratio(palette[key], palette[background]) if verbose: print(f" text {key:19s} on {background:17s} {ratio:5.2f}") if ratio < TEXT_MIN_RATIO: failures.append( f"{theme}: text {key} on {background} = {ratio:.2f} " f"(need {TEXT_MIN_RATIO})" ) for key, backgrounds in GRAPHIC_PAIRS: for background in backgrounds: ratio = contrast_ratio(palette[key], palette[background]) if verbose: print(f" graph {key:19s} on {background:17s} {ratio:5.2f}") if ratio < GRAPHIC_MIN_RATIO: failures.append( f"{theme}: graphic {key} on {background} = {ratio:.2f} " f"(need {GRAPHIC_MIN_RATIO})" ) for index, first in enumerate(VIZ_KEYS): for second in VIZ_KEYS[index + 1 :]: gap = _hue_gap(palette[first], palette[second]) high = max(relative_luminance(palette[first]), relative_luminance(palette[second])) low = min(relative_luminance(palette[first]), relative_luminance(palette[second])) ratio = (high + 0.05) / (low + 0.05) if verbose: print( f" viz {first:19s} vs {second:17s} " f"dHue={gap:5.1f} LrelRatio={ratio:4.2f}" ) if gap < VIZ_MIN_HUE_GAP and ratio < VIZ_MIN_LUMINANCE_RATIO: failures.append( f"{theme}: {first}/{second} are too similar " f"(hue gap {gap:.0f} deg, luminance ratio {ratio:.2f})" ) stops = [stop.strip() for stop in palette["flow_ramp"].split(",") if stop.strip()] if len(stops) >= 2: luminances = [relative_luminance(stop) for stop in stops] if verbose: print(f" ramp luminances {[round(value, 4) for value in luminances]}") if not all( luminances[i] < luminances[i + 1] for i in range(len(luminances) - 1) ): failures.append(f"{theme}: flow_ramp is not monotone in luminance") for stop in stops[RAMP_BLEND_STOPS:]: ratio = contrast_ratio(stop, palette["arrow_canvas"]) if ratio < GRAPHIC_MIN_RATIO: failures.append( f"{theme}: flow_ramp stop {stop} on arrow_canvas = " f"{ratio:.2f} (need {GRAPHIC_MIN_RATIO})" ) for theme, palette in palettes.items(): if palette["brand_bar_bg"] == palette["window_bg"]: failures.append(f"{theme}: brand_bar_bg collides with window_bg") if verbose: print(f" -- {theme} documented exemptions --") for key, backgrounds in DECORATIVE_PAIRS: for background in backgrounds: print( f" exempt {key:19s} on {background:17s} " f"{contrast_ratio(palette[key], palette[background]):5.2f}" ) return failures def main() -> int: from PyQt5.QtWidgets import QApplication from orisys_app_shared.force.calibration_io import CalSample from orisys_app_shared.i18n import set_locale, t from orisys_app_shared.theme import build_qss, load_colors from orisys_studio.app_window import 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") check( window.sensor_picker.isVisible(), f"{theme}: the sensor picker is not on the status rail", ) check( window.sensor_picker.type_label.text() == t("sensor.type") and window.sensor_picker.config_label.text() == t("sensor.config"), 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( window.sensor_picker.thumb_label.text() == "BOX", f"{theme}: the product badge does not name the sensor: " f"{window.sensor_picker.thumb_label.text()!r}", ) check( window.sensor_picker.config_label.toolTip() != "", f"{theme}: the sensor config picker has no explanation", ) check( window.toolbar.sensor_summary.text() != "", f"{theme}: the rail does not name the capture source", ) 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( "sensor.json" in window.device_label.text(), f"{theme}: the status bar does not name the config file: " f"{window.device_label.text()!r}", ) # 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.source_combo.isVisible(), f"{theme}: the source control is unreachable in the settings area", ) check( window.sensor_picker.isVisible(), f"{theme}: the sensor picker is unreachable in the settings area", ) 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( "FINGER" == english_window.sensor_picker.thumb_label.text(), f"the product badge is truncated: " f"{english_window.sensor_picker.thumb_label.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", ) check( window.force_plots.mode() == "browse", "the replay view did not open on the whole session", ) 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("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", ) 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", ) window_texts = plots.stats_label.text() check(window_texts != "", "no selection statistics in browse mode") check("峰" in window_texts, f"the peak is missing from the readout: {window_texts!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 user gesture enters browse; "back to latest" must actually leave it. plots._syncing = False plots.plots[0].setXRange(5.0, 8.0, padding=0.0) app.processEvents() check(plots.mode() == "browse", "a user pan did not enter browse mode") plots.follow_latest() check(plots.mode() == "follow", "back-to-latest did not restore live mode") 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", ) # 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", ) window.close() app.processEvents() stuck.release() section("layout preset") 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() check(window._layout_preset == "default", "the layout preset did not load") check(window.mesh3d_panel.isVisible(), "the 3D panel is hidden in the free layout") window._on_layout_preset("2d") app.processEvents() check(window._layout_preset == "2d", "the 2D preset did not apply") check(not window.mesh3d_panel.isVisible(), "the 2D preset keeps the 3D panel") check( window.arrow_panel.view_mode() == "depth", "the 2D preset does not show the depth map", ) check(not window.cal_panel.isVisible(), "the 2D preset keeps calibration expanded") check(not window.act_finger_3d.isEnabled(), "the 3D toggle is live in the 2D preset") check(window.act_layout_2d.isChecked(), "the 2D menu entry is not checked") window._on_layout_preset("default") app.processEvents() check(window.mesh3d_panel.isVisible(), "the free layout did not restore the 3D panel") check(window.act_finger_3d.isEnabled(), "the 3D toggle stayed disabled") window._persist_prefs() check(window.prefs.get("layout_preset") == "default", "the preset was not persisted") window.close() app.processEvents() stuck.release() section("replay") replay_root = APP_ROOT / "outputs" / "_replay_check" _shutil.rmtree(replay_root, ignore_errors=True) replay_root.mkdir(parents=True, exist_ok=True) from orisys_app_shared.record import ExperimentRecorder, RecordConfig recorder = ExperimentRecorder( RecordConfig(video=True, fps=30.0, segment_seconds=30.0) ) recording = recorder.start( root=replay_root, session_id=1, metadata={ "device": { "sensor_type": "box", "config_ref": str(replay_root / "sensor.json"), "config_sha256": "recorded", }, "versions": {"sdk": "0.5.1"}, "depth": {"stored": False, "replay": "recompute"}, }, ) (replay_root / "sensor.json").write_text('{"camera": {}}\n', encoding="utf-8") for index in range(40): recorder.on_packet(_fake_packet(index)) recorder.stop() from orisys_studio.ui.playback_dialog import PlaybackDialog playback = PlaybackDialog(theme="dark", directory=recording) playback.resize(1000, 620) playback.show() app.processEvents() check( playback.position() == 0.0, f"replay did not start at the beginning: {playback.position()}", ) check( all(int(curve.getData()[0].size) == 40 for curve in playback.curves), "replay did not draw every force sample", ) # 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(playback.slider.isEnabled(), "the replay position slider is disabled") check( t("playback.warn_depth_recomputed").split("{")[0].strip()[:4] in playback.warning_label.text(), f"the replay notes are missing: {playback.warning_label.text()!r}", ) moved: list[float] = [] playback.cursor_moved.connect(moved.append) playback.set_position(1.0) app.processEvents() check( [round(line.value(), 2) for line in playback.cursor_lines] == [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( playback.video_label.pixmap() is not None and not playback.video_label.pixmap().isNull(), "no video frame was rendered at the cursor position", ) # The depth recomputation must never run on the UI thread. A stub that blocks # proves the call returns immediately instead of freezing the window. import threading as _threading import time as _time class _Blocking: released = _threading.Event() def preflight(self) -> str: return "" def depth_at(self, _t): _Blocking.released.wait(timeout=5.0) return type("R", (), {"ok": False, "error": "stub", "depth": None})() def close(self) -> None: return None playback._recomputer = _Blocking() # noqa: SLF001 - exercising the contract started = _time.perf_counter() playback._on_recompute_depth() elapsed = _time.perf_counter() - started check(elapsed < 0.5, f"the depth recompute blocked the UI for {elapsed:.2f}s") check( playback._depth_thread is not None and playback._depth_thread.is_alive(), "the depth recompute did not run on a worker thread", ) check( playback._depth_thread.daemon, "the depth worker is not a daemon, so a blocked SDK call would keep the " "process alive at exit", ) playback._on_recompute_depth() # a second request is refused, not queued check( playback.depth_label.text() == t("playback.depth_busy"), f"a concurrent recompute was not refused: {playback.depth_label.text()!r}", ) _Blocking.released.set() for _ in range(40): app.processEvents() _time.sleep(0.05) if not playback._depth_busy(): # noqa: SLF001 break check(not playback._depth_busy(), "the depth worker never finished") # Deliver the worker result deterministically instead of waiting for the # 120 ms poll timer. playback._poll_depth() app.processEvents() check( t("playback.depth_failed").split("{")[0].strip()[:4] in playback.depth_label.text(), f"a failed recomputation was not reported: {playback.depth_label.text()!r}", ) playback.close() app.processEvents() # The plan must state where to start, which reset it is anchored to and # whether the state is really recoverable; and a request must belong to the # recording it started in. from orisys_app_shared.record.depth import ( STATE_APPROXIMATE, STATE_PREROLL, DepthRecomputer, ) from orisys_app_shared.record.playback import RecordingReader plan_reader = RecordingReader(recording) planner = DepthRecomputer(plan_reader) frames = plan_reader.info().frames check(bool(frames), "the replay recording has no frame index") plan = planner.plan_for(frames[10]) check( plan.state == STATE_APPROXIMATE and plan.anchor == "segment_start", f"a recording with no reset/pre-roll is not reported as approximate: {plan.state}", ) check( plan.steps and plan.steps[0].source == plan_reader.segment_path(frames[10].segment), "the plan does not name the video to read", ) check( "state_approximate" in plan.warnings, f"the approximate replay is not reported: {plan.warnings}", ) # A pre-roll makes the start of a recording reproducible. session_path = recording / "session.json" session_doc = _json.loads(session_path.read_text(encoding="utf-8")) prerecord = recording / "preroll_0001.mp4" prerecord.write_bytes(b"") session_doc["preroll_frames"] = 30 session_doc["preroll_segment"] = prerecord.name session_path.write_text(_json.dumps(session_doc), encoding="utf-8") planner.close() planner = DepthRecomputer(RecordingReader(recording)) plan = planner.plan_for(planner.reader.info().frames[5]) check( plan.state == STATE_PREROLL and plan.preroll == prerecord, f"the pre-roll was not used for the start of the recording: {plan.state}", ) check(plan.frames_to_read == 30 + 6, f"unexpected replay cost: {plan.frames_to_read}") planner.close() # A reset boundary is re-applied, and a request is bound to its recording. _Blocking.released.clear() second = PlaybackDialog(theme="dark", directory=recording) second.show() app.processEvents() serial_before = second._record_serial blocking = _Blocking() second._recomputer = blocking second._on_recompute_depth() check(second._depth_pending_record == serial_before, "the request has no recording id") check( second._depth_pending_seq is not None, "the request carries no sequence number", ) # Opening another recording invalidates the running request... second.open_recording(recording) check( second._record_serial == serial_before + 1, "opening a recording did not invalidate the running request", ) check( second._recomputer is not blocking, "the invalidated processor was not replaced", ) _Blocking.released.set() for _ in range(40): app.processEvents() _time.sleep(0.05) if not second._depth_busy(): break second._poll_depth() app.processEvents() pixmap = second.depth_label.pixmap() check( pixmap is None or pixmap.isNull(), "an answer for a previous recording was drawn into the new one", ) second.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())