diff --git a/src/aare/gui/panels/prediction_metrics_panel.py b/src/aare/gui/panels/prediction_metrics_panel.py new file mode 100644 index 00000000..4df322e5 --- /dev/null +++ b/src/aare/gui/panels/prediction_metrics_panel.py @@ -0,0 +1,791 @@ +""" +Prediction Metrics Panel - Real-time feedback on ML prediction stream. + +Displays: +- Confidence histogram distribution +- Object counts by class +- Frame timing / FPS +- Rolling statistics over time +- Optional ground-truth comparison (IoU, FP/FN tracking) +""" + +import csv +import time +from collections import deque +from dataclasses import dataclass, field + +from PySide6.QtCharts import ( + QChart, + QChartView, + QBarSeries, + QBarSet, + QBarCategoryAxis, + QValueAxis, + QLineSeries, +) +from PySide6.QtCore import Qt, Slot, QTimer +from PySide6.QtGui import QPainter, QColor, QPen +from PySide6.QtWidgets import ( + QWidget, + QVBoxLayout, + QHBoxLayout, + QLabel, + QPushButton, + QGroupBox, + QGridLayout, + QCheckBox, + QSpinBox, + QFileDialog, + QSplitter, + QScrollArea, + QFrame, +) + +from aare.common.logger_config import setup_logger +from aare.common.models import MLBoxType + +logger = setup_logger("aareGUI") + + +# ───────────────────────────────────────────────────────────────────────────── +# Data Models +# ───────────────────────────────────────────────────────────────────────────── + + +@dataclass +class PredictionFrame: + """Single frame of prediction data.""" + timestamp: float + boxes: list[dict] = field(default_factory=list) + frame_time_ms: float = 0.0 + class_counts: dict[str, int] = field(default_factory=dict) + confidences: list[float] = field(default_factory=list) + mean_confidence: float = 0.0 + max_confidence: float = 0.0 + + +@dataclass +class GroundTruthComparison: + """Comparison result against ground truth.""" + iou_scores: list[float] = field(default_factory=list) + false_positives: int = 0 + false_negatives: int = 0 + misclassifications: list[dict] = field(default_factory=list) + + +# ───────────────────────────────────────────────────────────────────────────── +# Class Name Utilities +# ───────────────────────────────────────────────────────────────────────────── + + +# Colors matching the bounding box colors in camera_image.py +CLASS_COLORS = { + "Crystal": "#0000ff", # blue + "Loop_face": "#ffff00", # yellow + "Loop_all": "#00ff00", # green + "Pin": "#ff0000", # red + "Ice": "#00ffff", # cyan + "Needle": "#ff00ff", # magenta +} + + +def get_class_name(cls_id: int) -> str: + """Convert class ID to human-readable name.""" + try: + return MLBoxType(cls_id).name + except (ValueError, KeyError): + return f"Class_{cls_id}" + + +def get_class_color(class_name: str) -> str: + """Get color for a class name (matches bounding box colors).""" + return CLASS_COLORS.get(class_name, "#888888") + + +# ───────────────────────────────────────────────────────────────────────────── +# Confidence Histogram Widget +# ───────────────────────────────────────────────────────────────────────────── + + +class ConfidenceHistogramWidget(QWidget): + """Real-time histogram of prediction confidence scores.""" + + BINS = [0.0, 0.2, 0.4, 0.6, 0.8, 1.0] + BIN_COLORS = ["#d62728", "#ff7f0e", "#ffbb78", "#98df8a", "#2ca02c"] + + def __init__(self, parent=None): + super().__init__(parent) + self._setup_ui() + + def _setup_ui(self): + layout = QVBoxLayout(self) + layout.setContentsMargins(0, 0, 0, 0) + + self.bar_set = QBarSet("Detections") + self.bar_set.append([0] * (len(self.BINS) - 1)) + + # Color each bar by confidence level + for i, color in enumerate(self.BIN_COLORS): + self.bar_set.setColor(QColor(color)) + + series = QBarSeries() + series.append(self.bar_set) + series.setBarWidth(0.8) + + self.chart = QChart() + self.chart.addSeries(series) + self.chart.setTitle("Confidence Distribution") + self.chart.setAnimationOptions(QChart.AnimationOption.NoAnimation) + self.chart.legend().setVisible(False) + + categories = [ + f"{self.BINS[i]:.1f}-{self.BINS[i + 1]:.1f}" + for i in range(len(self.BINS) - 1) + ] + self.axis_x = QBarCategoryAxis() + self.axis_x.append(categories) + self.chart.addAxis(self.axis_x, Qt.AlignmentFlag.AlignBottom) + series.attachAxis(self.axis_x) + + self.axis_y = QValueAxis() + self.axis_y.setRange(0, 10) + self.axis_y.setTitleText("Count") + self.axis_y.setLabelFormat("%d") + self.chart.addAxis(self.axis_y, Qt.AlignmentFlag.AlignLeft) + series.attachAxis(self.axis_y) + + chart_view = QChartView(self.chart) + chart_view.setRenderHint(QPainter.RenderHint.Antialiasing) + chart_view.setMinimumHeight(200) + + layout.addWidget(chart_view) + + def update_histogram(self, confidences: list[float]): + """Update histogram from list of confidence values.""" + counts = [0] * (len(self.BINS) - 1) + + for conf in confidences: + for i in range(len(self.BINS) - 1): + if self.BINS[i] <= conf < self.BINS[i + 1]: + counts[i] += 1 + break + # Handle conf == 1.0 edge case + if conf >= 1.0: + counts[-1] += 1 + + for i, count in enumerate(counts): + self.bar_set.replace(i, count) + + # Auto-scale Y axis + max_count = max(counts) if counts else 1 + self.axis_y.setRange(0, max(10, max_count + 2)) + + +# ───────────────────────────────────────────────────────────────────────────── +# Object Count Widget +# ───────────────────────────────────────────────────────────────────────────── + + +class ObjectCountWidget(QWidget): + """Displays real-time object counts by class with colored labels matching bounding boxes.""" + + def __init__(self, parent=None): + super().__init__(parent) + self._class_labels: dict[str, QLabel] = {} + self._setup_ui() + + def _setup_ui(self): + layout = QVBoxLayout(self) + layout.setContentsMargins(0, 0, 0, 0) + + group = QGroupBox("Object Counts") + self._grid = QGridLayout(group) + self._grid.setSpacing(8) + + # Pre-create labels for known classes with matching bounding box colors + known_classes = ["Crystal", "Loop_face", "Loop_all", "Pin", "Ice", "Needle"] + for i, cls_name in enumerate(known_classes): + color = get_class_color(cls_name) + + # Create colored indicator square + indicator = QLabel("■") + indicator.setStyleSheet(f"color: {color}; font-size: 16px;") + indicator.setFixedWidth(20) + + # Class name label + name_label = QLabel(f"{cls_name}:") + name_label.setStyleSheet("font-weight: bold;") + + # Count label with matching color + count_label = QLabel("0") + count_label.setStyleSheet(f"color: {color}; font-size: 14px; font-weight: bold;") + count_label.setAlignment(Qt.AlignmentFlag.AlignRight) + + row = i // 2 + col = (i % 2) * 3 + + self._grid.addWidget(indicator, row, col) + self._grid.addWidget(name_label, row, col + 1) + self._grid.addWidget(count_label, row, col + 2) + self._class_labels[cls_name] = count_label + + layout.addWidget(group) + + def update_counts(self, class_counts: dict[str, int]): + """Update displayed counts.""" + # Reset all to 0 first + for label in self._class_labels.values(): + label.setText("0") + + # Update with actual counts + for cls_name, count in class_counts.items(): + if cls_name in self._class_labels: + self._class_labels[cls_name].setText(str(count)) + + +# ───────────────────────────────────────────────────────────────────────────── +# Timing Stats Widget +# ───────────────────────────────────────────────────────────────────────────── + + +class TimingStatsWidget(QWidget): + """Displays frame timing, FPS, and processing latency.""" + + def __init__(self, parent=None): + super().__init__(parent) + self._setup_ui() + + def _setup_ui(self): + layout = QVBoxLayout(self) + layout.setContentsMargins(0, 0, 0, 0) + + group = QGroupBox("Timing") + grid = QGridLayout(group) + + # Frame time + grid.addWidget(QLabel("Frame time:"), 0, 0) + self.frame_time_label = QLabel("- ms") + self.frame_time_label.setStyleSheet("font-weight: bold;") + grid.addWidget(self.frame_time_label, 0, 1) + + # FPS + grid.addWidget(QLabel("FPS:"), 0, 2) + self.fps_label = QLabel("-") + self.fps_label.setStyleSheet("font-weight: bold;") + grid.addWidget(self.fps_label, 0, 3) + + # Detection count + grid.addWidget(QLabel("Detections:"), 1, 0) + self.detection_count_label = QLabel("0") + self.detection_count_label.setStyleSheet("font-weight: bold;") + grid.addWidget(self.detection_count_label, 1, 1) + + # Mean confidence + grid.addWidget(QLabel("Mean conf:"), 1, 2) + self.mean_conf_label = QLabel("-") + self.mean_conf_label.setStyleSheet("font-weight: bold;") + grid.addWidget(self.mean_conf_label, 1, 3) + + layout.addWidget(group) + + def update_stats( + self, + frame_time_ms: float, + fps: float, + detection_count: int, + mean_confidence: float, + ): + """Update timing statistics display.""" + self.frame_time_label.setText(f"{frame_time_ms:.1f} ms") + self.fps_label.setText(f"{fps:.1f}") + self.detection_count_label.setText(str(detection_count)) + + if mean_confidence > 0: + self.mean_conf_label.setText(f"{mean_confidence:.2f}") + else: + self.mean_conf_label.setText("-") + + +# ───────────────────────────────────────────────────────────────────────────── +# Error Tracking Widget +# ───────────────────────────────────────────────────────────────────────────── + + +class ErrorTrackingWidget(QWidget): + """ + Tracks and displays error metrics. + + Note: FP/FN tracking requires ground truth data to be meaningful. + Without ground truth, this shows detection statistics instead. + """ + + def __init__(self, parent=None): + super().__init__(parent) + self._false_positives = 0 + self._false_negatives = 0 + self._empty_frame_count = 0 + self._total_frames = 0 + self._low_confidence_count = 0 + self._confidence_threshold = 0.5 + self._setup_ui() + + def _setup_ui(self): + layout = QVBoxLayout(self) + layout.setContentsMargins(0, 0, 0, 0) + + group = QGroupBox("Detection Quality") + grid = QGridLayout(group) + + # Empty frames (no detections) + grid.addWidget(QLabel("Empty frames:"), 0, 0) + self.empty_frames_label = QLabel("0") + self.empty_frames_label.setStyleSheet("color: #d62728; font-weight: bold;") + grid.addWidget(self.empty_frames_label, 0, 1) + + # Low confidence detections + grid.addWidget(QLabel("Low conf (<0.5):"), 0, 2) + self.low_conf_label = QLabel("0") + self.low_conf_label.setStyleSheet("color: #ff7f0e; font-weight: bold;") + grid.addWidget(self.low_conf_label, 0, 3) + + # Detection rate + grid.addWidget(QLabel("Detection rate:"), 1, 0) + self.detection_rate_label = QLabel("-") + self.detection_rate_label.setStyleSheet("color: #2ca02c; font-weight: bold;") + grid.addWidget(self.detection_rate_label, 1, 1) + + # Reset button + self.reset_button = QPushButton("Reset") + self.reset_button.clicked.connect(self.reset_counters) + grid.addWidget(self.reset_button, 1, 3) + + layout.addWidget(group) + + def update_from_frame(self, frame: PredictionFrame): + """Update error tracking from a prediction frame.""" + self._total_frames += 1 + + if not frame.boxes: + self._empty_frame_count += 1 + + for conf in frame.confidences: + if conf < self._confidence_threshold: + self._low_confidence_count += 1 + + self._update_display() + + def _update_display(self): + self.empty_frames_label.setText(str(self._empty_frame_count)) + self.low_conf_label.setText(str(self._low_confidence_count)) + + if self._total_frames > 0: + rate = (self._total_frames - self._empty_frame_count) / self._total_frames + self.detection_rate_label.setText(f"{rate * 100:.1f}%") + else: + self.detection_rate_label.setText("-") + + def reset_counters(self): + """Reset all counters.""" + self._false_positives = 0 + self._false_negatives = 0 + self._empty_frame_count = 0 + self._total_frames = 0 + self._low_confidence_count = 0 + self._update_display() + + +# ───────────────────────────────────────────────────────────────────────────── +# Rolling Statistics Chart +# ───────────────────────────────────────────────────────────────────────────── + + +class RollingStatsChart(QWidget): + """Time-series chart showing rolling detection statistics.""" + + def __init__(self, window_seconds: float = 60.0, parent=None): + super().__init__(parent) + self._window_s = window_seconds + self._data: deque[PredictionFrame] = deque(maxlen=1000) + self._setup_ui() + + def _setup_ui(self): + layout = QVBoxLayout(self) + layout.setContentsMargins(0, 0, 0, 0) + + # Detection count series (left Y axis) + self.count_series = QLineSeries() + self.count_series.setName("Detections") + self.count_series.setPen(QPen(QColor("#1f77b4"), 2)) + + # Mean confidence series (right Y axis) + self.conf_series = QLineSeries() + self.conf_series.setName("Mean Confidence") + self.conf_series.setPen(QPen(QColor("#2ca02c"), 2)) + + self.chart = QChart() + self.chart.addSeries(self.count_series) + self.chart.addSeries(self.conf_series) + self.chart.setTitle("Rolling Statistics") + self.chart.setAnimationOptions(QChart.AnimationOption.NoAnimation) + self.chart.legend().setVisible(True) + + # Time axis + self.axis_x = QValueAxis() + self.axis_x.setTitleText("Time [s ago]") + self.axis_x.setRange(-self._window_s, 0) + self.chart.addAxis(self.axis_x, Qt.AlignmentFlag.AlignBottom) + + # Left Y axis for detection counts + self.axis_y_count = QValueAxis() + self.axis_y_count.setTitleText("Detections") + self.axis_y_count.setRange(0, 20) + self.chart.addAxis(self.axis_y_count, Qt.AlignmentFlag.AlignLeft) + + # Right Y axis for confidence (0-1 scale) + self.axis_y_conf = QValueAxis() + self.axis_y_conf.setTitleText("Confidence") + self.axis_y_conf.setRange(0, 1) + self.chart.addAxis(self.axis_y_conf, Qt.AlignmentFlag.AlignRight) + + # Attach series to axes + self.count_series.attachAxis(self.axis_x) + self.count_series.attachAxis(self.axis_y_count) + + self.conf_series.attachAxis(self.axis_x) + self.conf_series.attachAxis(self.axis_y_conf) + + chart_view = QChartView(self.chart) + chart_view.setRenderHint(QPainter.RenderHint.Antialiasing) + chart_view.setMinimumHeight(250) + + layout.addWidget(chart_view) + + def add_frame(self, frame: PredictionFrame): + """Add a prediction frame to the rolling data.""" + self._data.append(frame) + self._trim_old_data(frame.timestamp) + + def _trim_old_data(self, now: float): + """Remove data older than the window.""" + cutoff = now - self._window_s + while self._data and self._data[0].timestamp < cutoff: + self._data.popleft() + + def refresh(self): + """Refresh the chart display using bulk operations.""" + if not self._data: + return + + from PySide6.QtCore import QPointF + + now = self._data[-1].timestamp + + count_points = [] + conf_points = [] + + for frame in self._data: + t = frame.timestamp - now + count_points.append(QPointF(t, len(frame.boxes))) + conf_points.append(QPointF(t, frame.mean_confidence)) + + # Bulk replace for performance + self.count_series.replace(count_points) + self.conf_series.replace(conf_points) + + # Auto-scale detection count axis + if count_points: + max_count = max((p.y() for p in count_points), default=0) + self.axis_y_count.setRange(0, max(10, max_count + 2)) + + +# ───────────────────────────────────────────────────────────────────────────── +# Main Panel +# ───────────────────────────────────────────────────────────────────────────── + + +class PredictionMetricsPanel(QWidget): + """ + Comprehensive panel for real-time ML prediction feedback. + + Connect to PredictionSubscriber.prediction signal: + prediction_subscriber.prediction.connect(panel.update_from_prediction) + """ + + def __init__(self, parent=None): + super().__init__(parent) + self._last_frame_ts: float | None = None + self._frame_times: deque[float] = deque(maxlen=30) + self._history: deque[PredictionFrame] = deque(maxlen=5000) + self._paused = False + + self._setup_ui() + self._setup_refresh_timer() + + def _setup_ui(self): + layout = QVBoxLayout(self) + + # ─── Top Controls ─── + controls = QHBoxLayout() + + self.pause_button = QPushButton("Pause") + self.pause_button.setCheckable(True) + self.pause_button.toggled.connect(self._set_paused) + controls.addWidget(self.pause_button) + + self.export_button = QPushButton("Export CSV") + self.export_button.clicked.connect(self._export_csv) + controls.addWidget(self.export_button) + + self.clear_button = QPushButton("Clear History") + self.clear_button.clicked.connect(self._clear_history) + controls.addWidget(self.clear_button) + + controls.addSpacing(20) + + controls.addWidget(QLabel("Rolling window:")) + self.window_spin = QSpinBox() + self.window_spin.setRange(10, 300) + self.window_spin.setValue(60) + self.window_spin.setSuffix(" s") + self.window_spin.valueChanged.connect(self._update_window) + controls.addWidget(self.window_spin) + + controls.addStretch() + + self.status_label = QLabel("Waiting for predictions...") + self.status_label.setStyleSheet("color: #888;") + controls.addWidget(self.status_label) + + layout.addLayout(controls) + + # ─── Main Content with Splitter ─── + splitter = QSplitter(Qt.Orientation.Vertical) + + # Top section: stats widgets + top_widget = QWidget() + top_layout = QHBoxLayout(top_widget) + top_layout.setContentsMargins(0, 0, 0, 0) + + # Left column + left_col = QVBoxLayout() + self.histogram = ConfidenceHistogramWidget() + left_col.addWidget(self.histogram) + + self.object_counts = ObjectCountWidget() + left_col.addWidget(self.object_counts) + + top_layout.addLayout(left_col, stretch=1) + + # Right column + right_col = QVBoxLayout() + self.timing_stats = TimingStatsWidget() + right_col.addWidget(self.timing_stats) + + self.error_tracking = ErrorTrackingWidget() + right_col.addWidget(self.error_tracking) + + right_col.addStretch() + top_layout.addLayout(right_col, stretch=1) + + splitter.addWidget(top_widget) + + # Bottom section: rolling chart + self.rolling_chart = RollingStatsChart(window_seconds=60.0) + splitter.addWidget(self.rolling_chart) + + splitter.setSizes([300, 300]) + layout.addWidget(splitter) + + def _setup_refresh_timer(self): + """Setup timer to refresh charts periodically.""" + self._refresh_timer = QTimer(self) + self._refresh_timer.setInterval(500) # Reduce to 2 Hz (was 4 Hz) + self._refresh_timer.timeout.connect(self._refresh_display) + self._refresh_timer.start() + + # Separate slower timer for the expensive rolling chart + self._chart_refresh_timer = QTimer(self) + self._chart_refresh_timer.setInterval(1000) # 1 Hz for heavy chart + self._chart_refresh_timer.timeout.connect(self.rolling_chart.refresh) + self._chart_refresh_timer.start() + + def _set_paused(self, paused: bool): + """Pause/resume updates.""" + self._paused = paused + self.pause_button.setText("Resume" if paused else "Pause") + self._update_status() + + def _update_window(self, seconds: int): + """Update rolling window size.""" + self.rolling_chart._window_s = float(seconds) + + def _clear_history(self): + """Clear all historical data.""" + self._history.clear() + self._frame_times.clear() + self._last_frame_ts = None + self.error_tracking.reset_counters() + self.rolling_chart._data.clear() + self._refresh_display() + + def _update_status(self): + """Update status label.""" + if self._paused: + self.status_label.setText("Paused") + self.status_label.setStyleSheet("color: #ff7f0e;") + elif self._history: + count = len(self._history) + self.status_label.setText(f"Live: {count} frames recorded") + self.status_label.setStyleSheet("color: #2ca02c;") + else: + self.status_label.setText("Waiting for predictions...") + self.status_label.setStyleSheet("color: #888;") + + @Slot(dict) + def update_from_prediction(self, payload: dict): + """ + Update panel from prediction payload. + + Expected payload format: + { + "boxes": [ + {"cls": 0, "conf": 0.95, "x1": 100, "y1": 100, "x2": 200, "y2": 200}, + ... + ], + "frame_id": 12345, # optional + "time": 1234567890.123 # optional + } + """ + if self._paused: + return + + now = time.monotonic() + + # Calculate frame time + frame_time_ms = 0.0 + if self._last_frame_ts is not None: + frame_time_ms = (now - self._last_frame_ts) * 1000 + self._frame_times.append(frame_time_ms) + self._last_frame_ts = now + + # Parse boxes + boxes = payload.get("boxes", []) + + # Handle both list and dict box formats + if isinstance(boxes, dict): + boxes = list(boxes.values()) + + # Extract data + class_counts: dict[str, int] = {} + confidences: list[float] = [] + + for box in boxes: + if isinstance(box, dict): + cls_id = box.get("cls", box.get("class", -1)) + conf = box.get("conf", box.get("confidence", 0.0)) + else: + # Handle object with attributes + cls_id = getattr(box, "cls", -1) + conf = getattr(box, "conf", 0.0) + if hasattr(cls_id, "value"): + cls_id = cls_id.value + + cls_name = get_class_name(int(cls_id)) + class_counts[cls_name] = class_counts.get(cls_name, 0) + 1 + confidences.append(float(conf)) + + mean_conf = sum(confidences) / len(confidences) if confidences else 0.0 + max_conf = max(confidences) if confidences else 0.0 + + # Create frame record + frame = PredictionFrame( + timestamp=now, + boxes=boxes, + frame_time_ms=frame_time_ms, + class_counts=class_counts, + confidences=confidences, + mean_confidence=mean_conf, + max_confidence=max_conf, + ) + + self._history.append(frame) + self.rolling_chart.add_frame(frame) + self.error_tracking.update_from_frame(frame) + + self._update_status() + + def _refresh_display(self): + """Refresh all display components except rolling chart.""" + if not self._history: + return + + latest = self._history[-1] + + # Update histogram + self.histogram.update_histogram(latest.confidences) + + # Update object counts + self.object_counts.update_counts(latest.class_counts) + + # Update timing stats + avg_frame_time = ( + sum(self._frame_times) / len(self._frame_times) + if self._frame_times + else 0.0 + ) + fps = 1000.0 / avg_frame_time if avg_frame_time > 0 else 0.0 + + self.timing_stats.update_stats( + frame_time_ms=avg_frame_time, + fps=fps, + detection_count=len(latest.boxes), + mean_confidence=latest.mean_confidence, + ) + + def _export_csv(self): + """Export history to CSV file.""" + if not self._history: + logger.info("No prediction history to export") + return + + path, _ = QFileDialog.getSaveFileName( + self, + "Export Prediction Metrics", + "prediction_metrics.csv", + "CSV files (*.csv)", + ) + if not path: + return + + try: + with open(path, "w", newline="") as f: + writer = csv.writer(f) + writer.writerow([ + "timestamp", + "frame_time_ms", + "detection_count", + "mean_confidence", + "max_confidence", + "crystal_count", + "loop_face_count", + "loop_all_count", + "pin_count", + ]) + + for frame in self._history: + writer.writerow([ + f"{frame.timestamp:.6f}", + f"{frame.frame_time_ms:.2f}", + len(frame.boxes), + f"{frame.mean_confidence:.4f}", + f"{frame.max_confidence:.4f}", + frame.class_counts.get("Crystal", 0), + frame.class_counts.get("Loop_face", 0), + frame.class_counts.get("Loop_all", 0), + frame.class_counts.get("Pin", 0), + ]) + + logger.info(f"Exported {len(self._history)} frames to {path}") + + except Exception as e: + logger.error(f"Failed to export prediction metrics: {e}") \ No newline at end of file diff --git a/src/aare/gui/panels/target_stability_panel.py b/src/aare/gui/panels/target_stability_panel.py new file mode 100644 index 00000000..e65fc739 --- /dev/null +++ b/src/aare/gui/panels/target_stability_panel.py @@ -0,0 +1,1400 @@ +import csv +import math +import time +import threading +from collections import deque + +from PySide6.QtCharts import QChart, QChartView, QLineSeries, QValueAxis +from PySide6.QtCore import Qt, Slot, QTimer, QPointF, QPoint +from PySide6.QtGui import QPainter, QPen, QColor, QMouseEvent, QWheelEvent +from PySide6.QtWidgets import ( + QWidget, + QVBoxLayout, + QHBoxLayout, + QLabel, + QPushButton, + QDoubleSpinBox, + QFileDialog, + QCheckBox, + QComboBox, + QMessageBox, + QGroupBox, +) + +from aare.common.logger_config import setup_logger + +logger = setup_logger("aareGUI") + + +class InteractiveChartView(QChartView): + def __init__(self, chart: QChart, panel: "TargetStabilityPanel"): + super().__init__(chart) + self._panel = panel + self._last_pos: QPoint | None = None + self.setRenderHint(QPainter.RenderHint.Antialiasing) + self.setRubberBand(QChartView.RubberBand.NoRubberBand) + self.setMouseTracking(True) + + def wheelEvent(self, event: QWheelEvent) -> None: + delta = event.angleDelta().y() + if delta == 0: + event.ignore() + return + + factor = 0.85 if delta > 0 else 1.0 / 0.85 + + if event.modifiers() & Qt.KeyboardModifier.ShiftModifier: + self._panel._zoom_score_axis(factor) + event.accept() + return + + self._panel._zoom_selected_wheel_axis(factor) + event.accept() + + def mousePressEvent(self, event: QMouseEvent) -> None: + if event.button() == Qt.MouseButton.LeftButton: + click_pos = event.position() + chart_pos = self.chart().mapToValue(click_pos.toPoint()) + + if self._panel._try_select_series_at_point(click_pos.toPoint()): + event.accept() + return + + self._last_pos = event.position().toPoint() + self.setCursor(Qt.CursorShape.ClosedHandCursor) + event.accept() + return + + if event.button() == Qt.MouseButton.RightButton: + self._panel._reset_view() + event.accept() + return + + super().mousePressEvent(event) + + def mouseMoveEvent(self, event: QMouseEvent) -> None: + if self._last_pos is not None and (event.buttons() & Qt.MouseButton.LeftButton): + pos = event.position().toPoint() + delta = pos - self._last_pos + self._last_pos = pos + self._panel._pan_selected_axes(delta.x(), delta.y(), self.viewport().width(), self.viewport().height()) + event.accept() + return + + super().mouseMoveEvent(event) + + def mouseReleaseEvent(self, event: QMouseEvent) -> None: + if event.button() == Qt.MouseButton.LeftButton: + self._last_pos = None + self.setCursor(Qt.CursorShape.ArrowCursor) + event.accept() + return + + super().mouseReleaseEvent(event) + + +class TargetStabilityPanel(QWidget): + SIGMA_COLOR = "#1f77b4" + SIGMA_X_COLOR = "#6baed6" + SIGMA_Y_COLOR = "#9ecae1" + + DISTANCE_COLOR = "#d62728" + DX_COLOR = "#ff9896" + DY_COLOR = "#c43c39" + + SCORE_COLOR = "#ff7f0e" + STEP_COLOR = "#17becf" + + TARGET_COLOR = "#2ca02c" + BEAM_COLOR = "#9467bd" + + SCORE_FROM_STEP_XY = "Step XY" + SCORE_FROM_SIGMA_XY = "Sigma XY" + + WHEEL_X = "X axis" + WHEEL_LEFT_Y = "Left Y" + WHEEL_RIGHT_Y = "Right Y" + WHEEL_SCORE = "Score" + + PAN_ALL = "All visible" + PAN_X = "X only" + PAN_LEFT_Y = "Left Y only" + PAN_RIGHT_Y = "Right Y only" + PAN_SCORE = "Score only" + + TRACE_TARGET_SIGMA = "sigma" + TRACE_TARGET_SIGMA_X = "sigma_x" + TRACE_TARGET_SIGMA_Y = "sigma_y" + TRACE_TARGET_DISTANCE = "distance" + TRACE_TARGET_DX = "dx" + TRACE_TARGET_DY = "dy" + TRACE_TARGET_SCORE = "score" + TRACE_TARGET_STEP = "step" + + def __init__(self, parent=None): + super().__init__(parent) + + self._live_window_s = 60.0 + self._sigma_window_s = 1.0 + self._paused = False + self._beam_center: tuple[float, float] | None = None + self._latest_target_point: tuple[float, float] | None = None + self._auto_scale_enabled = True + self._active_trace_target = self.TRACE_TARGET_SIGMA + + self._metrics_update_interval_s = 0.5 + self._last_metrics_update_ts = 0.0 + self._stability_ref_px = 1.0 + + self._samples = deque(maxlen=10000) + self._plot_points = deque(maxlen=10000) + self._sigma_samples = deque(maxlen=10000) + + self._rolling_count = 0 + self._rolling_sum_dx = 0.0 + self._rolling_sum_dy = 0.0 + self._rolling_sum_dx2 = 0.0 + self._rolling_sum_dy2 = 0.0 + + self._kahan_c_dx = 0.0 + self._kahan_c_dy = 0.0 + self._kahan_c_dx2 = 0.0 + self._kahan_c_dy2 = 0.0 + + self._collecting_until: float | None = None + self._collected_samples: list[dict] = [] + self._frozen_plot_points: list[dict] | None = None + + self._data_lock = threading.Lock() + + layout = QVBoxLayout(self) + + controls = QHBoxLayout() + self.pause_button = QPushButton("Pause") + self.pause_button.setCheckable(True) + self.pause_button.toggled.connect(self._set_paused) + + self.seconds_spin = QDoubleSpinBox() + self.seconds_spin.setRange(1.0, 600.0) + self.seconds_spin.setDecimals(1) + self.seconds_spin.setSingleStep(1.0) + self.seconds_spin.setValue(10.0) + self.seconds_spin.setSuffix(" s") + self.seconds_spin.valueChanged.connect(self._update_action_labels) + + self.avg_spin = QDoubleSpinBox() + self.avg_spin.setRange(0.02, 60.0) + self.avg_spin.setDecimals(2) + self.avg_spin.setSingleStep(0.02) + self.avg_spin.setValue(1.0) + self.avg_spin.setSuffix(" s") + self.avg_spin.valueChanged.connect(self._set_sigma_window) + + self.save_button = QPushButton() + self.save_button.clicked.connect(self._save_last_x_seconds) + + self.collect_button = QPushButton() + self.collect_button.clicked.connect(self._collect_next_x_seconds) + + self.auto_scale_button = QPushButton("Auto scale") + self.auto_scale_button.clicked.connect(self._reset_view) + + self.help_button = QPushButton("Metrics help") + self.help_button.clicked.connect(self._show_metrics_help) + + self.score_basis_combo = QComboBox() + self.score_basis_combo.addItems([self.SCORE_FROM_STEP_XY, self.SCORE_FROM_SIGMA_XY]) + self.score_basis_combo.setCurrentText(self.SCORE_FROM_STEP_XY) + self.score_basis_combo.currentTextChanged.connect(self._on_score_basis_changed) + + self.wheel_mode_combo = QComboBox() + self.wheel_mode_combo.addItems([ + self.WHEEL_X, + self.WHEEL_LEFT_Y, + self.WHEEL_RIGHT_Y, + self.WHEEL_SCORE, + ]) + self.wheel_mode_combo.setCurrentText(self.WHEEL_X) + self.wheel_mode_combo.currentTextChanged.connect(lambda _text: self._update_status_label()) + + self.pan_mode_combo = QComboBox() + self.pan_mode_combo.addItems([ + self.PAN_ALL, + self.PAN_X, + self.PAN_LEFT_Y, + self.PAN_RIGHT_Y, + self.PAN_SCORE, + ]) + self.pan_mode_combo.setCurrentText(self.PAN_ALL) + self.pan_mode_combo.currentTextChanged.connect(lambda _text: self._update_status_label()) + + self.status_label = QLabel("Waiting for target-point data...") + self.status_label.setAlignment(Qt.AlignmentFlag.AlignLeft | Qt.AlignmentFlag.AlignVCenter) + + self.metrics_label = QLabel("Target: (-, -) | Beam: (-, -) | Distance: - px | Std dev: - px") + self.metrics_label.setAlignment(Qt.AlignmentFlag.AlignLeft | Qt.AlignmentFlag.AlignVCenter) + self.metrics_label.setTextFormat(Qt.TextFormat.RichText) + self.metrics_label.setWordWrap(True) + + self.controls_legend_label = QLabel( + "Mouse: wheel=selected axis zoom | Shift+wheel=score | left-drag=selected pan mode | right-click=reset | click a trace point to target its axis" + ) + self.controls_legend_label.setAlignment(Qt.AlignmentFlag.AlignLeft | Qt.AlignmentFlag.AlignVCenter) + self.controls_legend_label.setWordWrap(True) + + controls.addWidget(self.pause_button) + controls.addWidget(QLabel("Window:")) + controls.addWidget(self.seconds_spin) + controls.addWidget(self.save_button) + controls.addWidget(self.collect_button) + controls.addWidget(self.auto_scale_button) + controls.addWidget(self.help_button) + controls.addSpacing(8) + controls.addWidget(QLabel("Score from:")) + controls.addWidget(self.score_basis_combo) + controls.addSpacing(8) + controls.addWidget(QLabel("Wheel:")) + controls.addWidget(self.wheel_mode_combo) + controls.addWidget(QLabel("Avg:")) + controls.addWidget(self.avg_spin) + controls.addWidget(QLabel("Pan:")) + controls.addWidget(self.pan_mode_combo) + controls.addStretch() + + self.series = QLineSeries() + self.series.setName("Rolling σ") + self.series.setPointsVisible(True) + self.series.setPen(QPen(QColor(self.SIGMA_COLOR), 2.0)) + + self.sigma_x_series = QLineSeries() + self.sigma_x_series.setName("σx") + self.sigma_x_series.setPointsVisible(True) + self.sigma_x_series.setPen(QPen(QColor(self.SIGMA_X_COLOR), 2.0)) + + self.sigma_y_series = QLineSeries() + self.sigma_y_series.setName("σy") + self.sigma_y_series.setPointsVisible(True) + self.sigma_y_series.setPen(QPen(QColor(self.SIGMA_Y_COLOR), 2.0)) + + self.distance_series = QLineSeries() + self.distance_series.setName("Distance") + self.distance_series.setPointsVisible(True) + self.distance_series.setPen(QPen(QColor(self.DISTANCE_COLOR), 2.0)) + + self.dx_series = QLineSeries() + self.dx_series.setName("Distance X (dx)") + self.dx_series.setPointsVisible(True) + self.dx_series.setPen(QPen(QColor(self.DX_COLOR), 2.0)) + + self.dy_series = QLineSeries() + self.dy_series.setName("Distance Y (dy)") + self.dy_series.setPointsVisible(True) + self.dy_series.setPen(QPen(QColor(self.DY_COLOR), 2.0)) + + self.score_series = QLineSeries() + self.score_series.setName("Stability score") + self.score_series.setPointsVisible(True) + self.score_series.setPen(QPen(QColor(self.SCORE_COLOR), 2.0)) + + self.step_series = QLineSeries() + self.step_series.setName("Rolling RMS step motion") + self.step_series.setPointsVisible(True) + self.step_series.setPen(QPen(QColor(self.STEP_COLOR), 2.0)) + + self.chart = QChart() + for series in ( + self.series, + self.sigma_x_series, + self.sigma_y_series, + self.distance_series, + self.dx_series, + self.dy_series, + self.score_series, + self.step_series, + ): + self.chart.addSeries(series) + + self.chart.legend().setVisible(True) + self.chart.setTitle("Target stability, distance, score, and step motion (last 60 s)") + + self.axis_x = QValueAxis() + self.axis_x.setTitleText("Time [s ago]") + self.axis_x.setRange(-self._live_window_s, 0.0) + + self.axis_y = QValueAxis() + self.axis_y.setTitleText("Sigma / step [px]") + self.axis_y.setRange(0.0, 10.0) + + self.axis_y_distance = QValueAxis() + self.axis_y_distance.setTitleText("Distance / offset [px]") + self.axis_y_distance.setRange(-10.0, 10.0) + + self.axis_y_score = QValueAxis() + self.axis_y_score.setTitleText("Score [0-100]") + self.axis_y_score.setRange(0.0, 100.0) + + self.chart.addAxis(self.axis_x, Qt.AlignmentFlag.AlignBottom) + self.chart.addAxis(self.axis_y, Qt.AlignmentFlag.AlignLeft) + self.chart.addAxis(self.axis_y_distance, Qt.AlignmentFlag.AlignRight) + self.chart.addAxis(self.axis_y_score, Qt.AlignmentFlag.AlignRight) + + for series in (self.series, self.sigma_x_series, self.sigma_y_series, self.step_series): + series.attachAxis(self.axis_x) + series.attachAxis(self.axis_y) + + for series in (self.distance_series, self.dx_series, self.dy_series): + series.attachAxis(self.axis_x) + series.attachAxis(self.axis_y_distance) + + self.score_series.attachAxis(self.axis_x) + self.score_series.attachAxis(self.axis_y_score) + + self._connect_series_selection(self.series, self.TRACE_TARGET_SIGMA) + self._connect_series_selection(self.sigma_x_series, self.TRACE_TARGET_SIGMA_X) + self._connect_series_selection(self.sigma_y_series, self.TRACE_TARGET_SIGMA_Y) + self._connect_series_selection(self.distance_series, self.TRACE_TARGET_DISTANCE) + self._connect_series_selection(self.dx_series, self.TRACE_TARGET_DX) + self._connect_series_selection(self.dy_series, self.TRACE_TARGET_DY) + self._connect_series_selection(self.score_series, self.TRACE_TARGET_SCORE) + self._connect_series_selection(self.step_series, self.TRACE_TARGET_STEP) + + self.show_sigma_cb = QCheckBox("Rolling σ") + self.show_sigma_cb.setChecked(True) + self.show_sigma_cb.toggled.connect(self._update_trace_visibility) + self._style_trace_checkbox(self.show_sigma_cb, self.SIGMA_COLOR) + + self.show_sigma_x_cb = QCheckBox("σx") + self.show_sigma_x_cb.setChecked(False) + self.show_sigma_x_cb.toggled.connect(self._update_trace_visibility) + self._style_trace_checkbox(self.show_sigma_x_cb, self.SIGMA_X_COLOR) + + self.show_sigma_y_cb = QCheckBox("σy") + self.show_sigma_y_cb.setChecked(False) + self.show_sigma_y_cb.toggled.connect(self._update_trace_visibility) + self._style_trace_checkbox(self.show_sigma_y_cb, self.SIGMA_Y_COLOR) + + self.show_distance_cb = QCheckBox("Distance") + self.show_distance_cb.setChecked(True) + self.show_distance_cb.toggled.connect(self._update_trace_visibility) + self._style_trace_checkbox(self.show_distance_cb, self.DISTANCE_COLOR) + + self.show_dx_cb = QCheckBox("Distance X (dx)") + self.show_dx_cb.setChecked(False) + self.show_dx_cb.toggled.connect(self._update_trace_visibility) + self._style_trace_checkbox(self.show_dx_cb, self.DX_COLOR) + + self.show_dy_cb = QCheckBox("Distance Y (dy)") + self.show_dy_cb.setChecked(False) + self.show_dy_cb.toggled.connect(self._update_trace_visibility) + self._style_trace_checkbox(self.show_dy_cb, self.DY_COLOR) + + self.show_score_cb = QCheckBox("Score") + self.show_score_cb.setChecked(False) + self.show_score_cb.toggled.connect(self._update_trace_visibility) + self._style_trace_checkbox(self.show_score_cb, self.SCORE_COLOR) + + self.show_step_cb = QCheckBox("Step XY") + self.show_step_cb.setChecked(False) + self.show_step_cb.toggled.connect(self._update_trace_visibility) + self._style_trace_checkbox(self.show_step_cb, self.STEP_COLOR) + + trace_group = QGroupBox("Traces") + trace_layout = QVBoxLayout(trace_group) + trace_layout.setContentsMargins(8, 8, 8, 8) + trace_layout.setSpacing(6) + trace_layout.addWidget(self.show_sigma_cb) + trace_layout.addWidget(self.show_sigma_x_cb) + trace_layout.addWidget(self.show_sigma_y_cb) + trace_layout.addWidget(self.show_distance_cb) + trace_layout.addWidget(self.show_dx_cb) + trace_layout.addWidget(self.show_dy_cb) + trace_layout.addWidget(self.show_step_cb) + trace_layout.addWidget(self.show_score_cb) + trace_layout.addStretch() + + right_panel = QVBoxLayout() + right_panel.addWidget(trace_group) + right_panel.addStretch() + + self.chart_view = InteractiveChartView(self.chart, self) + + plot_row = QHBoxLayout() + plot_row.addWidget(self.chart_view, 1) + plot_row.addLayout(right_panel) + + layout.addLayout(controls) + layout.addWidget(self.status_label) + layout.addWidget(self.metrics_label) + layout.addWidget(self.controls_legend_label) + layout.addLayout(plot_row) + + self._update_action_labels() + self._update_trace_visibility() + self._update_metrics_label(force=True) + + self._refresh_timer = QTimer(self) + self._refresh_timer.setInterval(250) + self._refresh_timer.timeout.connect(self._refresh_chart) + self._refresh_timer.start() + + def _style_trace_checkbox(self, checkbox: QCheckBox, color_hex: str) -> None: + checkbox.setStyleSheet( + f""" + QCheckBox {{ + color: {color_hex}; + font-weight: 600; + }} + """ + ) + + def _connect_series_selection(self, series: QLineSeries, target_name: str) -> None: + series.clicked.connect(lambda _point, target=target_name: self._select_trace_target(target)) + + def _try_select_series_at_point(self, view_pos: QPoint) -> bool: + """Check if click is near a visible series point and select it. Returns True if found.""" + hit_threshold = 10 # pixels + + series_map = { + self.series: (self.TRACE_TARGET_SIGMA, self.show_sigma_cb), + self.sigma_x_series: (self.TRACE_TARGET_SIGMA_X, self.show_sigma_x_cb), + self.sigma_y_series: (self.TRACE_TARGET_SIGMA_Y, self.show_sigma_y_cb), + self.distance_series: (self.TRACE_TARGET_DISTANCE, self.show_distance_cb), + self.dx_series: (self.TRACE_TARGET_DX, self.show_dx_cb), + self.dy_series: (self.TRACE_TARGET_DY, self.show_dy_cb), + self.score_series: (self.TRACE_TARGET_SCORE, self.show_score_cb), + self.step_series: (self.TRACE_TARGET_STEP, self.show_step_cb), + } + + for series, (target_name, checkbox) in series_map.items(): + if not checkbox.isChecked(): + continue + for i in range(series.count()): + point = series.at(i) + screen_pos = self.chart.mapToPosition(point, series) + dx = abs(screen_pos.x() - view_pos.x()) + dy = abs(screen_pos.y() - view_pos.y()) + if dx < hit_threshold and dy < hit_threshold: + self._select_trace_target(target_name) + return True + return False + + def _select_trace_target(self, target_name: str) -> None: + self._active_trace_target = target_name + + if target_name in { + self.TRACE_TARGET_SIGMA, + self.TRACE_TARGET_SIGMA_X, + self.TRACE_TARGET_SIGMA_Y, + self.TRACE_TARGET_STEP, + }: + self.wheel_mode_combo.setCurrentText(self.WHEEL_LEFT_Y) + self.pan_mode_combo.setCurrentText(self.PAN_LEFT_Y) + elif target_name in { + self.TRACE_TARGET_DISTANCE, + self.TRACE_TARGET_DX, + self.TRACE_TARGET_DY, + }: + self.wheel_mode_combo.setCurrentText(self.WHEEL_RIGHT_Y) + self.pan_mode_combo.setCurrentText(self.PAN_RIGHT_Y) + elif target_name == self.TRACE_TARGET_SCORE: + self.wheel_mode_combo.setCurrentText(self.WHEEL_SCORE) + self.pan_mode_combo.setCurrentText(self.PAN_SCORE) + + self._update_status_label() + + def _set_paused(self, paused: bool) -> None: + self._paused = paused + if paused: + self.pause_button.setText("Resume (Live)" if self._collected_samples else "Resume") + else: + self._unfreeze_plot() + self.pause_button.setText("Pause") + self._update_status_label() + self._update_metrics_label(force=True) + + def _seconds_text(self) -> str: + seconds = float(self.seconds_spin.value()) + if seconds.is_integer(): + return f"{int(seconds)}" + return f"{seconds:.1f}" + + def _update_action_labels(self) -> None: + seconds_text = self._seconds_text() + self.save_button.setText(f"Save last {seconds_text} s") + self.collect_button.setText(f"Collect next {seconds_text} s") + + def _format_xy(self, point: tuple[float, float] | None) -> str: + if point is None: + return "(-, -)" + return f"({point[0]:.2f}, {point[1]:.2f})" + + def _stability_score(self, value_px: float) -> float: + ref = max(1e-6, self._stability_ref_px) + return 100.0 / (1.0 + (value_px / ref)) + + def _score_source_label(self) -> str: + return self.score_basis_combo.currentText() + + def _show_metrics_help(self) -> None: + msg = QMessageBox(self) + msg.setWindowTitle("Target stability metrics") + msg.setIcon(QMessageBox.Icon.Information) + msg.setTextFormat(Qt.TextFormat.RichText) + msg.setText( + "Definitions

" + "target = detected target point from the incoming payload
" + "beam = beam center / beam marker previously set for this panel

" + "Offsets
" + "dx = target_x - beam_x
" + "dy = target_y - beam_y
" + "distance = sqrt(dx² + dy²)

" + "Metrics
" + "Distance: sqrt(dx² + dy²)
" + "Distance X: dx
" + "Distance Y: dy

" + "σx: std(dx) over the sigma window
" + "σy: std(dy) over the sigma window
" + "Rolling σ: sqrt(σx² + σy²)
" + "These use target relative to beam via dx and dy

" + "Step X / Step Y: frame-to-frame motion relative to itself
" + "Δdx_i = dx_i - dx_(i-1)
" + "Δdy_i = dy_i - dy_(i-1)

" + "Step XY: sqrt(mean(Δdx² + Δdy²))
" + "This is the best direct metric for how much the target is moving relative to itself

" + "Score: 100 / (1 + metric / ref_px)
" + f"Current score source: {self._score_source_label()}

" + "Interaction
" + "Click a visible point on a trace to target wheel/pan to that trace's axis group." + ) + msg.exec() + + def _current_sigma_stats(self) -> dict[str, float]: + n = self._rolling_count + if n < 2: + return { + "std_dx": 0.0, + "std_dy": 0.0, + "sigma": 0.0, + } + + mean_dx = self._rolling_sum_dx / n + mean_dy = self._rolling_sum_dy / n + var_dx = max(0.0, (self._rolling_sum_dx2 / n) - (mean_dx * mean_dx)) + var_dy = max(0.0, (self._rolling_sum_dy2 / n) - (mean_dy * mean_dy)) + + std_dx = math.sqrt(var_dx) + std_dy = math.sqrt(var_dy) + sigma = math.hypot(std_dx, std_dy) + + return { + "std_dx": std_dx, + "std_dy": std_dy, + "sigma": sigma, + } + + def _current_step_jitter_stats(self) -> dict[str, float]: + if len(self._sigma_samples) < 2: + return { + "rms_step_dx": 0.0, + "rms_step_dy": 0.0, + "rms_step_xy": 0.0, + } + + step_dx2_sum = 0.0 + step_dy2_sum = 0.0 + step_xy2_sum = 0.0 + count = 0 + + prev = None + for sample in self._sigma_samples: + if prev is not None: + ddx = float(sample["dx"]) - float(prev["dx"]) + ddy = float(sample["dy"]) - float(prev["dy"]) + step_dx2_sum += ddx * ddx + step_dy2_sum += ddy * ddy + step_xy2_sum += ddx * ddx + ddy * ddy + count += 1 + prev = sample + + if count <= 0: + return { + "rms_step_dx": 0.0, + "rms_step_dy": 0.0, + "rms_step_xy": 0.0, + } + + return { + "rms_step_dx": math.sqrt(step_dx2_sum / count), + "rms_step_dy": math.sqrt(step_dy2_sum / count), + "rms_step_xy": math.sqrt(step_xy2_sum / count), + } + + def _current_score_value(self) -> float: + sigma_stats = self._current_sigma_stats() + step_stats = self._current_step_jitter_stats() + + if self.score_basis_combo.currentText() == self.SCORE_FROM_SIGMA_XY: + return self._stability_score(sigma_stats["sigma"]) + return self._stability_score(step_stats["rms_step_xy"]) + + def _update_metrics_label(self, force: bool = False) -> None: + now = time.monotonic() + if not force and (now - self._last_metrics_update_ts) < self._metrics_update_interval_s: + return + self._last_metrics_update_ts = now + + target_text = self._format_xy(self._latest_target_point) + beam_text = self._format_xy(self._beam_center) + + if self._samples: + latest = self._samples[-1] + distance_text = f"{float(latest['distance']):.3f} px" + dx_text = f"{float(latest['dx']):+.3f} px" + dy_text = f"{float(latest['dy']):+.3f} px" + else: + distance_text = "- px" + dx_text = "- px" + dy_text = "- px" + + sigma_stats = self._current_sigma_stats() + step_stats = self._current_step_jitter_stats() + score_value = self._current_score_value() + + self.metrics_label.setText( + f"Target: {target_text} | " + f"Beam: {beam_text} | " + f"Distance: {distance_text} | " + f"dx: {dx_text} | " + f"dy: {dy_text}
" + f"σ: {sigma_stats['sigma']:.3f} px | " + f"σx: {sigma_stats['std_dx']:.3f} px | " + f"σy: {sigma_stats['std_dy']:.3f} px
" + f"Step X: {step_stats['rms_step_dx']:.3f} px | " + f"Step Y: {step_stats['rms_step_dy']:.3f} px | " + f"Step XY: {step_stats['rms_step_xy']:.3f} px | " + f"Score ({self._score_source_label()}): " + f"{score_value:.1f}" + ) + + def _update_trace_visibility(self) -> None: + self.series.setVisible(self.show_sigma_cb.isChecked()) + self.sigma_x_series.setVisible(self.show_sigma_x_cb.isChecked()) + self.sigma_y_series.setVisible(self.show_sigma_y_cb.isChecked()) + + self.distance_series.setVisible(self.show_distance_cb.isChecked()) + self.dx_series.setVisible(self.show_dx_cb.isChecked()) + self.dy_series.setVisible(self.show_dy_cb.isChecked()) + + self.step_series.setVisible(self.show_step_cb.isChecked()) + self.score_series.setVisible(self.show_score_cb.isChecked()) + + self.axis_y.setVisible( + self.show_sigma_cb.isChecked() + or self.show_sigma_x_cb.isChecked() + or self.show_sigma_y_cb.isChecked() + or self.show_step_cb.isChecked() + ) + self.axis_y_distance.setVisible( + self.show_distance_cb.isChecked() + or self.show_dx_cb.isChecked() + or self.show_dy_cb.isChecked() + ) + self.axis_y_score.setVisible(self.show_score_cb.isChecked()) + + self._update_status_label() + + def _on_score_basis_changed(self) -> None: + for point in self._plot_points: + if "sigma" in point and "step_xy" in point: + if self.score_basis_combo.currentText() == self.SCORE_FROM_SIGMA_XY: + point["score"] = self._stability_score(float(point["sigma"])) + else: + point["score"] = self._stability_score(float(point["step_xy"])) + + self._update_metrics_label(force=True) + self._refresh_chart() + + def _zoom_selected_wheel_axis(self, factor: float) -> None: + mode = self.wheel_mode_combo.currentText() + + if mode == self.WHEEL_LEFT_Y: + self._zoom_left_y_axis(factor) + return + if mode == self.WHEEL_RIGHT_Y: + self._zoom_right_y_axis(factor) + return + if mode == self.WHEEL_SCORE: + self._zoom_score_axis(factor) + return + + self._zoom_x_axis(factor) + + def _update_status_label(self) -> None: + if self._paused: + self.status_label.setText("Paused") + return + + if self._beam_center is None: + self.status_label.setText("Waiting for beam center...") + return + + if self._collecting_until is not None: + remaining = max(0.0, self._collecting_until - time.monotonic()) + self.status_label.setText(f"Collecting... {remaining:.1f} s remaining") + return + + sample_count = len(self._samples) + if sample_count == 0: + self.status_label.setText("Waiting for target-point data...") + else: + visible = [] + if self.show_sigma_cb.isChecked(): + visible.append("σ") + if self.show_sigma_x_cb.isChecked(): + visible.append("σx") + if self.show_sigma_y_cb.isChecked(): + visible.append("σy") + if self.show_distance_cb.isChecked(): + visible.append("distance") + if self.show_dx_cb.isChecked(): + visible.append("dx") + if self.show_dy_cb.isChecked(): + visible.append("dy") + if self.show_step_cb.isChecked(): + visible.append("stepXY") + if self.show_score_cb.isChecked(): + visible.append("score") + + self.status_label.setText( + f"Live: {sample_count} samples in last {self._live_window_s:.0f} s | " + f"view={'auto' if self._auto_scale_enabled else 'manual'} | " + f"traces={', '.join(visible) if visible else 'none'} | " + f"score={self._score_source_label()} | " + f"wheel={self.wheel_mode_combo.currentText()} | " + f"pan={self.pan_mode_combo.currentText()} | " + f"active={self._active_trace_target}" + ) + + def set_beam_center(self, x: float, y: float) -> None: + self._beam_center = (float(x), float(y)) + self._update_status_label() + self._update_metrics_label(force=True) + + @Slot(dict) + def update_target_point(self, payload: dict) -> None: + if self._paused or self._beam_center is None: + return + + raw = payload.get("target_point") + target_xy = self._coerce_target_point(raw) + if target_xy is None: + return + + self._latest_target_point = target_xy + + ts = time.monotonic() + tx, ty = target_xy + bx, by = self._beam_center + dx = tx - bx + dy = ty - by + distance = math.hypot(dx, dy) + + sample = { + "ts": ts, + "dx": dx, + "dy": dy, + "distance": distance, + } + + should_finish_collection = False + with self._data_lock: + self._samples.append(sample) + self._append_plot_point(sample) + self._trim_samples(ts) + + if self._collecting_until is not None: + self._collected_samples.append(sample.copy()) + if ts >= self._collecting_until: + should_finish_collection = True + + if should_finish_collection: + self._finish_collection() + + self._update_status_label() + + def _set_sigma_window(self, seconds: float) -> None: + self._sigma_window_s = max(0.02, float(seconds)) + self._rebuild_plot_points() + self._update_metrics_label(force=True) + + def _rebuild_plot_points(self) -> None: + self._plot_points.clear() + self._sigma_samples.clear() + + self._rolling_count = 0 + self._rolling_sum_dx = 0.0 + self._rolling_sum_dy = 0.0 + self._rolling_sum_dx2 = 0.0 + self._rolling_sum_dy2 = 0.0 + self._kahan_c_dx = 0.0 + self._kahan_c_dy = 0.0 + self._kahan_c_dx2 = 0.0 + self._kahan_c_dy2 = 0.0 + + for sample in self._samples: + self._append_plot_point(sample) + + def _append_plot_point(self, sample: dict) -> None: + self._sigma_samples.append(sample) + + dx = float(sample["dx"]) + dy = float(sample["dy"]) + + self._rolling_count += 1 + + # Kahan summation for dx + y = dx - self._kahan_c_dx + t = self._rolling_sum_dx + y + self._kahan_c_dx = (t - self._rolling_sum_dx) - y + self._rolling_sum_dx = t + + # Kahan summation for dy + y = dy - self._kahan_c_dy + t = self._rolling_sum_dy + y + self._kahan_c_dy = (t - self._rolling_sum_dy) - y + self._rolling_sum_dy = t + + # Kahan summation for dx² + dx2 = dx * dx + y = dx2 - self._kahan_c_dx2 + t = self._rolling_sum_dx2 + y + self._kahan_c_dx2 = (t - self._rolling_sum_dx2) - y + self._rolling_sum_dx2 = t + + # Kahan summation for dy² + dy2 = dy * dy + y = dy2 - self._kahan_c_dy2 + t = self._rolling_sum_dy2 + y + self._kahan_c_dy2 = (t - self._rolling_sum_dy2) - y + self._rolling_sum_dy2 = t + + self._trim_sigma_samples(float(sample["ts"])) + + sigma_stats = self._current_sigma_stats() + step_xy = self._current_step_jitter_stats()["rms_step_xy"] + + if self.score_basis_combo.currentText() == self.SCORE_FROM_SIGMA_XY: + score = self._stability_score(sigma_stats["sigma"]) + else: + score = self._stability_score(step_xy) + + self._plot_points.append({ + "ts": float(sample["ts"]), + "sigma": sigma_stats["sigma"], + "sigma_x": sigma_stats["std_dx"], + "sigma_y": sigma_stats["std_dy"], + "distance": float(sample["distance"]), + "dx": dx, + "dy": dy, + "step_xy": step_xy, + "score": score, + }) + + def _remove_oldest_sigma_sample(self) -> None: + if not self._sigma_samples: + return + + sample = self._sigma_samples.popleft() + dx = float(sample["dx"]) + dy = float(sample["dy"]) + + self._rolling_count -= 1 + + # Kahan subtraction for dx + y = -dx - self._kahan_c_dx + t = self._rolling_sum_dx + y + self._kahan_c_dx = (t - self._rolling_sum_dx) - y + self._rolling_sum_dx = t + + # Kahan subtraction for dy + y = -dy - self._kahan_c_dy + t = self._rolling_sum_dy + y + self._kahan_c_dy = (t - self._rolling_sum_dy) - y + self._rolling_sum_dy = t + + # Kahan subtraction for dx² + dx2 = dx * dx + y = -dx2 - self._kahan_c_dx2 + t = self._rolling_sum_dx2 + y + self._kahan_c_dx2 = (t - self._rolling_sum_dx2) - y + self._rolling_sum_dx2 = t + + # Kahan subtraction for dy² + dy2 = dy * dy + y = -dy2 - self._kahan_c_dy2 + t = self._rolling_sum_dy2 + y + self._kahan_c_dy2 = (t - self._rolling_sum_dy2) - y + self._rolling_sum_dy2 = t + + if self._rolling_count <= 0: + self._rolling_count = 0 + self._rolling_sum_dx = 0.0 + self._rolling_sum_dy = 0.0 + self._rolling_sum_dx2 = 0.0 + self._rolling_sum_dy2 = 0.0 + self._kahan_c_dx = 0.0 + self._kahan_c_dy = 0.0 + self._kahan_c_dx2 = 0.0 + self._kahan_c_dy2 = 0.0 + + def _trim_sigma_samples(self, now: float) -> None: + cutoff = now - self._sigma_window_s + while self._sigma_samples and float(self._sigma_samples[0]["ts"]) < cutoff: + self._remove_oldest_sigma_sample() + + def _remove_oldest_sample(self) -> None: + if not self._samples: + return + self._samples.popleft() + + def _trim_samples(self, now: float) -> None: + cutoff = now - self._live_window_s + + while self._samples and float(self._samples[0]["ts"]) < cutoff: + self._remove_oldest_sample() + + while self._plot_points and float(self._plot_points[0]["ts"]) < cutoff: + self._plot_points.popleft() + + def _data_limits(self) -> tuple[float, float, float, float, float, float, float, float]: + left_values: list[float] = [] + right_values: list[float] = [] + + for point in self._plot_points: + left_values.extend([ + float(point["sigma"]), + float(point["sigma_x"]), + float(point["sigma_y"]), + float(point["step_xy"]), + ]) + right_values.extend([ + float(point["distance"]), + float(point["dx"]), + float(point["dy"]), + ]) + + left_min = min(left_values, default=0.0) + left_max = max(left_values, default=1.0) + right_min = min(right_values, default=-1.0) + right_max = max(right_values, default=1.0) + + left_span = max(1.0, left_max - left_min) + right_span = max(1.0, right_max - right_min) + + score_min = min((float(point["score"]) for point in self._plot_points), default=0.0) + score_max = max((float(point["score"]) for point in self._plot_points), default=100.0) + score_lo = max(0.0, score_min - 5.0) + score_hi = min(100.0, max(score_lo + 10.0, score_max + 5.0)) + + return ( + -self._live_window_s, + 0.0, + left_min - 0.075 * left_span, + left_max + 0.075 * left_span, + right_min - 0.075 * right_span, + right_max + 0.075 * right_span, + score_lo, + score_hi, + ) + + def _reset_view(self) -> None: + self._auto_scale_enabled = True + xmin, xmax, left_ymin, left_ymax, right_ymin, right_ymax, score_lo, score_hi = self._data_limits() + self.axis_x.setRange(xmin, xmax) + self.axis_y.setRange(left_ymin, left_ymax) + self.axis_y_distance.setRange(right_ymin, right_ymax) + self.axis_y_score.setRange(score_lo, score_hi) + self._update_status_label() + + def _zoom_axis(self, axis: QValueAxis, factor: float) -> None: + amin = axis.min() + amax = axis.max() + center = (amin + amax) / 2.0 + half = (amax - amin) * factor / 2.0 + if half <= 1e-9: + return + axis.setRange(center - half, center + half) + + def _zoom_x_axis(self, factor: float) -> None: + self._auto_scale_enabled = False + self._zoom_axis(self.axis_x, factor) + self._clamp_axes() + self._update_status_label() + + def _zoom_left_y_axis(self, factor: float) -> None: + self._auto_scale_enabled = False + if ( + self.show_sigma_cb.isChecked() + or self.show_sigma_x_cb.isChecked() + or self.show_sigma_y_cb.isChecked() + or self.show_step_cb.isChecked() + ): + self._zoom_axis(self.axis_y, factor) + self._clamp_axes() + self._update_status_label() + + def _zoom_right_y_axis(self, factor: float) -> None: + self._auto_scale_enabled = False + if ( + self.show_distance_cb.isChecked() + or self.show_dx_cb.isChecked() + or self.show_dy_cb.isChecked() + ): + self._zoom_axis(self.axis_y_distance, factor) + self._clamp_axes() + self._update_status_label() + + def _zoom_score_axis(self, factor: float) -> None: + self._auto_scale_enabled = False + if self.show_score_cb.isChecked(): + self._zoom_axis(self.axis_y_score, factor) + self._clamp_axes() + self._update_status_label() + + def _pan_selected_axes(self, dx_pixels: int, dy_pixels: int, width: int, height: int) -> None: + if width <= 0 or height <= 0: + return + + self._auto_scale_enabled = False + + x_span = self.axis_x.max() - self.axis_x.min() + y_span = self.axis_y.max() - self.axis_y.min() + y2_span = self.axis_y_distance.max() - self.axis_y_distance.min() + y3_span = self.axis_y_score.max() - self.axis_y_score.min() + + x_span = max(x_span, 1e-9) + y_span = max(y_span, 1e-9) + y2_span = max(y2_span, 1e-9) + y3_span = max(y3_span, 1e-9) + + x_shift = -(dx_pixels / width) * x_span + y_shift = (dy_pixels / height) * y_span + y2_shift = (dy_pixels / height) * y2_span + y3_shift = (dy_pixels / height) * y3_span + + mode = self.pan_mode_combo.currentText() + + if mode == self.PAN_X: + self.axis_x.setRange(self.axis_x.min() + x_shift, self.axis_x.max() + x_shift) + + elif mode == self.PAN_LEFT_Y: + if ( + self.show_sigma_cb.isChecked() + or self.show_sigma_x_cb.isChecked() + or self.show_sigma_y_cb.isChecked() + or self.show_step_cb.isChecked() + ): + self.axis_y.setRange(self.axis_y.min() + y_shift, self.axis_y.max() + y_shift) + + elif mode == self.PAN_RIGHT_Y: + if ( + self.show_distance_cb.isChecked() + or self.show_dx_cb.isChecked() + or self.show_dy_cb.isChecked() + ): + self.axis_y_distance.setRange( + self.axis_y_distance.min() + y2_shift, + self.axis_y_distance.max() + y2_shift, + ) + + elif mode == self.PAN_SCORE: + if self.show_score_cb.isChecked(): + self.axis_y_score.setRange( + self.axis_y_score.min() + y3_shift, + self.axis_y_score.max() + y3_shift, + ) + + else: + self.axis_x.setRange(self.axis_x.min() + x_shift, self.axis_x.max() + x_shift) + + if ( + self.show_sigma_cb.isChecked() + or self.show_sigma_x_cb.isChecked() + or self.show_sigma_y_cb.isChecked() + or self.show_step_cb.isChecked() + ): + self.axis_y.setRange(self.axis_y.min() + y_shift, self.axis_y.max() + y_shift) + + if ( + self.show_distance_cb.isChecked() + or self.show_dx_cb.isChecked() + or self.show_dy_cb.isChecked() + ): + self.axis_y_distance.setRange( + self.axis_y_distance.min() + y2_shift, + self.axis_y_distance.max() + y2_shift, + ) + + if self.show_score_cb.isChecked(): + self.axis_y_score.setRange( + self.axis_y_score.min() + y3_shift, + self.axis_y_score.max() + y3_shift, + ) + + self._clamp_axes() + self._update_status_label() + + def _clamp_axes(self) -> None: + full_xmin = -self._live_window_s + full_xmax = 0.0 + + xmin = self.axis_x.min() + xmax = self.axis_x.max() + xspan = xmax - xmin + + if xspan >= (full_xmax - full_xmin): + self.axis_x.setRange(full_xmin, full_xmax) + else: + if xmin < full_xmin: + self.axis_x.setRange(full_xmin, full_xmin + xspan) + elif xmax > full_xmax: + self.axis_x.setRange(full_xmax - xspan, full_xmax) + + score_min = self.axis_y_score.min() + score_max = self.axis_y_score.max() + score_span = max(score_max - score_min, 1.0) + + if score_min < 0.0: + self.axis_y_score.setRange(0.0, min(100.0, score_span)) + elif score_max > 100.0: + self.axis_y_score.setRange(max(0.0, 100.0 - score_span), 100.0) + + def _refresh_chart(self) -> None: + now = time.monotonic() + + with self._data_lock: + self._trim_samples(now) + + if not self.isVisible(): + self._update_status_label() + self._update_metrics_label() + return + + if self._paused and self._collecting_until is None and self._frozen_plot_points is None: + self._update_status_label() + self._update_metrics_label() + return + + # Use frozen data if available, otherwise use live data + if self._frozen_plot_points is not None: + plot_data = list(self._frozen_plot_points) + ref_time = plot_data[0]["ts"] if plot_data else now + else: + plot_data = list(self._plot_points) + ref_time = now + + sigma_points = [ + QPointF(float(point["ts"]) - ref_time, float(point["sigma"])) + for point in plot_data + ] + sigma_x_points = [ + QPointF(float(point["ts"]) - ref_time, float(point["sigma_x"])) + for point in plot_data + ] + sigma_y_points = [ + QPointF(float(point["ts"]) - ref_time, float(point["sigma_y"])) + for point in plot_data + ] + distance_points = [ + QPointF(float(point["ts"]) - ref_time, float(point["distance"])) + for point in plot_data + ] + dx_points = [ + QPointF(float(point["ts"]) - ref_time, float(point["dx"])) + for point in plot_data + ] + dy_points = [ + QPointF(float(point["ts"]) - ref_time, float(point["dy"])) + for point in plot_data + ] + score_points = [ + QPointF(float(point["ts"]) - ref_time, float(point["score"])) + for point in plot_data + ] + step_points = [ + QPointF(float(point["ts"]) - ref_time, float(point["step_xy"])) + for point in plot_data + ] + + self.series.replace(sigma_points) + self.sigma_x_series.replace(sigma_x_points) + self.sigma_y_series.replace(sigma_y_points) + self.distance_series.replace(distance_points) + self.dx_series.replace(dx_points) + self.dy_series.replace(dy_points) + self.score_series.replace(score_points) + self.step_series.replace(step_points) + + if self._auto_scale_enabled: + xmin, xmax, left_ymin, left_ymax, right_ymin, right_ymax, score_lo, score_hi = self._data_limits() + self.axis_x.setRange(xmin, xmax) + self.axis_y.setRange(left_ymin, left_ymax) + self.axis_y_distance.setRange(right_ymin, right_ymax) + self.axis_y_score.setRange(score_lo, score_hi) + else: + self._clamp_axes() + + self._update_status_label() + self._update_metrics_label() + + def _save_last_x_seconds(self) -> None: + seconds = float(self.seconds_spin.value()) + now = time.monotonic() + rows = [s for s in self._samples if float(s["ts"]) >= now - seconds] + self._save_rows(rows, suggested_name=f"target_stability_last_{self._seconds_text()}s.csv") + + def _collect_next_x_seconds(self) -> None: + if self._paused: + logger.info("Cannot collect target stability data while paused") + return + + seconds = float(self.seconds_spin.value()) + self._collected_samples = [] + self._collecting_until = time.monotonic() + seconds + self._update_status_label() + logger.info(f"Collecting target stability data for {seconds:.1f} s") + + def _finish_collection(self) -> None: + """Freeze the plot with collected data and prompt to save.""" + self._collecting_until = None + + # Build plot points from collected samples for display + self._frozen_plot_points = [] + temp_sigma_samples = deque(maxlen=10000) + temp_rolling_count = 0 + temp_rolling_sum_dx = 0.0 + temp_rolling_sum_dy = 0.0 + temp_rolling_sum_dx2 = 0.0 + temp_rolling_sum_dy2 = 0.0 + + for sample in self._collected_samples: + temp_sigma_samples.append(sample) + dx = float(sample["dx"]) + dy = float(sample["dy"]) + + temp_rolling_count += 1 + temp_rolling_sum_dx += dx + temp_rolling_sum_dy += dy + temp_rolling_sum_dx2 += dx * dx + temp_rolling_sum_dy2 += dy * dy + + # Trim old samples from rolling window + cutoff = float(sample["ts"]) - self._sigma_window_s + while temp_sigma_samples and float(temp_sigma_samples[0]["ts"]) < cutoff: + old = temp_sigma_samples.popleft() + old_dx = float(old["dx"]) + old_dy = float(old["dy"]) + temp_rolling_count -= 1 + temp_rolling_sum_dx -= old_dx + temp_rolling_sum_dy -= old_dy + temp_rolling_sum_dx2 -= old_dx * old_dx + temp_rolling_sum_dy2 -= old_dy * old_dy + + # Compute sigma stats + if temp_rolling_count >= 2: + mean_dx = temp_rolling_sum_dx / temp_rolling_count + mean_dy = temp_rolling_sum_dy / temp_rolling_count + var_dx = max(0.0, (temp_rolling_sum_dx2 / temp_rolling_count) - (mean_dx * mean_dx)) + var_dy = max(0.0, (temp_rolling_sum_dy2 / temp_rolling_count) - (mean_dy * mean_dy)) + std_dx = math.sqrt(var_dx) + std_dy = math.sqrt(var_dy) + sigma = math.hypot(std_dx, std_dy) + else: + std_dx, std_dy, sigma = 0.0, 0.0, 0.0 + + # Compute step jitter + step_xy = 0.0 + if len(temp_sigma_samples) >= 2: + step_sum = 0.0 + count = 0 + prev = None + for s in temp_sigma_samples: + if prev is not None: + ddx = float(s["dx"]) - float(prev["dx"]) + ddy = float(s["dy"]) - float(prev["dy"]) + step_sum += ddx * ddx + ddy * ddy + count += 1 + prev = s + if count > 0: + step_xy = math.sqrt(step_sum / count) + + score = self._stability_score( + sigma if self.score_basis_combo.currentText() == self.SCORE_FROM_SIGMA_XY else step_xy) + + self._frozen_plot_points.append({ + "ts": float(sample["ts"]), + "sigma": sigma, + "sigma_x": std_dx, + "sigma_y": std_dy, + "distance": float(sample["distance"]), + "dx": dx, + "dy": dy, + "step_xy": step_xy, + "score": score, + }) + + self._paused = True + self.pause_button.setChecked(True) + self.pause_button.setText("Resume (Live)") + + logger.info("Finished collecting target stability data - plot frozen") + self._update_status_label() + + # Schedule the save dialog to run after the current event processing + # This avoids blocking while any locks might be held + file_name = f"target_stability_collected_{self._seconds_text()}s.csv" + QTimer.singleShot(0, lambda: self._save_rows(self._collected_samples, suggested_name=file_name)) + self._update_status_label() + + def _unfreeze_plot(self) -> None: + """Return to live data display.""" + self._frozen_plot_points = None + self._collected_samples = [] + self._update_status_label() + + def _save_rows(self, rows: list[dict], suggested_name: str) -> None: + if not rows: + logger.info("No target stability data to save") + return + + path, _ = QFileDialog.getSaveFileName( + self, + "Save Target Stability Data", + suggested_name, + "CSV files (*.csv)", + ) + if not path: + return + + with open(path, "w", newline="") as f: + writer = csv.writer(f) + writer.writerow(["t_monotonic_s", "dx_px", "dy_px", "distance_px"]) + for row in rows: + writer.writerow([ + f"{float(row['ts']):.6f}", + f"{float(row['dx']):.6f}", + f"{float(row['dy']):.6f}", + f"{float(row['distance']):.6f}", + ]) + + @staticmethod + def _coerce_target_point(raw) -> tuple[float, float] | None: + try: + if isinstance(raw, dict): + if "x" in raw and "y" in raw: + return float(raw["x"]), float(raw["y"]) + if isinstance(raw, (list, tuple)) and len(raw) >= 2: + return float(raw[0]), float(raw[1]) + except Exception as e: + logger.warning(f"Failed to parse target point {raw}: {e}") + return None \ No newline at end of file