"""UI for running and reviewing center finding workflows."""
from coseda.logging_utils import log_print
from PyQt6.QtWidgets import (
QDialog,
QLabel,
QLineEdit,
QPushButton,
QVBoxLayout,
QHBoxLayout,
QGroupBox,
QFormLayout,
QMessageBox,
QSizePolicy,
QComboBox,
QCheckBox,
QSpinBox,
QDoubleSpinBox,
QTreeWidget,
QTreeWidgetItem,
QProgressBar,
QWidget,
)
from PyQt6.QtCore import Qt, pyqtSignal, QTimer, QObject, QThread
import h5py
from PyQt6.QtGui import QIntValidator, QDoubleValidator, QPixmap
import configparser
import os
import threading
import re
import time
import matplotlib
matplotlib.use('Agg') # Use non-interactive backend
import matplotlib.pyplot as plt
import numpy as np
from matplotlib.backends.backend_qtagg import FigureCanvasQTAgg as FigureCanvas
from matplotlib.figure import Figure
from coseda.io import read_config, config_to_paths
from coseda.centerfinding.findcenters_direct import write_centerfinderdirectsettings, run_findcenters_direct
# Focus-aware line edit used to request coordinate picking on focus
[docs]
class FocusableLineEdit(QLineEdit):
focus_in = pyqtSignal()
focus_out = pyqtSignal()
[docs]
def focusInEvent(self, event):
super().focusInEvent(event)
self.focus_in.emit()
[docs]
def focusOutEvent(self, event):
super().focusOutEvent(event)
self.focus_out.emit()
"""
# Popup dialog for linear fit plot
class LinearFitDialog(QDialog):
def __init__(self, npz_path, parent=None):
super().__init__(parent)
self.setWindowTitle("Linear Fit")
layout = QVBoxLayout(self)
# Create matplotlib figure & canvas
fig = Figure(figsize=(6, 4))
canvas = FigureCanvas(fig)
ax = fig.subplots()
# Load data and plot
data = np.load(npz_path)
ux = data['updated_x']
uy = data['updated_y']
# Determine x-axis: use frame_positions if available
if 'frame_positions' in data:
frames = data['frame_positions']
else:
frames = np.arange(len(ux))
# Plot fitted line
ax.plot(frames, ux, label='Fitted Line X')
ax.plot(frames, uy, label='Fitted Line Y')
ax.set_xlabel("Frame Index")
ax.set_ylabel("Center Value")
ax.legend()
layout.addWidget(canvas)
canvas.draw()"""
# -------------------- PreviewWindow for Pop-out Preview --------------------
[docs]
class PreviewWindow(QDialog):
"""Pop-out preview showing per-batch centerfinding plots and metadata."""
def __init__(self, ini_file_path, parent=None):
super().__init__(parent)
self.ini_file_path = ini_file_path
# Track which run is selected in the dropdown
self.current_run = None
self.setWindowTitle(f"Centerfinding for {os.path.basename(ini_file_path)}")
self.setAttribute(Qt.WidgetAttribute.WA_DeleteOnClose)
# Layout
layout = QVBoxLayout(self)
# Tree
self.preview_tree = QTreeWidget()
self.preview_tree.setHeaderLabels(["Batch", "Iteration", "File"])
self.preview_tree.setColumnWidth(0, 80)
self.preview_tree.setColumnWidth(1, 80)
self.preview_tree.setColumnWidth(2, 240)
self.preview_tree.itemClicked.connect(self.display_selected_plot)
layout.addWidget(self.preview_tree)
# Canvas with two subplots: scatter (left) and linear fit (right)
self.preview_canvas = FigureCanvas(Figure(figsize=(8, 4)))
self.plot_dev_ax, self.plot_fit_ax = self.preview_canvas.figure.subplots(1, 2)
layout.addWidget(self.preview_canvas)
# Label to show mean deviation of selected batch (left)
self.deviation_label = QLabel("Deviation: N/A")
# New label for scope-of-drift (middle)
self.drift_scope_label = QLabel("Scope of drift: N/A")
# New label for equations (right)
self.equation_label = QLabel("Linear fit: N/A")
# Arrange the three labels in one horizontal row
label_row = QHBoxLayout()
label_row.addWidget(self.deviation_label)
label_row.addWidget(self.drift_scope_label)
label_row.addWidget(self.equation_label)
layout.addLayout(label_row)
# Aliases for plotting
self.plot_canvas = self.preview_canvas
# Monitoring
self.monitor_start_time = time.time()
self.plots_loaded = set()
# Start monitoring for new plots
self.monitor_timer = QTimer(self)
self.monitor_timer.setInterval(1000) # check every second
self.monitor_timer.timeout.connect(self.update_preview)
self.monitor_timer.start()
# Populate initial plot list
self.update_preview()
# Auto-display initial plot
self.auto_display_initial()
[docs]
def auto_display_initial(self):
"""Automatically select and display the first available plot."""
if self.preview_tree.topLevelItemCount() > 0:
first_batch = self.preview_tree.topLevelItem(0)
if first_batch.childCount() > 0:
child_item = first_batch.child(0)
self.preview_tree.setCurrentItem(child_item)
self.display_selected_plot(child_item, 0)
[docs]
def set_run(self, run_name):
self.plots_loaded.clear()
self.preview_tree.clear()
self.current_run = run_name
# Stop auto-monitoring to preserve manual selection
self.monitor_timer.stop()
# Update the table first
self.update_preview()
# Then schedule displaying the selected plot
QTimer.singleShot(0, self.auto_display_initial)
[docs]
def update_preview(self):
"""Refresh the tree with available plots and redraw the selected run."""
outputfolder, outputfolder_path, logfile, logfile_path, h5file, h5file_path = config_to_paths(self.ini_file_path)
config = read_config(self.ini_file_path)
# Determine graphics folder path based on selected run or latest
if self.current_run:
graphicsfolder_path = os.path.join(outputfolder_path, self.current_run)
else:
try:
subdirs = sorted(d for d in os.listdir(outputfolder_path) if d.startswith("findcenters_"))
graphicsfolder_path = os.path.join(outputfolder_path, subdirs[-1])
except Exception:
graphicsfolder_path = os.path.join(outputfolder_path, 'plots')
# If the folder doesn't exist, abort
if not os.path.isdir(graphicsfolder_path):
return
# --- Direct method: single-plot mode, hide tree ---
direct_raw = os.path.join(graphicsfolder_path, 'direct_centers_raw.npy')
if os.path.exists(direct_raw):
# Hide the batch tree
self.preview_tree.hide()
# Hide the first label (deviation) for Direct mode
self.deviation_label.hide()
# Show only the middle and right labels
self.drift_scope_label.show()
self.equation_label.show()
# Reconfigure canvas to one axis
self.preview_canvas.figure.clear()
ax = self.preview_canvas.figure.add_subplot(1, 1, 1)
# --- Simplified, correct loading and plotting for Direct mode ---
data = np.load(direct_raw)
frames = data[:, 0] # original indices explicitly stored in the numpy file
xs = data[:, 1]
ys = data[:, 2]
min_length = min(len(frames), len(xs), len(ys))
frames, xs, ys = frames[:min_length], xs[:min_length], ys[:min_length]
ax.scatter(frames, xs, s=1, label='Center X')
ax.scatter(frames, ys, s=1, label='Center Y')
# Overlay fit if available
direct_fit = os.path.join(graphicsfolder_path, 'direct_centers_fit.npy')
if os.path.exists(direct_fit):
fit = np.load(direct_fit)
vals_x = fit[0]
vals_y = fit[1]
sx, ix = vals_x[0], vals_x[1]
sy, iy = vals_y[0], vals_y[1]
# Plot lines
ax.plot(frames, sx * frames + ix, label='Fit X')
ax.plot(frames, sy * frames + iy, label='Fit Y')
# Compute total drift over the full frame range
fmin, fmax = frames.min(), frames.max()
x_start = sx * fmin + ix
x_end = sx * fmax + ix
y_start = sy * fmin + iy
y_end = sy * fmax + iy
ax.set_xlabel('Frame Index')
ax.set_ylabel('Center (px)')
ax.legend()
# Set only the two labels for Direct mode (drift_scope_label and equation_label)
if os.path.exists(direct_fit):
self.drift_scope_label.setText(
"Scope of drift:\n"
f"ΔX={x_end - x_start:.2f}px\n"
f"ΔY={y_end - y_start:.2f}px"
)
# New dynamic multiline Linear fit equations with R² and optional σ
vals_x = fit[0]
vals_y = fit[1]
eq_label = "Linear fit:\n"
eq_label += f"X = {vals_x[0]:.6f}*f + {vals_x[1]:.2f} (R²={vals_x[2]:.3f})"
if len(vals_x) == 4:
eq_label += f" σ={vals_x[3]:.4f}"
eq_label += "\n"
eq_label += f"Y = {vals_y[0]:.6f}*f + {vals_y[1]:.2f} (R²={vals_y[2]:.3f})"
if len(vals_y) == 4:
eq_label += f" σ={vals_y[3]:.4f}"
self.equation_label.setText(eq_label)
else:
self.drift_scope_label.setText(
"Scope of drift:\nN/A"
)
self.equation_label.setText(
"Linear fit:\nN/A"
)
self.preview_canvas.draw()
return
# For non-Direct runs, ensure tree and dual-axes are visible
self.preview_tree.show()
# Ensure all three labels are visible for Beamstop mode
self.deviation_label.show()
self.drift_scope_label.show()
self.equation_label.show()
# Scan NPZs, sorted by batch and iteration
dev_regex = re.compile(r'^batch(\d+)_it(\d{3})\.npz$')
dev_files = []
for fn in os.listdir(graphicsfolder_path):
m = dev_regex.match(fn)
if m:
batch = int(m.group(1)); it = int(m.group(2))
dev_files.append((batch, it, fn))
# Sort by batch, then iteration
dev_files.sort(key=lambda x: (x[0], x[1]))
for batch, it, fn in dev_files:
file_path = os.path.join(graphicsfolder_path, fn)
if file_path in self.plots_loaded:
continue
item = NumericSortTreeWidgetItem([str(batch), str(it), fn])
item.setData(2, Qt.ItemDataRole.UserRole, file_path)
self.preview_tree.addTopLevelItem(item)
self.plots_loaded.add(file_path)
# (linearfit.npz and batchcenters.npz are intentionally not added for non-Direct runs)
# Sort entries by batch number (ascending)
self.preview_tree.sortItems(0, Qt.SortOrder.AscendingOrder)
current_item = self.preview_tree.currentItem()
if current_item and current_item.childCount() == 0:
self.display_selected_plot(current_item, 0)
else:
self.auto_display_initial()
[docs]
def display_selected_plot(self, item, column):
# Only proceed if it's a plot item (no children)
if item.childCount() != 0:
return
# Reset canvas and recreate both axes for fresh plotting
self.preview_canvas.figure.clear()
self.plot_dev_ax, self.plot_fit_ax = self.preview_canvas.figure.subplots(1, 2)
# Ensure both axes are visible
self.plot_dev_ax.set_visible(True)
self.plot_fit_ax.set_visible(True)
file_path = item.data(2, Qt.ItemDataRole.UserRole)
dirname = os.path.dirname(file_path)
basename = os.path.basename(file_path)
# --- Direct method raw scatter or fit overlay ---
if basename == 'direct_centers_raw.npy':
# Hide second axis
self.plot_fit_ax.clear()
self.plot_fit_ax.set_visible(False)
# Load raw data
data = np.load(file_path)
frames = data[:, 0]
xs = data[:, 1]
ys = data[:, 2]
# Plot raw centers
self.plot_dev_ax.clear()
self.plot_dev_ax.scatter(frames, xs, s=1, label='Center X')
self.plot_dev_ax.scatter(frames, ys, s=1, label='Center Y')
self.plot_dev_ax.set_xlabel('Frame Index')
self.plot_dev_ax.set_ylabel('Center (px)')
self.plot_dev_ax.legend()
self.plot_canvas.draw()
return
if basename == 'direct_centers_fit.npy':
# Ensure raw is plotted first
raw_path = os.path.join(dirname, 'direct_centers_raw.npy')
if os.path.exists(raw_path):
raw = np.load(raw_path)
frames = raw[:, 0]
xs = raw[:, 1]
ys = raw[:, 2]
self.plot_dev_ax.clear()
self.plot_dev_ax.scatter(frames, xs, s=1)
self.plot_dev_ax.scatter(frames, ys, s=1)
# Load fit parameters
fit = np.load(file_path)
vals_x = fit[0]
vals_y = fit[1]
sx, ix, r2x = vals_x[:3]
sy, iy, r2y = vals_y[:3]
# Compute fitted lines
xs_fit = sx * frames + ix
ys_fit = sy * frames + iy
# Plot fits
self.plot_dev_ax.plot(frames, xs_fit, label=f'Fit X (R²={r2x:.2f})')
self.plot_dev_ax.plot(frames, ys_fit, label=f'Fit Y (R²={r2y:.2f})')
self.plot_dev_ax.set_xlabel('Frame Index')
self.plot_dev_ax.set_ylabel('Center (px)')
self.plot_dev_ax.legend()
# Set the equation label dynamically as in update_preview
eq_label = "Linear fit:\n"
eq_label += f"X = {vals_x[0]:.6f}*f + {vals_x[1]:.2f} (R²={vals_x[2]:.3f})"
if len(vals_x) == 4:
eq_label += f" σ={vals_x[3]:.4f}"
eq_label += "\n"
eq_label += f"Y = {vals_y[0]:.6f}*f + {vals_y[1]:.2f} (R²={vals_y[2]:.3f})"
if len(vals_y) == 4:
eq_label += f" σ={vals_y[3]:.4f}"
self.equation_label.setText(eq_label)
self.plot_canvas.draw()
return
# Plot scatter for selected batch .npz
if basename.endswith('.npz') and basename.startswith('batch'):
data = np.load(file_path)
dev = data['deviations']
mean = data['mean']
if dev.size > 0:
self.plot_dev_ax.scatter(dev[:, 0], dev[:, 1], s=0.5)
# Plot marker at mean deviation (no legend)
if mean.size > 0:
self.plot_dev_ax.scatter(mean[0], mean[1], marker='x', color='red')
self.plot_dev_ax.set_xlabel("X-deviation")
self.plot_dev_ax.set_ylabel("Y-deviation")
# Ensure equal scaling
self.plot_dev_ax.set_aspect('equal', 'box')
# Annotate mean coordinates
if mean.size > 0:
mx, my = mean[0], mean[1]
self.plot_dev_ax.annotate(
f"({mx:.3f}, {my:.3f})",
xy=(mx, my),
xytext=(5, 5),
textcoords='offset points',
color='red'
)
# Compute base batch deviation and center
if mean.size > 0:
base_dev = f"Deviation: ({mean[0]:.3f}, {mean[1]:.3f})"
else:
base_dev = "Deviation: N/A"
bc_path = os.path.join(dirname, 'batchcenters.npz')
try:
bc_data = np.load(bc_path)
bx = float(bc_data['updated_x'][0])
by = float(bc_data['updated_y'][0])
base_center = f"Batch center: ({bx:.3f}, {by:.3f})"
except Exception:
base_center = "Batch center: N/A"
# Left label: both deviation and center
self.deviation_label.setText(f"{base_dev}\n{base_center}")
# Compute and show drift scope and linear-fit equations if files exist
raw_path = os.path.join(dirname, 'beamstop_centers_raw.npy')
fit_path = os.path.join(dirname, 'beamstop_centers_fit.npy')
if os.path.exists(raw_path) and os.path.exists(fit_path):
raw = np.load(raw_path)
frames = raw[:, 0]
ffirst, flast = frames[0], frames[-1]
fit = np.load(fit_path)
sx, ix, r2x = fit[0]
sy, iy, r2y = fit[1]
# Calculate drift
x_start = sx * ffirst + ix
x_end = sx * flast + ix
y_start = sy * ffirst + iy
y_end = sy * flast + iy
self.drift_scope_label.setText(
"Scope of drift:\n"
f"ΔX={x_end - x_start:.2f}px\n"
f"ΔY={y_end - y_start:.2f}px"
)
self.equation_label.setText(
"Linear fit:\n"
f"X = {sx:.6f}*f + {ix:.2f} (R²={r2x:.3f})\n"
f"Y = {sy:.6f}*f + {iy:.2f} (R²={r2y:.3f})"
)
else:
self.drift_scope_label.setText("Scope of drift: N/A")
self.equation_label.setText("Linear fit: N/A")
# --- Begin new block: try to load batchcenters.npz, else fall back to beamstop_centers_raw.npy ---
# This block replaces the old raw loading and scatter logic.
# Loads batchcenters.npz if available, else falls back to beamstop_centers_raw.npy
import h5py
self.plot_fit_ax.clear()
outputfolder, outputfolder_path, logfile, logfile_path, h5file, h5file_path = config_to_paths(self.ini_file_path)
with h5py.File(h5file_path, 'r') as hf:
index_map = hf['entry/data/index'][:]
self.plot_fit_ax.set_xlim(0, len(index_map) - 1)
frames = xs = ys = None
bc_path = os.path.join(dirname, 'batchcenters.npz')
if os.path.exists(bc_path):
try:
bc_data = np.load(bc_path)
frames = bc_data['frame_positions']
xs = bc_data['updated_x']
ys = bc_data['updated_y']
except Exception:
frames = xs = ys = np.array([])
else:
raw_path = os.path.join(dirname, 'beamstop_centers_raw.npy')
if os.path.exists(raw_path):
raw = np.load(raw_path)
frames = raw[:, 0]; xs = raw[:, 1]; ys = raw[:, 2]
else:
frames, xs, ys = np.array([]), np.array([]), np.array([])
if frames is not None and len(frames) > 0:
self.plot_fit_ax.scatter(frames, xs, s=1, label='Center X')
self.plot_fit_ax.scatter(frames, ys, s=1, label='Center Y')
y_vals = [xs, ys]
if os.path.exists(fit_path) and len(frames) > 0:
fit = np.load(fit_path)
sx, ix, _ = fit[0]
sy, iy, _ = fit[1]
xs_fit = sx * frames + ix
ys_fit = sy * frames + iy
self.plot_fit_ax.plot(frames, xs_fit, label='Fit X')
self.plot_fit_ax.plot(frames, ys_fit, label='Fit Y')
y_vals.extend([xs_fit, ys_fit])
# Set y-axis limits to include both raw centers and fit lines
all_y = np.concatenate(y_vals)
ymin, ymax = all_y.min(), all_y.max()
yrange = ymax - ymin if ymax > ymin else abs(ymin) * 0.1
margin = yrange * 0.05
self.plot_fit_ax.set_ylim(ymin - margin, ymax + margin)
# --- End new block ---
self.plot_fit_ax.set_xlabel("Frame Index")
self.plot_fit_ax.set_ylabel("Center Value")
self.plot_fit_ax.legend()
# Render the canvas
self.plot_canvas.draw()
return
from coseda.initialize import write_centerfindersettings
from coseda.centerfinding.findcenters_beamstop import quartering_batch
from coseda.centerrefinement.cancellation import RefinementCancelled
import threading
# Define the Worker class as above
[docs]
class Worker(QObject):
"""Threaded worker that runs centerfinding and reports progress."""
finished = pyqtSignal(str) # Emitted when processing is finished successfully
error = pyqtSignal(str) # Emitted when an error occurs
progress = pyqtSignal(int, int) # Emitted with (done, total)
cancelled = pyqtSignal(str) # Emitted when user cancels
def __init__(self, ini_file, method, batch_size=None, stop_event=None):
super().__init__()
self.ini_file = ini_file
self.method = method
self.batch_size = batch_size
self.stop_event = stop_event or threading.Event()
[docs]
def request_cancel(self):
"""Signal cancellation to the running task."""
self.stop_event.set()
[docs]
def run(self):
"""Execute the center finding process."""
try:
if self.method == "Beamstop":
quartering_batch(
input_path=self.ini_file,
usedask=True,
batch_size=self.batch_size,
progress_callback=self.progress.emit,
stop_event=self.stop_event
)
elif self.method == "Direct":
# Run the direct center-finding method with progress callback
run_findcenters_direct(
self.ini_file,
progress_callback=lambda done, total: self.progress.emit(done, total),
stop_event=self.stop_event
)
else:
raise ValueError(f"Unknown method: {self.method}")
# Emit the finished signal with a success message
self.finished.emit(f"Center finding completed successfully for {os.path.basename(self.ini_file)}.")
except RefinementCancelled as cancel_exc:
self.cancelled.emit(str(cancel_exc))
except MemoryError:
# Notify GUI that memory was exhausted
self.error.emit("Processing failed: ran out of memory. Try reducing the batch size.")
return
except NotImplementedError as nie:
# Emit the error signal with the exception message
self.error.emit(str(nie))
except Exception as e:
# Emit the error signal with the exception message
self.error.emit(f"An error occurred during center finding:\n{str(e)}")
[docs]
class CenterFinderSettingsWindow(QDialog):
"""Settings dialog for configuring and running centerfinding across datasets."""
# Define custom signals
settings_applied = pyqtSignal(dict)
processing_finished = pyqtSignal()
progress_updated = pyqtSignal(int, int) # done, total
coordinate_selection_requested = pyqtSignal()
def __init__(self, parent, apply_callback, ini_directory, ini_file_path):
"""Initialize dialog state, UI, timers, and worker tracking."""
super().__init__(parent)
self.parent = parent
self.apply_callback = apply_callback
self.ini_directory = ini_directory
self.current_ini_file_path = ini_file_path
self.setWindowTitle("Find Centers")
self.setModal(True)
self._stop_event = threading.Event()
self._stop_requested = False
self._last_progress = (0, 0) # done, total
self.init_ui()
# Track the current method for progress handling
self.current_method = None
# Initialize monitoring variables
self.monitor_timer = QTimer()
self.monitor_timer.setInterval(1000) # 1 second
self.monitor_timer.timeout.connect(self.update_plot_viewer)
self.monitor_start_time = None
self.plots_loaded = set()
# To keep references to threads and workers
self.threads = []
self.workers = []
# Connect progress updates
self.progress_updated.connect(self.on_progress_updated)
[docs]
def on_field_focused(self):
"""Request coordinate selection from the main viewer when x0/y0 gains focus."""
self.coordinate_selection_requested.emit()
[docs]
def receive_coordinate(self, x_value: int, y_value: int):
"""Receive picked coordinates from the main viewer and populate x0/y0 fields."""
if 'x0' in self.optional_entries:
self.optional_entries['x0'].setText(str(x_value))
if 'y0' in self.optional_entries:
self.optional_entries['y0'].setText(str(y_value))
# Shift focus to avoid re-triggering
if self.required_entries:
next(iter(self.required_entries.values())).setFocus()
def _start_next_centers(self):
"""Start the next center-finding task sequentially."""
if self._stop_event.is_set():
self._reset_run_controls()
return
ini_file = self.all_ini_files[self.current_center_index]
idx = self.current_center_index + 1
total = len(self.all_ini_files)
# Update label and show indeterminate bar until first update
fname = os.path.basename(ini_file)
self.progress_label.setText(f"{fname} ({idx}/{total})")
self.progress_bar.setRange(0, 0)
from PyQt6.QtWidgets import QApplication
QApplication.processEvents()
# Create thread/worker for this file
thread = QThread()
try:
# macOS can crash with SIGBUS on deep C stacks in worker threads.
thread.setStackSize(16 * 1024 * 1024)
except Exception:
pass
worker = Worker(ini_file, self.current_method, self.batch_size_all, stop_event=self._stop_event)
worker.moveToThread(thread)
thread.started.connect(worker.run)
worker.finished.connect(self._on_centerfind_step_finished)
worker.error.connect(self.on_processing_error)
worker.cancelled.connect(self.on_processing_cancelled)
worker.progress.connect(self.progress_updated)
worker.finished.connect(thread.quit)
worker.finished.connect(worker.deleteLater)
worker.cancelled.connect(thread.quit)
worker.cancelled.connect(worker.deleteLater)
thread.finished.connect(thread.deleteLater)
thread.start()
self.threads.append(thread)
self.workers.append(worker)
def _on_centerfind_step_finished(self, message):
"""After one file finishes, trigger next or finish."""
# Handle per-file completion (refresh runs, etc.)
self.on_processing_finished(message)
# Queue next
if self._stop_event.is_set():
self._reset_run_controls()
return
if self.current_center_index < len(self.all_ini_files) - 1:
self.current_center_index += 1
self._start_next_centers()
[docs]
def init_ui(self):
"""Build the settings UI (method selection, params, progress, preview)."""
main_layout = QVBoxLayout()
main_layout.setContentsMargins(10, 10, 10, 10) # Reduced margins
main_layout.setSpacing(10) # Reduced spacing between widgets
# -------------------- Method Selection Group --------------------
method_group = QGroupBox("Center Finding Method")
method_layout = QHBoxLayout()
method_layout.setContentsMargins(10, 5, 10, 5)
method_layout.setSpacing(10)
self.method_combo = QComboBox()
self.method_combo.setToolTip(
"Currently only 'Beamstop' is fully implemented.\n"
)
self.method_combo.addItems(["Beamstop", "Direct"])
self.method_combo.setCurrentIndex(0) # Default selection
self.method_combo.currentIndexChanged.connect(self.on_method_change)
method_layout.addWidget(QLabel("Select Method:"))
method_layout.addWidget(self.method_combo)
method_group.setLayout(method_layout)
main_layout.addWidget(method_group)
# -------------------- Direct Method Settings --------------------
self.direct_group = QGroupBox("Direct Method Settings")
direct_layout = QFormLayout()
self.direct_interval_spinbox = QSpinBox()
self.direct_interval_spinbox.setRange(1, 1000000)
direct_layout.addRow("Interval (frames):", self.direct_interval_spinbox)
self.direct_box_size_spinbox = QSpinBox()
self.direct_box_size_spinbox.setRange(1, 100000)
direct_layout.addRow("Box size:", self.direct_box_size_spinbox)
self.direct_search_radius_spinbox = QSpinBox()
self.direct_search_radius_spinbox.setRange(0, 100000)
direct_layout.addRow("Search radius:", self.direct_search_radius_spinbox)
self.direct_bin_size_spinbox = QSpinBox()
self.direct_bin_size_spinbox.setRange(1, 100000)
direct_layout.addRow("Bin size:", self.direct_bin_size_spinbox)
self.direct_auto_sample_count_spinbox = QSpinBox()
self.direct_auto_sample_count_spinbox.setRange(1, 1000000)
direct_layout.addRow("Auto sample count:", self.direct_auto_sample_count_spinbox)
self.direct_auto_patch_size_spinbox = QSpinBox()
self.direct_auto_patch_size_spinbox.setRange(1, 100000)
direct_layout.addRow("Auto patch size:", self.direct_auto_patch_size_spinbox)
self.direct_auto_factor_spinbox = QDoubleSpinBox()
self.direct_auto_factor_spinbox.setRange(0.0, 1000.0)
self.direct_auto_factor_spinbox.setSingleStep(0.1)
direct_layout.addRow("Auto factor:", self.direct_auto_factor_spinbox)
self.direct_group.setLayout(direct_layout)
main_layout.addWidget(self.direct_group)
self.direct_group.hide()
# Prepopulate Direct settings from INI if available
config = read_config(self.current_ini_file_path)
sec = 'Parameters'
if config.has_option(sec, 'findcentersdirect_interval'):
self.direct_interval_spinbox.setValue(config.getint(sec, 'findcentersdirect_interval'))
if config.has_option(sec, 'findcentersdirect_box_size'):
self.direct_box_size_spinbox.setValue(config.getint(sec, 'findcentersdirect_box_size'))
if config.has_option(sec, 'findcentersdirect_search_radius'):
self.direct_search_radius_spinbox.setValue(config.getint(sec, 'findcentersdirect_search_radius'))
if config.has_option(sec, 'findcentersdirect_bin_size'):
self.direct_bin_size_spinbox.setValue(config.getint(sec, 'findcentersdirect_bin_size'))
if config.has_option(sec, 'findcentersdirect_auto_sample_count'):
self.direct_auto_sample_count_spinbox.setValue(config.getint(sec, 'findcentersdirect_auto_sample_count'))
if config.has_option(sec, 'findcentersdirect_auto_patch_size'):
self.direct_auto_patch_size_spinbox.setValue(config.getint(sec, 'findcentersdirect_auto_patch_size'))
if config.has_option(sec, 'findcentersdirect_auto_factor'):
self.direct_auto_factor_spinbox.setValue(config.getfloat(sec, 'findcentersdirect_auto_factor'))
# -------------------- Settings Group (Required Parameters) --------------------
self.settings_group = QGroupBox("Settings")
settings_layout = QFormLayout()
settings_layout.setLabelAlignment(Qt.AlignmentFlag.AlignRight)
settings_layout.setFormAlignment(Qt.AlignmentFlag.AlignLeft | Qt.AlignmentFlag.AlignTop)
settings_layout.setHorizontalSpacing(10)
settings_layout.setVerticalSpacing(5)
# Dictionary to hold QLineEdit widgets for required settings
self.required_entries = {}
# Define required settings with their labels, default values, and value types
self.required_settings_definitions = {
"tolerance": ("Tolerance", "10", int),
"min_peaks": ("Min Peaks", "10", int),
"resolution_limit": ("Resolution Limit", "200", int),
"min_samples_fraction": ("Min Samples Fraction", "0.1", float),
"dbscan_eps": ("DBSCAN eps", "5.0", float),
}
# Load current settings from INI file
config = read_config(self.current_ini_file_path)
for key, (label_text, default, value_type) in self.required_settings_definitions.items():
value = self.get_parameter_from_config(
config,
'Parameters',
f'centerfinding_{key}',
default,
is_optional=False,
value_type=value_type,
aliases=['dbscan_eps'] if key == 'dbscan_eps' else None,
)
line_edit = QLineEdit(str(value))
# Set specific tooltips for each required setting
if key == "tolerance":
line_edit.setToolTip("This defines the radius around the intial center guess we consider to find the true center. The more accurate your intial guess is, the smaller the radius can be. Using a smaller radius speeds up the search process.")
elif key == "min_peaks":
line_edit.setToolTip("Minimum number of reflections a frame must have in order to be used for centerfinding.")
elif key == "resolution_limit":
line_edit.setToolTip("Resolution limit in detector pixels for reflections to be considered. Usually considering only low resolution reflections is sufficient, and speeds up the process.")
elif key == "min_samples_fraction":
line_edit.setToolTip("Fraction of datapoints of a batch in the largest cluster to be accepted as a cluster center.")
elif key == "dbscan_eps":
line_edit.setToolTip("DBSCAN neighborhood radius for clustering Friedel-pair deviations. Existing configs without this value use 5.0.")
line_edit.setMaximumWidth(200) # Adjusted width
line_edit.setFixedHeight(24) # Uniform height
line_edit.setSizePolicy(QSizePolicy.Policy.Fixed, QSizePolicy.Policy.Fixed)
if value_type == int:
line_edit.setValidator(QIntValidator())
elif value_type == float:
if key == "min_samples_fraction":
line_edit.setValidator(QDoubleValidator(0.0, 1.0, 5))
else:
line_edit.setValidator(QDoubleValidator(0.000001, 1000000.0, 6))
self.required_entries[key] = line_edit
settings_layout.addRow(QLabel(label_text + ":"), line_edit)
# Linear-fit toggles
force_default = config.getboolean('Parameters', 'centerfinding_force_linear_fit', fallback=None)
if force_default is None:
# Backward compatibility with legacy key
force_default = config.getboolean('Parameters', 'force_linear_fit', fallback=False)
self.force_linear_checkbox = QCheckBox("Force linear fit")
self.force_linear_checkbox.setToolTip("Force center estimation using a linear fit even if R² is low")
self.force_linear_checkbox.setChecked(force_default)
settings_layout.addRow(self.force_linear_checkbox)
skip_default = config.getboolean('Parameters', 'centerfinding_skip_linear_fit', fallback=False)
# Enforce mutual exclusivity on load: skip wins
if skip_default and self.force_linear_checkbox.isChecked():
self.force_linear_checkbox.setChecked(False)
self.skip_linear_checkbox = QCheckBox("Skip linear fit")
self.skip_linear_checkbox.setToolTip("Always skip linear drift fitting and use per-batch means")
self.skip_linear_checkbox.setChecked(skip_default)
settings_layout.addRow(self.skip_linear_checkbox)
# Wire mutual exclusivity at runtime
self.skip_linear_checkbox.toggled.connect(self.on_skip_linear_toggled)
self.force_linear_checkbox.toggled.connect(self.on_force_linear_toggled)
self.settings_group.setLayout(settings_layout)
main_layout.addWidget(self.settings_group)
# -------------------- Optional Settings Group (x0, y0) --------------------
self.optional_group = QGroupBox("Optional Settings")
optional_layout = QFormLayout()
optional_layout.setLabelAlignment(Qt.AlignmentFlag.AlignRight)
optional_layout.setFormAlignment(Qt.AlignmentFlag.AlignLeft | Qt.AlignmentFlag.AlignTop)
optional_layout.setHorizontalSpacing(10)
optional_layout.setVerticalSpacing(5)
# Dictionary to hold QLineEdit widgets for optional settings
self.optional_entries = {}
# Define optional settings with their labels and default values (empty by default)
self.optional_settings_definitions = {
"x0": ("Center guess x", ""), # Empty by default
"y0": ("Center guess y", ""), # Empty by default
}
for key, (label_text, default) in self.optional_settings_definitions.items():
# Determine default from peakfinding if centerfinding missing
peak_key = f'peakfinding_{key}'
peak_default = default
if config.has_option('Parameters', peak_key):
try:
peak_default = config.get('Parameters', peak_key)
except Exception:
peak_default = default
value = self.get_parameter_from_config(
config,
'Parameters',
f'centerfinding_{key}',
peak_default,
is_optional=True
)
line_edit = FocusableLineEdit(str(value))
# Set specific tooltips for each optional setting
if key == "x0":
line_edit.setToolTip("Initial guess of beam center (px).")
elif key == "y0":
line_edit.setToolTip("Initial guess of beam center (px).")
line_edit.setMaximumWidth(200) # Adjusted width
line_edit.setFixedHeight(24) # Uniform height
line_edit.setSizePolicy(QSizePolicy.Policy.Fixed, QSizePolicy.Policy.Fixed)
line_edit.setValidator(QIntValidator()) # Ensure only integers are entered
line_edit.focus_in.connect(self.on_field_focused)
self.optional_entries[key] = line_edit
optional_layout.addRow(QLabel(label_text + ":"), line_edit)
self.optional_group.setLayout(optional_layout)
main_layout.addWidget(self.optional_group)
# -------------------- Batch Size Field --------------------
self.batch_size_group = QGroupBox("Batch Size")
self.batch_size_group.setEnabled(True) # Enabled by default since default method is Dask Enabled
batch_size_layout = QFormLayout()
batch_size_layout.setLabelAlignment(Qt.AlignmentFlag.AlignRight)
batch_size_layout.setFormAlignment(Qt.AlignmentFlag.AlignLeft | Qt.AlignmentFlag.AlignTop)
batch_size_layout.setHorizontalSpacing(10)
batch_size_layout.setVerticalSpacing(5)
self.batch_size_entry = QLineEdit("3000") # Default batch size
self.batch_size_entry.setToolTip(
"Number of frames per batch when using parallel processing. Ususally 3000 is a good number to start with. However, if you notice that your system runs out of memory during the processing, try to reduce the number. Keep in mind that you need a certain number of reflections per batch in order to allow an accurate determination of the batch center. If you can not clearly identify clusters on the scatterplot, try to increase the batch size. "
)
self.batch_size_entry.setMaximumWidth(200)
self.batch_size_entry.setFixedHeight(24)
self.batch_size_entry.setSizePolicy(QSizePolicy.Policy.Fixed, QSizePolicy.Policy.Fixed)
self.batch_size_entry.setValidator(QIntValidator(1, 1000000)) # Limit batch size
self.batch_size_entry.setPlaceholderText("Enter batch size (e.g., 10000)")
batch_size_layout.addRow(QLabel("Batch Size:"), self.batch_size_entry)
self.batch_size_group.setLayout(batch_size_layout)
main_layout.addWidget(self.batch_size_group)
# -------------------- Save Settings Group --------------------
save_group = QGroupBox("Save Settings")
save_layout = QHBoxLayout()
save_layout.setContentsMargins(10, 5, 10, 5)
save_layout.setSpacing(20)
# Consistent button size
button_width = 120
button_height = 35
save_current_button = QPushButton("Save Current")
save_current_button.clicked.connect(self.on_save_current)
save_all_button = QPushButton("Save All")
save_all_button.clicked.connect(self.on_save_all)
save_layout.addWidget(save_current_button)
save_layout.addWidget(save_all_button)
save_group.setLayout(save_layout)
main_layout.addWidget(save_group)
# -------------------- Find Centers Group --------------------
find_centers_group = QGroupBox("Find Centers")
find_centers_layout = QVBoxLayout()
find_centers_layout.setContentsMargins(10, 5, 10, 5)
find_centers_layout.setSpacing(20)
find_current_button = QPushButton("Find Current")
find_current_button.clicked.connect(self.on_find_centers_current)
self.find_current_button = find_current_button
find_all_button = QPushButton("Find All")
find_all_button.clicked.connect(self.on_find_centers_all)
self.find_all_button = find_all_button
self.stop_button = QPushButton("Stop")
self.stop_button.setEnabled(False)
self.stop_button.clicked.connect(self.on_stop_clicked)
# Row for Find buttons
button_row = QHBoxLayout()
button_row.addWidget(find_current_button)
button_row.addWidget(find_all_button)
button_row.addWidget(self.stop_button)
find_centers_layout.addLayout(button_row)
# Progress label for filename and count (own row)
self.progress_label = QLabel("")
find_centers_layout.addWidget(self.progress_label)
# Progress Bar for center finding (own row)
self.progress_bar = QProgressBar()
self.progress_bar.setRange(0, 100)
find_centers_layout.addWidget(self.progress_bar)
find_centers_group.setLayout(find_centers_layout)
main_layout.addWidget(find_centers_group)
# Embed the results preview in the main window in a labeled frame
self.preview = PreviewWindow(self.current_ini_file_path, parent=self)
self.preview.setWindowFlags(Qt.WindowType.Widget)
# Create Results group box for the run selector and preview
results_group = QGroupBox("Results")
results_layout = QVBoxLayout()
# Populate run selector dropdown, include runs with direct_centers_raw.npy or any .npz
_, outputfolder_path, _, _, _, _ = config_to_paths(self.current_ini_file_path)
runs = []
for d in os.listdir(outputfolder_path):
if not d.startswith("findcenters_"):
continue
run_dir = os.path.join(outputfolder_path, d)
if not os.path.isdir(run_dir):
continue
# Include Beamstop (.npz) or Direct (.npy) runs by filename
files = os.listdir(run_dir)
if 'direct_centers_raw.npy' in files or any(fn.endswith('.npz') for fn in files):
runs.append(d)
runs.sort()
self.run_combo = QComboBox()
self.run_combo.addItems(runs)
self.run_combo.setToolTip("Select which findcenters run to preview")
self.run_combo.currentTextChanged.connect(self.preview.set_run)
results_layout.addWidget(self.run_combo)
results_layout.addWidget(self.preview)
results_group.setLayout(results_layout)
# Combine settings and the Results group in a horizontal layout
container_layout = QHBoxLayout()
# Wrap left settings layout in a widget to align it with the Results frame
left_widget = QWidget()
left_widget.setLayout(main_layout)
container_layout.addWidget(left_widget)
container_layout.setAlignment(left_widget, Qt.AlignmentFlag.AlignTop)
container_layout.addWidget(results_group)
container_layout.setAlignment(results_group, Qt.AlignmentFlag.AlignTop)
self.setLayout(container_layout)
[docs]
def on_method_change(self, index):
"""
Toggle visibility of settings panels depending on method selection.
"""
selected_method = self.method_combo.currentText()
if selected_method == "Direct":
# Hide beamstop settings
self.settings_group.hide()
self.optional_group.hide()
self.batch_size_group.hide()
self.direct_group.show()
else:
# Show beamstop settings
self.settings_group.show()
self.optional_group.show()
self.batch_size_group.show()
self.direct_group.hide()
[docs]
def get_parameter_from_config(
self,
config,
section,
parameter,
default_value,
is_optional=False,
value_type=int,
aliases=None,
):
"""
Retrieve parameter from config. If optional and not present, return empty string.
Otherwise, return value converted to specified type or default.
"""
try:
keys = [parameter]
if aliases:
keys.extend(aliases)
value = None
for key in keys:
if config.has_option(section, key):
value = config.get(section, key)
break
if value is None:
return default_value
if is_optional and value.strip() == "":
return "" # Empty string for optional parameters
else:
if value_type == int:
return int(value)
elif value_type == float:
return float(value)
else:
return value # Return as string
except (configparser.NoSectionError, configparser.NoOptionError, ValueError):
return default_value
[docs]
def on_skip_linear_toggled(self, checked: bool):
"""Skip and force are mutually exclusive; skip wins."""
if checked and self.force_linear_checkbox.isChecked():
self.force_linear_checkbox.setChecked(False)
[docs]
def on_force_linear_toggled(self, checked: bool):
"""Force cannot stay enabled when skip is active."""
if checked and self.skip_linear_checkbox.isChecked():
self.skip_linear_checkbox.setChecked(False)
[docs]
def on_save_current(self):
settings = {}
# Gather required settings
for key in self.required_settings_definitions.keys():
settings[key] = self.required_entries[key].text().strip()
# Gather optional settings
for key in self.optional_settings_definitions.keys():
settings[key] = self.optional_entries[key].text().strip()
# Get selected method
selected_method = self.method_combo.currentText()
settings['method'] = selected_method
# Convert settings to appropriate types
processed_settings = {}
# Handle required settings
for key, (label_text, default, value_type) in self.required_settings_definitions.items():
value = settings[key]
try:
if value_type == int:
processed_settings[key] = int(value)
elif value_type == float:
processed_settings[key] = float(value)
else:
processed_settings[key] = value
except ValueError:
QMessageBox.warning(self, "Invalid Input", f"Please enter a valid {value_type.__name__} for {key}.")
return
# Handle optional settings
for key in self.optional_settings_definitions.keys():
value = settings[key]
if value == "":
processed_settings[key] = None # Set to None if empty
else:
try:
processed_settings[key] = int(value)
except ValueError:
QMessageBox.warning(self, "Invalid Input", f"Please enter a valid integer for {key}.")
return
# Include the selected method
processed_settings['method'] = selected_method
processed_settings['force_linear_fit'] = self.force_linear_checkbox.isChecked()
processed_settings['skip_linear_fit'] = self.skip_linear_checkbox.isChecked()
if processed_settings['skip_linear_fit']:
processed_settings['force_linear_fit'] = False
# Save settings to the current INI file
try:
if selected_method == "Direct":
write_centerfinderdirectsettings(
self.current_ini_file_path,
interval=self.direct_interval_spinbox.value(),
box_size=self.direct_box_size_spinbox.value(),
search_radius=self.direct_search_radius_spinbox.value(),
bin_size=self.direct_bin_size_spinbox.value(),
auto_sample_count=self.direct_auto_sample_count_spinbox.value(),
auto_patch_size=self.direct_auto_patch_size_spinbox.value(),
auto_factor=self.direct_auto_factor_spinbox.value(),
)
QMessageBox.information(self, "Settings Saved", "Direct method settings have been saved.")
return
else:
write_centerfindersettings(
self.current_ini_file_path,
tolerance=processed_settings['tolerance'],
min_peaks=processed_settings['min_peaks'],
resolution_limit=processed_settings['resolution_limit'],
min_samples_fraction=processed_settings['min_samples_fraction'],
dbscan_eps=processed_settings['dbscan_eps'],
force_linear_fit=processed_settings['force_linear_fit'],
skip_linear_fit=processed_settings['skip_linear_fit'],
x0=processed_settings.get('x0'),
y0=processed_settings.get('y0')
)
QMessageBox.information(self, "Settings Saved", "Center Finder settings have been saved to the current file.")
self.apply_callback(processed_settings) # Notify the main window
except Exception as e:
QMessageBox.critical(self, "Error", f"Failed to save settings:\n{str(e)}")
[docs]
def on_save_all(self):
settings = {}
# Gather required settings
for key in self.required_settings_definitions.keys():
settings[key] = self.required_entries[key].text().strip()
# Gather optional settings
for key in self.optional_settings_definitions.keys():
settings[key] = self.optional_entries[key].text().strip()
# Get selected method
selected_method = self.method_combo.currentText()
settings['method'] = selected_method
# Convert settings to appropriate types
processed_settings = {}
# Handle required settings
for key, (label_text, default, value_type) in self.required_settings_definitions.items():
value = settings[key]
try:
if value_type == int:
processed_settings[key] = int(value)
elif value_type == float:
processed_settings[key] = float(value)
else:
processed_settings[key] = value
except ValueError:
QMessageBox.warning(self, "Invalid Input", f"Please enter a valid {value_type.__name__} for {key}.")
return
# Handle optional settings
for key in self.optional_settings_definitions.keys():
value = settings[key]
if value == "":
processed_settings[key] = None # Set to None if empty
else:
try:
processed_settings[key] = int(value)
except ValueError:
QMessageBox.warning(self, "Invalid Input", f"Please enter a valid integer for {key}.")
return
# Include the selected method
processed_settings['method'] = selected_method
processed_settings['force_linear_fit'] = self.force_linear_checkbox.isChecked()
processed_settings['skip_linear_fit'] = self.skip_linear_checkbox.isChecked()
if processed_settings['skip_linear_fit']:
processed_settings['force_linear_fit'] = False
# Save settings to all INI files in the workspace list from main window
try:
# Use workspace list from main window instead of directory scan
for file_path in self.parent.workspace_ini_paths.values():
if not file_path.endswith(".ini"):
continue
# Save settings to this INI file
if selected_method == "Direct":
write_centerfinderdirectsettings(
file_path,
interval=self.direct_interval_spinbox.value(),
box_size=self.direct_box_size_spinbox.value(),
search_radius=self.direct_search_radius_spinbox.value(),
bin_size=self.direct_bin_size_spinbox.value(),
auto_sample_count=self.direct_auto_sample_count_spinbox.value(),
auto_patch_size=self.direct_auto_patch_size_spinbox.value(),
auto_factor=self.direct_auto_factor_spinbox.value(),
)
else:
write_centerfindersettings(
file_path,
tolerance=processed_settings['tolerance'],
min_peaks=processed_settings['min_peaks'],
resolution_limit=processed_settings['resolution_limit'],
min_samples_fraction=processed_settings['min_samples_fraction'],
dbscan_eps=processed_settings['dbscan_eps'],
force_linear_fit=processed_settings['force_linear_fit'],
skip_linear_fit=processed_settings['skip_linear_fit'],
x0=processed_settings.get('x0'),
y0=processed_settings.get('y0')
)
QMessageBox.information(self, "Settings Saved", "Center Finder settings have been saved to all files.")
self.apply_callback(processed_settings) # Notify the main window
except Exception as e:
QMessageBox.critical(self, "Error", f"Failed to save settings to all files:\n{str(e)}")
[docs]
def on_find_centers_current(self):
"""
Initiates center finding on the current INI file using the selected method.
"""
self._stop_event.clear()
self._stop_requested = False
self._last_progress = (0, 0)
self.stop_button.setEnabled(True)
self.find_current_button.setEnabled(False)
self.find_all_button.setEnabled(False)
# Indeterminate until first progress update
self.progress_bar.setRange(0, 0)
selected_method = self.method_combo.currentText()
self.current_method = selected_method
ini_file = self.current_ini_file_path
# Get batch size if applicable
batch_size = None
if selected_method == "Beamstop":
batch_size_text = self.batch_size_entry.text().strip()
if not batch_size_text.isdigit():
QMessageBox.warning(self, "Invalid Input", "Please enter a valid integer for Batch Size.")
return
batch_size = int(batch_size_text)
elif selected_method == "Beamstop":
# Beamstop without Dask does not require batch size
batch_size = None
# Create a new thread and worker
thread = QThread()
try:
# macOS can crash with SIGBUS on deep C stacks in worker threads.
thread.setStackSize(16 * 1024 * 1024)
except Exception:
pass
worker = Worker(ini_file, selected_method, batch_size, stop_event=self._stop_event)
worker.moveToThread(thread)
# Connect signals and slots
thread.started.connect(worker.run)
worker.finished.connect(self.on_processing_finished)
worker.error.connect(self.on_processing_error)
worker.cancelled.connect(self.on_processing_cancelled)
# Ensure thread stops if a memory error (or any error) occurs
worker.error.connect(thread.quit)
worker.finished.connect(thread.quit)
worker.cancelled.connect(thread.quit)
worker.finished.connect(worker.deleteLater)
worker.cancelled.connect(worker.deleteLater)
thread.finished.connect(thread.deleteLater)
# Connect progress updates from the worker
worker.progress.connect(self.progress_updated)
# Start the thread
thread.start()
# Keep references to avoid garbage collection
self.threads.append(thread)
self.workers.append(worker)
# Update progress label for single file
fname = os.path.basename(self.current_ini_file_path)
self.progress_label.setText(f"{fname} (1/1)")
[docs]
def on_find_centers_all(self):
"""
Initiates center finding on all INI files in the directory using the selected method.
"""
self._stop_event.clear()
self._stop_requested = False
self._last_progress = (0, 0)
self.stop_button.setEnabled(True)
self.find_current_button.setEnabled(False)
self.find_all_button.setEnabled(False)
# Indeterminate until first progress update
self.progress_bar.setRange(0, 0)
selected_method = self.method_combo.currentText()
ini_directory = self.ini_directory
# Get batch size if applicable
batch_size = None
if selected_method == "Beamstop":
batch_size_text = self.batch_size_entry.text().strip()
if not batch_size_text.isdigit():
QMessageBox.warning(self, "Invalid Input", "Please enter a valid integer for Batch Size.")
return
batch_size = int(batch_size_text)
elif selected_method == "Direct":
# Beamstop without Dask does not require batch size
batch_size = None
# Use workspace list from main window instead of directory scan
ini_files = list(self.parent.workspace_ini_paths.values())
if not ini_files:
QMessageBox.warning(self, "No INI Files", "No INI files found in the specified directory.")
return
# Prepare sequential queue
self.current_method = self.method_combo.currentText()
self.all_ini_files = ini_files
self.current_center_index = 0
self.batch_size_all = batch_size
# Start sequential processing
self._start_next_centers()
[docs]
def on_processing_finished(self, message):
"""
Slot to handle successful processing.
"""
# QMessageBox.information(self, "Center Finding", message)
self.processing_finished.emit()
# Refresh run list and jump to latest
self.update_run_list()
if self.current_method == "Direct":
# Reset progress bar to complete state
self.progress_bar.setRange(0, 100)
self.progress_bar.setValue(100)
self._reset_run_controls()
[docs]
def on_processing_error(self, error_message):
"""
Slot to handle processing errors.
"""
# Suggest reducing batch size on memory errors
message = error_message
if "memory" in message.lower():
message += "\nTip: Try reducing the batch size in the Settings panel."
QMessageBox.critical(self, "Center Finding Error", message)
self.processing_finished.emit()
self._reset_run_controls()
[docs]
def on_processing_cancelled(self, message):
"""
Slot to handle user-requested cancellation.
"""
QMessageBox.information(self, "Center Finding", message)
self.processing_finished.emit()
self._reset_run_controls(label="Cancelled")
[docs]
def on_stop_clicked(self):
"""User requested stop."""
self._stop_event.set()
self._stop_requested = True
self.stop_button.setEnabled(False)
# Indicate we're draining already submitted batches
done, total = self._last_progress
if total:
self.progress_label.setText(f"Stopping after submitted batches... ({done}/{total})")
else:
self.progress_label.setText("Stopping after submitted batches...")
for worker in self.workers:
if hasattr(worker, "request_cancel"):
worker.request_cancel()
def _reset_run_controls(self, label=None):
"""Re-enable run controls after completion/cancel/error."""
self.stop_button.setEnabled(False)
self.find_current_button.setEnabled(True)
self.find_all_button.setEnabled(True)
self._stop_event.clear()
self._stop_requested = False
self._last_progress = (0, 0)
# Reset progress bar and label to idle state
self.progress_bar.setRange(0, 100)
self.progress_bar.setValue(0)
if label is not None:
self.progress_label.setText(label)
elif self.progress_label.text() in ("", "Stopping..."):
self.progress_label.setText("Ready")
[docs]
def update_run_list(self):
"""
Refresh the run dropdown to include any new ``findcenters_`` folders containing
either ``direct_centers_raw.npy`` or any ``.npz`` file, auto-selecting the latest run and updating the preview.
"""
_, outputfolder_path, _, _, _, _ = config_to_paths(self.current_ini_file_path)
runs = []
for d in os.listdir(outputfolder_path):
if not d.startswith("findcenters_"):
continue
run_dir = os.path.join(outputfolder_path, d)
if not os.path.isdir(run_dir):
continue
files = os.listdir(run_dir)
if 'direct_centers_raw.npy' in files or any(fn.endswith('.npz') for fn in files):
runs.append(d)
runs.sort()
# Update combo safely
self.run_combo.blockSignals(True)
self.run_combo.clear()
self.run_combo.addItems(runs)
if runs:
latest = runs[-1]
self.run_combo.setCurrentText(latest)
self.preview.set_run(latest)
self.run_combo.blockSignals(False)
[docs]
def update_plot_viewer(self):
"""
Scans the plots folder for new .npz files created after the monitoring started.
Adds them to the plot viewer.
"""
if not self.monitor_start_time:
return # Not monitoring
### New version
outputfolder, outputfolder_path, logfile, logfile_path, h5file, h5file_path = config_to_paths(self.ini_file_path)
# Get config
config = read_config(self.ini_file_path)
# Read outputfolder from INI
#config = configparser.ConfigParser()
#config.read(self.ini_file_path)
#try:
# outputfolder = config.get('Paths', 'outputfolder')
#except:
# return
#base_output = os.path.join(os.path.dirname(self.ini_file_path), outputfolder)
# Define the plot folder path
#config = configparser.ConfigParser()
#config.read(self.current_ini_file_path)
#try:
# outputfolder = config.get('Paths', 'outputfolder')
#except (configparser.NoSectionError, configparser.NoOptionError):
# QMessageBox.warning(self, "Configuration Error", "The 'outputfolder' path is not defined in the INI file.")
# self.monitor_timer.stop()
# return
# Determine the folder containing latest findcenters output
try:
subdirs = [
d for d in os.listdir(outputfolder_path)
if d.startswith("findcenters_") and os.path.isdir(os.path.join(outputfolder_path, d))
]
if subdirs:
latest = sorted(subdirs)[-1]
graphicsfolder_path = os.path.join(outputfolder_path, latest)
else:
graphicsfolder_path = os.path.join(outputfolder_path, 'plots')
except FileNotFoundError:
# Fallback to 'plots' if base_output is missing
graphicsfolder_path = os.path.join(outputfolder_path, 'plots')
if not os.path.exists(graphicsfolder_path):
log_print(f"Graphics folder does not exist in: {graphicsfolder_path}")
return # No plots yet
# Scan for deviation .npz files
dev_regex = re.compile(r'^batch(\d+)_it(\d{3})\.npz$')
for filename in os.listdir(graphicsfolder_path):
match = dev_regex.match(filename)
if match:
batch_num = int(match.group(1))
iter_num = int(match.group(2))
file_path = os.path.join(graphicsfolder_path, filename)
creation_time = os.path.getmtime(file_path)
if creation_time >= self.monitor_start_time and file_path not in self.plots_loaded:
self.add_plot_to_viewer(batch_num, iter_num, file_path)
self.plots_loaded.add(file_path)
# Also scan for linearfit and batchcenters .npz
for special in ['linearfit.npz', 'batchcenters.npz']:
file_path = os.path.join(graphicsfolder_path, special)
if os.path.exists(file_path) and file_path not in self.plots_loaded:
# Use batch 0 and iteration 0 for top-level special plots
self.add_plot_to_viewer(0, 0, file_path)
self.plots_loaded.add(file_path)
[docs]
def add_plot_to_viewer(self, batch_num, iter_num, file_path):
"""
Adds a plot to the QTreeWidget under the appropriate batch and iteration.
"""
# Find if the batch already exists
batch_item = None
for i in range(self.plot_tree.topLevelItemCount()):
item = self.plot_tree.topLevelItem(i)
if int(item.text(0)) == batch_num:
batch_item = item
break
if not batch_item:
# Create a new batch item
batch_item = QTreeWidgetItem([str(batch_num), "", ""])
self.plot_tree.addTopLevelItem(batch_item)
# Add the iteration as a child
plot_item = QTreeWidgetItem([str(batch_num), f"it{iter_num:03d}", os.path.basename(file_path)])
plot_item.setData(2, Qt.ItemDataRole.UserRole, file_path) # Store the file path
batch_item.addChild(plot_item)
batch_item.setExpanded(True)
[docs]
def display_selected_plot(self, item, column):
"""
Displays the selected plot from saved numpy arrays.
"""
# Only proceed if it's a plot item (no children)
if item.childCount() != 0:
return
file_path = item.data(2, Qt.ItemDataRole.UserRole)
if not file_path or not os.path.exists(file_path):
self.plot_ax.clear()
self.plot_ax.text(0.5, 0.5, "Plot file does not exist.", ha='center', va='center')
self.plot_canvas.draw()
return
# Clear previous plot
self.plot_ax.clear()
dirname = os.path.dirname(file_path)
basename = os.path.basename(file_path)
# Handle .npz plot types
if basename.endswith('.npz') and basename.startswith('batch'):
data = np.load(file_path)
dev = data['deviations']
mean = data['mean']
if dev.size > 0:
self.plot_ax.scatter(dev[:, 0], dev[:, 1], s=0.5)
if mean.size > 0:
self.plot_ax.scatter(mean[0], mean[1], marker='x', color='red')
self.plot_ax.set_xlabel("X-deviation / px")
self.plot_ax.set_ylabel("Y-deviation / px")
elif basename == 'linearfit.npz':
# Inline plot linear fit with equation annotation
data = np.load(file_path)
frames = data.get('frame_positions', np.arange(len(data['updated_x'])))
ux = data['updated_x']
uy = data['updated_y']
# Plot the lines
self.plot_ax.plot(frames, ux, label='Center X')
self.plot_ax.plot(frames, uy, label='Center Y')
# Compute and annotate equations
coeffs_x = np.polyfit(frames, ux, 1)
coeffs_y = np.polyfit(frames, uy, 1)
m_x, b_x = coeffs_x
m_y, b_y = coeffs_y
eq_text = (
f"X = {m_x:.3e}·f + {b_x:.3e}\n"
f"Y = {m_y:.3e}·f + {b_y:.3e}"
)
self.plot_ax.text(
0.05, 0.95, eq_text,
transform=self.plot_ax.transAxes,
fontsize=8, verticalalignment='top'
)
self.plot_ax.set_xlabel("Frame Index")
self.plot_ax.set_ylabel("Center Coordinates / px")
self.plot_ax.legend()
elif basename == 'batchcenters.npz':
data = np.load(file_path)
frames = data['frame_positions']
ux = data['updated_x']
uy = data['updated_y']
self.plot_ax.scatter(frames, ux, s=1, label='Center X')
self.plot_ax.scatter(frames, uy, s=1, label='Center Y')
self.plot_ax.set_xlabel("Frame Index")
self.plot_ax.set_ylabel("Center Value")
self.plot_ax.legend()
else:
# Fallback: load image
img = plt.imread(file_path)
self.plot_ax.imshow(img)
self.plot_ax.axis('off')
# Render the canvas
self.plot_canvas.draw()
[docs]
def stop_monitoring(self):
"""
Stops the plot viewer monitoring.
"""
self.monitor_timer.stop()
self.monitor_start_time = None
self.plots_loaded = set()
[docs]
def closeEvent(self, event):
"""
Ensures that the timer is stopped when the dialog is closed.
"""
self.monitor_timer.stop()
event.accept()
[docs]
def on_progress_updated(self, done, total):
# Remember latest progress for stop-label updates
self._last_progress = (done, total)
# If still indeterminate and no progress yet, keep busy indicator
if self.progress_bar.minimum() == 0 and self.progress_bar.maximum() == 0:
if done == 0:
return
# Switch to determinate once first progress arrives
self.progress_bar.setRange(0, 100)
# Update the progress bar percentage
if total > 0:
percent = int(done / total * 100)
else:
percent = 0
self.progress_bar.setValue(percent)
# If stop was requested, show that we are draining submitted work
if self._stop_requested:
self.progress_label.setText(f"Stopping after submitted batches... ({done}/{total})")