orisys-tech/orisys-app
3
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: 拆出 Studio 状态显示并记录分步迁移计划

Browse Source 
让状态显示独立于主窗口，保留录制快照注入入口。验证结果和完整门禁限制见 docs/architecture/studio-window-refactor-plan.zh.md。
This commit is contained in:
ykcheungab
2026-10-18 20:32:32 +08:00
parent cf3e638176
commit 0998009ec3
5 changed files with 580 additions and 203 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/studio/src/orisys_studio/app_window.py
+41 -170
View File
@@ -21,7 +21,7 @@ from orisys_app_shared.acquisition.image_backend import (
SOURCE_RECOMPUTE,
SOURCE_REPLAY,
ImageFrame,
render_image, # re-exported: tests monkeypatch ``app_window.render_image``
render_image as render_image, # re-exported: tests monkeypatch ``app_window.render_image``
)
from orisys_app_shared.force.calibration_recorder import CalibrationRecorder
from orisys_app_shared.force.filters import FilterSettings
@@ -31,7 +31,7 @@ from orisys_app_shared.logging import report_error
from orisys_app_shared.paths import logo_path, recordings_dir, sensor_json_path
from orisys_app_shared.prefs import save_preferences
from orisys_app_shared.record import RecordConfig
from orisys_app_shared.theme import build_qss, clear_theme_cache, color
from orisys_app_shared.theme import build_qss, clear_theme_cache
from PySide6.QtCore import Qt, QTimer
from PySide6.QtGui import QAction, QActionGroup, QIcon
from PySide6.QtWidgets import (
@@ -54,13 +54,13 @@ from orisys_studio.replay import ReplayEngine
from orisys_studio.replay_controller import ReplayController
from orisys_studio.scene3d.prefs import settings_to_prefs
from orisys_studio.session_controller import SessionController
from orisys_studio.ui import record_status as record_states
from orisys_studio.status_presenter import RailState, StatusPresenter
from orisys_studio.ui.about_dialog import AboutDialog
from orisys_studio.ui.analysis_progress import AnalysisProgress
from orisys_studio.ui.arrow_panel import ArrowPanel, bgr_to_qimage
from orisys_studio.ui.brand_bar import BrandBar
from orisys_studio.ui.cal_panel import CalPanel
from orisys_studio.ui.force_plots import CHANNEL_LABELS, ForcePlotsPanel
from orisys_studio.ui.force_plots import ForcePlotsPanel
from orisys_studio.ui.mesh3d_panel import Mesh3dPanel
from orisys_studio.ui.finger_orientation_dialog import FingerOrientationDialog
from orisys_studio.ui.plot_settings_dialog import (
@@ -250,6 +250,7 @@ class StudioMainWindow(QMainWindow):
self.toolbar = ToolbarWidget(theme=self._theme)
root.addWidget(self.toolbar)
self.status_chip = self.toolbar.status_chip
self.status_presenter = StatusPresenter(self.toolbar)
# Part B: the analysis pass a replay is served from. The surface sits directly
# under the rail, so "what is it doing and can I stop it" is answered where the
# transport is; it is hidden until a pass (or a cache hit) has something to say.
@@ -1431,184 +1432,54 @@ class StudioMainWindow(QMainWindow):
self.statusBar().showMessage(message, int(ms))
def _refresh_status_chip(self) -> None:
chip = getattr(self, "status_chip", None)
if chip is None:
return
# The session normalises the phase once, when it is set, so there is nothing
# to sanitise here: a second normalisation is a second rule to keep in step.
phase = self._session_phase
token = f"status_{phase}"
# One dot carries the colour; the text says what the state is, so the
# colour is never the only encoding.
chip.setText(f"● {t(f'status.phase_{phase}')}")
chip.setStyleSheet(
f"QLabel#statusChip {{ color: {color(self._theme, token)};"
" background: transparent; font-weight: 600; padding: 0 10px 0 4px; }"
)
self.status_presenter.refresh_phase(self.session.phase, theme=self._theme)
# ------------------------------------------------------ status rail state
def _refresh_status_rail(self) -> None:
"""Sensor, source detail, calibration, validity, FPS and device identity.
The rail is the single place these facts are written; the bottom status
bar keeps only transient messages.
"""
config = self.sensor_picker.config_name()
self.toolbar.set_sensor_summary(
t(
"sensor.summary",
type=self.sensor_picker.sensor_type().upper(),
config=Path(config).name if config else t("status.config_none"),
)
)
# The picker on the rail already names the camera, so the summary carries
# only what it cannot: the frame rate, and — **in the replay view only** —
# the video part being played. Naming the video in the live view was how a
# replay's file stayed on the rail after switching back to the camera.
detail = ""
result_type = ""
result_progress = None
if self._view_preset == "replay":
path = self.toolbar.video_path()
if path:
key = (
"status.rail_video_name"
if self.replay.has_video()
else "status.rail_recent_video"
)
detail = t(key, name=Path(path).name)
# Name the *result's* profile and whether it was reused or just finished, so
# changing the sensor picker cannot look like the open figures came from the
# newly selected product.
session = self.replay_ctl.available_result()
if session is not None:
provenance = getattr(getattr(session, "reader", None), "provenance", None)
result = self.replay_ctl.available_result()
if result is not None:
provenance = getattr(getattr(result, "reader", None), "provenance", None)
result_type = str(getattr(provenance, "sensor_type", "") or "").strip()
if result_type:
result_label = t("status.rail_result_profile", type=result_type)
detail = f"{result_label} · {detail}" if detail else result_label
progress = getattr(self.replay_ctl, "_progress", None)
progress_state = (
str(progress.state()) if progress is not None and hasattr(progress, "state") else ""
)
if progress_state == "reused":
status_text = t("status.rail_result_reused")
elif progress_state == "complete":
status_text = t("status.rail_result_complete")
elif progress_state == "running":
status_text = t("status.rail_result_running")
else:
status_text = t("status.rail_result_ready")
detail = f"{detail} · {status_text}" if detail else status_text
if self._show_fps and self._latest_fps is not None:
fps = f"{self._latest_fps:.1f} FPS"
# The cost goes next to the rate because one without the other cannot
# be read: "16.7 FPS" is either a slow camera or a pipeline that
# cannot keep up, and only this number tells the two apart.
if self._latest_process_ms:
fps = f"{fps} · {self._latest_process_ms:.0f} ms"
# ...and the camera's own rate settles it: a capture rate well above
# what we consumed means the frames were there and we did not take
# them (the drop counter says how many).
if self._latest_capture_fps:
fps = f"{fps} · {t('status.rail_camera_fps', fps=f'{self._latest_capture_fps:.1f}')}"
if self._latest_capture_dropped:
fps = f"{fps} (-{self._latest_capture_dropped})"
detail = f"{detail} · {fps}" if detail else fps
self.toolbar.set_source_summary(detail)
# The full path and the replay's notes do not fit on a 30 px rail, so
# they live in the summary's tooltip (the bar that used to show them is
# gone: the two views share one layout).
tip_parts = [detail, self.toolbar.video_path()]
if self._view_preset == "replay" and self.replay.has_video():
# Includes the "this is a video, not a recording" note for a bare video.
tip_parts.append(self.replay.warnings_text())
self.toolbar.set_source_summary_tip(
"\n".join(part for part in tip_parts if part)
)
self.toolbar.set_calibration_summary(self._calibration_summary())
if self._last_sample_valid is None:
self.toolbar.set_data_state("")
elif self._last_sample_valid:
self.toolbar.set_data_state(t("status.data_valid"), "success")
else:
self.toolbar.set_data_state(
t("status.data_invalid", count=self._record_error_count), "warning"
)
result_progress = str(self.analysis_progress.state())
self.status_presenter.refresh_rail(RailState(
sensor_type=self.sensor_picker.sensor_type(),
config_name=self.sensor_picker.config_name(),
view_preset=self._view_preset,
video_path=self.toolbar.video_path(),
has_video=self.replay.has_video(),
replay_warnings=(
self.replay.warnings_text()
if self._view_preset == "replay" and self.replay.has_video() else ""
),
calibration_summary=self._calibration_summary(),
result_sensor_type=result_type,
result_progress=result_progress,
show_fps=self._show_fps,
fps=self.session.latest_fps,
process_ms=self.session.latest_process_ms,
capture_fps=self.session.latest_capture_fps,
capture_dropped=self.session.latest_capture_dropped,
sample_valid=self.session.last_sample_valid,
record_error_count=self.session.record_error_count,
))
def _calibration_summary(self) -> str:
"""What the calibration state means for the three axes.
Before the first sample the only evidence is whether the device
confirmed a calibration version. Afterwards the display decision is the
evidence: an axis without a newton curve is an axis without calibration,
and the rail must not claim three calibrated axes for a one-axis device.
"""
if self._last_display_fallback is None:
return (
t("status.calibrated")
if self._active_calibration_id
else t("status.uncalibrated")
)
if self._last_display_fallback or self._last_display_available is None:
return t("status.uncalibrated")
axes = self._last_display_available
if all(axes):
return t("status.calibrated_all")
names = [CHANNEL_LABELS[index] for index, ok in enumerate(axes) if ok]
if not names:
return t("status.uncalibrated")
return t("status.calibrated_axes", axes="+".join(names))
return self.status_presenter.calibration_summary(
self.recording.active_calibration,
self._last_display_fallback,
self._last_display_available,
)
# ---------------------------------------------------------- record state
def _record_stats(self):
"""``(state, stats, directory)`` — the single read of the record channel.
Kept as one seam so a preview/demo driver can inject a record state
without touching the recorder or the writer thread.
"""
recorder = self.recorder
raw_directory = getattr(recorder, "directory", None)
directory = Path(raw_directory).name if raw_directory else ""
if self.recording.is_recording:
return record_states.RECORDING, recorder.stats(), directory
if self.recording.report_pending:
return record_states.CLOSING, recorder.stats(), directory
report = self._record_last_report
if report is not None:
stats = report.stats
state = (
record_states.ERROR
if (stats.failed or stats.write_errors)
else record_states.DONE
)
return state, stats, directory
return record_states.IDLE, None, directory
"""Keep the snapshot injection seam used by UI checks and demo screenshots."""
return self.status_presenter.record_stats(self.recording)
def _refresh_record_state(self) -> None:
"""Push the recording's live numbers (duration, drops, write errors)."""
state, stats, directory = self._record_stats()
if stats is None:
self.toolbar.set_record_state(state, directory=directory)
return
self.toolbar.set_record_state(
state,
duration_s=stats.duration_s,
frames=stats.frames,
segments=stats.segments,
# An unrecorded gap is a lost row too: reporting only the two
# queue-overflow counters would understate the loss.
dropped=(
stats.dropped_data
+ stats.dropped_video
+ stats.dropped_gaps
+ stats.unrecorded_gaps
),
dropped_data=stats.dropped_data,
dropped_video=stats.dropped_video,
dropped_gaps=stats.dropped_gaps + stats.unrecorded_gaps,
errors=stats.write_errors,
detail=stats.last_error,
directory=directory,
)
self.status_presenter.refresh_record(*self._record_stats())
# ------------------------------------------------------------------ nav
def _set_theme(self, theme: str) -> None:
apps/studio/src/orisys_studio/status_presenter.py
+189
View File
@@ -0,0 +1,189 @@
"""Render shell status from read-only facts on the GUI thread.
The presenter holds only the toolbar. It neither owns session state nor calls
back into the main window; a rail snapshot is discarded after each refresh.
"""
from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path
from typing import TYPE_CHECKING
from orisys_app_shared.i18n import t
from orisys_app_shared.theme import color
from orisys_studio.ui import record_status as record_states
from orisys_studio.ui.force_plots import CHANNEL_LABELS
if TYPE_CHECKING:
from orisys_app_shared.record import RecordStats
from orisys_studio.recording import RecordingCoordinator
from orisys_studio.ui.toolbar import ToolbarWidget
@dataclass(frozen=True)
class RailState:
"""One GUI-thread read of the facts needed to paint the status rail."""
sensor_type: str
config_name: str
view_preset: str
video_path: str
has_video: bool
replay_warnings: str
calibration_summary: str
result_sensor_type: str = ""
#: None means no result; an empty string means a result with no progress state.
result_progress: str | None = None
show_fps: bool = False
fps: float | None = None
process_ms: float | None = None
capture_fps: float = 0.0
capture_dropped: int = 0
sample_valid: bool | None = None
record_error_count: int = 0
class StatusPresenter:
"""Paint status without changing the acquisition or recording lifecycle."""
def __init__(self, toolbar: ToolbarWidget) -> None:
self._toolbar = toolbar
def refresh_phase(self, phase: str, *, theme: str) -> None:
# The session normalises phase; colour is never the only encoding.
chip = self._toolbar.status_chip
chip.setText(f"● {t(f'status.phase_{phase}')}")
token = f"status_{phase}"
chip.setStyleSheet(
f"QLabel#statusChip {{ color: {color(theme, token)};"
" background: transparent; font-weight: 600; padding: 0 10px 0 4px; }"
)
def refresh_rail(self, state: RailState) -> None:
toolbar = self._toolbar
toolbar.set_sensor_summary(
t(
"sensor.summary",
type=state.sensor_type.upper(),
config=Path(state.config_name).name
if state.config_name
else t("status.config_none"),
)
)
detail = ""
if state.view_preset == "replay":
if state.video_path:
key = (
"status.rail_video_name"
if state.has_video
else "status.rail_recent_video"
)
detail = t(key, name=Path(state.video_path).name)
# The displayed result keeps its own identity after a picker change.
if state.result_progress is not None:
if state.result_sensor_type:
label = t(
"status.rail_result_profile", type=state.result_sensor_type
)
detail = f"{label} · {detail}" if detail else label
if state.result_progress == "reused":
status_text = t("status.rail_result_reused")
elif state.result_progress == "complete":
status_text = t("status.rail_result_complete")
elif state.result_progress == "running":
status_text = t("status.rail_result_running")
else:
status_text = t("status.rail_result_ready")
detail = f"{detail} · {status_text}" if detail else status_text
if state.show_fps and state.fps is not None:
fps = f"{state.fps:.1f} FPS"
# Processing cost and capture rate distinguish a slow chain from a slow camera.
if state.process_ms:
fps = f"{fps} · {state.process_ms:.0f} ms"
if state.capture_fps:
fps = f"{fps} · {t('status.rail_camera_fps', fps=f'{state.capture_fps:.1f}')}"
if state.capture_dropped:
fps = f"{fps} (-{state.capture_dropped})"
detail = f"{detail} · {fps}" if detail else fps
toolbar.set_source_summary(detail)
tip_parts = [detail, state.video_path]
if state.view_preset == "replay" and state.has_video:
tip_parts.append(state.replay_warnings)
toolbar.set_source_summary_tip("\n".join(part for part in tip_parts if part))
toolbar.set_calibration_summary(state.calibration_summary)
if state.sample_valid is None:
toolbar.set_data_state("")
elif state.sample_valid:
toolbar.set_data_state(t("status.data_valid"), "success")
else:
toolbar.set_data_state(
t("status.data_invalid", count=state.record_error_count), "warning"
)
@staticmethod
def calibration_summary(
active_id: str,
fallback: bool | None,
available: tuple[bool, bool, bool] | None,
) -> str:
"""Confirmed identity is evidence until the first per-axis display decision."""
if fallback is None:
return t("status.calibrated") if active_id else t("status.uncalibrated")
if fallback or available is None:
return t("status.uncalibrated")
if all(available):
return t("status.calibrated_all")
names = [CHANNEL_LABELS[index] for index, ok in enumerate(available) if ok]
if not names:
return t("status.uncalibrated")
return t("status.calibrated_axes", axes="+".join(names))
@staticmethod
def record_stats(
recording: RecordingCoordinator,
) -> tuple[str, RecordStats | None, str]:
"""Read live stats while closing; the final report is authoritative afterwards."""
recorder = recording.recorder
raw_directory = getattr(recorder, "directory", None)
directory = Path(raw_directory).name if raw_directory else ""
if recording.is_recording:
return record_states.RECORDING, recorder.stats(), directory
if recording.report_pending:
return record_states.CLOSING, recorder.stats(), directory
report = recording.last_report
if report is not None:
stats = report.stats
state = (
record_states.ERROR
if (stats.failed or stats.write_errors)
else record_states.DONE
)
return state, stats, directory
return record_states.IDLE, None, directory
def refresh_record(
self, state: str, stats: RecordStats | None, directory: str
) -> None:
if stats is None:
self._toolbar.set_record_state(state, directory=directory)
return
self._toolbar.set_record_state(
state,
duration_s=stats.duration_s,
frames=stats.frames,
segments=stats.segments,
# A gap that could not be recorded is also a lost row.
dropped=stats.dropped_data
+ stats.dropped_video
+ stats.dropped_gaps
+ stats.unrecorded_gaps,
dropped_data=stats.dropped_data,
dropped_video=stats.dropped_video,
dropped_gaps=stats.dropped_gaps + stats.unrecorded_gaps,
errors=stats.write_errors,
detail=stats.last_error,
directory=directory,
)
apps/studio/tests/test_status_presenter.py
+172
View File
@@ -0,0 +1,172 @@
"""Status presentation without a main window or running acquisition."""
from dataclasses import replace
from types import SimpleNamespace
import pytest
from orisys_app_shared.i18n import t
from orisys_app_shared.record.experiment import RecordStats
from orisys_studio.status_presenter import RailState, StatusPresenter
from orisys_studio.ui.toolbar import ToolbarWidget
@pytest.fixture
def surface(qt_app):
toolbar = ToolbarWidget(theme="dark")
yield toolbar, StatusPresenter(toolbar)
toolbar.close()
toolbar.deleteLater()
def rail(**changes):
return replace(
RailState(
sensor_type="finger",
config_name="finger/r01/sensor.json",
view_preset="live",
video_path="D:/clips/box.mp4",
has_video=True,
replay_warnings="video warning",
calibration_summary="calibration",
),
**changes,
)
def test_live_summary_keeps_video_details_out_and_reports_capture_cost(surface):
toolbar, presenter = surface
state = rail(
show_fps=True,
fps=16.67,
process_ms=58.4,
capture_fps=30.0,
capture_dropped=7,
)
presenter.refresh_rail(state)
text = toolbar.source_summary.text()
assert "box.mp4" not in text
assert "video warning" not in toolbar.source_summary.toolTip()
assert "16.7 FPS" in text and "58 ms" in text
assert "30.0" in text and "(-7)" in text
assert state.view_preset == "live"
@pytest.mark.parametrize(
"loaded,key",
[
(True, "status.rail_video_name"),
(False, "status.rail_recent_video"),
],
)
def test_replay_distinguishes_open_video_from_remembered_path(surface, loaded, key):
toolbar, presenter = surface
presenter.refresh_rail(rail(view_preset="replay", has_video=loaded))
assert toolbar.source_summary.text() == t(key, name="box.mp4")
assert ("video warning" in toolbar.source_summary.toolTip()) is loaded
@pytest.mark.parametrize(
"progress,key",
[
("reused", "status.rail_result_reused"),
("complete", "status.rail_result_complete"),
("running", "status.rail_result_running"),
("", "status.rail_result_ready"),
],
)
def test_result_identity_belongs_to_the_result_not_the_picker(surface, progress, key):
toolbar, presenter = surface
presenter.refresh_rail(
rail(
view_preset="replay",
result_sensor_type="box",
result_progress=progress,
)
)
text = toolbar.source_summary.text()
assert text.startswith(t("status.rail_result_profile", type="box"))
assert text.endswith(t(key))
assert "FINGER" in toolbar.btn_sensor.text()
@pytest.mark.parametrize(
"version,fallback,axes,key,labels",
[
("", None, None, "status.uncalibrated", ""),
("confirmed", None, None, "status.calibrated", ""),
("confirmed", True, (True, True, True), "status.uncalibrated", ""),
("confirmed", False, None, "status.uncalibrated", ""),
("confirmed", False, (False, False, False), "status.uncalibrated", ""),
("confirmed", False, (True, True, True), "status.calibrated_all", ""),
("confirmed", False, (True, False, True), "status.calibrated_axes", "FN+FY"),
],
)
def test_calibration_summary_preserves_unknown_raw_and_partial_axes(
version,
fallback,
axes,
key,
labels,
):
assert StatusPresenter.calibration_summary(version, fallback, axes) == t(
key, axes=labels
)
def test_phase_repaints_after_a_theme_change_without_changing_the_phase(surface):
toolbar, presenter = surface
presenter.refresh_phase("connecting", theme="dark")
assert toolbar.status_chip.text() == f"● {t('status.phase_connecting')}"
presenter.refresh_phase("connecting", theme="light")
assert toolbar.status_chip.text() == f"● {t('status.phase_connecting')}"
def test_record_close_and_all_gap_counts_remain_visible(surface):
toolbar, presenter = surface
stats = RecordStats(
frames=12,
duration_s=2.5,
dropped_data=1,
dropped_video=2,
dropped_gaps=3,
unrecorded_gaps=4,
write_errors=1,
last_error="disk full",
)
recording = SimpleNamespace(
recorder=SimpleNamespace(directory="D:/recordings/run", stats=lambda: stats),
is_recording=False,
report_pending=True,
last_report=SimpleNamespace(stats=RecordStats()),
)
snapshot = presenter.record_stats(recording)
assert snapshot == ("closing", stats, "run")
presenter.refresh_record(*snapshot)
displayed = toolbar.record_chip.stats()
assert displayed["state"] == "closing"
assert displayed["dropped"] == 10
assert displayed["dropped_gaps"] == 7
assert displayed["errors"] == 1
@pytest.mark.parametrize(
"failed,errors,expected",
[
(False, 0, "done"),
(True, 0, "error"),
(False, 1, "error"),
],
)
def test_finished_recording_uses_the_final_report(surface, failed, errors, expected):
toolbar, presenter = surface
stats = RecordStats(failed=failed, write_errors=errors)
recording = SimpleNamespace(
recorder=SimpleNamespace(directory=None),
is_recording=False,
report_pending=False,
last_report=SimpleNamespace(stats=stats),
)
assert presenter.record_stats(recording) == (expected, stats, "")
presenter.refresh_record(expected, stats, "")
assert toolbar.record_chip.state() == expected