import os
import re
from PyQt6.QtCore import Qt
from PyQt6.QtWidgets import (
QCheckBox,
QComboBox,
QDoubleSpinBox,
QFileDialog,
QFormLayout,
QGroupBox,
QLabel,
QLineEdit,
QMessageBox,
QSpinBox,
QStackedWidget,
QTextEdit,
QVBoxLayout,
QWidget,
)
from coseda.logging_utils import log_print
from coseda.nexus.process import write_nxprocess_indexing
[docs]
class IndexSettingsMixin:
MAX_INDEXER_THREADS = 1
def _init_index_integrate_tab(self):
"""Build the Index & Integrate tab with key CrystFEL parameters."""
from PyQt6.QtGui import QDoubleValidator
self.index_tab = QWidget()
layout = QVBoxLayout()
form_layout = QFormLayout()
form_layout.setLabelAlignment(Qt.AlignmentFlag.AlignLeft)
form_layout.setFormAlignment(Qt.AlignmentFlag.AlignLeft | Qt.AlignmentFlag.AlignTop)
form_layout.setFieldGrowthPolicy(QFormLayout.FieldGrowthPolicy.AllNonFixedFieldsGrow)
self.worker_threads_spin = QSpinBox()
self.worker_threads_spin.setMinimum(1)
self.worker_threads_spin.setMaximum(128)
self.worker_threads_spin.setValue(12)
self.worker_threads_spin.setToolTip("Total indexamajig worker threads (-j flag).")
form_layout.addRow("Worker threads:", self.worker_threads_spin)
# Restore push resolution and integration radius rows, with validators/placeholders
self.push_res_edit = QLineEdit("0.5")
# Ensure field is editable and accepts floating point numbers (up to 6 decimals)
self.push_res_edit.setReadOnly(False)
dv = QDoubleValidator(0.0, 1e12, 6, self)
dv.setNotation(QDoubleValidator.Notation.StandardNotation)
self.push_res_edit.setValidator(dv)
self.push_res_edit.setPlaceholderText("0.5")
self.push_res_edit.setToolTip("Integrate beyond apparent resolution of each pattern (1/nm)")
form_layout.addRow("Push resolution:", self.push_res_edit)
self.int_radius_edit = QLineEdit("4,5,8")
self.int_radius_edit.setPlaceholderText("e.g. 4,5,8")
self.int_radius_edit.setToolTip("Outer radii for inner, middle and outer intergration rings (px)")
form_layout.addRow("Integration radius:", self.int_radius_edit)
self.min_peaks_spin = QSpinBox()
self.min_peaks_spin.setValue(25)
form_layout.addRow("Min peaks:", self.min_peaks_spin)
self.tolerance_edit = QLineEdit("5,5,5,1.5")
self.tolerance_edit.setToolTip("Tolerance for unit cell comparison (%), default 5,5,5,1.5")
form_layout.addRow("Tolerance:", self.tolerance_edit)
# Insert new XGandalf tolerance field after tolerance
self.xgandalf_tolerance_edit = QLineEdit("0.02")
self.xgandalf_tolerance_edit.setToolTip(
"Relative tolerance of lattice vectors, default 0.02 (2%)"
)
form_layout.addRow("XGandalf tolerance:", self.xgandalf_tolerance_edit)
self.sampling_pitch_spin = QSpinBox()
self.sampling_pitch_spin.setValue(5)
self.sampling_pitch_spin.setToolTip(
"XGANDALF sampling density of reciprocal space, 0-7 (7 is dense)"
)
form_layout.addRow("Sampling pitch:", self.sampling_pitch_spin)
self.use_cell_cb = QCheckBox("Use cell as prior")
self.use_cell_cb.setChecked(True)
self.use_cell_cb.setToolTip(
"Pass the cell file to indexamajig via -p. XGandalf uses it as a prior for the "
"gradient descent, making the lattice vector length range irrelevant. "
"Uncheck to run a free-lattice search (useful for unknown unit cells)."
)
form_layout.addRow(self.use_cell_cb)
self.use_cell_cb.stateChanged.connect(self._on_use_cell_toggled)
self.use_cell_cb.stateChanged.connect(self._update_command_label)
self.min_lat_vec_len_spin = QSpinBox()
self.min_lat_vec_len_spin.setRange(0, 10000)
self.min_lat_vec_len_spin.setValue(3)
self.min_lat_vec_len_spin.setToolTip(
"Minimum possible lattice vector length (Å). Only used when no prior cell is provided."
)
self.min_lat_vec_len_spin.setEnabled(False)
form_layout.addRow("Min lattice vector length:", self.min_lat_vec_len_spin)
self.max_lat_vec_len_spin = QSpinBox()
self.max_lat_vec_len_spin.setRange(0, 10000)
self.max_lat_vec_len_spin.setValue(30)
self.max_lat_vec_len_spin.setToolTip(
"Maximum possible lattice vector length (Å). Only used when no prior cell is provided."
)
self.max_lat_vec_len_spin.setEnabled(False)
form_layout.addRow("Max lattice vector length:", self.max_lat_vec_len_spin)
self.grad_desc_iterations_spin = QSpinBox()
self.grad_desc_iterations_spin.setValue(2)
self.grad_desc_iterations_spin.setToolTip(
"Number of gradient-descent iterations for XGANDALF 0–5 (5 is many)"
)
form_layout.addRow("Gradient descent iterations:", self.grad_desc_iterations_spin)
self.fix_profile_radius_cb = QCheckBox("Use fixed profile radius")
self.fix_profile_radius_cb.setChecked(False)
self.fix_profile_radius_cb.setToolTip(
"Write --fix-profile-radius for indexamajig. Leave this off unless a known, validated radius is required."
)
form_layout.addRow(self.fix_profile_radius_cb)
self.fix_profile_radius_edit = QLineEdit("")
self.fix_profile_radius_edit.setEnabled(self.fix_profile_radius_cb.isChecked())
form_layout.addRow("Fixed profile radius:", self.fix_profile_radius_edit)
self.fix_profile_radius_cb.stateChanged.connect(self._on_fix_profile_radius_toggle)
# Connect to update command label
self.fix_profile_radius_edit.textChanged.connect(self._update_command_label)
# Add checkboxes for optional flags
self.no_non_hits_cb = QCheckBox("No non-hits in stream")
self.no_non_hits_cb.setChecked(True)
self.no_non_hits_cb.setToolTip("Do not include non-indexed patterns in the output stream. Keeps output smaller.")
form_layout.addRow(self.no_non_hits_cb)
self.no_non_hits_cb.stateChanged.connect(self._update_command_label)
self.no_revalidate_cb = QCheckBox("No revalidate")
self.no_revalidate_cb.setChecked(True)
self.no_revalidate_cb.setToolTip("Skip filtering peaks too close to detector edge, doublepeaks or are saturated.")
form_layout.addRow(self.no_revalidate_cb)
self.no_revalidate_cb.stateChanged.connect(self._update_command_label)
self.no_check_peaks_cb = QCheckBox("No check peaks")
self.no_check_peaks_cb.setChecked(True)
self.no_check_peaks_cb.setToolTip("Disable peak sanity checks when reading peaks.")
form_layout.addRow(self.no_check_peaks_cb)
self.no_check_peaks_cb.stateChanged.connect(self._update_command_label)
self.no_half_pixel_shift_cb = QCheckBox("No half-pixel shift")
self.no_half_pixel_shift_cb.setChecked(True)
self.no_half_pixel_shift_cb.setToolTip("Disable half-pixel offset when reading images.")
form_layout.addRow(self.no_half_pixel_shift_cb)
self.no_half_pixel_shift_cb.stateChanged.connect(self._update_command_label)
self.no_retry_cb = QCheckBox("No retry")
self.no_retry_cb.setChecked(True)
self.no_retry_cb.setToolTip("Skip retrying with 10% of weakest reflections removed.")
form_layout.addRow(self.no_retry_cb)
self.no_retry_cb.stateChanged.connect(self._update_command_label)
self.no_check_cell_cb = QCheckBox("No check cell")
self.no_check_cell_cb.setChecked(True)
self.no_check_cell_cb.setToolTip("Don't check cell parmameters against reference cell.")
form_layout.addRow(self.no_check_cell_cb)
self.no_check_cell_cb.stateChanged.connect(self._update_command_label)
self.no_refine_cb = QCheckBox("No refine")
self.no_refine_cb.setChecked(True)
self.no_refine_cb.setToolTip("Don't refine the unit cell after indexing.")
form_layout.addRow(self.no_refine_cb)
self.no_refine_cb.stateChanged.connect(self._update_command_label)
self.overpredict_cb = QCheckBox("Overpredict")
self.overpredict_cb.setChecked(True)
self.overpredict_cb.setToolTip("Predict reflections beyond the observed peaks.")
form_layout.addRow(self.overpredict_cb)
self.overpredict_cb.stateChanged.connect(self._update_command_label)
layout.addLayout(form_layout)
self.index_tab.setLayout(layout)
self.settings_tab = QWidget()
settings_layout = QVBoxLayout(self.settings_tab)
self.settings_stack = QStackedWidget()
self.settings_stack.addWidget(self.index_tab)
self.ici_settings_tab = QWidget()
self.settings_stack.addWidget(self.ici_settings_tab)
settings_layout.addWidget(self.settings_stack)
self.tabs.addTab(self.settings_tab, "Settings")
# Connect signals to update the command label when values change
self.worker_threads_spin.valueChanged.connect(self._update_command_label)
self.push_res_edit.textChanged.connect(self._update_command_label)
self.int_radius_edit.textChanged.connect(self._update_command_label)
self.min_peaks_spin.valueChanged.connect(self._update_command_label)
self.tolerance_edit.textChanged.connect(self._update_command_label)
self.xgandalf_tolerance_edit.textChanged.connect(self._update_command_label)
self.sampling_pitch_spin.valueChanged.connect(self._update_command_label)
self.min_lat_vec_len_spin.valueChanged.connect(self._update_command_label)
self.max_lat_vec_len_spin.valueChanged.connect(self._update_command_label)
self.grad_desc_iterations_spin.valueChanged.connect(self._update_command_label)
# Already connected: self.fix_profile_radius_edit.textChanged.connect(self._update_command_label)
def _on_use_cell_toggled(self):
using_cell = self.use_cell_cb.isChecked()
self.min_lat_vec_len_spin.setEnabled(not using_cell)
self.max_lat_vec_len_spin.setEnabled(not using_cell)
def _on_fix_profile_radius_toggle(self):
enabled = self.fix_profile_radius_cb.isChecked()
self.fix_profile_radius_edit.setEnabled(enabled)
self._update_command_label()
def _snapshot_index_settings(self):
"""Collect current Index & Integrate settings into a serializable dict."""
worker_threads = self.worker_threads_spin.value()
return {
'worker_threads': worker_threads,
'max_indexer_threads': self.MAX_INDEXER_THREADS,
'threads': worker_threads,
'push_res': self.push_res_edit.text().strip(),
'int_radius': self.int_radius_edit.text().strip(),
'min_peaks': self.min_peaks_spin.value(),
'tolerance': self.tolerance_edit.text().strip(),
'xg_tolerance': self.xgandalf_tolerance_edit.text().strip(),
'sampling_pitch': self.sampling_pitch_spin.value(),
'min_lat_vec_len': int(self.min_lat_vec_len_spin.value()),
'max_lat_vec_len': int(self.max_lat_vec_len_spin.value()),
'grad_desc_iterations': self.grad_desc_iterations_spin.value(),
'use_cell': bool(self.use_cell_cb.isChecked()),
'fix_profile_radius': self.fix_profile_radius_edit.text().strip(),
'fix_profile_radius_enabled': bool(self.fix_profile_radius_cb.isChecked()),
'no_non_hits': bool(self.no_non_hits_cb.isChecked()),
'no_revalidate': bool(self.no_revalidate_cb.isChecked()),
'no_check_peaks': bool(self.no_check_peaks_cb.isChecked()),
'no_half_pixel_shift': bool(self.no_half_pixel_shift_cb.isChecked()),
'no_retry': bool(self.no_retry_cb.isChecked()),
'no_check_cell': bool(self.no_check_cell_cb.isChecked()),
'no_refine': bool(self.no_refine_cb.isChecked()),
'overpredict': bool(self.overpredict_cb.isChecked()),
}
def _extract_h5_paths_from_list(self, list_path: str):
if not list_path or not os.path.exists(list_path):
return []
base_dir = os.path.dirname(list_path)
paths = []
try:
with open(list_path, "r") as handle:
for line in handle:
stripped = line.strip()
if not stripped or stripped.startswith("#"):
continue
token = stripped.split()[0].strip("\"'")
if token.startswith("file://"):
token = token[7:]
if not os.path.isabs(token):
token = os.path.abspath(os.path.join(base_dir, token))
paths.append(token)
except Exception:
return []
unique = []
seen = set()
for path in paths:
if path in seen:
continue
seen.add(path)
unique.append(path)
return unique
def _write_indexing_nxprocess(self, cmd: str) -> None:
h5_paths = self._extract_h5_paths_from_list(getattr(self, "list_filepath", None))
if not h5_paths and getattr(self, "h5_path", None):
h5_paths = [self.h5_path]
if not h5_paths:
return
params = self._snapshot_index_settings()
params.update({
"command": cmd,
"geom_file": getattr(self, "geom_filepath", ""),
"cell_file": getattr(self, "cell_filepath", ""),
"list_file": getattr(self, "list_filepath", ""),
"stream_file": getattr(self, "stream_filepath", ""),
"indexing": "xgandalf",
"integration": "rings",
})
for h5_path in h5_paths:
if not h5_path or not os.path.exists(h5_path):
continue
try:
import h5py
with h5py.File(h5_path, "r+") as h5file:
write_nxprocess_indexing(
h5file,
program="cosedaUI.indexintegrate_window",
method="crystfel.indexamajig",
parameters=params,
input_path=getattr(self, "list_filepath", None),
output_path=getattr(self, "stream_filepath", None),
)
except Exception as exc:
log_print(f"DEBUG: Failed to write NXprocess indexing for {h5_path}: {exc}")
def _apply_index_settings(self, cfg):
"""Apply a previously saved settings dict to UI widgets (signals suppressed)."""
# Block signals during bulk apply to avoid redundant command rebuilds
self.worker_threads_spin.blockSignals(True)
self.push_res_edit.blockSignals(True)
self.int_radius_edit.blockSignals(True)
self.min_peaks_spin.blockSignals(True)
self.tolerance_edit.blockSignals(True)
self.xgandalf_tolerance_edit.blockSignals(True)
self.sampling_pitch_spin.blockSignals(True)
self.min_lat_vec_len_spin.blockSignals(True)
self.max_lat_vec_len_spin.blockSignals(True)
self.grad_desc_iterations_spin.blockSignals(True)
self.fix_profile_radius_edit.blockSignals(True)
self.fix_profile_radius_cb.blockSignals(True)
self.use_cell_cb.blockSignals(True)
self.no_non_hits_cb.blockSignals(True)
self.no_revalidate_cb.blockSignals(True)
self.no_check_peaks_cb.blockSignals(True)
self.no_half_pixel_shift_cb.blockSignals(True)
self.no_retry_cb.blockSignals(True)
self.no_check_cell_cb.blockSignals(True)
self.no_refine_cb.blockSignals(True)
self.overpredict_cb.blockSignals(True)
try:
if 'worker_threads' in cfg:
self.worker_threads_spin.setValue(int(cfg['worker_threads']))
elif 'threads' in cfg:
self.worker_threads_spin.setValue(int(cfg['threads']))
if 'push_res' in cfg: self.push_res_edit.setText(str(cfg['push_res']))
if 'int_radius' in cfg: self.int_radius_edit.setText(str(cfg['int_radius']))
if 'min_peaks' in cfg: self.min_peaks_spin.setValue(int(cfg['min_peaks']))
if 'tolerance' in cfg: self.tolerance_edit.setText(str(cfg['tolerance']))
if 'xg_tolerance' in cfg: self.xgandalf_tolerance_edit.setText(str(cfg['xg_tolerance']))
if 'sampling_pitch' in cfg: self.sampling_pitch_spin.setValue(int(cfg['sampling_pitch']))
if 'min_lat_vec_len' in cfg: self.min_lat_vec_len_spin.setValue(int(cfg['min_lat_vec_len']))
if 'max_lat_vec_len' in cfg: self.max_lat_vec_len_spin.setValue(int(cfg['max_lat_vec_len']))
if 'grad_desc_iterations' in cfg: self.grad_desc_iterations_spin.setValue(int(cfg['grad_desc_iterations']))
if 'fix_profile_radius' in cfg: self.fix_profile_radius_edit.setText(str(cfg['fix_profile_radius']))
if 'fix_profile_radius_enabled' in cfg:
val = cfg['fix_profile_radius_enabled']
if isinstance(val, bool):
enabled = val
else:
enabled = str(val).strip().lower() in ("1", "true", "yes", "y", "on")
self.fix_profile_radius_cb.setChecked(enabled)
else:
self.fix_profile_radius_cb.setChecked(False)
if 'use_cell' in cfg:
self.use_cell_cb.setChecked(bool(cfg['use_cell']))
if 'no_non_hits' in cfg: self.no_non_hits_cb.setChecked(bool(cfg['no_non_hits']))
if 'no_revalidate' in cfg: self.no_revalidate_cb.setChecked(bool(cfg['no_revalidate']))
if 'no_check_peaks' in cfg: self.no_check_peaks_cb.setChecked(bool(cfg['no_check_peaks']))
if 'no_half_pixel_shift' in cfg: self.no_half_pixel_shift_cb.setChecked(bool(cfg['no_half_pixel_shift']))
if 'no_retry' in cfg: self.no_retry_cb.setChecked(bool(cfg['no_retry']))
if 'no_check_cell' in cfg: self.no_check_cell_cb.setChecked(bool(cfg['no_check_cell']))
if 'no_refine' in cfg: self.no_refine_cb.setChecked(bool(cfg['no_refine']))
if 'overpredict' in cfg: self.overpredict_cb.setChecked(bool(cfg['overpredict']))
finally:
# Unblock signals
self.worker_threads_spin.blockSignals(False)
self.push_res_edit.blockSignals(False)
self.int_radius_edit.blockSignals(False)
self.min_peaks_spin.blockSignals(False)
self.tolerance_edit.blockSignals(False)
self.xgandalf_tolerance_edit.blockSignals(False)
self.sampling_pitch_spin.blockSignals(False)
self.min_lat_vec_len_spin.blockSignals(False)
self.max_lat_vec_len_spin.blockSignals(False)
self.grad_desc_iterations_spin.blockSignals(False)
self.fix_profile_radius_edit.blockSignals(False)
self.fix_profile_radius_cb.blockSignals(False)
self.use_cell_cb.blockSignals(False)
self.no_non_hits_cb.blockSignals(False)
self.no_revalidate_cb.blockSignals(False)
self.no_check_peaks_cb.blockSignals(False)
self.no_half_pixel_shift_cb.blockSignals(False)
self.no_retry_cb.blockSignals(False)
self.no_check_cell_cb.blockSignals(False)
self.no_refine_cb.blockSignals(False)
self.overpredict_cb.blockSignals(False)
self.fix_profile_radius_edit.setEnabled(self.fix_profile_radius_cb.isChecked())
self._on_use_cell_toggled()
self._last_index_settings = cfg
self._update_command_label()
def _run_group_key(self):
"""Return the run folder name (e.g., 'indexingintegration_YYYYMMDD_HHMMSS') if current run_dir is set."""
if not getattr(self, 'run_dir', None):
return None
return os.path.basename(self.run_dir.rstrip(os.sep))
def _create_fresh_run_for_current_ini(
self,
group: str | None = None,
seed_input_lst: bool = True,
settings: dict | None = None,
) -> str | None:
"""
Create a brand-new timestamped run directory for the CURRENT INI,
apply the current GUI settings to that run (cell/geom/settings),
and point the UI to it. Returns the run_dir or None on failure.
"""
import os, json
from datetime import datetime as _dt
ini = getattr(self, 'ini_path', None)
if not ini or not os.path.exists(ini):
return None
if not group:
group = f"indexingintegration_{_dt.now().strftime('%Y%m%d_%H%M%S')}"
run_base = self._resolve_run_root(ini)
rd = os.path.join(run_base, group)
try:
os.makedirs(rd, exist_ok=True)
except Exception:
return None
cfg = dict(settings) if settings is not None else self._snapshot_index_settings()
# Point UI to this run (sets paths like cell/geom/list/stream) without
# letting a same-second/reused run directory overwrite pending edits.
self._apply_run_dir(rd, load_index_settings=False)
self._apply_index_settings(cfg)
# Persist current GUI settings (no broadcast)
try:
with open(os.path.join(rd, 'index_settings.json'), 'w') as jf:
json.dump(cfg, jf, indent=2)
except Exception:
pass
# Seed input.lst from the current INI’s detected H5/CXI/EMD if possible
if seed_input_lst:
try:
lst = os.path.join(rd, 'input.lst')
if not os.path.exists(lst):
h5 = self._find_h5_path_from_ini(ini)
if h5:
with open(lst, 'w') as f:
f.write(os.path.abspath(h5))
except Exception:
pass
# Ensure files reflect current UI values
if not self._save_cell_file(show_errors=True):
return None
if not self._save_geom_file(show_errors=True):
return None
self._update_command_label()
return rd
def _broadcast_index_settings(self, cfg):
try:
import json, os, glob
group = self._run_group_key()
if not group:
return
ini_paths = self._get_workspace_ini_paths()
for ini in ini_paths:
try:
base_dir = os.path.dirname(ini)
run_base = self._resolve_run_root(ini)
candidates = sorted(
set(glob.glob(os.path.join(run_base, group))) |
set(glob.glob(os.path.join(base_dir, group)))
)
for candidate in candidates:
if os.path.isdir(candidate):
with open(os.path.join(candidate, 'index_settings.json'), 'w') as jf:
json.dump(cfg, jf, indent=2)
except Exception:
continue
except Exception:
pass
def _load_index_settings_from_run(self, run_dir=None):
"""Load settings JSON from a run dir (if present) and apply to the UI. Returns True if applied."""
import os, json
rd = run_dir or getattr(self, 'run_dir', None)
if not rd:
return False
js = os.path.join(rd, 'index_settings.json')
if not os.path.exists(js):
return False
try:
with open(js, 'r') as f:
cfg = json.load(f)
if isinstance(cfg, dict):
self._apply_index_settings(cfg)
# Remember for sessions where another file has no JSON yet
self._last_index_settings = cfg
return True
except Exception:
return False
return False
def _save_index_settings(self):
from coseda.initialize import write_xgandalfsettings
try:
# Ensure a run exists to anchor settings/paths
if not getattr(self, 'run_dir', None) or not os.path.isdir(self.run_dir):
self._create_new_run_for_current_file()
if not getattr(self, 'run_dir', None) or not os.path.isdir(self.run_dir):
raise RuntimeError('Failed to create/select a run directory for saving settings.')
from configparser import ConfigParser
# Save updated geom and cell file paths back to ini if changed
# (No longer using QLineEdit for these fields, so skip updating from edits)
# pass separate XGandalf tolerance if supported by write_xgandalfsettings
try:
write_xgandalfsettings(
self.ini_path,
tolerance=self.tolerance_edit.text(),
xgandalf_tolerance=self.xgandalf_tolerance_edit.text(),
sampling_pitch=self.sampling_pitch_spin.value(),
min_lattice_vector_length=self.min_lat_vec_len_spin.value(),
max_lattice_vector_length=self.max_lat_vec_len_spin.value(),
grad_desc_iterations=self.grad_desc_iterations_spin.value(),
tolerance_5d=0.2,
fix_profile_radius=self.fix_profile_radius_cb.isChecked()
)
except TypeError:
# fallback if original function does not accept xgandalf_tolerance
write_xgandalfsettings(
self.ini_path,
tolerance=self.tolerance_edit.text(),
sampling_pitch=self.sampling_pitch_spin.value(),
min_lattice_vector_length=self.min_lat_vec_len_spin.value(),
max_lattice_vector_length=self.max_lat_vec_len_spin.value(),
grad_desc_iterations=self.grad_desc_iterations_spin.value(),
tolerance_5d=0.2,
fix_profile_radius=self.fix_profile_radius_cb.isChecked()
)
# Snapshot and broadcast settings to runs with the same timestamped name (batch group)
cfg = self._snapshot_index_settings()
self._last_index_settings = cfg
self._broadcast_index_settings(cfg)
QMessageBox.information(self, "Saved", "Index & Integrate settings saved.")
except Exception as e:
QMessageBox.critical(self, "Error", str(e))
# --- Cell File Editor Tab ---