"""Dialog and worker to import MRC files/folders, create INIs, and convert to H5."""
from coseda.logging_utils import log_print
from PyQt6.QtWidgets import (
QDialog, QVBoxLayout, QHBoxLayout, QPushButton, QListWidget,
QFileDialog, QTextEdit, QMessageBox, QGroupBox
)
from PyQt6.QtCore import Qt, QThread, pyqtSignal
import os
[docs]
class ImportMRCWindow(QDialog):
"""UI to batch-import MRC folders and kick off conversion to H5."""
def __init__(self, parent=None):
super().__init__(parent)
self.parent = parent # Reference to main window
self.setWindowTitle("Import MRC Files")
self.setMinimumSize(500, 400)
self.setWindowFlags(Qt.WindowType.Window)
self.setAcceptDrops(True)
self.init_ui()
[docs]
def init_ui(self):
"""Build the import dialog controls and wiring."""
main_layout = QVBoxLayout()
main_layout.setSpacing(10)
main_layout.setContentsMargins(10, 10, 10, 10)
# Source selection
folder_group = QGroupBox("MRC Files or Folders")
folder_layout = QVBoxLayout()
# List Widget to display folders
self.folder_list_widget = QListWidget()
folder_layout.addWidget(self.folder_list_widget)
# Buttons to add/remove folders
folder_buttons_layout = QHBoxLayout()
add_file_button = QPushButton("Add File")
add_file_button.clicked.connect(self.add_mrc_file)
add_folder_button = QPushButton("Add Folder")
add_folder_button.clicked.connect(self.add_mrc_folder)
remove_folder_button = QPushButton("Remove Selected")
remove_folder_button.clicked.connect(self.remove_selected_folders)
folder_buttons_layout.addWidget(add_file_button)
folder_buttons_layout.addWidget(add_folder_button)
folder_buttons_layout.addWidget(remove_folder_button)
folder_buttons_layout.addStretch()
folder_layout.addLayout(folder_buttons_layout)
folder_group.setLayout(folder_layout)
main_layout.addWidget(folder_group)
# Parameters (Add any additional parameters if needed)
# For MRC import, there may not be additional parameters required
# Buttons
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)
# Status Text
self.status_text_edit = QTextEdit()
self.status_text_edit.setReadOnly(True)
main_layout.addWidget(self.status_text_edit)
self.setLayout(main_layout)
[docs]
def add_mrc_file(self):
"""Add one or more .mrc files to the list."""
files, _ = QFileDialog.getOpenFileNames(
self,
"Select MRC File(s)",
os.getcwd(),
"MRC Files (*.mrc *.MRC)"
)
for file_path in files:
self._add_source_path(file_path)
[docs]
def add_mrc_folder(self):
"""Add a folder containing MRC files to the list."""
folder = QFileDialog.getExistingDirectory(self, "Select MRC Folder", os.getcwd())
if folder:
self._add_source_path(folder)
def _add_source_path(self, source_path):
"""Insert source path if not already present."""
if not any(self.folder_list_widget.item(i).text() == source_path for i in range(self.folder_list_widget.count())):
self.folder_list_widget.addItem(source_path)
else:
QMessageBox.information(self, "Entry Exists", f"Already added:\n{source_path}")
[docs]
def remove_selected_folders(self):
"""Remove selected folder entries from the list."""
selected_items = self.folder_list_widget.selectedItems()
if not selected_items:
return
for item in selected_items:
self.folder_list_widget.takeItem(self.folder_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('.mrc')):
self._add_source_path(path)
[docs]
def start_import(self):
"""Validate selection and start the worker thread to convert MRCs."""
# Get the list of input paths (folders and/or files)
input_paths = [self.folder_list_widget.item(i).text() for i in range(self.folder_list_widget.count())]
if not input_paths:
QMessageBox.warning(self, "No Inputs Selected", "Please add at least one MRC file or folder.")
return
# Disable the import button
self.import_button.setEnabled(False)
self.status_text_edit.clear()
self.update_status("Starting import...")
# Use QThread to run the import process in a separate thread
self.import_thread = MRCImportThread(
input_paths,
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()
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 status messages to the log box."""
self.status_text_edit.append(message)
[docs]
def import_finished(self, success, message, ini_files):
"""Handle completion of the import process and update workspace on success."""
if self.parent and hasattr(self.parent, '_clear_status_message'):
self.parent._clear_status_message()
if success:
self.update_status("Import completed successfully.")
QMessageBox.information(self, "Import Complete", "MRC files have been successfully imported.")
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 hasattr(self.parent, "open_file"):
self.parent.open_file(result["added"][0])
self.import_button.setEnabled(True)
[docs]
class MRCImportThread(QThread):
"""Worker thread to create INI files from MRC sources and run batch conversion."""
status_update = pyqtSignal(str)
file_progress = pyqtSignal(int, int) # (current_index, total_sources)
finished = pyqtSignal(bool, str, list) # Add 'list' to include ini file paths
def __init__(self, input_paths, parent=None):
super().__init__(parent)
self.input_paths = input_paths
self.parent = parent
[docs]
def run(self):
"""Perform INI creation and MRC conversion across all selected folders."""
try:
from coseda.importers.import_mrc import (
create_insfiles_mrc_batch,
mrc_folder_conversion_batch
)
overall_success = True
error_messages = []
converted_ini_files = []
total_paths = len(self.input_paths)
for path_idx, source_path in enumerate(self.input_paths, 1):
self.file_progress.emit(path_idx, total_paths)
self.status_update.emit(f"Processing source: {source_path}")
# Create INI files for MRC source(s)
self.status_update.emit("Creating INI files...")
ini_files = create_insfiles_mrc_batch(source_path)
log_print(ini_files)
if not ini_files:
self.status_update.emit("No valid MRC files found for this source.")
overall_success = False
error_messages.append(f"No valid MRC files found for source: {source_path}")
continue
# Start MRC folder conversion
self.status_update.emit("Starting MRC folder conversion...")
success, conversion_errors = mrc_folder_conversion_batch(
ini_files,
continue_on_error=True,
return_errors=True
)
if success:
self.status_update.emit("MRC folder conversion completed successfully.")
for ini_file in ini_files:
if ini_file not in converted_ini_files:
converted_ini_files.append(ini_file)
else:
self.status_update.emit("MRC folder conversion encountered errors. Continuing with remaining sources.")
overall_success = False
if conversion_errors:
error_messages.extend(conversion_errors)
else:
error_messages.append(f"Conversion failed for source: {source_path}")
# Signal the overall success or failure
if overall_success:
self.finished.emit(True, "", converted_ini_files)
else:
self.finished.emit(False, "\n".join(error_messages), converted_ini_files)
except Exception as e:
self.status_update.emit(f"Error: {str(e)}")
self.finished.emit(False, str(e), [])