"""UI for stripping empty/low-peak frames from HDF5 datasets."""
import os
from coseda.pipeline.strip_h5 import strip_h5_batch
from coseda.initialize import write_stripsettings
from coseda.io import config_to_paths
from PyQt6.QtWidgets import (
QDialog, QVBoxLayout, QHBoxLayout, QLabel, QSpinBox, QDoubleSpinBox, QPushButton, QCheckBox, QMessageBox, QProgressBar, QComboBox
)
from PyQt6.QtCore import Qt
from PyQt6.QtCore import QThread, pyqtSignal
[docs]
class StripWorker(QThread):
"""Worker thread to strip a single HDF5 file based on peak threshold."""
progress = pyqtSignal(int)
finished = pyqtSignal(str)
def __init__(self, h5_path, min_peaks, force):
"""Worker to strip a single HDF5 file based on peak count."""
super().__init__()
self.h5_path = h5_path
self.min_peaks = min_peaks
self.force = force
self._abort = False
[docs]
def abort(self):
self._abort = True
self.terminate()
[docs]
def run(self):
from coseda.pipeline.strip_h5 import strip_h5
try:
strip_h5(
self.h5_path,
threshold=self.min_peaks,
force=self.force,
progress_callback=self.progress.emit
)
self.finished.emit("Stripping complete!")
except Exception as e:
self.finished.emit(f"Error: {e}")
[docs]
class StripWorkerBatch(QThread):
"""Worker thread to strip batches of INIs/HDF5 files, emitting per-file progress."""
progress = pyqtSignal(int)
finished = pyqtSignal(str)
file_progress = pyqtSignal(str, int, int)
def __init__(self, input_args, progress_callback, finished_callback):
"""Worker to strip batches of INIs/HDF5 files, emitting per-file progress."""
super().__init__()
self.input_args = input_args
self.progress_callback = progress_callback
self.finished_callback = finished_callback
[docs]
def run(self):
try:
if isinstance(self.input_args, str):
ini_list = [self.input_args]
else:
ini_list = self.input_args
total_files = len(ini_list)
for idx, ini_path in enumerate(ini_list, 1):
self.file_progress.emit(ini_path, idx, total_files)
strip_h5_batch(
ini_path,
progress_callback=self.progress_callback
)
self.finished.emit("Stripping complete!")
except Exception as e:
self.finished.emit(f"Error: {e}")
[docs]
def abort(self):
# No built-in abort for batch
pass
[docs]
class StripFilesWindow(QDialog):
def __init__(self, parent=None, ini_directory=None, ini_file_path=None):
"""Dialog to strip empty/low-peak frames from HDF5 files for one or all workspace INIs."""
super().__init__(parent)
# Store reference to main window for workspace access
self.main_window = parent
self.setWindowTitle("Remove Empty Frames")
self.setMinimumWidth(400)
self.ini_directory = ini_directory
self.ini_file_path = ini_file_path
self.last_ini_file = None # track the most recently processed INI
self._init_ui()
def _init_ui(self):
"""Build and wire the Remove Empty Frames UI controls."""
layout = QVBoxLayout(self)
# File selection mode
filemode_layout = QHBoxLayout()
filemode_label = QLabel("Mode:")
self.filemode_combo = QComboBox()
self.filemode_combo.addItems(["Current file only", "All files in workspace"])
self.filemode_combo.setToolTip("Choose whether to strip only the current file or every INI in the workspace.")
filemode_layout.addWidget(filemode_label)
filemode_layout.addWidget(self.filemode_combo)
layout.addLayout(filemode_layout)
# nPeaks threshold
threshold_layout = QHBoxLayout()
threshold_label = QLabel("Minimum Peaks per Frame:")
self.threshold_spin = QSpinBox()
self.threshold_spin.setMinimum(0)
self.threshold_spin.setMaximum(1000)
self.threshold_spin.setValue(1)
self.threshold_spin.setToolTip("Frames with fewer detected peaks than this are removed.")
threshold_layout.addWidget(threshold_label)
threshold_layout.addWidget(self.threshold_spin)
layout.addLayout(threshold_layout)
# Remove empty frames only (optional checkbox)
self.empty_only_checkbox = QCheckBox("Only remove frames with 0 peaks")
self.empty_only_checkbox.setChecked(True)
self.empty_only_checkbox.setToolTip("When checked, only zero-peak frames are removed (threshold locked to 1).")
layout.addWidget(self.empty_only_checkbox)
self.empty_only_checkbox.stateChanged.connect(self.on_empty_only_checkbox_changed)
# Option to force even if most frames are empty
self.force_checkbox = QCheckBox("Force even if many frames are removed (danger!)")
self.force_checkbox.setChecked(False)
self.force_checkbox.setToolTip("Allow stripping even if a large portion of frames will be removed.")
layout.addWidget(self.force_checkbox)
# Start stripping button
self.start_btn = QPushButton("Start Stripping")
self.start_btn.setToolTip("Write strip settings and start removing empty/low-peak frames.")
self.start_btn.clicked.connect(self.start_stripping)
layout.addWidget(self.start_btn)
# Status label
self.status_label = QLabel("")
layout.addWidget(self.status_label)
# Total batch progress bar (file X of Y)
self.total_progress_label = QLabel("Overall:")
self.total_progress_label.setVisible(False)
layout.addWidget(self.total_progress_label)
self.total_progress_bar = QProgressBar()
self.total_progress_bar.setMinimum(0)
self.total_progress_bar.setMaximum(100)
self.total_progress_bar.setValue(0)
self.total_progress_bar.setVisible(False)
self.total_progress_bar.setToolTip("Shows overall progress across all files.")
layout.addWidget(self.total_progress_bar)
# Per-file frame progress bar
self.file_progress_label = QLabel("Current file:")
self.file_progress_label.setVisible(False)
layout.addWidget(self.file_progress_label)
self.progress_bar = QProgressBar()
self.progress_bar.setMinimum(0)
self.progress_bar.setMaximum(100)
self.progress_bar.setValue(0)
self.progress_bar.setVisible(False)
self.progress_bar.setToolTip("Shows progress within the current file.")
layout.addWidget(self.progress_bar)
self.abort_btn = QPushButton("Abort and Restore Backup")
self.abort_btn.setEnabled(False)
self.abort_btn.setToolTip("Stop processing and restore the last backup if available.")
self.abort_btn.clicked.connect(self.abort_stripping)
layout.addWidget(self.abort_btn)
layout.addStretch()
self.on_empty_only_checkbox_changed(None)
[docs]
def on_empty_only_checkbox_changed(self, state):
if self.empty_only_checkbox.isChecked():
self.threshold_spin.setValue(1)
self.threshold_spin.setEnabled(False)
else:
self.threshold_spin.setEnabled(True)
[docs]
def start_stripping(self):
if self.main_window and getattr(self.main_window, "_peakfinding_active", False):
QMessageBox.warning(
self,
"Peakfinding Running",
"Peakfinding is still running. Please wait until it finishes before stripping frames.",
)
return
min_peaks = self.threshold_spin.value()
force = self.force_checkbox.isChecked()
# Confirmation if force is checked on multiple INIs
if force and self.filemode_combo.currentText() == "All files in workspace":
reply = QMessageBox.question(
self, "Confirm Dangerous Action",
"You are about to remove a large number of frames. Are you sure?",
QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No
)
if reply != QMessageBox.StandardButton.Yes:
return
# Determine input for write_stripsettings / strip_h5_batch
if self.filemode_combo.currentText() == "All files in workspace":
# Use only files currently open in the workspace
workspace = getattr(self.main_window, 'workspace_ini_paths', {})
ini_files = list(workspace.values())
if not ini_files:
QMessageBox.warning(
self,
"No Files",
"No INI files are currently open in the workspace."
)
return
input_args = ini_files
else:
input_args = self.ini_file_path
if not isinstance(input_args, list):
input_args = [input_args]
# Record HDF5 file paths and initial total size for storage savings calculation
self.h5_paths = []
for ini in input_args:
_, _, _, _, _, h5file_path = config_to_paths(ini)
self.h5_paths.append(h5file_path)
self.initial_total_size = sum(os.path.getsize(p) for p in self.h5_paths if os.path.exists(p))
# Write strip settings into INI(s)
try:
write_stripsettings(
input_args,
strip_threshold=min_peaks,
strip_force=force
)
except Exception as e:
QMessageBox.critical(self, "Settings Error", f"Failed to write settings:\n{e}")
return
# Start batch stripping
is_batch = len(input_args) > 1
self.progress_bar.setVisible(True)
self.progress_bar.setValue(0)
self.file_progress_label.setVisible(is_batch)
self.total_progress_bar.setVisible(is_batch)
self.total_progress_bar.setValue(0)
self.total_progress_label.setVisible(is_batch)
self.abort_btn.setEnabled(True)
self.status_label.setText("Starting stripping...")
if self.main_window and hasattr(self.main_window, '_set_status_message'):
self.main_window._set_status_message("Stripping empty frames", animate=True)
def on_finished(message):
# Compute storage saved
final_total = sum(os.path.getsize(p) for p in self.h5_paths if os.path.exists(p))
saved_bytes = self.initial_total_size - final_total
saved_mb = saved_bytes / (1024 * 1024 * 1024)
summary = f"{message} Freed {saved_mb:.2f} GB of storage."
self.status_label.setText(summary)
self.abort_btn.setEnabled(False)
self.progress_bar.setVisible(False)
self.file_progress_label.setVisible(False)
self.total_progress_bar.setVisible(False)
self.total_progress_label.setVisible(False)
if self.main_window and hasattr(self.main_window, '_clear_status_message'):
self.main_window._clear_status_message()
# Create and start worker
self.worker = StripWorkerBatch(
input_args,
progress_callback=self.progress_bar.setValue,
finished_callback=on_finished
)
# Track which INI is currently being processed
self.worker.file_progress.connect(lambda path, idx, total: setattr(self, 'last_ini_file', path))
# Update status label with current file progress
self.worker.file_progress.connect(
lambda path, idx, total: self.status_label.setText(
f"Stripping {os.path.basename(path)} ({idx}/{total})"
)
)
# Reset frame-level progress bar and update total progress when starting a new file
self.worker.file_progress.connect(lambda path, idx, total: self.progress_bar.setValue(0))
self.worker.file_progress.connect(
lambda path, idx, total: self.total_progress_bar.setValue(int((idx - 1) / total * 100))
)
self.worker.finished.connect(on_finished)
self.worker.progress.connect(self.progress_bar.setValue)
self.worker.start()
[docs]
def abort_stripping(self):
# Attempt to stop the worker
if hasattr(self, "worker") and self.worker.isRunning():
self.worker.abort()
self.status_label.setText("Aborted. Attempting to restore backup...")
# Determine HDF5 path for the last processed INI
if self.filemode_combo.currentText() == "All files in workspace":
ini_path = self.last_ini_file
else:
ini_path = self.ini_file_path
if ini_path:
utputfolder, outputfolder_path, logfile, logfile_path, h5file, h5file_path= config_to_paths(ini_path)
else:
h5file_path = None
backup_path = h5file_path + ".backup" if h5file_path else None
if backup_path and os.path.exists(backup_path):
try:
os.replace(backup_path, h5file_path)
self.status_label.setText("Backup restored.")
except Exception as e:
self.status_label.setText(f"Backup restore failed: {e}")
else:
self.status_label.setText("No backup file found.")
self.abort_btn.setEnabled(False)