#!/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 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 = "", ): """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, fps=30.0, process_ms=1.0, normal_force=(2.0 + frame_id * 0.01) if valid else nan, shear_x=0.5 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, 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() 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 reset(self) -> None: return None 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")), ("danger", ("card_bg", "panel_bg")), ("danger_hover", ("card_bg", "panel_bg")), ("warning", ("card_bg", "panel_bg")), ("info", ("card_bg", "panel_bg")), ("success", ("card_bg", "panel_bg")), ("status_idle", ("panel_bg",)), ("status_connecting", ("panel_bg",)), ("status_running", ("panel_bg",)), ("status_error", ("panel_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", ) # Three more points, one of them off-trend, so the fit has a residual # that excluding the outlier must reduce and enough left over to hold # one out as a validation point. 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") worst = window.cal_recorder.fit("fn").max_residual_n check(worst > 0.05, f"the off-trend point produced no residual: {worst}") window.cal_panel.table.setCurrentCell(1, 0) window.cal_panel._on_toggle_excluded() app.processEvents() check( window.cal_recorder.fit("fn").max_residual_n < worst, "excluding an outlier did not tighten the fit", ) check( window.cal_panel.table.item(1, 5).text() == t("cal.state_excluded"), "the excluded row does not show its state", ) # A validation point is held out of the fit and reported separately. window.cal_panel.table.setCurrentCell(2, 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", ) 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("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("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() window._on_reset() # event: zeroing 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)}") 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("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", ) 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() _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())