"""Dialog and worker to import EMD files/folders and convert them to H5/INI."""
from coseda.logging_utils import log_print
from PyQt6.QtWidgets import (
QDialog, QVBoxLayout, QHBoxLayout, QLabel, QPushButton, QListWidget,
QFileDialog, QTextEdit, QComboBox, QMessageBox, QGroupBox, QFormLayout,
QLineEdit, QCheckBox, QSpinBox
)
from PyQt6.QtCore import Qt, QThread, pyqtSignal, QMimeData, QTimer
import os
import time
[docs]
class ImportEMDWindow(QDialog):
def __init__(self, parent=None):
"""Dialog to import EMD files/folders and convert them to H5/INI via a worker thread."""
super().__init__(parent)
self.parent = parent # Reference to main window
self.setWindowTitle("Import EMD Files")
self.setMinimumSize(500, 400)
self.setWindowFlags(Qt.WindowType.Window)
self.setAcceptDrops(True)
self.import_thread = None
self.monitor_active = False
self.monitor_timer = QTimer(self)
self.monitor_timer.timeout.connect(self._scan_monitor_sources)
self.monitor_candidate_stats = {}
self.monitor_processed_paths = set()
self.monitor_queue = []
self.monitor_stable_seconds = 60
self.monitor_bin_factor = 1
self.monitor_sample_class = "reference"
self.monitor_current_path = None
self.monitor_invalid_sources = set()
self._current_import_from_monitor = False
self.init_ui()
[docs]
def init_ui(self):
"""Build the import dialog UI controls and wiring."""
# Create main layout
main_layout = QVBoxLayout()
main_layout.setSpacing(10)
main_layout.setContentsMargins(10, 10, 10, 10)
# File selection group
file_group = QGroupBox("EMD Files or Folders")
file_layout = QVBoxLayout()
# List widget to display selected files/folders
self.file_list_widget = QListWidget()
file_layout.addWidget(self.file_list_widget)
# Buttons to add/remove files/folders
file_buttons_layout = QHBoxLayout()
add_file_button = QPushButton("Add File")
add_file_button.clicked.connect(self.add_emd_file)
add_folder_button = QPushButton("Add Folder")
add_folder_button.clicked.connect(self.add_emd_folder)
remove_selected_button = QPushButton("Remove Selected")
remove_selected_button.clicked.connect(self.remove_selected_items)
file_buttons_layout.addWidget(add_file_button)
file_buttons_layout.addWidget(add_folder_button)
file_buttons_layout.addWidget(remove_selected_button)
file_buttons_layout.addStretch()
file_layout.addLayout(file_buttons_layout)
file_group.setLayout(file_layout)
main_layout.addWidget(file_group)
# Parameters group
parameters_group = QGroupBox("Conversion Parameters")
parameters_layout = QFormLayout()
# bin_factor input
self.bin_factor_input = QLineEdit()
self.bin_factor_input.setText("1") # Default value
parameters_layout.addRow("Bin Factor:", self.bin_factor_input)
# sample_class selection
self.sample_class_combo = QComboBox()
self.sample_class_combo.addItems(['reference', 'MOF', 'COF', 'zeolite', 'protein', 'other'])
parameters_layout.addRow("Sample Class:", self.sample_class_combo)
parameters_group.setLayout(parameters_layout)
main_layout.addWidget(parameters_group)
# Monitor group
monitor_group = QGroupBox("Monitor Mode")
monitor_layout = QFormLayout()
self.monitor_checkbox = QCheckBox("Watch selected folders for new EMD files")
monitor_layout.addRow(self.monitor_checkbox)
self.scan_interval_spin = QSpinBox()
self.scan_interval_spin.setRange(1, 3600)
self.scan_interval_spin.setValue(5)
self.scan_interval_spin.setSuffix(" s")
monitor_layout.addRow("Scan Interval:", self.scan_interval_spin)
self.stable_time_spin = QSpinBox()
self.stable_time_spin.setRange(1, 3600)
self.stable_time_spin.setValue(60)
self.stable_time_spin.setSuffix(" s")
monitor_layout.addRow("Stable Time:", self.stable_time_spin)
monitor_group.setLayout(monitor_layout)
main_layout.addWidget(monitor_group)
# Status Text
self.status_text_edit = QTextEdit()
self.status_text_edit.setReadOnly(True)
main_layout.addWidget(self.status_text_edit)
# Buttons layout
buttons_layout = QHBoxLayout()
self.import_button = QPushButton("Import")
self.import_button.clicked.connect(self.start_import)
cancel_button = QPushButton("Close")
cancel_button.clicked.connect(self.close)
buttons_layout.addStretch()
buttons_layout.addWidget(self.import_button)
buttons_layout.addWidget(cancel_button)
main_layout.addLayout(buttons_layout)
self.setLayout(main_layout)
# Methods to handle adding/removing files/folders
[docs]
def add_emd_file(self):
"""Add one or more .emd files to the selection list, avoiding duplicates."""
file_dialog = QFileDialog(self, "Select EMD File(s)")
file_dialog.setFileMode(QFileDialog.FileMode.ExistingFiles)
file_dialog.setNameFilter("EMD Files (*.emd)")
if file_dialog.exec():
selected_files = file_dialog.selectedFiles()
for file in selected_files:
if not any(self.file_list_widget.item(i).text() == file for i in range(self.file_list_widget.count())):
self.file_list_widget.addItem(file)
else:
QMessageBox.information(self, "File Exists", "This file is already in the list.")
[docs]
def add_emd_folder(self):
"""Add a folder path; later expanded to include all .emd files inside."""
folder = QFileDialog.getExistingDirectory(self, "Select EMD Folder", os.getcwd())
if folder:
if not any(self.file_list_widget.item(i).text() == folder for i in range(self.file_list_widget.count())):
self.file_list_widget.addItem(folder)
else:
QMessageBox.information(self, "Folder Exists", "This folder is already in the list.")
[docs]
def remove_selected_items(self):
"""Remove the currently selected rows from the selection list."""
selected_items = self.file_list_widget.selectedItems()
if not selected_items:
return
for item in selected_items:
self.file_list_widget.takeItem(self.file_list_widget.row(item))
[docs]
def dragEnterEvent(self, event):
if event.mimeData().hasUrls():
event.acceptProposedAction()
[docs]
def dropEvent(self, event):
for url in event.mimeData().urls():
path = url.toLocalFile()
if not path:
continue
if os.path.isdir(path) or (os.path.isfile(path) and path.lower().endswith('.emd')):
if not any(self.file_list_widget.item(i).text() == path for i in range(self.file_list_widget.count())):
self.file_list_widget.addItem(path)
[docs]
def closeEvent(self, event):
"""Stop monitor cleanly and keep the dialog alive while a worker runs."""
if self._is_import_running():
QMessageBox.warning(
self,
"Import Running",
"Please wait for the current EMD import to finish before closing this window.",
)
event.ignore()
return
if self.monitor_active:
self.stop_monitor()
event.accept()
[docs]
def start_import(self):
"""Validate inputs, resolve .emd files, and start the conversion worker thread."""
if self.monitor_active:
self.stop_monitor()
return
input_paths = self._selected_input_paths()
if not input_paths:
QMessageBox.warning(self, "No Files or Folders Selected", "Please add at least one EMD file or folder.")
return
if self.monitor_checkbox.isChecked():
self.start_monitor(input_paths)
return
try:
resolved_paths = self._resolve_emd_paths(input_paths)
except ValueError as exc:
QMessageBox.warning(self, "Invalid Input", str(exc))
return
if not resolved_paths:
QMessageBox.warning(self, "No .emd Files Found", "No valid .emd files were found in the provided input.")
return
parameters = self._get_conversion_parameters()
if parameters is None:
return
bin_factor, sample_class = parameters
log_print(f"Selected Sample Class: {sample_class}") # Debugging statement
self.import_button.setEnabled(False)
self.status_text_edit.clear()
self.update_status("Starting import...")
self._start_import_thread(resolved_paths, bin_factor, sample_class, from_monitor=False)
def _selected_input_paths(self):
"""Return source paths currently listed in the dialog."""
return [self.file_list_widget.item(i).text() for i in range(self.file_list_widget.count())]
def _resolve_emd_paths(self, input_paths):
"""Resolve files/folders into immediate .emd file paths."""
resolved_paths = []
for path in input_paths:
if os.path.isfile(path) and path.lower().endswith('.emd'):
resolved_paths.append(path)
elif os.path.isdir(path):
resolved_paths.extend(
[
os.path.join(path, file)
for file in os.listdir(path)
if file.lower().endswith('.emd') and not file.startswith("._")
]
)
else:
raise ValueError(f"The path {path} is not valid.")
return resolved_paths
def _get_conversion_parameters(self):
"""Return validated conversion parameters or None if invalid."""
bin_factor_text = self.bin_factor_input.text()
try:
bin_factor = int(bin_factor_text)
except ValueError:
QMessageBox.warning(self, "Invalid Bin Factor", "Please enter a valid integer for bin factor.")
return None
sample_class = self.sample_class_combo.currentText()
if not sample_class:
QMessageBox.warning(self, "Invalid Sample Class", "Please select a valid sample class.")
return None
return bin_factor, sample_class
def _start_import_thread(self, resolved_paths, bin_factor, sample_class, from_monitor=False):
"""Start the EMD conversion worker for already-resolved file paths."""
self._current_import_from_monitor = from_monitor
self.import_thread = EMDImportThread(
resolved_paths,
bin_factor,
sample_class,
parent=self
)
self.import_thread.status_update.connect(self.update_status)
self.import_thread.file_progress.connect(self._update_main_window_status)
self.import_thread.finished.connect(self.import_finished)
self.import_thread.start()
[docs]
def start_monitor(self, input_paths):
"""Start polling selected EMD sources and import stable files."""
invalid_sources = [
path for path in input_paths
if not os.path.isdir(path) and not (os.path.isfile(path) and path.lower().endswith(".emd"))
]
if invalid_sources:
QMessageBox.warning(
self,
"Invalid Monitor Source",
"Monitor mode can only watch folders or .emd files:\n" + "\n".join(invalid_sources),
)
return
parameters = self._get_conversion_parameters()
if parameters is None:
return
self.monitor_bin_factor, self.monitor_sample_class = parameters
self.monitor_stable_seconds = self.stable_time_spin.value()
scan_interval_ms = self.scan_interval_spin.value() * 1000
self.monitor_active = True
self.monitor_candidate_stats = {}
self.monitor_processed_paths = set()
self.monitor_queue = []
self.monitor_current_path = None
self.monitor_invalid_sources = set()
self._current_import_from_monitor = False
self.status_text_edit.clear()
self.update_status("Starting EMD monitor...")
self.update_status(
f"Scan interval: {self.scan_interval_spin.value()} s; "
f"stable time: {self.monitor_stable_seconds} s"
)
self.import_button.setText("Stop Monitor")
self.import_button.setEnabled(True)
self.monitor_timer.start(scan_interval_ms)
self._scan_monitor_sources()
[docs]
def stop_monitor(self):
"""Stop scanning for new EMD files."""
self.monitor_active = False
self.monitor_timer.stop()
self.monitor_queue.clear()
self.monitor_candidate_stats.clear()
self.update_status("EMD monitor stopped.")
if self._is_import_running():
self.import_button.setEnabled(False)
self.update_status("Current import will finish before controls are re-enabled.")
else:
self.import_button.setText("Import")
self.import_button.setEnabled(True)
if self.parent and hasattr(self.parent, '_clear_status_message'):
self.parent._clear_status_message()
def _scan_monitor_sources(self):
"""Poll selected paths and queue EMD files whose size/mtime are stable."""
if not self.monitor_active:
return
candidates = self._monitor_candidates()
queued_paths = set(self.monitor_queue)
now = time.monotonic()
for path in candidates:
if (
path in self.monitor_processed_paths
or path in queued_paths
or path == self.monitor_current_path
):
continue
try:
stat = os.stat(path)
except OSError:
self.monitor_candidate_stats.pop(path, None)
continue
previous = self.monitor_candidate_stats.get(path)
current = {
"size": stat.st_size,
"mtime": stat.st_mtime,
"stable_since": now,
}
if previous is None:
self.monitor_candidate_stats[path] = current
self.update_status(f"Detected EMD candidate: {path}")
continue
if previous["size"] != stat.st_size or previous["mtime"] != stat.st_mtime:
self.monitor_candidate_stats[path] = current
continue
stable_for = now - previous["stable_since"]
if stable_for < self.monitor_stable_seconds:
continue
if not self._is_readable_emd(path):
continue
self.monitor_queue.append(path)
self.monitor_processed_paths.add(path)
self.monitor_candidate_stats.pop(path, None)
queued_paths.add(path)
self.update_status(f"Queued stable EMD file: {path}")
self._start_next_monitor_import()
def _monitor_candidates(self):
"""Return immediate .emd candidates from the current monitor sources."""
candidates = []
for source in self._selected_input_paths():
if os.path.isfile(source) and source.lower().endswith(".emd"):
candidates.append(source)
continue
if os.path.isdir(source):
try:
for name in os.listdir(source):
if name.startswith("._") or not name.lower().endswith(".emd"):
continue
candidates.append(os.path.join(source, name))
except OSError as exc:
if source not in self.monitor_invalid_sources:
self.monitor_invalid_sources.add(source)
self.update_status(f"Could not scan monitor folder {source}: {exc}")
continue
if source not in self.monitor_invalid_sources:
self.monitor_invalid_sources.add(source)
self.update_status(f"Monitor source no longer exists: {source}")
return candidates
def _is_readable_emd(self, path):
"""Return True once the EMD/HDF5 container can be opened for reading."""
try:
import h5py
with h5py.File(path, "r"):
return True
except Exception:
return False
def _start_next_monitor_import(self):
"""Start the next queued monitor import if the worker is idle."""
if not self.monitor_active or self._is_import_running() or not self.monitor_queue:
return
path = self.monitor_queue.pop(0)
self.monitor_current_path = path
self.update_status(f"Importing monitored EMD file: {path}")
self._start_import_thread(
[path],
self.monitor_bin_factor,
self.monitor_sample_class,
from_monitor=True,
)
def _is_import_running(self):
return bool(self.import_thread and self.import_thread.isRunning())
def _update_main_window_status(self, idx, total):
if self.parent and hasattr(self.parent, '_set_status_message'):
self.parent._set_status_message(f"Importing ({idx}/{total})", animate=True)
[docs]
def update_status(self, message):
"""Append a status message to the log text area."""
self.status_text_edit.append(message)
[docs]
def import_finished(self, success, message, ini_files):
"""Handle completion of the import, reporting results and adding INIs to the workspace."""
from_monitor = self._current_import_from_monitor
if self.parent and hasattr(self.parent, '_clear_status_message'):
self.parent._clear_status_message()
if success and from_monitor:
self.update_status("Monitored EMD import completed successfully.")
elif success:
self.update_status("Import completed successfully.")
QMessageBox.information(self, "Import Complete", "EMD files have been successfully imported.")
elif from_monitor:
self.update_status(f"Monitored EMD import failed: {message}")
else:
self.update_status(f"Import failed: {message}")
QMessageBox.critical(self, "Import Failed", f"Import failed: {message}")
if ini_files and self.parent and hasattr(self.parent, "add_generated_ini_paths"):
result = self.parent.add_generated_ini_paths(ini_files)
if result["added"]:
self.update_status("Added to workspace:\n" + "\n".join(result["added"]))
if from_monitor:
self.monitor_current_path = None
self._current_import_from_monitor = False
self.import_thread = None
if self.monitor_active:
self._start_next_monitor_import()
self.import_button.setText("Stop Monitor")
self.import_button.setEnabled(True)
else:
self.import_button.setText("Import")
self.import_button.setEnabled(True)
return
self.import_thread = None
self.import_button.setText("Import")
self.import_button.setEnabled(True)
[docs]
class EMDImportThread(QThread):
"""Worker thread that converts EMD inputs to H5/INI artifacts and reports status."""
status_update = pyqtSignal(str)
file_progress = pyqtSignal(int, int) # (current_index, total_files)
finished = pyqtSignal(bool, str, list)
def __init__(self, input_paths, bin_factor, sample_class, parent=None):
"""Worker that converts EMD files to H5/INI artifacts without blocking the UI."""
super().__init__(parent)
self.input_paths = input_paths
self.bin_factor = bin_factor
self.sample_class = sample_class
self.parent = parent
self.limit_frames = False
self.created_ini_files = []
[docs]
def run(self):
"""Perform EMD conversion work, emitting progress and final result/INI list."""
try:
from coseda import initialize
from coseda.importers import h5convert
log_print(f"Running import with sample_class: {self.sample_class}")
log_print(f"Limit Frames: {self.limit_frames}")
overall_success = True # Flag to track overall success
error_messages = [] # List to collect error messages
total_paths = len(self.input_paths)
for path_idx, input_path in enumerate(self.input_paths, 1):
self.file_progress.emit(path_idx, total_paths)
self.status_update.emit(f"Processing: {input_path}")
# Determine the type of input (file or folder)
if os.path.isfile(input_path) and input_path.lower().endswith('.emd'):
# Single .emd file
input_type = "emd"
create_insfiles_input = [input_path]
elif os.path.isdir(input_path):
# Folder containing .emd files
input_type = "emd"
create_insfiles_input = [
os.path.join(input_path, file) for file in os.listdir(input_path) if file.lower().endswith('.emd')
]
else:
self.status_update.emit(f"Invalid input path: {input_path}")
overall_success = False
error_messages.append(f"Invalid input path: {input_path}")
continue
if not create_insfiles_input:
self.status_update.emit(f"No .emd files found in {input_path}.")
overall_success = False
error_messages.append(f"No .emd files found in {input_path}.")
continue
self.status_update.emit(f"Input for create_insfiles: {create_insfiles_input}")
# Create INS files
self.status_update.emit("Creating INS file...")
create_insfiles_result = initialize.create_insfiles(create_insfiles_input, input_type)
if not create_insfiles_result:
self.status_update.emit("Failed to create INS file.")
overall_success = False
error_messages.append(f"Failed to create INS file for {input_path}")
continue # Continue processing other inputs
self.status_update.emit(f"INS file created successfully: {create_insfiles_result}")
# Write conversion settings with the correct parameters
self.status_update.emit("Writing conversion settings...")
initialize.write_h5conversionsettings(
input_path=create_insfiles_result,
limit_frames=self.limit_frames, # Always False
bin_factor=self.bin_factor,
sample_class=self.sample_class
)
# Convert EMD to H5 - Process each INS file individually
self.status_update.emit("Converting EMD to H5...")
for ini_file in create_insfiles_result:
self.status_update.emit(f"Converting {ini_file} to H5...")
result = h5convert.emd_to_h5(ini_file)
if result is not None:
self.status_update.emit(f"Error processing {ini_file}: {result}")
overall_success = False
error_messages.append(f"Error processing {ini_file}: {result}")
# Continue processing other files
continue
else:
self.status_update.emit(f"Successfully converted {ini_file} to H5.")
if ini_file not in self.created_ini_files:
self.created_ini_files.append(ini_file)
self.status_update.emit(f"Finished processing: {input_path}")
log_print(f"Finished for {create_insfiles_result}")
# Signal success or failure based on overall_success
if overall_success:
self.finished.emit(True, "", self.created_ini_files)
else:
error_message_combined = "\n".join(error_messages)
self.finished.emit(False, error_message_combined, self.created_ini_files)
except Exception as e:
self.status_update.emit(f"Error: {str(e)}")
self.finished.emit(False, str(e), self.created_ini_files)