"""UI for configuring and running center refinement for beamstop/direct methods."""
from PyQt6.QtWidgets import (
QDialog, QLabel, QLineEdit, QPushButton, QVBoxLayout,
QHBoxLayout, QGroupBox, QFormLayout, QMessageBox,
QSizePolicy, QComboBox, QTreeWidget, QTreeWidgetItem, QVBoxLayout, QProgressBar,
QSplitter, QWidget, QStackedLayout,
QSpinBox, QDoubleSpinBox, QCheckBox
)
[docs]
class TrimmingDoubleSpinBox(QDoubleSpinBox):
[docs]
def textFromValue(self, value: float) -> str:
txt = super().textFromValue(value)
# strip trailing zeros and optional decimal point
if '.' in txt:
txt = txt.rstrip('0').rstrip('.')
return txt
from PyQt6.QtCore import Qt, pyqtSignal, QTimer, QThread, QObject
from PyQt6.QtGui import QIntValidator, QDoubleValidator, QPixmap, QGuiApplication
import configparser
import os
import re
import time
import traceback
import sys # Added for platform detection
import subprocess # Added for opening files
import zipfile
import numpy as np
from matplotlib.backends.backend_qtagg import FigureCanvasQTAgg as FigureCanvas
from matplotlib.figure import Figure
from PyQt6.QtWidgets import QDialog, QVBoxLayout, QTreeWidgetItem
from PyQt6.QtCore import QTimer
import threading
from coseda.initialize import write_centerrefinementsettings
from coseda.centerrefinement.refinecenters_beamstop import refinecenters_batch
from coseda.centerrefinement.refinecenters_direct import refinecenters_direct_batch, write_centerrefinement_direct_settings
from coseda.io import config_to_paths
from coseda.centerrefinement.cancellation import RefinementCancelled
import h5py
[docs]
def envelope_downsample(x: np.ndarray, y: np.ndarray, n_bins: int):
"""
Downsample by capturing first, last, min, and max in each of n_bins equally-sized bins.
Returns (downsampled_x, downsampled_y).
"""
# Indices to split into bins
bin_edges = np.linspace(0, len(x), n_bins + 1, dtype=int)
out_x, out_y = [], []
for i in range(len(bin_edges) - 1):
sl = slice(bin_edges[i], bin_edges[i + 1])
xi, yi = x[sl], y[sl]
if xi.size == 0:
continue
# capture first, last, min, and max
out_x.extend([xi[0], xi[-1], xi[yi.argmin()], xi[yi.argmax()]])
out_y.extend([yi[0], yi[-1], yi.min(), yi.max()])
# sort by x to preserve order
order = np.argsort(out_x)
return np.array(out_x)[order], np.array(out_y)[order]
# --- Simple uniform decimation for fast plotting ---
[docs]
def decimate(x: np.ndarray, y: np.ndarray, max_points: int):
"""
Downsample by taking every nth point so that at most max_points remain.
"""
length = len(x)
stride = max(1, length // max_points)
return x[::stride], y[::stride]
[docs]
class ClickableLabel(QLabel):
"""Label that emits a clicked signal when the user releases the left mouse button."""
clicked = pyqtSignal()
[docs]
def mouseReleaseEvent(self, event):
if event.button() == Qt.MouseButton.LeftButton:
self.clicked.emit()
super().mouseReleaseEvent(event)
[docs]
class Worker(QObject):
"""Threaded worker that runs center refinement and reports success/errors."""
finished = pyqtSignal(str)
error = pyqtSignal(str)
cancelled = pyqtSignal(str)
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 the worker's underlying job to stop."""
self.stop_event.set()
[docs]
def run(self):
"""Execute the configured center refinement workflow."""
try:
if self.method == "Beamstop":
refinecenters_batch(
input_path=self.ini_file,
usedask=True,
batch_size=self.batch_size,
stop_event=self.stop_event
)
elif self.method == "Direct":
refinecenters_direct_batch(
input_path=self.ini_file,
stop_event=self.stop_event
)
elif self.method == "No Beamstop":
raise NotImplementedError("No Beamstop method is currently not implemented.")
else:
raise ValueError(f"Unknown method: {self.method}")
self.finished.emit(f"Center refinement completed successfully for {os.path.basename(self.ini_file)}.")
except RefinementCancelled as cancel_exc:
self.cancelled.emit(str(cancel_exc))
except NotImplementedError as nie:
self.error.emit(str(nie))
except Exception as e:
error_trace = traceback.format_exc()
self.error.emit(f"An error occurred during center refinement:\n{str(e)}\n{error_trace}")
# -------------------- PreviewWindow Class --------------------
[docs]
class PreviewWindow(QDialog):
"""Non-modal preview window for inspecting refinecenters outputs and runs."""
def __init__(self, ini_file_path, parent=None):
super().__init__(parent)
# Make this dialog non-modal and behave as a normal window
self.setModal(False)
self.setWindowModality(Qt.WindowModality.NonModal)
# Ensure it's a top-level window, not always-on-top
flags = self.windowFlags()
self.setWindowFlags(flags & ~Qt.WindowType.WindowStaysOnTopHint | Qt.WindowType.Window)
self.ini_file_path = ini_file_path
self.setWindowTitle(f"Refine Preview: {os.path.basename(ini_file_path)}")
self.setAttribute(Qt.WidgetAttribute.WA_DeleteOnClose)
layout = QVBoxLayout(self)
# Dropdown to select specific refine run
self.run_selector = QComboBox()
self.run_selector.addItem("Select Run...")
self.run_selector.setEnabled(False)
layout.addWidget(self.run_selector)
self.run_selector.currentIndexChanged.connect(self.on_run_selected)
self.available_runs = {} # map run name to folder path
# Canvas with three vertical subplots: deviations, LOWESS X/Y, and centers X/Y
self.preview_canvas = FigureCanvas(Figure(figsize=(16, 18)))
axes = self.preview_canvas.figure.subplots(3, 1)
# Subplot 0: deviations + LOWESS fit
self.dev_ax = axes[0]
# Subplot 1: center X and LOWESS X-deviation
self.top_ax = axes[1]
self.top_ax2 = self.top_ax.twinx()
# Subplot 2: center Y and LOWESS Y-deviation
self.bottom_ax = axes[2]
self.bottom_ax2 = self.bottom_ax.twinx()
layout.addWidget(self.preview_canvas)
# Navigation controls
nav_layout = QHBoxLayout()
nav_layout.setAlignment(Qt.AlignmentFlag.AlignHCenter)
self.prev_btn = QPushButton("Previous")
self.next_btn = QPushButton("Next")
self.iter_label = QLabel("0/0")
# Narrow navigation buttons
self.prev_btn.setFixedWidth(80)
self.next_btn.setFixedWidth(80)
nav_layout.addWidget(self.prev_btn)
nav_layout.addWidget(self.iter_label)
nav_layout.addWidget(self.next_btn)
layout.addLayout(nav_layout)
self.prev_btn.clicked.connect(self.on_prev)
self.next_btn.clicked.connect(self.on_next)
# Monitoring
self.monitor_start_time = time.time()
self.plots_loaded = set()
self.monitor_timer = QTimer(self)
self.monitor_timer.setInterval(1000)
self.monitor_timer.timeout.connect(self.update_preview)
self.monitor_timer.start()
# Initialize iteration navigation
self.iterations = [] # List of (itnum, file_path)
self.current_iter_index = 0
self._update_nav_buttons()
self._raw_frame_count = self._determine_raw_frame_count()
def _determine_raw_frame_count(self):
"""Cache the total raw-frame count to avoid reopening the HDF5 file during refinement."""
try:
_, _, _, _, _, h5file_path = config_to_paths(self.ini_file_path)
with h5py.File(h5file_path, 'r') as f:
if 'entry/data/index' in f:
return int(f['entry/data/index'].shape[0])
return int(f['entry/data/center_x'].shape[0])
except Exception:
return None
[docs]
def update_preview(self):
"""
Monitors for ``refinecenters_`` folders, populates dropdown with all runs,
and auto-selects the newest run created after monitoring started.
Only supports NPZ iteration detection.
"""
# Use config_to_paths to get the absolute output folder path
_, base_output, _, _, _, _ = config_to_paths(self.ini_file_path)
if not os.path.isdir(base_output):
return
# Find all refinecenters_ directories
try:
subdirs = [d for d in os.listdir(base_output) if d.startswith("refinecenters_")]
except OSError:
return
if not subdirs:
return
# Sort all runs by modification time (oldest first)
subdirs_sorted = sorted(
subdirs,
key=lambda d: os.path.getmtime(os.path.join(base_output, d))
)
# Populate dropdown with any new run entries
for run_name in subdirs_sorted:
if run_name not in self.available_runs:
run_path = os.path.join(base_output, run_name)
self.available_runs[run_name] = run_path
self.run_selector.addItem(run_name, run_path)
# Enable selector now that runs exist
self.run_selector.setEnabled(True)
# Auto-select the newest run when it appears (fallback to latest even if monitor_start_time is None)
latest_run = subdirs_sorted[-1]
latest_mtime = os.path.getmtime(os.path.join(base_output, latest_run))
start_time = self.monitor_start_time if self.monitor_start_time is not None else 0
if latest_mtime >= start_time:
idx = self.run_selector.findText(latest_run)
if idx != -1 and self.run_selector.currentIndex() != idx:
self.run_selector.setCurrentIndex(idx)
# Refresh iteration files for the currently selected run
current_idx = self.run_selector.currentIndex()
if current_idx > 0:
run_path = self.run_selector.itemData(current_idx)
# Scan for iteration files in the selected run
new_items = []
for fn in os.listdir(run_path):
m = re.match(r'^lowess_and_centers_it(\d{3})\.npz$', fn)
if m:
itnum = int(m.group(1))
new_items.append((itnum, os.path.join(run_path, fn)))
# Sort and update if changed
new_items.sort(key=lambda x: x[0])
if new_items and new_items != self.iterations:
self.iterations = new_items
# Always show latest iteration when new files appear
self.current_iter_index = len(self.iterations) - 1
self._update_nav_buttons()
self._display_current()
# --- Navigation and display helpers ---
def _update_nav_buttons(self):
total = len(self.iterations)
if total > 0:
self.iter_label.setText(f"Iteration {self.current_iter_index+1}/{total}")
else:
self.iter_label.setText("Iteration 0/0")
self.prev_btn.setEnabled(self.current_iter_index > 0)
self.next_btn.setEnabled(self.current_iter_index < total - 1)
def _display_current(self):
"""Plot deviations + LOWESS fit and previous centers for the current iteration."""
# Determine iteration and paths
if not self.iterations:
return
itnum, data_path = self.iterations[self.current_iter_index]
n_raw_frames = self._raw_frame_count
# Always treat data_path as the NPZ file path
lowess_path = data_path
folder = os.path.dirname(lowess_path)
# Paths to data
dev_path = os.path.join(folder, f'deviations_it{itnum:03d}.npz')
# Clear all axes
self.dev_ax.clear()
self.top_ax.clear()
self.top_ax2.clear()
self.bottom_ax.clear()
self.bottom_ax2.clear()
# Load LOWESS and centers data
data_low = np.load(lowess_path)
lowess_x = data_low['lowess_x']
lowess_y = data_low['lowess_y']
centers_x = data_low['existing_center_x']
centers_y = data_low['existing_center_y']
# Existing raw frame indices from NPZ (fallback to sequential if missing)
if 'existing_indices' in data_low.files:
raw_indices = data_low['existing_indices'].astype(int)
else:
raw_indices = np.arange(len(centers_x), dtype=int)
if 'index_length' in data_low.files:
n_raw_frames = int(data_low['index_length'])
elif n_raw_frames is None and raw_indices.size > 0:
n_raw_frames = int(raw_indices.max()) + 1
elif n_raw_frames is None:
n_raw_frames = len(centers_x)
# Raw frame numbers for LOWESS curves
raw_lowess_x = lowess_x[:, 0].astype(int)
raw_lowess_y = lowess_y[:, 0].astype(int)
# Build full-length arrays for centers, handling Beamstop vs Direct formats
if 'existing_center_x' in data_low:
# Beamstop: show gaps as NaN
full_centers_x = np.full(n_raw_frames, np.nan)
full_centers_y = np.full(n_raw_frames, np.nan)
full_centers_x[raw_indices] = centers_x
full_centers_y[raw_indices] = centers_y
else:
# Direct: centers_x and centers_y already full-length, no NaN gaps
full_centers_x = centers_x
full_centers_y = centers_y
# LOWESS curves: always represent missing frames as NaN
full_lowess_x = np.full(n_raw_frames, np.nan)
full_lowess_y = np.full(n_raw_frames, np.nan)
if lowess_x.size:
full_lowess_x[raw_lowess_x] = lowess_x[:, 1]
if lowess_y.size:
full_lowess_y[raw_lowess_y] = lowess_y[:, 1]
# Plot deviations + LOWESS fit on dev_ax
if os.path.exists(dev_path):
try:
data_dev = np.load(dev_path)
except (EOFError, zipfile.BadZipFile, ValueError) as exc:
# Corrupted/partial NPZ (can happen if viewer races the writer); skip plotting
from coseda.logging_utils import log_print
log_print(f"[WARN] Could not load deviations file {dev_path} ({exc}); skipping preview for this iteration")
data_dev = None
if data_dev is not None:
dev_indices = data_dev['indices']
dev = data_dev['deviations']
# Guard against empty payloads
if dev.size == 0 or dev_indices.size == 0:
self.dev_ax.text(0.5, 0.5, "No deviations to plot", ha='center', va='center', transform=self.dev_ax.transAxes)
self.preview_canvas.draw()
return
# Use raw dev_indices directly
raw_dev_indices = dev_indices.astype(int)
# Decimate deviations for faster plotting
max_points = 5000
ds_x_idx, ds_x = decimate(raw_dev_indices, dev[:, 0], max_points)
ds_y_idx, ds_y = decimate(raw_dev_indices, dev[:, 1], max_points)
self.dev_ax.scatter(ds_x_idx, ds_x, s=1, alpha=1, label='X deviation', color='blue')
self.dev_ax.scatter(ds_y_idx, ds_y, s=1, alpha=1, label='Y deviation', color='orange')
# Plot LOESS fit lines: for Direct, compute LOESS on deviations; for Beamstop, use saved curves
config = configparser.ConfigParser()
config.read(self.ini_file_path)
if config.has_option('Parameters', 'centerrefinement_direct_loess_span'):
from coseda.centerrefinement.refinecenters_direct import lowess_xunit
loess_span = float(config.get('Parameters', 'centerrefinement_direct_loess_span'))
fitted_x = lowess_xunit(dev_indices, dev[:, 0], dev_indices, loess_span)
fitted_y = lowess_xunit(dev_indices, dev[:, 1], dev_indices, loess_span)
# Use fitted indices as raw frame numbers directly
raw_fitted_x = fitted_x[:, 0].astype(int)
raw_fitted_y = fitted_y[:, 0].astype(int)
self.dev_ax.plot(raw_fitted_x, fitted_x[:, 1], label='LOWESS X', color='red', linestyle='-')
self.dev_ax.plot(raw_fitted_y, fitted_y[:, 1], label='LOWESS Y', color='green', linestyle='--')
else:
# Use raw-lowess mappings computed earlier
self.dev_ax.plot(raw_lowess_x, lowess_x[:, 1], label='LOWESS X', color='red', linestyle='-')
self.dev_ax.plot(raw_lowess_y, lowess_y[:, 1], label='LOWESS Y', color='green', linestyle='--')
else:
# Deviations file may not be written yet; inform user and retry shortly.
self.dev_ax.text(0.5, 0.5, "Waiting for deviations file...", ha='center', va='center', transform=self.dev_ax.transAxes)
from PyQt6.QtCore import QTimer
QTimer.singleShot(1000, self._display_current)
self.preview_canvas.draw()
return
self.dev_ax.set_xlabel("Frame Index")
self.dev_ax.set_ylabel("Deviation")
self.dev_ax.legend(loc='upper right')
self.dev_ax.set_title(f"Deviations & LOWESS (it{itnum:03d})")
# Plot center X and LOWESS X-deviation on top_ax
ax1 = self.top_ax
ax1.plot(np.arange(n_raw_frames), full_centers_x, linestyle='-', alpha=1, label="Center X")
ax1.set_xlabel("Frame Index")
ax1.set_ylabel("X-coordinate")
ax1.legend(loc='upper left')
ax1b = self.top_ax2
ax1b.plot(np.arange(n_raw_frames), full_lowess_x, linestyle='-', label="LOWESS X-deviation", color='red')
ax1b.set_ylabel("X-deviation")
ax1b.yaxis.set_label_position("right")
ax1b.yaxis.tick_right()
ax1b.legend(loc='upper right')
ax1.set_title(f"Centers X & LOWESS X (it{itnum:03d})")
# Plot center Y and LOWESS Y-deviation on bottom_ax
ax2 = self.bottom_ax
ax2.plot(np.arange(n_raw_frames), full_centers_y, linestyle='-', alpha=1, label="Center Y")
ax2.set_xlabel("Frame Index")
ax2.set_ylabel("Y-coordinate")
ax2.legend(loc='upper left')
ax2b = self.bottom_ax2
ax2b.plot(np.arange(n_raw_frames), full_lowess_y, linestyle='-', label="LOWESS Y-deviation", color='green')
ax2b.set_ylabel("Y-deviation")
ax2b.yaxis.set_label_position("right")
ax2b.yaxis.tick_right()
ax2b.legend(loc='upper right')
ax2.set_title(f"Centers Y & LOWESS Y (it{itnum:03d})")
# Enforce full raw-frame x-axis range
self.dev_ax.set_xlim(0, n_raw_frames - 1)
self.top_ax.set_xlim(0, n_raw_frames - 1)
self.bottom_ax.set_xlim(0, n_raw_frames - 1)
# Layout
self.preview_canvas.figure.tight_layout(rect=[0, 0.03, 1, 0.95])
self.preview_canvas.draw()
self._update_nav_buttons()
[docs]
def on_prev(self):
if self.current_iter_index > 0:
self.current_iter_index -= 1
self._display_current()
[docs]
def on_next(self):
if self.current_iter_index < len(self.iterations) - 1:
self.current_iter_index += 1
self._display_current()
[docs]
def on_run_selected(self, index):
"""
Handles selection of a specific run from the dropdown.
"""
if index <= 0:
return
run_path = self.run_selector.itemData(index)
# Scan for iteration files in chosen run folder
items = []
for fn in os.listdir(run_path):
m = re.match(r'^lowess_and_centers_it(\d{3})\.npz$', fn)
if m:
itnum = int(m.group(1))
items.append((itnum, os.path.join(run_path, fn)))
items.sort(key=lambda x: x[0])
# Fallback for Direct method: if no npz iterations, check for raw .npy outputs
if not items:
indices_path = os.path.join(run_path, 'indices.npy')
coarse_path = os.path.join(run_path, 'coarse_centers.npy')
fine_path = os.path.join(run_path, 'fine_centers.npy')
if os.path.exists(indices_path) and os.path.exists(coarse_path) and os.path.exists(fine_path):
items = [(0, run_path)]
self.iterations = items
self.current_iter_index = 0
self._update_nav_buttons()
self._display_current()
[docs]
class CenterRefinementSettingsWindow(QDialog):
"""Settings dialog to launch center refinement across selected INIs."""
processing_finished = pyqtSignal()
def __init__(self, parent, ini_directory, ini_file_path):
"""Initialize state, UI, and timers for refinement previews."""
super().__init__(parent)
self.parent = parent
self.ini_directory = ini_directory
self.current_ini_file_path = ini_file_path
self.setWindowTitle("Refine Centers")
# Make settings window non-modal and normal stacking
self.setModal(False)
self.setWindowModality(Qt.WindowModality.NonModal)
flags = self.windowFlags()
self.setWindowFlags(flags & ~Qt.WindowType.WindowStaysOnTopHint | Qt.WindowType.Window)
self._stop_event = threading.Event()
self._stop_requested = False
self._last_progress = 0
self.init_ui()
# 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 = []
def _start_next_refinement(self):
"""Start the next refinement from the workspace list sequentially."""
if self._stop_event.is_set():
self._reset_run_controls()
return
ini_file = self.all_ini_files[self.current_refine_index]
idx = self.current_refine_index + 1
total = len(self.all_ini_files)
# Update label and reset progress bar
fname = os.path.basename(ini_file)
self.progress_label.setText(f"{fname} ({idx}/{total})")
# Indeterminate until first iteration file appears
self.progress_bar.setRange(0, 0)
# Force UI to refresh
from PyQt6.QtWidgets import QApplication
QApplication.processEvents()
# Create thread and 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.method_combo.currentText(), self.batch_size_all, stop_event=self._stop_event)
worker.moveToThread(thread)
thread.started.connect(worker.run)
worker.finished.connect(self._on_refinement_step_finished)
worker.error.connect(self.on_refinement_error)
worker.cancelled.connect(self.on_refinement_cancelled)
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()
# Keep references
self.threads.append(thread)
self.workers.append(worker)
def _on_refinement_step_finished(self, message):
"""
Slot after each file finishes. Starts the next or stops.
"""
# Adjust progress bar to full for this file
try:
_, base_output, _, _, _, _ = config_to_paths(self.current_ini_file_path)
runs = [d for d in os.listdir(base_output) if d.startswith('refinecenters_')]
if runs:
latest = max(runs, key=lambda d: os.path.getmtime(os.path.join(base_output, d)))
folder = os.path.join(base_output, latest)
count = sum(1 for fn in os.listdir(folder) if re.match(r'^lowess_and_centers_it\d{3}\.npz$', fn))
self.progress_bar.setMaximum(count)
self.progress_bar.setValue(count)
except Exception:
pass
# Move to next file or finish
if self._stop_event.is_set():
if hasattr(self, 'progress_monitor_timer'):
self.progress_monitor_timer.stop()
self.processing_finished.emit()
self._reset_run_controls()
return
if self.current_refine_index < len(self.all_ini_files) - 1:
self.current_refine_index += 1
self._start_next_refinement()
else:
# All done
if hasattr(self, 'progress_monitor_timer'):
self.progress_monitor_timer.stop()
self.processing_finished.emit()
self._reset_run_controls()
[docs]
def init_ui(self):
# Left column layout for settings
left_layout = QVBoxLayout()
left_layout.setContentsMargins(10, 10, 10, 10)
left_layout.setSpacing(10)
# -------------------- Settings Group (Required Parameters) --------------------
# Beamstop panel
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 widgets for required settings
self.required_entries = {}
# Load current settings from INI file
config = configparser.ConfigParser()
config.read(self.current_ini_file_path)
# Tolerance
tol_val = self.get_parameter_from_config(
config, 'Parameters', 'centerrefinement_tolerance', '5', is_optional=False
)
tol_spin = QSpinBox()
tol_spin.setRange(0, 1000000)
tol_spin.setSingleStep(1)
tol_spin.setValue(int(tol_val))
tol_spin.setToolTip("Pixels around the current center that will be searched for Friedel-pair deviations. Smaller radii assume a better initial center and run faster.")
settings_layout.addRow(QLabel("Tolerance:"), tol_spin)
self.required_entries['tolerance'] = tol_spin
# Min Peaks
peaks_val = self.get_parameter_from_config(
config, 'Parameters', 'centerrefinement_min_peaks', '15', is_optional=False
)
peaks_spin = QSpinBox()
peaks_spin.setRange(0, 1000000)
peaks_spin.setSingleStep(1)
peaks_spin.setValue(int(peaks_val))
peaks_spin.setToolTip("Minimum number of peaks a processed frame must contain to participate in center refinement.")
settings_layout.addRow(QLabel("Min Peaks:"), peaks_spin)
self.required_entries['min_peaks'] = peaks_spin
# Resolution Limit
res_val = self.get_parameter_from_config(
config, 'Parameters', 'centerrefinement_resolution_limit', '400', is_optional=False
)
res_spin = QSpinBox()
res_spin.setRange(0, 1000000)
res_spin.setSingleStep(1)
res_spin.setValue(int(res_val))
res_spin.setToolTip("Maximum distance from the current center (in detector pixels) for peaks that enter the refinement. Limits the fit to low-resolution reflections.")
settings_layout.addRow(QLabel("Resolution Limit:"), res_spin)
self.required_entries['resolution_limit'] = res_spin
# Max Iterations
iter_val = self.get_parameter_from_config(
config, 'Parameters', 'centerrefinement_max_iterations', '10', is_optional=False
)
iter_spin = QSpinBox()
iter_spin.setRange(0, 1000000)
iter_spin.setSingleStep(1)
iter_spin.setValue(int(iter_val))
iter_spin.setToolTip("Upper bound on refinement iterations. The loop stops once this count is reached even if the convergence criterion is not met.")
settings_layout.addRow(QLabel("Max Iterations:"), iter_spin)
self.required_entries['max_iterations'] = iter_spin
# Convergence Threshold
conv_val = self.get_parameter_from_config(
config, 'Parameters', 'centerrefinement_convergence_threshold', '0.01', is_optional=False
)
conv_spin = TrimmingDoubleSpinBox()
conv_spin.setRange(0.0, 1.0)
conv_spin.setDecimals(5)
conv_spin.setSingleStep(0.01)
conv_spin.setValue(float(conv_val))
conv_spin.setToolTip("Convergence threshold (px). When the LOWESS-smoothed deviations stay within this band for both axes the refinement stops.")
settings_layout.addRow(QLabel("Convergence Threshold:"), conv_spin)
self.required_entries['convergence_threshold'] = conv_spin
# LOWESS Fraction
lowess_val = self.get_parameter_from_config(
config, 'Parameters', 'centerrefinement_lowess_frac', '0.1', is_optional=True
)
lowess_spin = TrimmingDoubleSpinBox()
lowess_spin.setRange(0.0, 1.0)
lowess_spin.setDecimals(5)
lowess_spin.setSingleStep(0.01)
lowess_spin.setValue(float(lowess_val) if lowess_val is not None else 0.1)
lowess_spin.setToolTip("Define size of the LOWESS window as a fraction of the dataset. Ignored when a LOWESS window (absolute frame count) is specified.")
settings_layout.addRow(QLabel("LOWESS Fraction:"), lowess_spin)
self.required_entries['lowess_frac'] = lowess_spin
# LOWESS Window (frames)
lowess_window_val = self.get_parameter_from_config(
config, 'Parameters', 'centerrefinement_lowess_window', '0', is_optional=True
)
lowess_window_spin = QSpinBox()
lowess_window_spin.setRange(0, 1_000_000)
lowess_window_spin.setSingleStep(10)
lowess_window_spin.setValue(int(float(lowess_window_val)) if lowess_window_val is not None else 0)
lowess_window_spin.setToolTip("Absolute window size in frames for the LOWESS smoother. Overrides the fraction when greater than zero.")
settings_layout.addRow(QLabel("LOWESS Window:"), lowess_window_spin)
self.required_entries['lowess_window'] = lowess_window_spin
# Deviation aggregation for LOWESS
aggregation_val = config.get(
'Parameters',
'centerrefinement_deviation_aggregation',
fallback='median'
).strip().lower()
pair_lowess_checkbox = QCheckBox("Use pair-level deviations")
pair_lowess_checkbox.setChecked(aggregation_val in {"pair", "pairs", "none", "notebook"})
pair_lowess_checkbox.setToolTip(
"Checked fits LOWESS to every accepted Friedel-pair deviation, so frames with more pairs have more influence. "
"Unchecked first reduces each frame to its median deviation, so each frame contributes one robust value."
)
settings_layout.addRow(QLabel("LOWESS Input:"), pair_lowess_checkbox)
self.required_entries['pair_lowess'] = pair_lowess_checkbox
beam_widget = QWidget()
beam_widget.setLayout(settings_layout)
# -------------------- Method Selection Group --------------------
method_group = QGroupBox("Refinement Method")
method_layout = QHBoxLayout()
method_layout.setContentsMargins(10, 5, 10, 5)
method_layout.setSpacing(10)
self.method_combo = QComboBox()
self.method_combo.addItems(["Beamstop", "Direct"])
self.method_combo.setCurrentIndex(0)
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)
method_group.setSizePolicy(QSizePolicy.Policy.Preferred, QSizePolicy.Policy.Fixed)
# -------------------- Direct Parameters Group --------------------
self.direct_entries = {}
direct_layout = QFormLayout()
direct_layout.setLabelAlignment(Qt.AlignmentFlag.AlignRight)
direct_layout.setFormAlignment(Qt.AlignmentFlag.AlignLeft | Qt.AlignmentFlag.AlignTop)
direct_layout.setHorizontalSpacing(10)
direct_layout.setVerticalSpacing(5)
# Load current direct settings from INI
config = configparser.ConfigParser()
config.read(self.current_ini_file_path)
# Threshold Fraction
thresh_val = self.get_parameter_from_config(
config, 'Parameters', 'centerrefinement_direct_threshold_frac', '0.5', is_optional=False
)
thresh_spin = TrimmingDoubleSpinBox()
thresh_spin.setRange(0.0, 1.0)
thresh_spin.setDecimals(5)
thresh_spin.setSingleStep(0.01)
thresh_spin.setValue(float(thresh_val))
direct_layout.addRow(QLabel("Threshold Fraction:"), thresh_spin)
self.direct_entries['threshold_frac'] = thresh_spin
# Box1 Half
box_val = self.get_parameter_from_config(
config, 'Parameters', 'centerrefinement_direct_box1_half', '10', is_optional=False
)
box_spin = QSpinBox()
box_spin.setRange(1, 10000)
box_spin.setSingleStep(1)
box_spin.setValue(int(box_val))
direct_layout.addRow(QLabel("Box1 Half:"), box_spin)
self.direct_entries['box1_half'] = box_spin
# Step 1
step1_val = self.get_parameter_from_config(
config, 'Parameters', 'centerrefinement_direct_step1', '1.0', is_optional=False
)
step1_spin = TrimmingDoubleSpinBox()
step1_spin.setRange(0.001, 10000.0)
step1_spin.setDecimals(5)
step1_spin.setSingleStep(0.1)
step1_spin.setValue(float(step1_val))
direct_layout.addRow(QLabel("Step 1:"), step1_spin)
self.direct_entries['step1'] = step1_spin
# Step 2
step2_val = self.get_parameter_from_config(
config, 'Parameters', 'centerrefinement_direct_step2', '0.5', is_optional=False
)
step2_spin = TrimmingDoubleSpinBox()
step2_spin.setRange(0.001, 10000.0)
step2_spin.setDecimals(5)
step2_spin.setSingleStep(0.1)
step2_spin.setValue(float(step2_val))
direct_layout.addRow(QLabel("Step 2:"), step2_spin)
self.direct_entries['step2'] = step2_spin
# Sigma X
sigmax_val = self.get_parameter_from_config(
config, 'Parameters', 'centerrefinement_direct_sigma_x', '1.0', is_optional=False
)
sigmax_spin = TrimmingDoubleSpinBox()
sigmax_spin.setRange(0.0, 10000.0)
sigmax_spin.setDecimals(5)
sigmax_spin.setSingleStep(0.1)
sigmax_spin.setValue(float(sigmax_val))
direct_layout.addRow(QLabel("Sigma X:"), sigmax_spin)
self.direct_entries['sigma_x'] = sigmax_spin
# Sigma Y
sigmay_val = self.get_parameter_from_config(
config, 'Parameters', 'centerrefinement_direct_sigma_y', '1.0', is_optional=False
)
sigmay_spin = TrimmingDoubleSpinBox()
sigmay_spin.setRange(0.0, 10000.0)
sigmay_spin.setDecimals(5)
sigmay_spin.setSingleStep(0.1)
sigmay_spin.setValue(float(sigmay_val))
direct_layout.addRow(QLabel("Sigma Y:"), sigmay_spin)
self.direct_entries['sigma_y'] = sigmay_spin
# Direct: number of refinement cycles
cycles_val = self.get_parameter_from_config(
config, 'Parameters', 'centerrefinement_direct_cycles', '1', is_optional=True
)
cycles_spin = QSpinBox()
cycles_spin.setRange(1, 100)
cycles_spin.setValue(int(cycles_val))
direct_layout.addRow(QLabel("Cycles:"), cycles_spin)
self.direct_entries['cycles'] = cycles_spin
# Direct: convergence threshold
conv_val = self.get_parameter_from_config(
config, 'Parameters', 'centerrefinement_direct_convergence_threshold', '0.01', is_optional=True
)
conv_spin = TrimmingDoubleSpinBox()
conv_spin.setRange(0.0, 10.0)
conv_spin.setDecimals(5)
conv_spin.setSingleStep(0.01)
conv_spin.setValue(float(conv_val))
direct_layout.addRow(QLabel("Convergence Threshold:"), conv_spin)
self.direct_entries['convergence_threshold'] = conv_spin
direct_widget = QWidget()
direct_widget.setLayout(direct_layout)
# -------------------- Settings Group as Stacked Layout --------------------th
settings_group = QGroupBox("Settings")
self.settings_stack = QStackedLayout()
self.settings_stack.addWidget(beam_widget)
self.settings_stack.addWidget(direct_widget)
settings_group.setLayout(self.settings_stack)
settings_group.setSizePolicy(QSizePolicy.Policy.Preferred, QSizePolicy.Policy.Fixed)
# Add method group and settings group (stacked) to left layout
left_layout.addWidget(method_group)
left_layout.addWidget(settings_group)
# -------------------- Optional Settings Group --------------------
self.optional_entries = {}
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)
# LOESS Span (optional)
config = configparser.ConfigParser()
config.read(self.current_ini_file_path)
loess_frac_default = self.get_parameter_from_config(
config, 'Parameters', 'centerrefinement_direct_loess_frac', "0.1", is_optional=True
)
loess_frac_spin = TrimmingDoubleSpinBox()
loess_frac_spin.setRange(0.0, 1.0)
loess_frac_spin.setDecimals(5)
loess_frac_spin.setSingleStep(0.01)
loess_frac_spin.setValue(float(loess_frac_default) if loess_frac_default is not None else 0.1)
optional_layout.addRow(QLabel("LOESS Fraction:"), loess_frac_spin)
self.optional_entries['loess_frac'] = loess_frac_spin
loess_default = self.get_parameter_from_config(
config, 'Parameters', 'centerrefinement_direct_loess_span', "", is_optional=True
)
loess_edit = QLineEdit(str(loess_default) if loess_default is not None else "")
loess_edit.setMaximumWidth(200)
# Removed setFixedHeight(24)
loess_edit.setValidator(QDoubleValidator(0.0, 1.0, 5))
optional_layout.addRow(QLabel("LOESS Span:"), loess_edit)
self.optional_entries['loess_span'] = loess_edit
optional_group.setLayout(optional_layout)
optional_group.setSizePolicy(QSizePolicy.Policy.Preferred, QSizePolicy.Policy.Fixed)
optional_group.setVisible(False)
left_layout.addWidget(optional_group)
self.optional_group = optional_group
# -------------------- Batch Size Field --------------------
self.batch_size_group = QGroupBox("Batch Size")
self.batch_size_group.setEnabled(True)
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("500")
self.batch_size_entry.setMaximumWidth(200)
# Removed setFixedHeight(24)
# Removed setSizePolicy(QSizePolicy.Policy.Fixed, QSizePolicy.Policy.Fixed)
self.batch_size_entry.setValidator(QIntValidator(1, 1000000))
self.batch_size_entry.setToolTip("Controls how many frames are processed per chunk for performance tuning only – it does not affect results.")
self.batch_size_entry.setPlaceholderText("Enter batch size (e.g., 500)")
batch_size_layout.addRow(QLabel("Batch Size:"), self.batch_size_entry)
self.batch_size_group.setLayout(batch_size_layout)
self.batch_size_group.setSizePolicy(QSizePolicy.Policy.Preferred, QSizePolicy.Policy.Fixed)
left_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)
save_group.setSizePolicy(QSizePolicy.Policy.Preferred, QSizePolicy.Policy.Fixed)
left_layout.addWidget(save_group)
# -------------------- Find Refinement Group --------------------
find_refinement_group = QGroupBox("Find Centers")
find_refinement_layout = QVBoxLayout()
find_refinement_layout.setContentsMargins(10, 5, 10, 5)
find_refinement_layout.setSpacing(20)
find_current_button = QPushButton("Find Current")
find_current_button.clicked.connect(self.on_find_refinement_current)
self.find_current_button = find_current_button
find_all_button = QPushButton("Find All")
find_all_button.clicked.connect(self.on_find_refinement_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_refinement_layout.addLayout(button_row)
# Progress label row
self.progress_label = QLabel("")
find_refinement_layout.addWidget(self.progress_label)
# Progress bar row
self.progress_bar = QProgressBar()
find_refinement_layout.addWidget(self.progress_bar)
find_refinement_group.setLayout(find_refinement_layout)
find_refinement_group.setSizePolicy(QSizePolicy.Policy.Preferred, QSizePolicy.Policy.Fixed)
left_layout.addWidget(find_refinement_group)
left_layout.addStretch()
# Splitter to show settings on left and preview in a labeled frame
splitter = QSplitter(Qt.Orientation.Horizontal)
# Settings column
settings_container = QWidget()
settings_container.setLayout(left_layout)
splitter.addWidget(settings_container)
# Preview column inside a labeled Results frame
self.preview = PreviewWindow(self.current_ini_file_path, parent=self)
self.preview.setWindowFlags(Qt.WindowType.Widget)
results_group = QGroupBox("Results")
results_layout = QVBoxLayout()
results_layout.addWidget(self.preview)
results_group.setLayout(results_layout)
splitter.addWidget(results_group)
# Set the splitter as the main layout
container_layout = QHBoxLayout()
container_layout.addWidget(splitter)
self.setLayout(container_layout)
# Initialize controls based on the current combo selection
self.on_method_change(self.method_combo.currentIndex())
[docs]
def on_method_change(self, index):
selected = self.method_combo.currentText()
# Show correct panel
if selected == "Beamstop":
self.settings_stack.setCurrentIndex(0)
self.batch_size_group.setVisible(True)
self.optional_group.setVisible(False)
elif selected == "Direct":
self.settings_stack.setCurrentIndex(1)
self.batch_size_group.setVisible(True)
self.optional_group.setVisible(True)
else:
return
# Reload values from INI
config = configparser.ConfigParser()
config.read(self.current_ini_file_path)
if selected == "Beamstop":
be = self.required_entries
be['tolerance'].setValue(int(self.get_parameter_from_config(
config, 'Parameters', 'centerrefinement_tolerance', '5', False)))
be['min_peaks'].setValue(int(self.get_parameter_from_config(
config, 'Parameters', 'centerrefinement_min_peaks', '15', False)))
be['resolution_limit'].setValue(int(self.get_parameter_from_config(
config, 'Parameters', 'centerrefinement_resolution_limit', '400', False)))
be['max_iterations'].setValue(int(self.get_parameter_from_config(
config, 'Parameters', 'centerrefinement_max_iterations', '10', False)))
be['convergence_threshold'].setValue(float(self.get_parameter_from_config(
config, 'Parameters', 'centerrefinement_convergence_threshold', '0.01', False)))
be['lowess_frac'].setValue(float(self.get_parameter_from_config(
config, 'Parameters', 'centerrefinement_lowess_frac', '0.1', True)))
be['lowess_window'].setValue(int(float(self.get_parameter_from_config(
config, 'Parameters', 'centerrefinement_lowess_window', '0', True))))
aggregation_val = config.get(
'Parameters',
'centerrefinement_deviation_aggregation',
fallback='median'
).strip().lower()
be['pair_lowess'].setChecked(aggregation_val in {"pair", "pairs", "none", "notebook"})
elif selected == "Direct":
de = self.direct_entries
de['threshold_frac'].setValue(float(self.get_parameter_from_config(
config, 'Parameters', 'centerrefinement_direct_threshold_frac', '0.5', False)))
de['box1_half'].setValue(int(self.get_parameter_from_config(
config, 'Parameters', 'centerrefinement_direct_box1_half', '10', False)))
de['step1'].setValue(float(self.get_parameter_from_config(
config, 'Parameters', 'centerrefinement_direct_step1', '1.0', False)))
de['step2'].setValue(float(self.get_parameter_from_config(
config, 'Parameters', 'centerrefinement_direct_step2', '0.5', False)))
de['sigma_x'].setValue(float(self.get_parameter_from_config(
config, 'Parameters', 'centerrefinement_direct_sigma_x', '1.0', False)))
de['sigma_y'].setValue(float(self.get_parameter_from_config(
config, 'Parameters', 'centerrefinement_direct_sigma_y', '1.0', False)))
# Reload optional LOESS span
loess = self.get_parameter_from_config(
config, 'Parameters', 'centerrefinement_direct_loess_span', '', True)
self.optional_entries['loess_span'].setText(str(loess) if loess is not None else "")
loess_frac = self.get_parameter_from_config(
config, 'Parameters', 'centerrefinement_direct_loess_frac', '0.1', True)
self.optional_entries['loess_frac'].setValue(float(loess_frac) if loess_frac is not None else 0.1)
[docs]
def get_parameter_from_config(self, config, section, parameter, default_value, is_optional=False):
"""
Retrieve parameter from config. If optional and not present, return default.
Otherwise, return integer or float value or default.
"""
try:
value = config.get(section, parameter)
if is_optional and value.strip() == "":
return default_value # Return default if optional and empty
else:
# Determine if the parameter should be float or int
if parameter.startswith('centerrefinement_direct_') \
or parameter.endswith('convergence_threshold') \
or parameter == 'centerrefinement_lowess_frac':
return float(value)
else:
return int(value)
except (configparser.NoSectionError, configparser.NoOptionError, ValueError):
# Return None for optional empty defaults
if is_optional and default_value == "":
return None
# Try parsing default as int, then float; fall back to raw default_value
try:
return int(default_value)
except (ValueError, TypeError):
try:
return float(default_value)
except (ValueError, TypeError):
return default_value
def _selected_deviation_aggregation(self):
"""Return the beamstop LOWESS input mode selected in the GUI."""
return 'pair' if self.required_entries['pair_lowess'].isChecked() else 'median'
[docs]
def on_save_current(self):
settings = {}
# Get selected method
selected_method = self.method_combo.currentText()
settings['method'] = selected_method
try:
if selected_method == "Beamstop":
write_centerrefinementsettings(
self.current_ini_file_path,
tolerance=self.required_entries['tolerance'].value(),
min_peaks=self.required_entries['min_peaks'].value(),
resolution_limit=self.required_entries['resolution_limit'].value(),
max_iterations=self.required_entries['max_iterations'].value(),
convergence_threshold=self.required_entries['convergence_threshold'].value(),
deviation_aggregation=self._selected_deviation_aggregation(),
lowess_frac=self.required_entries['lowess_frac'].value(),
lowess_window=self.required_entries['lowess_window'].value()
)
elif selected_method == "Direct":
# Gather Direct parameters using .value()
de = self.direct_entries
thresh = de['threshold_frac'].value()
box = de['box1_half'].value()
s1 = de['step1'].value()
s2 = de['step2'].value()
sx = de['sigma_x'].value()
sy = de['sigma_y'].value()
loess_text = self.optional_entries['loess_span'].text().strip()
loess = float(loess_text) if loess_text else None
loess_frac = self.optional_entries['loess_frac'].value()
# Batch size override
bs_text = self.batch_size_entry.text().strip()
bs = int(bs_text) if bs_text.isdigit() else None
# Cycles and convergence threshold
cycles = self.direct_entries['cycles'].value()
conv_thresh = self.direct_entries['convergence_threshold'].value()
write_centerrefinement_direct_settings(
self.current_ini_file_path,
threshold_frac=thresh,
box1_half=box,
step1=s1,
step2=s2,
sigma_x=sx,
sigma_y=sy,
loess_span=loess,
loess_frac=loess_frac,
batch_size=bs,
cycles=cycles,
convergence_threshold=conv_thresh
)
else:
raise ValueError(f"Unknown method: {selected_method}")
QMessageBox.information(self, "Settings Saved", "Center Refinement settings have been saved to the current file.")
except Exception as e:
QMessageBox.critical(self, "Error", f"Failed to save settings:\n{str(e)}")
[docs]
def on_save_all(self):
settings = {}
# Get selected method
selected_method = self.method_combo.currentText()
settings['method'] = selected_method
# Save settings to all INI files in the directory
try:
# Use workspace list from main window instead of directory scan
ini_files = [
file_path for file_path in self.parent.workspace_ini_paths.values()
if file_path.endswith(".ini")
]
if not ini_files:
QMessageBox.warning(self, "No INI Files", "No INI files found in the specified directory.")
return
for file_path in ini_files:
if selected_method == "Beamstop":
write_centerrefinementsettings(
file_path,
tolerance=self.required_entries['tolerance'].value(),
min_peaks=self.required_entries['min_peaks'].value(),
resolution_limit=self.required_entries['resolution_limit'].value(),
max_iterations=self.required_entries['max_iterations'].value(),
convergence_threshold=self.required_entries['convergence_threshold'].value(),
deviation_aggregation=self._selected_deviation_aggregation(),
lowess_frac=self.required_entries['lowess_frac'].value(),
lowess_window=self.required_entries['lowess_window'].value()
)
elif selected_method == "Direct":
de = self.direct_entries
thresh = de['threshold_frac'].value()
box = de['box1_half'].value()
s1 = de['step1'].value()
s2 = de['step2'].value()
sx = de['sigma_x'].value()
sy = de['sigma_y'].value()
loess_text = self.optional_entries['loess_span'].text().strip()
loess = float(loess_text) if loess_text else None
loess_frac = self.optional_entries['loess_frac'].value()
bs_text = self.batch_size_entry.text().strip()
bs = int(bs_text) if bs_text.isdigit() else None
cycles = self.direct_entries['cycles'].value()
conv_thresh = self.direct_entries['convergence_threshold'].value()
write_centerrefinement_direct_settings(
file_path,
threshold_frac=thresh,
box1_half=box,
step1=s1,
step2=s2,
sigma_x=sx,
sigma_y=sy,
loess_span=loess,
loess_frac=loess_frac,
batch_size=bs,
cycles=cycles,
convergence_threshold=conv_thresh
)
else:
raise ValueError(f"Unknown method: {selected_method}")
# Persist selected method
config = configparser.ConfigParser()
config.read(file_path)
if not config.has_section('Parameters'):
config.add_section('Parameters')
config.set('Parameters', 'centerrefinement_method', selected_method)
with open(file_path, 'w') as f:
config.write(f)
QMessageBox.information(self, "Settings Saved", "Center Refinement settings have been saved to all files.")
except Exception as e:
QMessageBox.critical(self, "Error", f"Failed to save settings to all files:\n{str(e)}")
[docs]
def on_find_refinement_current(self):
"""
Initiates center refinement on the current INI file using the selected method.
"""
self._stop_event.clear()
self.stop_button.setEnabled(True)
self.find_current_button.setEnabled(False)
self.find_all_button.setEnabled(False)
selected_method = self.method_combo.currentText()
ini_file = self.current_ini_file_path
# Always get batch size from entry
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)
# Write settings to INI file before running refinement
try:
if selected_method == "Beamstop":
write_centerrefinementsettings(
ini_file,
tolerance=self.required_entries['tolerance'].value(),
min_peaks=self.required_entries['min_peaks'].value(),
resolution_limit=self.required_entries['resolution_limit'].value(),
max_iterations=self.required_entries['max_iterations'].value(),
convergence_threshold=self.required_entries['convergence_threshold'].value(),
deviation_aggregation=self._selected_deviation_aggregation(),
lowess_frac=self.required_entries['lowess_frac'].value(),
lowess_window=self.required_entries['lowess_window'].value()
)
elif selected_method == "Direct":
# use same gathering logic as in on_save_current for Direct, but use .value()
de = self.direct_entries
thresh = de['threshold_frac'].value()
box = de['box1_half'].value()
s1 = de['step1'].value()
s2 = de['step2'].value()
sx = de['sigma_x'].value()
sy = de['sigma_y'].value()
loess_text = self.optional_entries['loess_span'].text().strip()
loess = float(loess_text) if loess_text else None
loess_frac = self.optional_entries['loess_frac'].value()
cycles = self.direct_entries['cycles'].value()
conv_thresh = self.direct_entries['convergence_threshold'].value()
write_centerrefinement_direct_settings(
ini_file,
threshold_frac=thresh,
box1_half=box,
step1=s1,
step2=s2,
sigma_x=sx,
sigma_y=sy,
loess_span=loess,
loess_frac=loess_frac,
batch_size=batch_size,
cycles=cycles,
convergence_threshold=conv_thresh
)
else:
raise ValueError(f"Unknown method: {selected_method}")
except Exception as e:
QMessageBox.critical(self, "Error", f"Failed to write settings to INI file:\n{str(e)}")
return
# Initialize plot viewer monitoring
self.monitor_start_time = time.time()
self.plots_loaded = set()
# self.plot_tree.clear()
# self.plot_display_label.setText("No plot selected.")
# self.monitor_timer.start()
# 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_refinement_finished)
worker.error.connect(self.on_refinement_error)
worker.cancelled.connect(self.on_refinement_cancelled)
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)
# Start the thread
thread.start()
# Keep references to avoid garbage collection
self.threads.append(thread)
self.workers.append(worker)
# Initialize iteration progress monitoring for this file
fname = os.path.basename(self.current_ini_file_path)
# Show filename (1/1)
self.progress_label.setText(f"{fname} (1/1)")
# Make progress bar indeterminate until first update
self.progress_bar.setRange(0, 0)
# Timer to update progress by counting npz files
if hasattr(self, 'progress_monitor_timer'):
self.progress_monitor_timer.stop()
self.progress_monitor_timer = QTimer(self)
self.progress_monitor_timer.setInterval(1000)
self.progress_monitor_timer.timeout.connect(self.update_refine_progress)
self.progress_monitor_timer.start()
[docs]
def on_find_refinement_all(self):
"""
Initiates center refinement on all INI files in the directory using the selected method.
"""
self._stop_event.clear()
self._stop_requested = False
self.stop_button.setEnabled(True)
self.find_current_button.setEnabled(False)
self.find_all_button.setEnabled(False)
selected_method = self.method_combo.currentText()
ini_directory = self.ini_directory
# Always get batch size from entry
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)
# Use workspace list from main window instead of directory scan
ini_files = [
file_path for file_path in self.parent.workspace_ini_paths.values()
if file_path.endswith(".ini")
]
total_files = len(ini_files)
if not ini_files:
QMessageBox.warning(self, "No INI Files", "No INI files found in the specified directory.")
return
# Initialize monitoring for all refinements
self.monitor_start_time = time.time()
self.plots_loaded = set()
# Single timer for all files
if hasattr(self, 'progress_monitor_timer'):
self.progress_monitor_timer.stop()
else:
self.progress_monitor_timer = QTimer(self)
self.progress_monitor_timer.setInterval(1000)
self.progress_monitor_timer.timeout.connect(self.update_refine_progress)
self.progress_monitor_timer.start()
# Write settings to all INI files before running refinement
try:
for ini_file in ini_files:
if selected_method == "Beamstop":
write_centerrefinementsettings(
ini_file,
tolerance=self.required_entries['tolerance'].value(),
min_peaks=self.required_entries['min_peaks'].value(),
resolution_limit=self.required_entries['resolution_limit'].value(),
max_iterations=self.required_entries['max_iterations'].value(),
convergence_threshold=self.required_entries['convergence_threshold'].value(),
deviation_aggregation=self._selected_deviation_aggregation(),
lowess_frac=self.required_entries['lowess_frac'].value(),
lowess_window=self.required_entries['lowess_window'].value()
)
elif selected_method == "Direct":
de = self.direct_entries
thresh = de['threshold_frac'].value()
box = de['box1_half'].value()
s1 = de['step1'].value()
s2 = de['step2'].value()
sx = de['sigma_x'].value()
sy = de['sigma_y'].value()
loess_text = self.optional_entries['loess_span'].text().strip()
loess = float(loess_text) if loess_text else None
loess_frac = self.optional_entries['loess_frac'].value()
cycles = self.direct_entries['cycles'].value()
conv_thresh = self.direct_entries['convergence_threshold'].value()
write_centerrefinement_direct_settings(
ini_file,
threshold_frac=thresh,
box1_half=box,
step1=s1,
step2=s2,
sigma_x=sx,
sigma_y=sy,
loess_span=loess,
loess_frac=loess_frac,
batch_size=batch_size,
cycles=cycles,
convergence_threshold=conv_thresh
)
else:
raise ValueError(f"Unknown method: {selected_method}")
except Exception as e:
QMessageBox.critical(self, "Error", f"Failed to write settings to INI files:\n{str(e)}")
return
# Prepare sequential queue
self.all_ini_files = ini_files
self.current_refine_index = 0
self.batch_size_all = batch_size
# Kick off the sequential refinements
self._start_next_refinement()
[docs]
def on_stop_clicked(self):
"""
Request cancellation of any running refinement job.
"""
self._stop_event.set()
self._stop_requested = True
self.stop_button.setEnabled(False)
# Indicate we're finishing already submitted iterations/batches
if self._last_progress:
self.progress_label.setText(f"Stopping after submitted work... ({self._last_progress})")
else:
self.progress_label.setText("Stopping after submitted work...")
# Propagate cancel request to all known workers
for worker in self.workers:
if hasattr(worker, "request_cancel"):
worker.request_cancel()
# Do not immediately stop timers; let cancellation handler tidy up
def _reset_run_controls(self):
"""Re-enable run buttons and clear stop state."""
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
[docs]
def on_refinement_finished(self, message):
"""
Slot to handle successful refinement.
"""
if hasattr(self, 'progress_monitor_timer'):
self.progress_monitor_timer.stop()
# Adjust progress bar max to actual iterations
try:
_, base_output, _, _, _, _ = config_to_paths(self.current_ini_file_path)
runs = [d for d in os.listdir(base_output) if d.startswith('refinecenters_')]
if runs:
latest = max(runs, key=lambda d: os.path.getmtime(os.path.join(base_output, d)))
folder = os.path.join(base_output, latest)
count = sum(1 for fn in os.listdir(folder) if re.match(r'^lowess_and_centers_it\d{3}\.npz$', fn))
self.progress_bar.setMaximum(count)
self.progress_bar.setValue(count)
except Exception:
pass
# QMessageBox.information(self, "Center Refinement", message)
self.processing_finished.emit()
self.monitor_timer.stop()
self._reset_run_controls()
[docs]
def on_refinement_error(self, error_message):
"""
Slot to handle refinement errors.
"""
if hasattr(self, 'progress_monitor_timer'):
self.progress_monitor_timer.stop()
QMessageBox.critical(self, "Center Refinement Error", error_message)
self.processing_finished.emit()
self.monitor_timer.stop()
self._reset_run_controls()
[docs]
def on_refinement_cancelled(self, message):
"""
Slot to handle user-requested cancellation.
"""
if hasattr(self, 'progress_monitor_timer'):
self.progress_monitor_timer.stop()
self.progress_label.setText("Refinement cancelled.")
QMessageBox.information(self, "Center Refinement", message)
self.processing_finished.emit()
self.monitor_timer.stop()
self._reset_run_controls()
[docs]
def update_refine_progress(self):
"""
Periodically count completed iterations and update the progress bar,
ignoring runs predating the start time.
"""
# Switch from indeterminate to determinate once any progress is detected
count = None
try:
# Determine base output directory from config_to_paths
_, base_output, _, _, _, _ = config_to_paths(self.current_ini_file_path)
# Find all refinecenters_ runs created since monitoring started
runs = [d for d in os.listdir(base_output) if d.startswith("refinecenters_")]
valid = []
for run in runs:
path = os.path.join(base_output, run)
try:
if os.path.getmtime(path) >= self.monitor_start_time:
valid.append((run, path))
except OSError:
continue
if not valid:
return
# Pick the latest run
latest_path = max(valid, key=lambda x: os.path.getmtime(x[1]))[1]
# Count NPZ iteration files
count = sum(
1 for fn in os.listdir(latest_path)
if re.match(r'^lowess_and_centers_it\d{3}\.npz$', fn)
)
except Exception:
return
if count is not None:
# If still indeterminate and count>0, initialize determinate range
if self.progress_bar.minimum() == 0 and self.progress_bar.maximum() == 0 and count > 0:
max_iter = int(self.required_entries['max_iterations'].text())
self.progress_bar.setRange(0, max_iter)
# Now update the value
self.progress_bar.setValue(count)
self._last_progress = count
if self._stop_requested:
self.progress_label.setText(f"Stopping after submitted work... ({count}/{self.progress_bar.maximum() or '?'})")
if count >= self.progress_bar.maximum():
self.progress_monitor_timer.stop()
return
[docs]
def update_plot_viewer(self):
"""
Scans the plots folder for new plots created after the monitoring started.
Adds them to the plot viewer.
"""
if not self.monitor_start_time:
return
# 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
graphicsfolder_path = os.path.join(os.path.dirname(self.current_ini_file_path), outputfolder, 'plots')
if not os.path.exists(graphicsfolder_path):
return
# Regex to match the plot naming scheme
plot_regex = re.compile(r'^(lowess|lowess_and_centers)_it(\d+)\.png$')
for filename in os.listdir(graphicsfolder_path):
match = plot_regex.match(filename)
if match:
plot_type = 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:
# Add to the plot viewer
self.add_plot_to_viewer(iter_num, plot_type, file_path)
self.plots_loaded.add(file_path)
[docs]
def add_plot_to_viewer(self, iter_num, plot_type, file_path):
"""
Adds a plot to the QTreeWidget under the appropriate iteration and plot type.
"""
# Find or create the iteration item
iteration_item = None
for i in range(self.plot_tree.topLevelItemCount()):
item = self.plot_tree.topLevelItem(i)
if item.text(0) == f"Iteration {iter_num}":
iteration_item = item
break
if not iteration_item:
# Create a new iteration item
iteration_item = QTreeWidgetItem([f"Iteration {iter_num}", "", ""])
self.plot_tree.addTopLevelItem(iteration_item)
# Add the plot item under the iteration
plot_item = QTreeWidgetItem(["", plot_type.replace('_', ' ').capitalize(), os.path.basename(file_path)])
plot_item.setData(2, Qt.ItemDataRole.UserRole, file_path)
iteration_item.addChild(plot_item)
iteration_item.setExpanded(True)
[docs]
def display_selected_plot(self, item, column):
"""
Displays the selected plot in the QLabel.
"""
# Only proceed if it's a plot item (has a file path)
if item.childCount() == 0 and item.parent() is not None:
file_path = item.data(2, Qt.ItemDataRole.UserRole)
if file_path and os.path.exists(file_path):
self.current_plot_file_path = file_path # Store the file path
pixmap = QPixmap(file_path)
if not pixmap.isNull():
scaled_pixmap = pixmap.scaled(
self.plot_display_label.size(),
Qt.AspectRatioMode.KeepAspectRatio,
Qt.TransformationMode.SmoothTransformation
)
self.plot_display_label.setPixmap(scaled_pixmap)
else:
self.plot_display_label.setText("Unable to load image.")
else:
self.plot_display_label.setText("Plot file does not exist.")
else:
self.plot_display_label.setText("No plot selected.")
[docs]
def open_plot_image(self, item, column):
"""
Opens the selected plot image in the default image viewer.
"""
# Only proceed if it's a plot item (no children and has a parent)
if item.childCount() == 0 and item.parent() is not None:
file_path = item.data(2, Qt.ItemDataRole.UserRole)
if file_path and os.path.exists(file_path):
try:
if sys.platform.startswith('darwin'):
subprocess.call(['open', file_path])
elif sys.platform.startswith('win'):
os.startfile(file_path)
elif sys.platform.startswith('linux'):
subprocess.call(['xdg-open', file_path])
else:
QMessageBox.warning(self, "Unsupported OS", "Opening images is not supported on this OS.")
except Exception as e:
QMessageBox.critical(self, "Error", f"Failed to open image:\n{str(e)}")
else:
QMessageBox.warning(self, "File Not Found", "The plot image file does not exist.")
[docs]
def on_plot_display_label_clicked(self):
"""
Displays the current plot image in a larger dialog when the label is clicked.
"""
if hasattr(self, 'current_plot_file_path') and self.current_plot_file_path:
if os.path.exists(self.current_plot_file_path):
dialog = QDialog(self)
dialog.setWindowTitle("Image Viewer")
layout = QVBoxLayout()
label = QLabel()
pixmap = QPixmap(self.current_plot_file_path)
if not pixmap.isNull():
# Get screen size and scale the image appropriately
screen = QGuiApplication.primaryScreen()
screen_size = screen.size()
max_width = int(screen_size.width() * 0.8)
max_height = int(screen_size.height() * 0.8)
scaled_pixmap = pixmap.scaled(
max_width, max_height,
Qt.AspectRatioMode.KeepAspectRatio,
Qt.TransformationMode.SmoothTransformation
)
label.setPixmap(scaled_pixmap)
layout.addWidget(label)
dialog.setLayout(layout)
dialog.exec()
else:
QMessageBox.warning(self, "Image Error", "Unable to load image.")
else:
QMessageBox.warning(self, "File Not Found", "The plot image file does not exist.")
else:
QMessageBox.information(self, "No Image", "No image is currently displayed.")
[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()