"""Helpers for saving, loading, and manipulating workspace INI lists and view state."""
from coseda.logging_utils import log_print
import os
from PyQt6.QtWidgets import QFileDialog, QMessageBox
_WORKSPACE_SEARCH_PRUNE_DIRS = {
".git",
".pytest_cache",
"__pycache__",
"pr-logs",
}
def _should_prune_workspace_search_dir(name):
"""Skip large generated folders that cannot contain source INI entries."""
return (
name in _WORKSPACE_SEARCH_PRUNE_DIRS
or name.startswith("merge-run")
or "_merge_" in name
)
def _matching_suffix_score(original_path, candidate_path):
original_parts = os.path.normpath(original_path).split(os.sep)
candidate_parts = os.path.normpath(candidate_path).split(os.sep)
score = 0
for old, new in zip(reversed(original_parts), reversed(candidate_parts)):
if old != new:
break
score += 1
return score
def _build_ini_index(search_root):
"""Index INI files below the workspace folder by basename."""
index = {}
if not search_root or not os.path.isdir(search_root):
return index
for root, dirs, files in os.walk(search_root):
dirs[:] = [d for d in dirs if not _should_prune_workspace_search_dir(d)]
for name in files:
if not name.lower().endswith(".ini"):
continue
index.setdefault(name, []).append(os.path.join(root, name))
return index
def _resolve_workspace_ini_paths(workspace_path, ini_paths):
"""
Resolve INI paths from a workspace file.
Existing absolute paths are kept. Missing paths are repaired by searching
below the directory containing the workspace file for a matching INI
basename. If multiple candidates exist, the candidate with the longest
matching path suffix wins; ties are reported as ambiguous.
"""
workspace_dir = os.path.dirname(os.path.abspath(workspace_path)) if workspace_path else ""
ini_index = None
resolved = []
repaired = []
missing = []
ambiguous = []
for raw_path in ini_paths:
original_path = raw_path.strip()
if not original_path:
continue
if os.path.exists(original_path):
resolved.append(original_path)
continue
if not os.path.isabs(original_path):
workspace_relative = os.path.normpath(os.path.join(workspace_dir, original_path))
if os.path.exists(workspace_relative):
resolved.append(workspace_relative)
repaired.append((original_path, workspace_relative))
continue
if ini_index is None:
ini_index = _build_ini_index(workspace_dir)
basename = os.path.basename(original_path)
candidates = ini_index.get(basename, [])
if not candidates:
missing.append(original_path)
continue
ranked = sorted(
((_matching_suffix_score(original_path, candidate), candidate) for candidate in candidates),
key=lambda item: (item[0], -len(os.path.relpath(item[1], workspace_dir).split(os.sep))),
reverse=True,
)
best_score = ranked[0][0]
best_candidates = [candidate for score, candidate in ranked if score == best_score]
if len(best_candidates) == 1:
repaired_path = best_candidates[0]
resolved.append(repaired_path)
repaired.append((original_path, repaired_path))
else:
ambiguous.append((original_path, best_candidates))
return resolved, repaired, missing, ambiguous
def _show_workspace_path_resolution_warning(self, repaired, missing, ambiguous):
problems = []
if missing:
problems.append(f"{len(missing)} path(s) could not be found")
if ambiguous:
problems.append(f"{len(ambiguous)} path(s) matched multiple files")
if not problems:
return
details = []
if repaired:
details.append(f"Repaired {len(repaired)} moved path(s) automatically.")
details.append("Unresolved workspace entries were not loaded.")
if missing:
details.append("\nMissing:")
details.extend(f"- {path}" for path in missing[:10])
if len(missing) > 10:
details.append(f"- ... {len(missing) - 10} more")
if ambiguous:
details.append("\nAmbiguous:")
for original, candidates in ambiguous[:5]:
details.append(f"- {original}")
for candidate in candidates[:3]:
details.append(f" -> {candidate}")
if len(candidates) > 3:
details.append(f" -> ... {len(candidates) - 3} more")
if len(ambiguous) > 5:
details.append(f"- ... {len(ambiguous) - 5} more ambiguous entries")
QMessageBox.warning(
self,
"Workspace Paths Not Resolved",
"\n".join(problems + [""] + details),
)
[docs]
def save_workspace_state(self):
"""Save workspace paths (or workspace file) to the config file."""
try:
# Ensure configuration directory exists
config_dir = os.path.dirname(self.config_file_path)
os.makedirs(config_dir, exist_ok=True)
if not self.config.has_section('General'):
self.config.add_section('General')
# Save workspace: either file or list of files
if self.workspace_file_path:
# A named workspace was saved
self.config.set('General', 'last_workspace_file', self.workspace_file_path)
if self.config.has_option('General', 'last_workspace_files'):
self.config.remove_option('General', 'last_workspace_files')
else:
# No workspace file: save the current list of INIs
if self.workspace_ini_paths:
files = list(self.workspace_ini_paths.values())
self.config.set('General', 'last_workspace_files', ",".join(files))
if self.config.has_option('General', 'last_workspace_file'):
self.config.remove_option('General', 'last_workspace_file')
# Write config
with open(self.config_file_path, 'w') as configfile:
self.config.write(configfile)
except Exception as e:
log_print(f"Failed to save workspace state: {e}")
[docs]
def load_last_directory(self):
"""Restore last directory and last workspace (file or INI list) from config."""
try:
if os.path.exists(self.config_file_path):
self.config.read(self.config_file_path)
if self.config.has_section('General') and self.config.has_option('General', 'last_directory'):
last_dir = self.config.get('General', 'last_directory')
if os.path.isdir(last_dir):
self.ini_directory = last_dir
log_print(f"Loaded last directory: {self.ini_directory}")
# Restore last-opened workspace
# Priority to a saved workspace file
if self.config.has_option('General', 'last_workspace_file'):
wsp = self.config.get('General', 'last_workspace_file')
if os.path.exists(wsp):
self.workspace_file_path = wsp
# Load it into the listbox
self.open_workspace()
elif self.config.has_option('General', 'last_workspace_files'):
files_str = self.config.get('General', 'last_workspace_files')
for full_path in files_str.split(','):
if os.path.exists(full_path):
name = os.path.basename(full_path)
if name not in self.workspace_ini_paths:
self.workspace_ini_paths[name] = full_path
self.ini_listbox.addItem(name)
else:
log_print("No configuration file found. Starting fresh.")
except Exception as e:
log_print(f"Failed to load last directory: {e}")
def _add_ini_paths_to_workspace(self, paths):
"""Add INI paths to the workspace list and return added/skipped/invalid paths."""
result = {"added": [], "skipped": [], "invalid": []}
if not paths:
return result
for path in paths:
if not path:
continue
if not path.lower().endswith(".ini") or not os.path.exists(path):
result["invalid"].append(path)
continue
name = os.path.basename(path)
if name not in self.workspace_ini_paths:
self.workspace_ini_paths[name] = path
self.ini_listbox.addItem(name)
result["added"].append(path)
else:
result["skipped"].append(path)
if self.workspace_file_path:
self.save_workspace()
self.save_workspace_state()
return result
[docs]
def add_ini_paths(self, paths):
"""Add provided INI file paths to the workspace list (no file dialog)."""
result = _add_ini_paths_to_workspace(self, paths)
if result["invalid"]:
msg = "Only existing .ini files can be added. Skipped:\n"
msg += "\n".join(result["invalid"])
QMessageBox.warning(self, "Unsupported Items", msg)
if result["skipped"]:
msg = "These files were skipped because a file with the same name is already in the workspace:\n"
msg += "\n".join(result["skipped"])
QMessageBox.warning(self, "Duplicate INI names", msg)
return result
[docs]
def add_generated_ini_paths(self, paths, show_warnings=True):
"""
Add importer-generated INI files to the workspace.
Returns a dict with added/skipped/invalid path lists so import dialogs can
report concise status without duplicating workspace mutation logic.
"""
result = _add_ini_paths_to_workspace(self, paths)
if show_warnings and result["invalid"]:
msg = "Generated INI files were not added because they could not be found:\n"
msg += "\n".join(os.path.basename(p) for p in result["invalid"])
QMessageBox.warning(self, "Generated INIs Not Added", msg)
if show_warnings and result["skipped"]:
msg = "These generated INI files were skipped because a file with the same name is already in the workspace:\n"
msg += "\n".join(os.path.basename(p) for p in result["skipped"])
QMessageBox.warning(self, "Duplicate INI names", msg)
return result
[docs]
def add_ini_files(self):
"""Open a dialog to add INI files to the workspace list."""
paths, _ = QFileDialog.getOpenFileNames(
self, "Select INI files", "", "INI Files (*.ini);;All Files (*)"
)
self.add_ini_paths(paths)
[docs]
def remove_ini_files(self):
"""Remove selected INI files from the workspace list."""
for item in self.ini_listbox.selectedItems():
name = item.text()
row = self.ini_listbox.row(item)
self.ini_listbox.takeItem(row)
self.workspace_ini_paths.pop(name, None)
# Auto-save workspace if already loaded
if self.workspace_file_path:
self.save_workspace()
self.save_workspace_state()
[docs]
def save_workspace(self):
"""Save the current workspace INI paths to a .cosedawsp file."""
if not self.workspace_file_path:
# Ask for destination
path, _ = QFileDialog.getSaveFileName(
self, "Save Workspace", "", "COSEDA Workspace (*.cosedawsp)"
)
if not path:
return
if not path.endswith(".cosedawsp"):
path += ".cosedawsp"
self.workspace_file_path = path
try:
with open(self.workspace_file_path, "w") as f:
# Save view options (colormap and log normalization)
try:
f.write(f"# Colormap={self.cmap_combo.currentText()}\n")
f.write(f"# LogNormalize={int(self.log_checkbox.isChecked())}\n")
try:
f.write(f"# ShowBeamCenter={int(self.show_point_checkbox.isChecked())}\n")
f.write(f"# ShowPeaks={int(self.show_peaks_checkbox.isChecked())}\n")
try:
f.write(f"# ShowResolutionRings={int(self.show_resolution_rings_checkbox.isChecked())}\n")
except Exception:
pass
if hasattr(self, "show_strong_peaks_checkbox"):
f.write(f"# ShowStrongPeaks={int(self.show_strong_peaks_checkbox.isChecked())}\n")
f.write(f"# Gamma={self.gamma_slider.value()/10:.1f}\n")
f.write(f"# Brightness={self.brightness_slider.value()/10:.1f}\n")
f.write(f"# Contrast={self.contrast_slider.value()/10:.1f}\n")
except Exception:
pass
except Exception:
pass
for full_path in self.workspace_ini_paths.values():
f.write(full_path + "\n")
# Update workspace label to show saved workspace name
name = os.path.basename(self.workspace_file_path)
self.workspace_label.setText(f"Workspace: {name}")
# QMessageBox.information(
# self, "Workspace Saved", f"Workspace saved to:\n{self.workspace_file_path}"
# )
except Exception as e:
QMessageBox.critical(
self, "Save Error", f"Failed to save workspace:\n{e}"
)
self.save_workspace_state()
[docs]
def open_workspace(self):
"""Load a saved workspace file (.cosedawsp) and populate the INI list."""
path, _ = QFileDialog.getOpenFileName(
self, "Open Workspace", "", "COSEDA Workspace (*.cosedawsp);;All Files (*)"
)
if not path:
return
self.workspace_file_path = path
try:
with open(path, "r") as f:
raw_lines = [line.rstrip("\n") for line in f if line.strip()]
# Separate settings (starting with '#') and INI paths
view_lines = [l for l in raw_lines if l.startswith("#")]
ini_paths = [
l for l in raw_lines
if l.strip() and not l.lstrip().startswith("#")
]
ini_paths, repaired, missing, ambiguous = _resolve_workspace_ini_paths(path, ini_paths)
# Block signals to prevent GUI callbacks during restore
self.cmap_combo.blockSignals(True)
self.log_checkbox.blockSignals(True)
self.show_point_checkbox.blockSignals(True)
self.show_peaks_checkbox.blockSignals(True)
if hasattr(self, 'show_resolution_rings_checkbox'):
self.show_resolution_rings_checkbox.blockSignals(True)
if hasattr(self, "show_strong_peaks_checkbox"):
self.show_strong_peaks_checkbox.blockSignals(True)
self.gamma_slider.blockSignals(True)
self.brightness_slider.blockSignals(True)
self.contrast_slider.blockSignals(True)
# Parse view options with fallback
for v in view_lines:
try:
# Strip leading '#' and any whitespace, then split into key and value
kv = v.lstrip("#").strip().split("=", 1)
key = kv[0].strip().lower()
val = kv[1].strip() if len(kv) > 1 else ""
if key == "colormap":
self.cmap_combo.setCurrentText(val)
elif key == "lognormalize":
self.log_checkbox.setChecked(bool(int(val)))
elif key == "showbeamcenter":
self.show_point_checkbox.setChecked(bool(int(val)))
elif key == "showpeaks":
self.show_peaks_checkbox.setChecked(bool(int(val)))
elif key == "showstrongpeaks" and hasattr(self, "show_strong_peaks_checkbox"):
self.show_strong_peaks_checkbox.setChecked(bool(int(val)))
elif key == "showresolutionrings" and hasattr(self, 'show_resolution_rings_checkbox'):
self.show_resolution_rings_checkbox.setChecked(bool(int(val)))
elif key == "gamma":
try:
fval = float(val)
self.gamma_slider.setValue(int(round(fval * 10)))
except Exception:
pass
elif key == "brightness":
try:
fval = float(val)
self.brightness_slider.setValue(int(round(fval * 10)))
except Exception:
pass
elif key == "contrast":
try:
fval = float(val)
self.contrast_slider.setValue(int(round(fval * 10)))
except Exception:
pass
except Exception:
pass
# Unblock signals after view settings restored
self.cmap_combo.blockSignals(False)
self.log_checkbox.blockSignals(False)
self.show_point_checkbox.blockSignals(False)
self.show_peaks_checkbox.blockSignals(False)
if hasattr(self, "show_strong_peaks_checkbox"):
self.show_strong_peaks_checkbox.blockSignals(False)
if hasattr(self, 'show_resolution_rings_checkbox'):
self.show_resolution_rings_checkbox.blockSignals(False)
self.gamma_slider.blockSignals(False)
self.brightness_slider.blockSignals(False)
self.contrast_slider.blockSignals(False)
if hasattr(self, "_sync_frame_order_combo_with_mode"):
self._sync_frame_order_combo_with_mode()
# Clear existing workspace
self.ini_listbox.clear()
self.workspace_ini_paths.clear()
# Load INI paths
duplicates = []
for full_path in ini_paths:
if os.path.exists(full_path):
name = os.path.basename(full_path)
if name not in self.workspace_ini_paths:
self.workspace_ini_paths[name] = full_path
self.ini_listbox.addItem(name)
elif self.workspace_ini_paths.get(name) != full_path:
duplicates.append(full_path)
if repaired:
self.save_workspace()
# Refresh preview to apply restored view settings
self.update_image()
# Update workspace label to show loaded workspace name
name = os.path.basename(path)
self.workspace_label.setText(f"Workspace: {name}")
QMessageBox.information(
self, "Workspace Loaded", f"Workspace loaded from:\n{path}"
)
if duplicates:
msg = (
"Workspace contains multiple INI files with the same filename. "
"Only the first occurrence was loaded:\n"
)
msg += "\n".join(duplicates)
QMessageBox.warning(self, "Duplicate INI names", msg)
_show_workspace_path_resolution_warning(self, repaired, missing, ambiguous)
except Exception as e:
QMessageBox.critical(
self, "Load Error", f"Failed to load workspace:\n{e}"
)
self.save_workspace_state()
[docs]
def new_workspace(self):
"""Clear the current workspace list and reset state."""
self.ini_listbox.clear()
self.workspace_ini_paths.clear()
self.workspace_file_path = None
self.workspace_label.setText("Workspace: -")
self.save_workspace_state()
[docs]
def load_last_workspace_state(self):
"""Restore workspace state from configuration (named file or INI list)."""
try:
if os.path.exists(self.config_file_path):
self.config.read(self.config_file_path)
# Try named workspace file first
if self.config.has_option('General', 'last_workspace_file'):
wsp = self.config.get('General', 'last_workspace_file')
if os.path.exists(wsp):
self.workspace_file_path = wsp
# Read all lines, including comments
with open(wsp, 'r') as f:
raw_lines = [line.rstrip("\n") for line in f if line.strip()]
# Separate view settings and INI paths
view_lines = [l for l in raw_lines if l.lstrip().startswith("#")]
ini_paths = [l for l in raw_lines if not l.lstrip().startswith("#")]
ini_paths, repaired, missing, ambiguous = _resolve_workspace_ini_paths(wsp, ini_paths)
# Block signals to prevent GUI callbacks during restore
self.cmap_combo.blockSignals(True)
self.log_checkbox.blockSignals(True)
self.show_point_checkbox.blockSignals(True)
self.show_peaks_checkbox.blockSignals(True)
if hasattr(self, "show_strong_peaks_checkbox"):
self.show_strong_peaks_checkbox.blockSignals(True)
self.gamma_slider.blockSignals(True)
self.brightness_slider.blockSignals(True)
self.contrast_slider.blockSignals(True)
# Restore view options
for v in view_lines:
try:
kv = v.lstrip("#").strip().split("=", 1)
key = kv[0].strip().lower()
val = kv[1].strip() if len(kv) > 1 else ""
if key == "colormap":
self.cmap_combo.setCurrentText(val)
elif key == "lognormalize":
self.log_checkbox.setChecked(bool(int(val)))
elif key == "showbeamcenter":
self.show_point_checkbox.setChecked(bool(int(val)))
elif key == "showpeaks":
self.show_peaks_checkbox.setChecked(bool(int(val)))
elif key == "showstrongpeaks" and hasattr(self, "show_strong_peaks_checkbox"):
self.show_strong_peaks_checkbox.setChecked(bool(int(val)))
elif key == "gamma":
try:
fval = float(val)
self.gamma_slider.setValue(int(round(fval * 10)))
except Exception:
pass
elif key == "brightness":
try:
fval = float(val)
self.brightness_slider.setValue(int(round(fval * 10)))
except Exception:
pass
elif key == "contrast":
try:
fval = float(val)
self.contrast_slider.setValue(int(round(fval * 10)))
except Exception:
pass
except Exception:
pass
self.cmap_combo.blockSignals(False)
self.log_checkbox.blockSignals(False)
self.show_point_checkbox.blockSignals(False)
self.show_peaks_checkbox.blockSignals(False)
if hasattr(self, "show_strong_peaks_checkbox"):
self.show_strong_peaks_checkbox.blockSignals(False)
self.gamma_slider.blockSignals(False)
self.brightness_slider.blockSignals(False)
self.contrast_slider.blockSignals(False)
if hasattr(self, "_sync_frame_order_combo_with_mode"):
self._sync_frame_order_combo_with_mode()
# Populate INI list
duplicates = []
for full_path in ini_paths:
if os.path.exists(full_path):
name = os.path.basename(full_path)
if name not in self.workspace_ini_paths:
self.workspace_ini_paths[name] = full_path
self.ini_listbox.addItem(name)
elif self.workspace_ini_paths.get(name) != full_path:
duplicates.append(full_path)
if repaired:
self.save_workspace()
self.workspace_label.setText(f"Workspace: {os.path.basename(wsp)}")
# Refresh preview to apply restored view settings
self.update_image()
if duplicates:
msg = (
"Workspace contains multiple INI files with the same filename. "
"Only the first occurrence was loaded:\n"
)
msg += "\n".join(duplicates)
QMessageBox.warning(self, "Duplicate INI names", msg)
_show_workspace_path_resolution_warning(self, repaired, missing, ambiguous)
# Otherwise fallback to list of files
elif self.config.has_option('General', 'last_workspace_files'):
files_str = self.config.get('General', 'last_workspace_files')
for full_path in files_str.split(','):
if os.path.exists(full_path):
name = os.path.basename(full_path)
if name not in self.workspace_ini_paths:
self.workspace_ini_paths[name] = full_path
self.ini_listbox.addItem(name)
except Exception as e:
log_print(f"Failed to load workspace state: {e}")
self.save_workspace_state()