"""UI for configuring peakfinder settings and launching batch peakfinding runs."""
import numpy as np
import matplotlib.pyplot as plt
from matplotlib.backends.backend_qt5agg import FigureCanvasQTAgg as FigureCanvas
from PyQt6.QtWidgets import (
QDialog,
QLabel,
QLineEdit,
QVBoxLayout,
QHBoxLayout,
QMessageBox,
QSizePolicy,
QCheckBox,
QProgressBar,
QSpinBox,
QComboBox,
QPushButton,
QGroupBox,
QFormLayout,
)
from PyQt6.QtCore import Qt, pyqtSignal, QTimer
from PyQt6.QtCore import QLocale
from PyQt6.QtGui import QIntValidator, QDoubleValidator
import configparser
import os
import threading
import h5py
from coseda.initialize import write_peakfindersettings
from coseda.nexus.paths import get_mask_dataset
from coseda.peakfinding.findpeaks_dask import findpeaks_dask_batch
from coseda.peakfinding.peakfinder9 import findpeaks9_dask_batch
# FocusableLineEdit class
[docs]
class FocusableLineEdit(QLineEdit):
"""QLineEdit subclass that emits signals on focus in/out."""
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()
[docs]
class PeakFinderSettingsWindow(QDialog):
"""Settings dialog for configuring peakfinder_8 / peakfinder9 and running them."""
# Define signals
settings_applied = pyqtSignal(dict)
settings_closed = pyqtSignal()
coordinate_selection_requested = pyqtSignal()
peakfinding_finished = pyqtSignal()
peakfinding_error = pyqtSignal(str)
progress_updated = pyqtSignal(int, int)
file_progress_updated = pyqtSignal(str, int, int)
# -----------------------------------------------------------------------
# Algorithm parameter definitions
# -----------------------------------------------------------------------
# (label, default, type, tooltip) type in {'int', 'float'}
_PF8_SETTINGS = {
"threshold": (
"Threshold", "15", "int",
"Minimum absolute pixel intensity (ADU) for a pixel to be considered "
"as part of a peak. Pixels below this value are ignored entirely.",
),
"min_snr": (
"Min S/N", "3", "float",
"Minimum signal-to-noise ratio. The peak must exceed the local "
"background mean by at least this many standard deviations.",
),
"min_pix_count": (
"Min Pix", "4", "int",
"Minimum number of connected pixels above threshold that together "
"form a valid peak. Isolated single pixels are rejected.",
),
"max_pix_count": (
"Max Pix", "500", "int",
"Maximum number of connected pixels allowed per peak. Regions larger "
"than this (e.g. ice rings, hot regions) are rejected.",
),
"local_bg_radius": (
"BG Radius", "9", "int",
"Radius of the annular region (in pixels) used to estimate the local "
"background mean and standard deviation around each candidate peak.",
),
"min_res": (
"Min Res", "50", "int",
"Inner radius of the search annulus in pixels from the beam centre. "
"Pixels closer to the centre than this are masked and ignored.",
),
"max_res": (
"Max Res", "500", "int",
"Outer radius of the search annulus in pixels from the beam centre. "
"Pixels further from the centre than this are masked and ignored.",
),
}
_PF9_SETTINGS = {
"window_radius": (
"BG Radius", "2", "int_list",
"Half-side of the square background ring (r) — analogous to BG Radius "
"in pf8. Background mean and std are computed from the perimeter of the "
"(2r+1)×(2r+1) square centred on each candidate pixel.\n"
"Enter a single integer (e.g. 3) or a comma-separated list "
"(e.g. 2,3,4) to run multi-scale detection and merge results — "
"can be useful when the reflection diameter varies strongly across a dataset.",
),
"min_sigma": (
"Min BG σ", "5.0", "float",
"Floor applied to the background standard deviation before any SNR "
"comparison. Prevents false peaks in artificially quiet regions.",
),
"min_peak_over_neighbors": (
"Local Contrast", "10.0", "float",
"The candidate pixel must exceed every pixel on the background ring "
"by at least this many ADU. Unlike the absolute Threshold in pf8, this "
"is a relative criterion — the peak must stand out locally above its "
"immediate neighbourhood.",
),
"min_snr_max_pixel": (
"Min S/N", "5.0", "float",
"The candidate pixel must be at least this many sigma above the ring "
"mean — analogous to Min S/N in pf8. Applied to the single brightest "
"pixel of the peak.",
),
"min_snr_peak_pixels": (
"Pix S/N", "4.0", "float",
"Pixels within the window that exceed the ring mean by at least this "
"many sigma are counted as part of the peak and contribute to the "
"integrated peak mass.",
),
"min_snr_whole_peak": (
"Total S/N", "6.0", "float",
"Integrated peak mass (sum of qualifying pixel excesses above ring mean) "
"divided by the background std must exceed this threshold.",
),
"min_res": (
"Min Res", "50", "int",
"Inner radius of the search annulus in pixels from the beam centre. "
"Pixels closer to the centre than this are masked and ignored.",
),
"max_res": (
"Max Res", "500", "int",
"Outer radius of the search annulus in pixels from the beam centre. "
"Pixels further from the centre than this are masked and ignored.",
),
}
def __init__(self, parent, ini_directory, ini_file_path):
"""Initialize the peakfinder settings UI and wire signals."""
super().__init__(parent)
self.setWindowFlags(Qt.WindowType.Window)
self.parent = parent
self.ini_directory = ini_directory
self.current_ini_file_path = ini_file_path
self.setWindowTitle("Peak Finder Settings")
# Histogram controls (define before init_ui so update_run_list can use them)
self.hist_run_combo = QComboBox()
self.hist_bins_spin = QSpinBox()
self.hist_bins_spin.setRange(1, 500)
self.hist_bins_spin.setValue(10)
self.hist_run_combo.currentIndexChanged.connect(self.on_plot_histogram)
self.hist_bins_spin.valueChanged.connect(self.on_plot_histogram)
# Debounce timer for live preview
self.apply_timer = QTimer(self)
self.apply_timer.setInterval(500)
self.apply_timer.setSingleShot(True)
self.apply_timer.timeout.connect(self.apply_settings)
# Must exist before init_ui(), because init_ui() triggers button-state checks.
self._active_peakfinding_runs = 0
self.init_ui()
self.check_and_enable_findpeaks_button()
self.peakfinding_finished.connect(self.on_peakfinding_finished)
self.peakfinding_error.connect(self.on_peakfinding_error)
self.progress_updated.connect(self.on_progress_updated)
self.file_progress_updated.connect(self.on_file_progress_updated)
[docs]
def closeEvent(self, event):
"""Emit closed signal so the main window can re-enable buttons/menus."""
self.settings_closed.emit()
super().closeEvent(event)
# -----------------------------------------------------------------------
# UI construction
# -----------------------------------------------------------------------
[docs]
def init_ui(self):
"""Build the settings form, preview widgets, and buttons."""
main_layout = QHBoxLayout()
left_layout = QVBoxLayout()
right_layout = QVBoxLayout()
main_layout.setContentsMargins(10, 10, 10, 10)
main_layout.setSpacing(10)
config = configparser.ConfigParser()
config.read(self.current_ini_file_path)
# --- Algorithm selector -------------------------------------------
algo_group = QGroupBox("Algorithm")
algo_group_layout = QHBoxLayout()
algo_group_layout.setContentsMargins(5, 5, 5, 5)
self.algo_combo = QComboBox()
self.algo_combo.addItems(["peakfinder8", "peakfinder9"])
self.algo_combo.setToolTip(
"peakfinder8 — uses a circular background annulus and connected-pixel "
"counting (Diffractem / CrystFEL classic).\n"
"peakfinder9 — uses a square background ring with a strict local-maximum "
"check and integrated peak-mass SNR (calNG / CrystFEL pf9)."
)
# Restore persisted choice
saved_algo = self._get_str_from_config(
config, 'Parameters', 'peakfinding_algorithm', 'peakfinder8'
)
if saved_algo in ("peakfinder8", "peakfinder9"):
self.algo_combo.setCurrentText(saved_algo)
algo_group_layout.addWidget(self.algo_combo)
algo_group.setLayout(algo_group_layout)
left_layout.addWidget(algo_group)
# --- PF8 settings group ------------------------------------------
self.pf8_group = QGroupBox("Settings")
self.pf8_entries = self._build_settings_group(
self.pf8_group,
self._PF8_SETTINGS,
prefix='peakfinding_',
config=config,
)
left_layout.addWidget(self.pf8_group)
# --- PF9 settings group ------------------------------------------
self.pf9_group = QGroupBox("Settings")
self.pf9_entries = self._build_settings_group(
self.pf9_group,
self._PF9_SETTINGS,
prefix='pf9_',
config=config,
)
left_layout.addWidget(self.pf9_group)
# Show only the active algorithm's group
self._update_settings_visibility()
self.algo_combo.currentIndexChanged.connect(self._on_algo_changed)
# --- Optional settings (shared: x0, y0) --------------------------
optional_group = QGroupBox("Optional Settings")
optional_group_layout = QVBoxLayout()
optional_group_layout.setContentsMargins(5, 5, 5, 5)
optional_group_layout.setSpacing(5)
optional_layout = QFormLayout()
optional_layout.setLabelAlignment(Qt.AlignmentFlag.AlignRight)
optional_layout.setFormAlignment(Qt.AlignmentFlag.AlignLeft | Qt.AlignmentFlag.AlignTop)
optional_layout.setHorizontalSpacing(5)
optional_layout.setVerticalSpacing(3)
self.optional_entries = {}
self.optional_settings_definitions = {
"x0": ("x0", "",
"Beam centre x-coordinate (column index). Used to build the "
"radial Min/Max Res mask. Click on the image while this field "
"is focused to pick the position interactively."),
"y0": ("y0", "",
"Beam centre y-coordinate (row index). Used to build the "
"radial Min/Max Res mask. Click on the image while this field "
"is focused to pick the position interactively."),
}
for key, (label_text, default, tooltip) in self.optional_settings_definitions.items():
# pf9 saves under pf9_x0/pf9_y0; pf8 uses peakfinding_x0/peakfinding_y0.
if saved_algo == 'peakfinder9':
value = self._get_optional_int_from_config(
config, 'Parameters', f'pf9_{key}', default
)
if not value:
value = self._get_optional_int_from_config(
config, 'Parameters', f'peakfinding_{key}', default
)
else:
value = self._get_optional_int_from_config(
config, 'Parameters', f'peakfinding_{key}', default
)
line_edit = FocusableLineEdit(str(value))
line_edit.focus_in.connect(self.on_field_focused)
line_edit.setMaximumWidth(120)
line_edit.setFixedHeight(24)
line_edit.setSizePolicy(QSizePolicy.Policy.Fixed, QSizePolicy.Policy.Fixed)
line_edit.setValidator(QIntValidator())
line_edit.setToolTip(tooltip)
self.optional_entries[key] = line_edit
lbl = QLabel(label_text + ":")
lbl.setToolTip(tooltip)
optional_layout.addRow(lbl, line_edit)
line_edit.textChanged.connect(self.on_setting_changed)
optional_group_layout.addLayout(optional_layout)
optional_group.setLayout(optional_group_layout)
left_layout.addWidget(optional_group)
# Set x0/y0 to image centre if still blank
if not self.optional_entries['x0'].text() or not self.optional_entries['y0'].text():
hdf5_path = self.parent.hdf5_path
if hdf5_path and os.path.exists(hdf5_path):
try:
with h5py.File(hdf5_path, 'r') as f:
shape = f['entry/data/images'].shape[1:]
if not self.optional_entries['x0'].text():
self.optional_entries['x0'].setText(str(shape[1] // 2))
if not self.optional_entries['y0'].text():
self.optional_entries['y0'].setText(str(shape[0] // 2))
except Exception as e:
QMessageBox.warning(self, "File Error",
f"Cannot determine image center:\n{str(e)}")
else:
QMessageBox.warning(self, "File Error",
"HDF5 file not found. Cannot determine image center.")
# --- Preview group -----------------------------------------------
preview_group = QGroupBox("Preview")
preview_group_layout = QVBoxLayout()
preview_group_layout.setContentsMargins(5, 5, 5, 5)
preview_group_layout.setSpacing(5)
self.live_preview_checkbox = QCheckBox("Enable Live Preview")
self.live_preview_checkbox.setToolTip(
"When checked, peaks are re-detected and drawn on the current frame "
"every time a setting changes (500 ms debounce). Useful for tuning "
"parameters interactively, but adds a small per-frame delay."
)
live_preview_enabled = self._get_bool_from_config(
config, 'Parameters', 'peakfinding_live_preview', False
)
self.live_preview_checkbox.setChecked(live_preview_enabled)
preview_group_layout.addWidget(self.live_preview_checkbox)
self.live_preview_checkbox.stateChanged.connect(self.on_setting_changed)
preview_group.setLayout(preview_group_layout)
left_layout.addWidget(preview_group)
# --- Save settings group -----------------------------------------
save_group = QGroupBox("Save Settings")
save_group_layout = QVBoxLayout()
save_group_layout.setContentsMargins(5, 5, 5, 5)
save_group_layout.setSpacing(5)
save_buttons_layout = QHBoxLayout()
save_buttons_layout.setContentsMargins(0, 0, 0, 0)
save_buttons_layout.setSpacing(10)
save_current_button = QPushButton("Current")
save_current_button.clicked.connect(self.on_save_current)
save_buttons_layout.addWidget(save_current_button)
save_all_button = QPushButton("All Files")
save_all_button.clicked.connect(self.on_save_all)
save_buttons_layout.addWidget(save_all_button)
save_group_layout.addLayout(save_buttons_layout)
save_group.setLayout(save_group_layout)
left_layout.addWidget(save_group)
# --- Find Peaks group --------------------------------------------
find_peaks_group = QGroupBox("Find Peaks")
find_peaks_group_layout = QVBoxLayout()
find_peaks_group_layout.setContentsMargins(5, 5, 5, 5)
find_peaks_group_layout.setSpacing(5)
find_peaks_buttons_layout = QHBoxLayout()
find_peaks_buttons_layout.setContentsMargins(0, 0, 0, 0)
find_peaks_buttons_layout.setSpacing(10)
self.find_peaks_current_button = QPushButton("Current")
self.find_peaks_current_button.clicked.connect(self.on_find_peaks_current)
self.find_peaks_current_button.setEnabled(False)
find_peaks_buttons_layout.addWidget(self.find_peaks_current_button)
self.find_peaks_all_button = QPushButton("All Files")
self.find_peaks_all_button.clicked.connect(self.on_find_peaks_all)
self.find_peaks_all_button.setEnabled(False)
find_peaks_buttons_layout.addWidget(self.find_peaks_all_button)
find_peaks_group_layout.addLayout(find_peaks_buttons_layout)
find_peaks_group.setLayout(find_peaks_group_layout)
left_layout.addWidget(find_peaks_group)
# File progress label + bar
self.file_progress_label = QLabel("")
left_layout.addWidget(self.file_progress_label)
self.progress_bar = QProgressBar()
self.progress_bar.setRange(0, 100)
left_layout.addWidget(self.progress_bar)
# --- Statistics / histogram (right panel) ------------------------
output_group = QGroupBox("Peak Statistics")
output_group_layout = QVBoxLayout()
control_layout = QHBoxLayout()
control_layout.addWidget(QLabel("Run:"))
control_layout.addWidget(self.hist_run_combo)
control_layout.addWidget(QLabel("Bins:"))
control_layout.addWidget(self.hist_bins_spin)
output_group_layout.addLayout(control_layout)
self.canvas = FigureCanvas(plt.figure())
output_group_layout.addWidget(self.canvas)
output_group.setLayout(output_group_layout)
right_layout.addWidget(output_group)
self.update_run_list()
self.scan_timer = QTimer(self)
self.scan_timer.setInterval(2000)
self.scan_timer.timeout.connect(self.update_run_list)
self.scan_timer.start()
main_layout.addLayout(left_layout)
main_layout.addLayout(right_layout)
self.setLayout(main_layout)
self.check_and_enable_findpeaks_button()
def _build_settings_group(self, group_box, definitions, prefix, config):
"""Populate a QGroupBox with a QFormLayout of parameter fields.
Returns a dict mapping key → QLineEdit.
"""
layout = QVBoxLayout()
layout.setContentsMargins(5, 5, 5, 5)
layout.setSpacing(5)
form = QFormLayout()
form.setLabelAlignment(Qt.AlignmentFlag.AlignRight)
form.setFormAlignment(Qt.AlignmentFlag.AlignLeft | Qt.AlignmentFlag.AlignTop)
form.setHorizontalSpacing(5)
form.setVerticalSpacing(3)
entries = {}
for key, (label, default, dtype, tooltip) in definitions.items():
ini_key = f'{prefix}{key}'
raw = default
try:
raw = config.get('Parameters', ini_key)
except (configparser.NoSectionError, configparser.NoOptionError):
pass
line_edit = QLineEdit(str(raw))
line_edit.setMaximumWidth(120)
line_edit.setFixedHeight(24)
line_edit.setSizePolicy(QSizePolicy.Policy.Fixed, QSizePolicy.Policy.Fixed)
line_edit.setToolTip(tooltip)
if dtype == 'float':
v = QDoubleValidator(0.0, 1e9, 6, self)
v.setNotation(QDoubleValidator.Notation.StandardNotation)
v.setLocale(QLocale.c())
line_edit.setValidator(v)
elif dtype == 'int':
line_edit.setValidator(QIntValidator())
# int_list: no validator — accepts "2", "2,3,4", etc.
line_edit.textChanged.connect(self.on_setting_changed)
entries[key] = line_edit
lbl = QLabel(label + ":")
lbl.setToolTip(tooltip)
form.addRow(lbl, line_edit)
layout.addLayout(form)
group_box.setLayout(layout)
return entries
# -----------------------------------------------------------------------
# Algorithm switching
# -----------------------------------------------------------------------
def _on_algo_changed(self):
self._update_settings_visibility()
self.check_and_enable_findpeaks_button()
self.on_setting_changed()
def _update_settings_visibility(self):
is_pf9 = self.algo_combo.currentText() == "peakfinder9"
self.pf8_group.setVisible(not is_pf9)
self.pf9_group.setVisible(is_pf9)
def _active_entries(self):
"""Return the entry dict for the currently selected algorithm."""
if self.algo_combo.currentText() == "peakfinder9":
return self.pf9_entries
return self.pf8_entries
def _active_definitions(self):
"""Return the definitions dict for the currently selected algorithm."""
if self.algo_combo.currentText() == "peakfinder9":
return self._PF9_SETTINGS
return self._PF8_SETTINGS
# -----------------------------------------------------------------------
# Config helpers
# -----------------------------------------------------------------------
def _get_str_from_config(self, config, section, key, default):
try:
return config.get(section, key)
except (configparser.NoSectionError, configparser.NoOptionError):
return default
def _get_bool_from_config(self, config, section, key, default):
try:
return config.getboolean(section, key)
except (configparser.NoSectionError, configparser.NoOptionError, ValueError):
return default
def _get_optional_int_from_config(self, config, section, key, default):
try:
val = config.get(section, key).strip()
return int(val) if val else default
except (configparser.NoSectionError, configparser.NoOptionError, ValueError):
return default
# kept for backward-compat with any external callers
[docs]
def get_parameter_from_config(self, config, section, parameter, default_value, is_optional=False):
try:
value = config.get(section, parameter)
if parameter in ('peakfinding_min_snr',):
return value
elif is_optional:
if parameter == 'peakfinding_live_preview':
return config.getboolean(section, parameter)
elif value.strip() == "":
return ""
else:
return int(value)
else:
return int(value)
except (configparser.NoSectionError, configparser.NoOptionError, ValueError):
return default_value
# -----------------------------------------------------------------------
# Settings gathering / validation
# -----------------------------------------------------------------------
def _gather_active_settings(self):
"""Validate and return a dict of processed settings for the active algorithm.
Returns None and shows a warning dialog on validation failure.
The returned dict always contains 'algorithm', 'x0', 'y0', 'live_preview'
plus all algorithm-specific keys.
"""
definitions = self._active_definitions()
entries = self._active_entries()
processed = {}
for key, (_, _, dtype, *_) in definitions.items():
raw = entries[key].text().strip()
if dtype == 'int_list':
parts = [p.strip() for p in raw.split(',') if p.strip()]
try:
vals = [int(p) for p in parts]
except ValueError:
QMessageBox.warning(self, "Invalid Input",
f"Please enter one or more comma-separated "
f"integers for '{key}' (e.g. 2 or 2,3,4).")
return None
if not vals:
QMessageBox.warning(self, "Invalid Input",
f"'{key}' must not be empty.")
return None
processed[key] = vals[0] if len(vals) == 1 else vals
elif dtype == 'float':
try:
processed[key] = float(raw.replace(',', '.'))
except ValueError:
QMessageBox.warning(self, "Invalid Input",
f"Please enter a number for '{key}'.")
return None
else: # 'int'
try:
processed[key] = int(raw)
except ValueError:
QMessageBox.warning(self, "Invalid Input",
f"Please enter an integer for '{key}'.")
return None
# Optional shared fields
for key in self.optional_settings_definitions:
raw = self.optional_entries[key].text().strip()
if not raw:
continue
try:
processed[key] = int(raw)
except ValueError:
QMessageBox.warning(self, "Invalid Input",
f"Please enter a valid integer for '{key}'.")
return None
processed['algorithm'] = self.algo_combo.currentText()
processed['live_preview'] = self.live_preview_checkbox.isChecked()
return processed
# -----------------------------------------------------------------------
# INI persistence helpers
# -----------------------------------------------------------------------
def _write_pf9_settings(self, file_path, settings):
"""Write pf9_* keys (plus shared keys) into an INI file."""
config = configparser.ConfigParser()
config.read(file_path)
if not config.has_section('Parameters'):
config.add_section('Parameters')
pf9_map = {
'window_radius': 'pf9_window_radius',
'min_sigma': 'pf9_min_sigma',
'min_peak_over_neighbors': 'pf9_min_peak_over_neighbors',
'min_snr_max_pixel': 'pf9_min_snr_max_pixel',
'min_snr_peak_pixels': 'pf9_min_snr_peak_pixels',
'min_snr_whole_peak': 'pf9_min_snr_whole_peak',
'min_res': 'pf9_min_res',
'max_res': 'pf9_max_res',
}
for key, ini_key in pf9_map.items():
if key in settings:
val = settings[key]
config.set('Parameters', ini_key,
','.join(str(v) for v in val) if isinstance(val, list)
else str(val))
for opt in ('x0', 'y0'):
if opt in settings:
config.set('Parameters', f'pf9_{opt}', str(settings[opt]))
config.set('Parameters', 'peakfinding_algorithm', 'peakfinder9')
with open(file_path, 'w') as f:
config.write(f)
def _write_algo_choice(self, file_path, algo):
"""Persist the algorithm choice to an INI without touching other keys."""
config = configparser.ConfigParser()
config.read(file_path)
if not config.has_section('Parameters'):
config.add_section('Parameters')
config.set('Parameters', 'peakfinding_algorithm', algo)
with open(file_path, 'w') as f:
config.write(f)
# -----------------------------------------------------------------------
# Slots: setting changes / live preview
# -----------------------------------------------------------------------
[docs]
def on_setting_changed(self):
self.apply_timer.start()
[docs]
def apply_settings(self):
settings = self._gather_active_settings()
if settings is None:
return
self.settings_applied.emit(settings)
[docs]
def on_field_focused(self):
self.coordinate_selection_requested.emit()
[docs]
def receive_coordinate(self, x_value, y_value):
self.optional_entries['x0'].setText(str(x_value))
self.optional_entries['y0'].setText(str(y_value))
self.required_entries['threshold'].setFocus() if hasattr(self, 'required_entries') else None
# -----------------------------------------------------------------------
# Slots: Save buttons
# -----------------------------------------------------------------------
[docs]
def on_save_current(self):
settings = self._gather_active_settings()
if settings is None:
return
try:
if settings['algorithm'] == 'peakfinder9':
self._write_pf9_settings(self.current_ini_file_path, settings)
else:
write_peakfindersettings(self.current_ini_file_path, **{
k: v for k, v in settings.items()
if k not in ('algorithm', 'live_preview')
})
self._write_algo_choice(self.current_ini_file_path, 'peakfinder8')
self.check_and_enable_findpeaks_button()
except Exception as e:
QMessageBox.critical(self, "Error", f"Failed to save settings:\n{str(e)}")
[docs]
def on_save_all(self):
settings = self._gather_active_settings()
if settings is None:
return
try:
workspace = getattr(self.parent, 'workspace_ini_paths', {})
if workspace:
file_paths = list(workspace.values())
else:
file_paths = [
os.path.join(self.ini_directory, f)
for f in os.listdir(self.ini_directory)
if f.endswith(".ini") and not f.startswith(".")
]
for file_path in file_paths:
if settings['algorithm'] == 'peakfinder9':
self._write_pf9_settings(file_path, settings)
else:
write_peakfindersettings(file_path, **{
k: v for k, v in settings.items()
if k not in ('algorithm', 'live_preview')
})
self._write_algo_choice(file_path, 'peakfinder8')
self.check_and_enable_findpeaks_button()
except Exception as e:
QMessageBox.critical(self, "Error", f"Failed to save settings to all files:\n{str(e)}")
# -----------------------------------------------------------------------
# Slots: Find Peaks buttons
# -----------------------------------------------------------------------
def _set_parent_peakfinding_state(self, is_active: bool):
if self.parent is not None:
setattr(self.parent, "_peakfinding_active", bool(is_active))
def _set_parent_status(self, message: str, animate: bool = False):
if self.parent is not None and hasattr(self.parent, "_set_status_message"):
self.parent._set_status_message(message, animate=animate)
def _clear_parent_peakfinding_status(self):
if self.parent is None:
return
if not hasattr(self.parent, "_clear_status_message"):
return
base = getattr(self.parent, "_status_base_message", "") or ""
if base.startswith("Peakfinding"):
self.parent._clear_status_message()
def _begin_peakfinding_run(self):
self._active_peakfinding_runs += 1
self.find_peaks_current_button.setEnabled(False)
self.find_peaks_all_button.setEnabled(False)
self._set_parent_peakfinding_state(True)
self._set_parent_status("Peakfinding", animate=True)
def _end_peakfinding_run(self):
self._active_peakfinding_runs = max(0, self._active_peakfinding_runs - 1)
if self._active_peakfinding_runs == 0:
self._set_parent_peakfinding_state(False)
self._clear_parent_peakfinding_status()
self.check_and_enable_findpeaks_button()
def _resolve_h5_from_ini(self, ini_path: str):
candidates = []
def _add(path: str):
if not path:
return
ap = os.path.abspath(path)
if ap not in candidates:
candidates.append(ap)
ini_dir = os.path.dirname(os.path.abspath(ini_path))
cfg = configparser.ConfigParser()
try:
cfg.read(ini_path)
except Exception:
pass
if cfg.has_section("Paths"):
h5_rel = cfg.get("Paths", "h5file", fallback="").strip()
out_rel = cfg.get("Paths", "outputfolder", fallback="").strip()
if h5_rel:
if os.path.isabs(h5_rel):
_add(h5_rel)
else:
_add(os.path.join(ini_dir, h5_rel))
if out_rel:
_add(os.path.join(ini_dir, out_rel, h5_rel))
stem = os.path.splitext(os.path.abspath(ini_path))[0]
_add(stem + ".h5")
_add(stem + ".hdf5")
_add(stem + ".cxi")
for path in candidates:
if os.path.exists(path):
return path
return None
def _missing_mask_inis(self, ini_paths):
missing = []
for ini_path in ini_paths:
h5_path = self._resolve_h5_from_ini(ini_path)
if not h5_path:
missing.append((ini_path, "HDF5 file not found"))
continue
try:
with h5py.File(h5_path, "r") as f:
mask_ds = get_mask_dataset(f)
if mask_ds is None or mask_ds.ndim != 2:
missing.append((ini_path, f"No 2D mask dataset in {h5_path}"))
except Exception as exc:
missing.append((ini_path, f"Failed to read mask from {h5_path}: {exc}"))
return missing
[docs]
def on_find_peaks_current(self):
missing = self._missing_mask_inis([self.current_ini_file_path])
if missing:
QMessageBox.warning(
self,
"Mask Required",
"Peakfinding requires a mask dataset. Run Preflight Check and generate a full mask first.\n\n"
+ "\n".join(f"- {ini}: {reason}" for ini, reason in missing),
)
return
self._begin_peakfinding_run()
algo = self.algo_combo.currentText()
try:
threading.Thread(
target=self.run_findpeaks,
args=([self.current_ini_file_path], algo),
daemon=True,
).start()
except Exception:
self._end_peakfinding_run()
raise
[docs]
def on_find_peaks_all(self):
algo = self.algo_combo.currentText()
workspace = getattr(self.parent, 'workspace_ini_paths', {})
if workspace:
ini_files = list(workspace.values())
else:
ini_files = [
os.path.join(self.ini_directory, f)
for f in os.listdir(self.ini_directory)
if f.endswith(".ini") and not f.startswith(".")
]
missing = self._missing_mask_inis(ini_files)
if missing:
detail = "\n".join(f"- {ini}: {reason}" for ini, reason in missing[:20])
if len(missing) > 20:
detail += f"\n- ... and {len(missing) - 20} more"
QMessageBox.warning(
self,
"Mask Required",
"Peakfinding requires a mask dataset for every file. "
"Run Preflight Check and generate/apply full masks first.\n\n"
+ detail,
)
return
self._begin_peakfinding_run()
try:
threading.Thread(
target=self.run_findpeaks,
args=(ini_files, algo),
daemon=True,
).start()
except Exception:
self._end_peakfinding_run()
raise
[docs]
def run_findpeaks(self, input_paths, algo):
"""Execute peak finding with the chosen algorithm in a background thread."""
try:
if isinstance(input_paths, str):
if os.path.isdir(input_paths):
ini_list = sorted(
os.path.join(input_paths, f)
for f in os.listdir(input_paths)
if f.endswith(".ini")
)
else:
ini_list = [input_paths]
else:
ini_list = input_paths
batch_fn = findpeaks9_dask_batch if algo == 'peakfinder9' else findpeaks_dask_batch
folder_prefix = 'findpeaks9' if algo == 'peakfinder9' else 'findpeaks'
total_files = len(ini_list)
for idx, ini_path in enumerate(ini_list, start=1):
self.file_progress_updated.emit(ini_path, idx, total_files)
def progress_cb(done, total):
self.progress_updated.emit(done, total)
out = batch_fn(ini_path, progress_callback=progress_cb)
outputfolder = None
if isinstance(out, dict) and 'outputfolder' in out:
outputfolder = out['outputfolder']
else:
base = os.path.dirname(ini_path)
candidates = [
os.path.join(base, name)
for name in os.listdir(base)
if name.startswith(folder_prefix)
]
if candidates:
outputfolder = max(candidates, key=os.path.getmtime)
if outputfolder:
self.last_output_folder = (
os.path.dirname(outputfolder)
if os.path.basename(outputfolder).startswith(folder_prefix)
else outputfolder
)
else:
self.last_output_folder = os.path.dirname(ini_path)
self.peakfinding_finished.emit()
except Exception as e:
self.peakfinding_error.emit(str(e))
# -----------------------------------------------------------------------
# Button state management
# -----------------------------------------------------------------------
# -----------------------------------------------------------------------
# Statistics / histogram
# -----------------------------------------------------------------------
[docs]
def update_run_list(self):
"""Scan for findpeaks_* and findpeaks9_* result folders and populate the dropdown."""
current = self.hist_run_combo.currentText()
self.hist_run_combo.blockSignals(True)
self.hist_run_combo.clear()
base = self.ini_directory
if os.path.isdir(base):
for name in sorted(os.listdir(base)):
if name.startswith('findpeaks'):
self.hist_run_combo.addItem(name)
self.hist_run_combo.blockSignals(False)
if self.hist_run_combo.count() == 0:
self.canvas.figure.clear()
self.canvas.draw()
else:
idx = self.hist_run_combo.findText(current)
self.hist_run_combo.setCurrentIndex(idx if idx != -1 else 0)
self.on_plot_histogram()
[docs]
def on_plot_histogram(self):
"""Load peak_counts.npy from the selected run and plot a histogram."""
run = self.hist_run_combo.currentText()
if not run:
self.canvas.figure.clear()
self.canvas.draw()
return
base = getattr(self, 'last_output_folder', self.ini_directory)
data_file = os.path.join(base, run, 'peak_counts.npy')
if not os.path.exists(data_file):
self.canvas.figure.clear()
self.canvas.draw()
return
data = np.load(data_file)
counts = data[:, 1]
self.canvas.figure.clear()
ax = self.canvas.figure.subplots()
ax.hist(counts, bins=self.hist_bins_spin.value())
ax.set_xlabel("Number of Peaks")
ax.set_ylabel("Frequency")
self.canvas.draw()
# -----------------------------------------------------------------------
# Progress / status slots
# -----------------------------------------------------------------------
[docs]
def on_progress_updated(self, done, total):
self.progress_bar.setValue(int(done / total * 100) if total > 0 else 0)
[docs]
def on_peakfinding_finished(self):
self._end_peakfinding_run()
[docs]
def on_peakfinding_error(self, message):
self._end_peakfinding_run()
QMessageBox.critical(self, "Peak Finding Error",
f"An error occurred during peak finding:\n{message}")
[docs]
def on_file_progress_updated(self, filename, index, total):
self.file_progress_label.setText(
f"{os.path.basename(filename)} ({index}/{total})"
)
self._set_parent_status(
f"Peakfinding ({index}/{total}): {os.path.basename(filename)}",
animate=True,
)