mx/AareDAQ
0
1
Fork 0
Code Issues 52 Pull Requests 1 Actions Packages Projects Releases 65 Wiki Activity

GUI: added a target stability panel and a predicition metrics panel - needs tidying!

Browse Source
This commit is contained in:
appleb_m
2026-04-22 10:18:39 +02:00
parent 9a1030a2aa
commit 80008891ef
2 changed files with 2191 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/aare/gui/panels/prediction_metrics_panel.py
+791
View File
@@ -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}")
src/aare/gui/panels/target_stability_panel.py
+1400
View File
File diff suppressed because it is too large Load Diff