"""Qt/VisPy window used by ``s6 track -v``.""" import os import re from collections import deque from multiprocessing import Queue from typing import Any, Optional os.environ["KMP_DUPLICATE_LIB_OK"] = "TRUE" import numpy as np import pyqtgraph as pg import yaml from pydantic import BaseModel from PyQt6.QtCore import QSignalBlocker, Qt, QSize from PyQt6.QtGui import ( QAction, QColor, QFont, QIcon, QSyntaxHighlighter, QTextCharFormat, ) from PyQt6.QtWidgets import ( QFrame, QHBoxLayout, QLabel, QMainWindow, QPlainTextEdit, QSizePolicy, QFileDialog, QSlider, QStyle, QStyleOptionSlider, QToolBar, QVBoxLayout, QWidget, ) from vispy import app, scene from s6.app._common import CommandSet, PlaybackMode, iter_queue_nowait from s6.app.pipeline import BasePipeline from s6.app._track_runtime import TrackFrameMessage, TrackState from s6.rendering.visuals import VisualTree class _YamlContextSerializer: """Serialize a frame context for compact YAML display.""" LONG_STRING_LIMIT = 200 def _serialize_value(self, value: Any) -> Any: if isinstance(value, np.ndarray): shape = tuple(int(dim) for dim in value.shape) return f"image({shape})" if isinstance(value, np.generic): return value.item() if isinstance(value, BaseModel): return self._serialize_value(value.dict()) if isinstance(value, dict): keys = list(value.keys()) if "debug" in value: keys = ["debug"] + [key for key in keys if key != "debug"] return {key: self._serialize_value(value[key]) for key in keys} if isinstance(value, (list, tuple)): return [self._serialize_value(item) for item in value] if isinstance(value, str) and len(value) > self.LONG_STRING_LIMIT: return f"string({len(value)} chars)" return value def dumps(self, context: dict[str, Any]) -> str: processed = self._serialize_value(context) return yaml.safe_dump( processed, sort_keys=False, allow_unicode=True, default_flow_style=False, ).rstrip() class _YamlSyntaxHighlighter(QSyntaxHighlighter): """Apply basic YAML key/value highlighting in the context sidebar.""" _KEY_PATTERN = re.compile(r"^(\s*)(-\s+)?([^:#][^:]*)(:)(.*)$") _SEQUENCE_VALUE_PATTERN = re.compile(r"^(\s*-\s+)(.+)$") def __init__(self, document): super().__init__(document) self._key_format = QTextCharFormat() self._key_format.setForeground(QColor("#8cc8ff")) self._key_format.setFontWeight(QFont.Weight.Bold) self._separator_format = QTextCharFormat() self._separator_format.setForeground(QColor("#7a8794")) self._value_format = QTextCharFormat() self._value_format.setForeground(QColor("#d7dee8")) self._scalar_format = QTextCharFormat() self._scalar_format.setForeground(QColor("#f0c674")) def highlightBlock(self, text: str) -> None: match = self._KEY_PATTERN.match(text) if match is not None: key_start = len(match.group(1)) + len(match.group(2) or "") key_len = len(match.group(3)) separator_start = key_start + key_len value_start = separator_start + len(match.group(4)) self.setFormat(key_start, key_len, self._key_format) self.setFormat(separator_start, len(match.group(4)), self._separator_format) if match.group(5).strip(): self._highlight_value(text, value_start) return sequence_match = self._SEQUENCE_VALUE_PATTERN.match(text) if sequence_match is not None: self._highlight_value(text, len(sequence_match.group(1))) def _highlight_value(self, text: str, start: int) -> None: self.setFormat(start, len(text) - start, self._value_format) value = text[start:].strip() if not value: return if value in {"true", "false", "null"} or _looks_numeric(value): leading_space = len(text[start:]) - len(text[start:].lstrip()) self.setFormat( start + leading_space, len(value), self._scalar_format, ) def _looks_numeric(value: str) -> bool: try: float(value) except ValueError: return False return True class _JumpToClickSlider(QSlider): """Slider that jumps to the clicked groove position.""" def _event_position(self, event): if hasattr(event, "position"): return event.position() return event.pos() def _value_from_position(self, position) -> int: option = QStyleOptionSlider() self.initStyleOption(option) style = self.style() slider_length = style.pixelMetric( QStyle.PixelMetric.PM_SliderLength, option, self, ) if self.orientation() == Qt.Orientation.Horizontal: span = max(1, self.width() - slider_length) mouse_position = int(round(position.x() - slider_length / 2)) else: span = max(1, self.height() - slider_length) mouse_position = int(round(position.y() - slider_length / 2)) mouse_position = max(0, min(mouse_position, span)) return QStyle.sliderValueFromPosition( self.minimum(), self.maximum(), mouse_position, span, option.upsideDown, ) def _jump_to_event_position(self, event) -> int: value = self._value_from_position(self._event_position(event)) self.setSliderPosition(value) self.setValue(value) self.sliderMoved.emit(value) return value def mousePressEvent(self, event) -> None: if event.button() == Qt.MouseButton.LeftButton and self.isEnabled(): self.setSliderDown(True) self.sliderPressed.emit() self._jump_to_event_position(event) event.accept() return super().mousePressEvent(event) def mouseMoveEvent(self, event) -> None: if self.isSliderDown() and event.buttons() & Qt.MouseButton.LeftButton: self._jump_to_event_position(event) event.accept() return super().mouseMoveEvent(event) def mouseReleaseEvent(self, event) -> None: if self.isSliderDown() and event.button() == Qt.MouseButton.LeftButton: self._jump_to_event_position(event) self.setSliderDown(False) self.sliderReleased.emit() event.accept() return super().mouseReleaseEvent(event) class MainWindow(QMainWindow): """Render pipeline contexts and expose the compact track transport UI.""" TIMER_INTERVAL_SEC = 0.005 LEFT_PANEL_WIDTH = 450 PLOT_PANEL_WIDTH = 400 def __init__( self, pipeline: BasePipeline, replay: bool, enable_plots: bool = False, command_queue: Queue = None, dataset_path: str = "None", ): super().__init__() self.setWindowTitle("Visualizer") self.setGeometry(100, 100, 1920, 1080) self.pipeline = pipeline self.command_queue = command_queue self._is_replay_mode = bool(replay) self._enable_plots = bool(enable_plots) self._dataset_path = dataset_path self.q: Optional[Queue] = None self.timer = None self.has_set_total_frame = False self._total_frames = 0 self._current_frame = 0 self._is_scrubbing = False self._pending_seek_frame: Optional[int] = None self._last_seek_command_frame: Optional[int] = None self._is_recording = False self._playback_state = PlaybackMode.PLAY self._pending_playback_state: Optional[PlaybackMode] = None self._context_yaml_serializer = _YamlContextSerializer() self._create_canvas() self._setup_views() self._build_window() def _create_canvas(self) -> None: """Create the VisPy canvas and objects expected by pipeline view setup.""" self.visual_tree = VisualTree() self.canvas = scene.SceneCanvas(keys="interactive", size=(1920, 1080)) self.grid = self.canvas.central_widget.add_grid() def _setup_views(self) -> None: """Delegate view layout to the active pipeline.""" self.pipeline._setup_views(self) def _build_window(self) -> None: central_widget = QWidget(self) self.setCentralWidget(central_widget) root_layout = QVBoxLayout(central_widget) root_layout.setContentsMargins(0, 0, 0, 0) root_layout.setSpacing(0) content_layout = QHBoxLayout() content_layout.setContentsMargins(0, 0, 0, 0) content_layout.setSpacing(0) self.left_panel = self._build_left_panel() content_layout.addWidget(self.left_panel) canvas_widget = self.canvas.native canvas_widget.setSizePolicy( QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Expanding ) content_layout.addWidget(canvas_widget, stretch=1) if self._enable_plots: content_layout.addWidget(self._create_plot_panel()) root_layout.addLayout(content_layout, stretch=1) root_layout.addWidget(self._build_transport_bar()) def _build_left_panel(self) -> QWidget: """Return the current-frame YAML context sidebar.""" panel = QWidget(self) panel.setObjectName("leftPanel") panel.setFixedWidth(self.LEFT_PANEL_WIDTH) layout = QVBoxLayout(panel) layout.setContentsMargins(0, 0, 0, 0) layout.setSpacing(0) self.context_yaml_view = QPlainTextEdit(panel) self.context_yaml_view.setObjectName("contextYamlView") self.context_yaml_view.setReadOnly(True) self.context_yaml_view.setLineWrapMode( QPlainTextEdit.LineWrapMode.NoWrap ) self.context_yaml_view.setFont(QFont("monospace")) self.context_yaml_view.setPlaceholderText("Waiting for frame context...") self._context_yaml_highlighter = _YamlSyntaxHighlighter( self.context_yaml_view.document() ) self.context_yaml_view.setStyleSheet( """ QPlainTextEdit#contextYamlView { border: 0; background: rgba(16, 16, 16, 0.96); color: rgba(245, 245, 245, 0.92); selection-background-color: rgba(24, 144, 255, 0.45); } """ ) layout.addWidget(self.context_yaml_view) return panel def _build_transport_bar(self) -> QWidget: bar = QFrame(self) bar.setObjectName("bottomTransportBar") bar.setFrameShape(QFrame.Shape.NoFrame) bar.setStyleSheet( """ #bottomTransportBar { border-top: 1px solid rgba(255, 255, 255, 0.14); background: rgba(20, 20, 20, 0.92); } QToolButton { padding: 5px 7px; border-radius: 4px; border: 1px solid transparent; } QToolButton:hover { background: rgba(255, 255, 255, 0.08); border-color: rgba(255, 255, 255, 0.10); } QToolButton:pressed { background: rgba(255, 255, 255, 0.16); border-color: rgba(255, 255, 255, 0.20); } QToolButton:checked { background: rgba(24, 144, 255, 0.32); border-color: rgba(24, 144, 255, 0.60); } QSlider::groove:horizontal { height: 6px; background: rgba(255, 255, 255, 0.18); border-radius: 3px; } QSlider::sub-page:horizontal { background: rgba(24, 144, 255, 0.85); border-radius: 3px; } QSlider::add-page:horizontal { background: rgba(255, 255, 255, 0.10); border-radius: 3px; } QSlider::handle:horizontal { width: 14px; height: 14px; margin: -5px 0; border-radius: 7px; background: white; } """ ) layout = QHBoxLayout(bar) layout.setContentsMargins(10, 6, 12, 6) layout.setSpacing(8) layout.addWidget(self._build_dataset_toolbar()) layout.addWidget(self._build_playback_toolbar()) layout.addWidget(self._build_capture_toolbar()) self.lbl_index = QLabel("Live" if not self._is_replay_mode else "-- / 0", bar) self.lbl_index.setObjectName("frameIndexLabel") self.lbl_index.setMinimumWidth(72) self.lbl_index.setAlignment( Qt.AlignmentFlag.AlignRight | Qt.AlignmentFlag.AlignVCenter ) layout.addWidget(self.lbl_index) self.frame_slider = self._build_frame_slider(bar) layout.addWidget(self.frame_slider, stretch=1) self.lbl_pipeline_stats = QLabel("Pipeline: -- FPS | -- ms", bar) self.lbl_pipeline_stats.setObjectName("pipelineStatsLabel") self.lbl_pipeline_stats.setMinimumWidth(190) self.lbl_pipeline_stats.setAlignment( Qt.AlignmentFlag.AlignRight | Qt.AlignmentFlag.AlignVCenter ) layout.addWidget(self.lbl_pipeline_stats) return bar def _build_dataset_toolbar(self) -> QToolBar: toolbar = QToolBar("Dataset", self) toolbar.setObjectName("datasetToolbar") toolbar.setIconSize(QSize(22, 22)) toolbar.setMovable(False) toolbar.setFloatable(False) toolbar.setContextMenuPolicy(Qt.ContextMenuPolicy.PreventContextMenu) toolbar.setStyleSheet("QToolBar { border: 0; background: transparent; }") load_icon = QIcon.fromTheme("folder-open") if load_icon.isNull(): load_icon = self.style().standardIcon(QStyle.StandardPixmap.SP_DirOpenIcon) self.act_load_dataset = QAction(load_icon, "Load Dataset", self) self.act_load_dataset.setEnabled(self._is_replay_mode) self.act_load_dataset.triggered.connect(self._on_load_dataset_clicked) toolbar.addAction(self.act_load_dataset) return toolbar def _build_playback_toolbar(self) -> QToolBar: toolbar = QToolBar("Playback", self) toolbar.setObjectName("playbackToolbar") toolbar.setIconSize(QSize(24, 24)) toolbar.setMovable(False) toolbar.setFloatable(False) toolbar.setContextMenuPolicy(Qt.ContextMenuPolicy.PreventContextMenu) toolbar.setStyleSheet("QToolBar { border: 0; background: transparent; }") style = self.style() self.act_prev = QAction( style.standardIcon(QStyle.StandardPixmap.SP_MediaSeekBackward), "Backward", self, ) self._icon_play = style.standardIcon(QStyle.StandardPixmap.SP_MediaPlay) self._icon_pause = style.standardIcon(QStyle.StandardPixmap.SP_MediaPause) self.act_play_pause = QAction(self._icon_pause, "Pause", self) # Backwards-compatible alias for callers that access the old play action. self.act_play = self.act_play_pause self.act_stop = QAction( style.standardIcon(QStyle.StandardPixmap.SP_MediaStop), "Stop", self, ) self.act_next = QAction( style.standardIcon(QStyle.StandardPixmap.SP_MediaSeekForward), "Forward", self, ) self.act_play_pause.setCheckable(False) self.act_stop.setCheckable(False) self.act_prev.triggered.connect(self.toggle_backward) self.act_play_pause.triggered.connect(self.toggle_play_pause) self.act_stop.triggered.connect(self.toggle_stop) self.act_next.triggered.connect(self.toggle_forward) for action in ( self.act_prev, self.act_play_pause, self.act_stop, self.act_next, ): action.setEnabled(self._is_replay_mode) toolbar.addAction(action) return toolbar def _build_capture_toolbar(self) -> QToolBar: toolbar = QToolBar("Capture", self) toolbar.setObjectName("captureToolbar") toolbar.setIconSize(QSize(22, 22)) toolbar.setMovable(False) toolbar.setFloatable(False) toolbar.setContextMenuPolicy(Qt.ContextMenuPolicy.PreventContextMenu) toolbar.setStyleSheet("QToolBar { border: 0; background: transparent; }") snapshot_icon = QIcon.fromTheme("camera-photo") if snapshot_icon.isNull(): snapshot_icon = self.style().standardIcon(QStyle.StandardPixmap.SP_DialogSaveButton) record_icon = QIcon.fromTheme("media-record") if record_icon.isNull(): record_icon = self.style().standardIcon(QStyle.StandardPixmap.SP_DialogYesButton) self._icon_record = record_icon self._icon_stop = self.style().standardIcon(QStyle.StandardPixmap.SP_MediaStop) self.act_snapshot = QAction(snapshot_icon, "Snapshot", self) self.act_snapshot.triggered.connect(self._on_snapshot_clicked) toolbar.addAction(self.act_snapshot) self.act_record = QAction(self._icon_record, "Start Recording", self) self.act_record.setCheckable(True) self.act_record.triggered.connect(self._on_record_toggled) toolbar.addAction(self.act_record) return toolbar def _build_frame_slider(self, parent: QWidget) -> QSlider: slider = _JumpToClickSlider(Qt.Orientation.Horizontal, parent) slider.setObjectName("frameSlider") slider.setRange(0, 0) slider.setSingleStep(1) slider.setPageStep(30) slider.setTracking(False) slider.setToolTip("Frame") slider.setEnabled(False) slider.sliderPressed.connect(self._on_seek_pressed) slider.sliderMoved.connect(self._on_seek_moved) slider.sliderReleased.connect(self._on_seek_released) slider.valueChanged.connect(self._on_seek_value_changed) return slider def show_canvas(self, data_source: Queue) -> None: """Connect a queue and start the VisPy update loop.""" self.q = data_source self.canvas.show() self._start_timer() def connect(self, data_source: Queue) -> None: """Connect the window to the processed context queue.""" self.show_canvas(data_source) def _start_timer(self) -> None: self.timer = app.Timer(interval=self.TIMER_INTERVAL_SEC) self.timer.connect(self._update) self.timer.start() def _update(self, _event: Any) -> None: """Render the newest queued worker message.""" if self.q is None: return latest = None for item in iter_queue_nowait(self.q): latest = self._coerce_frame_message(item) if latest is None: return context, state = latest if self._should_ignore_pending_seek_message(context, state): return self._render_context(context, state) def _coerce_frame_message( self, item: Any ) -> Optional[tuple[dict[str, Any], Optional[TrackState]]]: if isinstance(item, TrackFrameMessage): return item.context, item.state if isinstance(item, dict): return item, None context = getattr(item, "context", None) state = getattr(item, "state", None) if isinstance(context, dict): return context, state return None def _render_context( self, context: dict[str, Any], state: Optional[TrackState] = None, ) -> None: self.visual_tree.update(context) self.canvas.update() if state is not None: self._apply_track_state(state) else: if not self.has_set_total_frame: self.set_total_frames(context.get("frame_length", 1)) self.has_set_total_frame = True frame_serial = int(context.get("frame_serial", 0)) self.update_current_frame(frame_serial) if self._enable_plots: self._update_plots(context) self._update_context_yaml(context) def _update_context_yaml(self, context: dict[str, Any]) -> None: try: yaml_text = self._context_yaml_serializer.dumps(context) except Exception as exc: yaml_text = yaml.safe_dump( {"context_yaml_error": str(exc)}, sort_keys=False, default_flow_style=False, ).rstrip() self._set_context_yaml_text(yaml_text) def _set_context_yaml_text(self, yaml_text: str) -> None: if self.context_yaml_view.toPlainText() == yaml_text: return vertical_scrollbar = self.context_yaml_view.verticalScrollBar() horizontal_scrollbar = self.context_yaml_view.horizontalScrollBar() vertical_value = vertical_scrollbar.value() horizontal_value = horizontal_scrollbar.value() self.context_yaml_view.setPlainText(yaml_text) vertical_scrollbar.setValue(min(vertical_value, vertical_scrollbar.maximum())) horizontal_scrollbar.setValue( min(horizontal_value, horizontal_scrollbar.maximum()) ) def _apply_track_state(self, state: TrackState) -> None: self._is_replay_mode = state.source_kind == "replay" playback_state = self._effective_playback_state(state) self._playback_state = playback_state self._set_replay_controls_enabled(self._is_replay_mode) self.set_total_frames(state.frame_length) self.update_current_frame(state.frame_serial) self._set_playback_checked(playback_state) self._set_recording_state(state.is_recording) self._set_pipeline_stats(state.pipeline_fps, state.pipeline_latency_ms) def _effective_playback_state(self, state: TrackState) -> PlaybackMode: if state.source_kind != "replay": self._pending_playback_state = None return state.playback_state if self._pending_playback_state is None: return state.playback_state if state.playback_state == self._pending_playback_state: self._pending_playback_state = None return state.playback_state return self._pending_playback_state def _put_command(self, command: CommandSet, payload: Any) -> bool: if self.command_queue is None: return False self.command_queue.put((command, payload)) return True def _context_frame_serial(self, context: dict[str, Any]) -> int: try: return int(context.get("frame_serial", 0)) except (TypeError, ValueError): return 0 def _should_ignore_pending_seek_message( self, context: dict[str, Any], state: Optional[TrackState], ) -> bool: if self._pending_seek_frame is None: return False if state is None: frame_serial = self._context_frame_serial(context) if frame_serial == self._pending_seek_frame: self._pending_seek_frame = None self._last_seek_command_frame = None return False return True if state.source_kind != "replay": self._pending_seek_frame = None self._last_seek_command_frame = None return False is_seek_ack = ( state.frame_serial == self._pending_seek_frame and state.playback_state != PlaybackMode.PLAY ) if is_seek_ack: self._pending_seek_frame = None self._last_seek_command_frame = None return False return True def toggle_play(self) -> None: self._pending_seek_frame = None self._last_seek_command_frame = None self._set_requested_playback_state(PlaybackMode.PLAY) self._put_command(CommandSet.PLAYBACKCONTROL, PlaybackMode.PLAY) def toggle_pause(self) -> None: self._pending_seek_frame = None self._last_seek_command_frame = None self._set_requested_playback_state(PlaybackMode.PAUSE) self._put_command(CommandSet.PLAYBACKCONTROL, PlaybackMode.PAUSE) def toggle_play_pause(self) -> None: if self._playback_state == PlaybackMode.PLAY: self.toggle_pause() else: self.toggle_play() def toggle_backward(self) -> None: self._pending_seek_frame = None self._last_seek_command_frame = None self._set_requested_playback_state(PlaybackMode.PAUSE) self._put_command(CommandSet.PLAYBACKCONTROL, PlaybackMode.BACKWARD) def toggle_forward(self) -> None: self._pending_seek_frame = None self._last_seek_command_frame = None self._set_requested_playback_state(PlaybackMode.PAUSE) self._put_command(CommandSet.PLAYBACKCONTROL, PlaybackMode.FORWARD) def toggle_stop(self) -> None: self._pending_seek_frame = None self._last_seek_command_frame = None self._set_requested_playback_state(PlaybackMode.STOP) self._put_command(CommandSet.PLAYBACKCONTROL, PlaybackMode.STOP) def _set_requested_playback_state(self, playback_state: PlaybackMode) -> None: self._pending_playback_state = playback_state self._playback_state = playback_state self._set_play_pause_action_state(playback_state) def _on_snapshot_clicked(self) -> None: self._put_command(CommandSet.TAKESNAPSHOT, None) def _on_record_toggled(self, checked: bool) -> None: self._is_recording = bool(checked) self.act_record.setText("Stop Recording" if checked else "Start Recording") self.act_record.setIcon(self._icon_stop if checked else self._icon_record) self._put_command(CommandSet.RECORD, bool(checked)) def _on_load_dataset_clicked(self) -> None: if not self._is_replay_mode: return dataset_dir = QFileDialog.getExistingDirectory( self, "Load Dataset", self._default_dataset_dialog_dir(), ) if not dataset_dir: return self._dataset_path = dataset_dir self._pending_seek_frame = None self._last_seek_command_frame = None self._pending_playback_state = None self.has_set_total_frame = False self._current_frame = 0 self.set_total_frames(0) self._set_pipeline_stats(None, None) self._put_command(CommandSet.CHANGEDATASET, dataset_dir) @staticmethod def _default_dataset_dialog_dir() -> str: temp_dir = os.path.abspath("temp") if os.path.isdir(temp_dir): return temp_dir return os.getcwd() def slider_value_change(self, frame_number: int) -> None: try: frame_number = int(frame_number) except (TypeError, ValueError): return if self._total_frames > 0: frame_number = max(0, min(frame_number, self._total_frames - 1)) blocker = QSignalBlocker(self.frame_slider) self.frame_slider.setValue(frame_number) self.frame_slider.setSliderPosition(frame_number) del blocker self._current_frame = frame_number self._set_frame_label(frame_number) if frame_number == self._last_seek_command_frame: return if self._put_command(CommandSet.GOTOFRAME, frame_number): self._pending_seek_frame = frame_number self._last_seek_command_frame = frame_number self._set_requested_playback_state(PlaybackMode.PAUSE) def _on_seek_pressed(self) -> None: self._is_scrubbing = True def _on_seek_moved(self, value: int) -> None: self.slider_value_change(value) def _on_seek_value_changed(self, value: int) -> None: if not self._is_scrubbing: self.slider_value_change(value) def _on_seek_released(self) -> None: self._is_scrubbing = False self.slider_value_change(self.frame_slider.sliderPosition()) def _set_frame_label(self, frame_index: int) -> None: if self._total_frames <= 0: self.lbl_index.setText("-- / 0" if self._is_replay_mode else "Live") return frame_index = max(0, min(int(frame_index), self._total_frames - 1)) self.lbl_index.setText(f"{frame_index + 1} / {self._total_frames}") def update_current_frame(self, frame_index: int) -> None: if self._is_scrubbing or self._total_frames <= 0: return frame_index = max(0, min(int(frame_index), self._total_frames - 1)) blocker = QSignalBlocker(self.frame_slider) self.frame_slider.setValue(frame_index) del blocker self._current_frame = frame_index self._set_frame_label(frame_index) def set_total_frames(self, total: int) -> None: try: total = int(total) except (TypeError, ValueError): total = 0 self._total_frames = max(0, total) if self._total_frames <= 0: blocker = QSignalBlocker(self.frame_slider) self.frame_slider.setRange(0, 0) del blocker self.frame_slider.setEnabled(False) self._set_frame_label(0) return blocker = QSignalBlocker(self.frame_slider) self.frame_slider.setRange(0, self._total_frames - 1) del blocker self.frame_slider.setEnabled(self._is_replay_mode) self.update_current_frame(min(self._current_frame, self._total_frames - 1)) def _set_replay_controls_enabled(self, enabled: bool) -> None: self.act_load_dataset.setEnabled(enabled) for action in ( self.act_prev, self.act_play_pause, self.act_stop, self.act_next, ): action.setEnabled(enabled) self.frame_slider.setEnabled(enabled and self._total_frames > 0) def _set_playback_checked(self, playback_state: PlaybackMode) -> None: self._set_play_pause_action_state(playback_state) def _set_play_pause_action_state(self, playback_state: PlaybackMode) -> None: if playback_state == PlaybackMode.PLAY: self.act_play_pause.setIcon(self._icon_pause) self.act_play_pause.setText("Pause") return self.act_play_pause.setIcon(self._icon_play) self.act_play_pause.setText("Play") def _set_recording_state(self, is_recording: bool) -> None: self._is_recording = bool(is_recording) blocker = QSignalBlocker(self.act_record) self.act_record.setChecked(self._is_recording) del blocker self.act_record.setText( "Stop Recording" if self._is_recording else "Start Recording" ) self.act_record.setIcon( self._icon_stop if self._is_recording else self._icon_record ) def _set_pipeline_stats( self, pipeline_fps: Optional[float], pipeline_latency_ms: Optional[float], ) -> None: fps_text = "--" if pipeline_fps is None else f"{pipeline_fps:.1f}" latency_text = ( "--" if pipeline_latency_ms is None else f"{pipeline_latency_ms:.1f}" ) self.lbl_pipeline_stats.setText( f"Pipeline: {fps_text} FPS | {latency_text} ms" ) def _create_plot_panel(self) -> QWidget: widget = QWidget(self) widget.setObjectName("plotPanel") widget.setFixedWidth(self.PLOT_PANEL_WIDTH) layout = QVBoxLayout(widget) title = QLabel("Tip Point Trace", widget) title.setAlignment(Qt.AlignmentFlag.AlignCenter) layout.addWidget(title) self.plot_x = pg.PlotWidget(title="X Position") self.plot_x.setLabel("left", "X (um)") self.plot_x.setLabel("bottom", "Time (s)") self.curve_x = self.plot_x.plot(pen="r") layout.addWidget(self.plot_x) self.plot_y = pg.PlotWidget(title="Y Position") self.plot_y.setLabel("left", "Y (um)") self.plot_y.setLabel("bottom", "Time (s)") self.curve_y = self.plot_y.plot(pen="g") layout.addWidget(self.plot_y) self.plot_z = pg.PlotWidget(title="Z Position") self.plot_z.setLabel("left", "Z (um)") self.plot_z.setLabel("bottom", "Time (s)") self.curve_z = self.plot_z.plot(pen="b") layout.addWidget(self.plot_z) self.time_data = deque(maxlen=100) self.x_data = deque(maxlen=100) self.y_data = deque(maxlen=100) self.z_data = deque(maxlen=100) return widget def _update_plots(self, context: dict[str, Any]) -> None: if not hasattr(self, "_t0"): self._t0 = context.get("timestamp") solver = context.get("solver", {}) tippoint = solver.get("tippoint_b") if isinstance(solver, dict) else None if tippoint is None: return timestamp = context.get("timestamp") t0 = self._t0 if timestamp is None or t0 is None: plot_time = len(self.time_data) else: plot_time = timestamp - t0 self.time_data.append(plot_time) self.x_data.append(tippoint.x * 1e6) self.y_data.append(tippoint.y * 1e6) self.z_data.append(tippoint.z * 1e6) self.curve_x.setData(list(self.time_data), list(self.x_data)) self.curve_y.setData(list(self.time_data), list(self.y_data)) self.curve_z.setData(list(self.time_data), list(self.z_data))