import os
import re
import signal
import subprocess
import time
import configparser
import json
from PyQt6.QtCore import Qt, QTimer
from PyQt6.QtWidgets import (
QFormLayout,
QHBoxLayout,
QLabel,
QLineEdit,
QMessageBox,
QFileDialog,
QPlainTextEdit,
QPushButton,
QStackedWidget,
QTextEdit,
QTreeWidgetItem,
QVBoxLayout,
QWidget,
QProgressBar,
)
from coseda.logging_utils import log_print
from coseda.io import config_to_paths
from coseda.ici.run_contract import (
canonical_stream_path,
resolve_final_stream_path,
update_final_stream_alias,
)
from coseda.crystfel_cell import (
crystfel_unique_axis_allowed,
normalize_crystfel_lattice_type,
)
from coseda.nexus.paths import get_mask_dataset
[docs]
class RunControlMixin:
def _is_ici_run_dir(self, run_dir: str) -> bool:
"""Return True when a run directory follows the ICI run layout."""
if not run_dir:
return False
name = os.path.basename(os.path.abspath(run_dir).rstrip(os.sep))
return (
name.startswith("ici_")
or os.path.exists(os.path.join(run_dir, "ici_settings.json"))
or os.path.isdir(os.path.join(run_dir, "run_000"))
)
def _resolve_run_stream_path(self, run_dir: str) -> str:
"""Resolve the stream path for simple indexamajig and ICI runs."""
if self._is_ici_run_dir(run_dir):
alias = update_final_stream_alias(run_dir)
if alias and os.path.exists(alias):
return alias
resolved = resolve_final_stream_path(run_dir)
if resolved:
return resolved
return canonical_stream_path(run_dir)
run_folder_name = os.path.basename(run_dir)
return os.path.join(run_dir, f'{run_folder_name}.stream')
def _load_ici_manifest(self, run_dir: str) -> dict:
path = os.path.join(run_dir, "ici_run_manifest.json")
if not os.path.isfile(path):
return {}
try:
with open(path, "r", encoding="utf-8") as handle:
data = json.load(handle)
return data if isinstance(data, dict) else {}
except Exception:
return {}
def _add_run_tree_child(self, parent_item, label: str, tooltip: str = ""):
child = QTreeWidgetItem([label])
if tooltip:
child.setToolTip(0, tooltip)
parent_item.addChild(child)
return child
def _populate_ici_run_tree_item(self, run_item, run_dir: str):
manifest = self._load_ici_manifest(run_dir)
status = manifest.get("status")
latest_run = manifest.get("latest_run")
summary = manifest.get("summary", {}) if isinstance(manifest.get("summary"), dict) else {}
detail = []
if status:
detail.append(f"status: {status}")
if latest_run is not None:
detail.append(f"latest run_{int(latest_run):03d}")
done = summary.get("done_events")
total = summary.get("total_events")
if done is not None and total:
detail.append(f"done events: {done}/{total}")
if detail:
self._add_run_tree_child(run_item, " | ".join(detail), run_dir)
for name in ("done.stream", "early_break.stream", os.path.basename(canonical_stream_path(run_dir))):
path = os.path.join(run_dir, name)
if os.path.exists(path):
self._add_run_tree_child(run_item, name, path)
for name in ("image_run_log.csv", "orchestrator.log", "ici_run_manifest.json"):
path = os.path.join(run_dir, name)
if os.path.exists(path):
self._add_run_tree_child(run_item, name, path)
try:
run_names = sorted(
name for name in os.listdir(run_dir)
if re.fullmatch(r"run_\d{3}", name) and os.path.isdir(os.path.join(run_dir, name))
)
except OSError:
run_names = []
for name in run_names:
self._add_run_tree_child(run_item, name, os.path.join(run_dir, name))
def _top_level_ini_item_for_tree_item(self, item):
current = item
while current is not None and current.parent() is not None:
current = current.parent()
return current
def _run_tree_item_for_tree_item(self, item):
current = item
parent = current.parent() if current is not None else None
while parent is not None and parent.parent() is not None:
current = parent
parent = current.parent()
return current if parent is not None else None
def _resolve_h5_for_current_run(self):
paths = []
seen = set()
def _add(path):
if not path:
return
ap = os.path.abspath(path)
if ap in seen:
return
seen.add(ap)
paths.append(ap)
list_path = getattr(self, "list_filepath", None)
if list_path and os.path.exists(list_path):
list_dir = os.path.dirname(os.path.abspath(list_path))
try:
with open(list_path, "r") as handle:
for raw in handle:
line = raw.strip()
if not line or line.startswith("#"):
continue
candidate = line.split()[0]
if not os.path.isabs(candidate):
candidate = os.path.normpath(os.path.join(list_dir, candidate))
if os.path.exists(candidate):
_add(candidate)
except Exception:
pass
if paths:
return paths
h5 = getattr(self, "h5_path", None)
if h5 and os.path.exists(h5):
_add(h5)
for inferred in self._resolve_h5_paths_from_ini(getattr(self, "ini_path", None)):
if inferred and os.path.exists(inferred):
_add(inferred)
return paths
def _resolve_h5_paths_from_ini(self, ini_path: str):
"""Resolve HDF5/CXI inputs declared by an INI before falling back to naming guesses."""
if not ini_path:
return []
paths = []
seen = set()
def _add(path):
if not path:
return
path = os.path.expanduser(os.path.expandvars(str(path).strip()))
if not path:
return
ap = os.path.abspath(path)
if ap in seen:
return
seen.add(ap)
paths.append(ap)
try:
ini_path = os.path.abspath(ini_path)
ini_dir = os.path.dirname(ini_path)
cfg = configparser.ConfigParser()
cfg.read(ini_path)
if cfg.has_section("Paths"):
h5_rel = cfg.get("Paths", "h5file", fallback="").strip()
out_rel = cfg.get("Paths", "outputfolder", fallback="").strip()
if h5_rel:
if os.path.isabs(h5_rel):
_add(h5_rel)
else:
_add(os.path.join(ini_dir, h5_rel))
if out_rel:
_add(os.path.join(ini_dir, out_rel, h5_rel))
try:
_, _, _, _, _, h5file_path = config_to_paths(ini_path)
_add(h5file_path)
except Exception:
pass
stem = os.path.splitext(ini_path)[0]
_add(stem + ".h5")
_add(stem + ".hdf5")
_add(stem + ".cxi")
except Exception as exc:
log_print(f"DEBUG: INI H5 resolve failed for {ini_path}: {exc}")
return [path for path in paths if os.path.isfile(path)]
def _h5_has_required_mask(self, h5_path: str):
if not h5_path:
return False, "No HDF5 file resolved from current INI."
if not os.path.exists(h5_path):
return False, f"HDF5 file not found: {h5_path}"
try:
import h5py
with h5py.File(h5_path, "r") as f:
ds = get_mask_dataset(f)
if ds is None:
return False, (
f"No mask dataset found in {h5_path} "
"('/entry/instrument/detector/mask' or '/mask')."
)
if ds.ndim != 2 or ds.shape[0] <= 0 or ds.shape[1] <= 0:
return False, f"Mask dataset in {h5_path} is not a valid 2D image mask."
except Exception as exc:
return False, f"Failed to read mask dataset from {h5_path}: {exc}"
return True, ""
@staticmethod
def _geom_has_mask_entries(path: str) -> bool:
if not path or not os.path.exists(path):
return False
try:
with open(path, "r") as handle:
text = handle.read()
except Exception:
return False
return re.search(r"^\s*[^\s#;]+/mask\s*=", text, re.MULTILINE) is not None
def _init_start_indexing_tab(self):
"""Build the run tab that shows inputs, options, and the constructed command."""
self.start_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)
# Use QLineEdit for horizontally scrollable, read-only paths
self.geom_path_edit = QLineEdit(self.geom_filepath or "")
self.geom_path_edit.setReadOnly(True)
form_layout.addRow("Geometry file:", self.geom_path_edit)
self.cell_path_edit = QLineEdit(self.cell_filepath or "")
self.cell_path_edit.setReadOnly(True)
form_layout.addRow("Cell file:", self.cell_path_edit)
self.list_path_edit = QLineEdit(self.list_filepath or "")
self.list_path_edit.setReadOnly(True)
form_layout.addRow("List file:", self.list_path_edit)
self.stream_path_edit = QLineEdit(self.stream_filepath or "")
self.stream_path_edit.setReadOnly(True)
form_layout.addRow("Output stream:", self.stream_path_edit)
# Command edit (read-only single-line QLineEdit)
self.command_edit = QLineEdit()
self.command_edit.setReadOnly(True)
# Run progress & stats (one per line)
stats_layout = QVBoxLayout()
self.hitrate_label = QLabel("Hit rate: -")
self.speed_label = QLabel("Throughput: -")
self.indexable_label = QLabel("Indexed: -")
self.index_rate_label = QLabel("Indexing rate: -")
stats_layout.addWidget(self.hitrate_label)
stats_layout.addWidget(self.speed_label)
stats_layout.addWidget(self.indexable_label)
stats_layout.addWidget(self.index_rate_label)
self.progress_bar = QProgressBar()
self.progress_bar.setTextVisible(True)
self.progress_bar.setMinimum(0)
self.progress_bar.setMaximum(0) # indeterminate until we know total frames
self._total_frames_estimate = None
self._reset_run_metrics()
# Add Start (current), Start All (batch) and Stop buttons
self.start_current_button = QPushButton("Start Current")
self.start_current_button.setToolTip("Create a fresh run for the current file and start indexing.")
self.start_current_button.clicked.connect(
lambda _checked=False: self._start_indexing(create_fresh_run=True)
)
form_layout.addRow(self.start_current_button)
self.start_all_button = QPushButton("Start All in Batch")
self.start_all_button.setToolTip("Create a fresh shared batch run for all workspace files and start indexing.")
self.start_all_button.clicked.connect(self._start_batch_indexing)
form_layout.addRow(self.start_all_button)
self.stop_button = QPushButton("Stop")
self.stop_button.clicked.connect(self._stop_indexing)
form_layout.addRow(self.stop_button)
# Unit cell histograms (opens main window dialog, synced to current run)
self.unit_cell_hist_button = QPushButton("Unit Cell Histograms")
self.unit_cell_hist_button.clicked.connect(lambda: self._open_unit_cell_hist_from_run(auto_refresh=False))
self.unit_cell_hist_button.setEnabled(False)
form_layout.addRow(self.unit_cell_hist_button)
# Status label to indicate which run is currently executing (for batch)
self.batch_status_label = QLabel("")
form_layout.addRow("Running:", self.batch_status_label)
# Add output_edit QTextEdit for logs/output
self.output_edit = QTextEdit()
self.output_edit.setReadOnly(True)
layout.addLayout(form_layout)
command_layout = QFormLayout()
command_layout.setLabelAlignment(Qt.AlignmentFlag.AlignLeft)
command_layout.setFormAlignment(Qt.AlignmentFlag.AlignLeft | Qt.AlignmentFlag.AlignTop)
command_layout.setFieldGrowthPolicy(QFormLayout.FieldGrowthPolicy.AllNonFixedFieldsGrow)
command_layout.addRow("Command:", self.command_edit)
layout.addLayout(command_layout)
layout.addLayout(stats_layout)
layout.addWidget(self.progress_bar)
layout.addWidget(self.output_edit)
self.start_tab.setLayout(layout)
self.run_tab = QWidget()
run_layout = QVBoxLayout(self.run_tab)
self.run_stack = QStackedWidget()
self.run_stack.addWidget(self.start_tab)
self.ici_run_tab = QWidget()
self.run_stack.addWidget(self.ici_run_tab)
run_layout.addWidget(self.run_stack)
self.tabs.addTab(self.run_tab, "Run")
def _start_indexing_for_run(self, run_dir: str, ini_path: str | None = None) -> bool:
"""Prepare UI/paths for the given run_dir and start indexing. Returns True if started."""
if not run_dir or not os.path.isdir(run_dir):
return False
if ini_path and os.path.exists(ini_path):
self.ini_path = ini_path
self.h5_path = self._find_h5_path_from_ini(ini_path) or None
# Point Start tab to this run and rebuild command
self._apply_run_dir(run_dir)
# Ensure command is up to date
self._update_command_label()
# Start the already prepared batch run. Do not create another
# timestamped run here; the batch queue owns the shared run name.
self._start_indexing(create_fresh_run=False)
return True
def _start_batch_indexing(self):
"""
Queue and run all INIs in the workspace with a NEW timestamped run name.
For each INI, create the run, seed input.lst if possible, and write the
current GUI settings into index_settings.json. Then execute sequentially.
"""
from datetime import datetime as _dt
import os, json
# Always create a NEW group name each time the button is pressed
group = f"indexingintegration_{_dt.now().strftime('%Y%m%d_%H%M%S')}"
cfg = self._snapshot_index_settings()
self._batch_queue = []
self._batch_run_to_ini = {}
created_dirs = 0
created_lists = 0
ini_paths = self._get_workspace_ini_paths()
# Prepare runs for all INIs
for ini in ini_paths:
try:
run_base = self._resolve_run_root(ini)
rd = os.path.join(run_base, group)
os.makedirs(rd, exist_ok=True)
created_dirs += 1
# Write current settings into each run
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 if possible
try:
lst_path = os.path.join(rd, 'input.lst')
if not os.path.exists(lst_path):
h5 = self._find_h5_path_from_ini(ini)
if h5:
with open(lst_path, 'w') as f:
f.write(os.path.abspath(h5))
created_lists += 1
except Exception:
pass
# Queue this run
self._batch_queue.append(rd)
self._batch_run_to_ini[rd] = ini
except Exception:
continue
# Switch UI to the current INI’s fresh run for visibility
cur_ini = getattr(self, 'ini_path', None)
if cur_ini and os.path.exists(cur_ini):
rd_cur = os.path.join(self._resolve_run_root(cur_ini), group)
if os.path.isdir(rd_cur):
self._apply_run_dir(rd_cur)
# Refresh tree so user sees the new runs
try:
self._refresh_runs_tree()
try:
self._select_tree_for_ini(getattr(self, 'ini_path', None))
except Exception:
pass
except Exception:
pass
if not self._batch_queue:
QMessageBox.information(self, "Batch Run", "No runs could be prepared for batch execution.")
return
info_bits = [f"created {created_dirs} run folder(s)"]
if created_lists:
info_bits.append(f"prepared {created_lists} input.lst file(s)")
self.batch_status_label.setText("; ".join(info_bits))
# Kick off the first; the rest chain via _check_process_done
self._batch_mode = True
self._start_next_in_batch()
def _start_next_in_batch(self):
if not self._batch_queue:
self.batch_status_label.setText("")
self._batch_mode = False
self._batch_run_to_ini = {}
QMessageBox.information(self, "Batch Run", "Batch finished.")
return
rd = self._batch_queue.pop(0)
ini = getattr(self, "_batch_run_to_ini", {}).get(rd)
# Update status and highlight in tree
self.batch_status_label.setText(rd)
self._select_tree_item_for_run(rd)
self._start_indexing_for_run(rd, ini_path=ini)
# Begin polling for completion
self._proc_poll_timer.start()
def _check_process_done(self):
# Called periodically while a process is running
if self.indexing_process is None:
# Nothing running; if we are in batch mode, schedule next
if self._batch_mode:
self._proc_poll_timer.stop()
self._start_next_in_batch()
return
try:
rc = self.indexing_process.poll()
if rc is not None:
# Process finished
self._proc_poll_timer.stop()
self.indexing_process = None
if self._batch_mode:
self._start_next_in_batch()
except Exception:
# On error, try to progress the batch
self._proc_poll_timer.stop()
self.indexing_process = None
if self._batch_mode:
self._start_next_in_batch()
def _select_tree_item_for_run(self, run_dir: str):
try:
top_count = self.runs_tree.topLevelItemCount()
for i in range(top_count):
top = self.runs_tree.topLevelItem(i)
if not top:
continue
for j in range(top.childCount()):
child = top.child(j)
if child and child.toolTip(0) == run_dir:
self.runs_tree.setCurrentItem(child)
idx = self.runs_tree.indexFromItem(child)
if idx.isValid():
self.runs_tree.scrollTo(idx)
return
except Exception:
pass
def _update_command_label(self):
"""(Re)build the indexamajig command preview.
If no run is currently selected, try to select the latest run for the
current INI. If path fields are still empty, prefill them based on the
run folder so the user always sees a concrete command string.
"""
import os
# 1) Ensure we point at some run (latest for current INI if none selected)
if not getattr(self, 'run_dir', None) or not isinstance(self.run_dir, str) or not os.path.isdir(self.run_dir):
self._ensure_run_selected_or_latest()
# 2) Prefill missing path fields from the run_dir so we can always show a command
rd = getattr(self, 'run_dir', None)
if rd and os.path.isdir(rd):
if not getattr(self, 'cell_filepath', None):
self.cell_filepath = os.path.join(rd, 'cellfile.cell')
if not getattr(self, 'geom_filepath', None):
self.geom_filepath = os.path.join(rd, 'geometry.geom')
if not getattr(self, 'list_filepath', None):
self.list_filepath = os.path.join(rd, 'input.lst')
if not getattr(self, 'stream_filepath', None):
run_folder_name = os.path.basename(rd)
self.stream_filepath = os.path.join(rd, f'{run_folder_name}.stream')
# Reflect into Start tab read-only edits if present
if hasattr(self, 'geom_path_edit') and self.geom_path_edit.text().strip() == '' and getattr(self, 'geom_filepath', ''):
self.geom_path_edit.setText(self.geom_filepath)
if hasattr(self, 'cell_path_edit') and self.cell_path_edit.text().strip() == '' and getattr(self, 'cell_filepath', ''):
self.cell_path_edit.setText(self.cell_filepath)
if hasattr(self, 'list_path_edit') and self.list_path_edit.text().strip() == '' and getattr(self, 'list_filepath', ''):
self.list_path_edit.setText(self.list_filepath)
if hasattr(self, 'stream_path_edit') and self.stream_path_edit.text().strip() == '' and getattr(self, 'stream_filepath', ''):
self.stream_path_edit.setText(self.stream_filepath)
# 3) Collect all inputs (may still be empty if no INI/run at all)
geom = getattr(self, 'geom_filepath', '') or ''
lst = getattr(self, 'list_filepath', '') or ''
stream = getattr(self, 'stream_filepath', '') or ''
cell = getattr(self, 'cell_filepath', '') or ''
# If we *still* have no context, clear and bail out gracefully
if not any([geom, lst, stream, cell]):
if hasattr(self, 'command_edit'):
self.command_edit.setText('')
return
# 4) Read current UI parameter values
worker_threads = self.worker_threads_spin.value()
max_indexer_threads = getattr(self, "MAX_INDEXER_THREADS", 1)
push_res = self.push_res_edit.text().strip()
int_radius = self.int_radius_edit.text().strip()
min_peaks = self.min_peaks_spin.value()
tol = self.tolerance_edit.text().strip()
xg_tol = self.xgandalf_tolerance_edit.text().strip()
samp_pitch = self.sampling_pitch_spin.value()
min_lat = self.min_lat_vec_len_spin.value()
max_lat = self.max_lat_vec_len_spin.value()
gd_iters = self.grad_desc_iterations_spin.value()
fix_profile_radius_enabled = self.fix_profile_radius_cb.isChecked()
fix_profile_radius = self.fix_profile_radius_edit.text().strip()
# 5) Optional flags
flags = []
if hasattr(self, 'no_non_hits_cb') and self.no_non_hits_cb.isChecked():
flags.append('--no-non-hits-in-stream')
if hasattr(self, 'no_revalidate_cb') and self.no_revalidate_cb.isChecked():
flags.append('--no-revalidate')
if hasattr(self, 'no_check_peaks_cb') and self.no_check_peaks_cb.isChecked():
flags.append('--no-check-peaks')
if hasattr(self, 'no_half_pixel_shift_cb') and self.no_half_pixel_shift_cb.isChecked():
flags.append('--no-half-pixel-shift')
if hasattr(self, 'no_retry_cb') and self.no_retry_cb.isChecked():
flags.append('--no-retry')
if hasattr(self, 'no_check_cell_cb') and self.no_check_cell_cb.isChecked():
flags.append('--no-check-cell')
if hasattr(self, 'no_refine_cb') and self.no_refine_cb.isChecked():
flags.append('--no-refine')
if hasattr(self, 'overpredict_cb') and self.overpredict_cb.isChecked():
flags.append('--overpredict')
flags_str = ' '.join(flags)
# 6) Build the command string even if files don't exist yet (paths suffice)
use_cell = hasattr(self, 'use_cell_cb') and self.use_cell_cb.isChecked()
cmd_parts = [
f"indexamajig -g {geom}",
f"-i {lst}",
f"-o {stream}",
f"-j {worker_threads}",
*([ f"-p {cell}" ] if use_cell else []),
f"--indexing=xgandalf",
f"--push-res={push_res}",
f"--integration=rings",
flags_str,
f"--int-radius={int_radius}",
f"--peaks=cxi",
f"--max-indexer-threads={max_indexer_threads}",
f"--min-peaks={min_peaks}",
f"--xgandalf-tolerance={xg_tol}",
f"--xgandalf-sampling-pitch={samp_pitch}",
*([ f"--xgandalf-min-lattice-vector-length={min_lat}",
f"--xgandalf-max-lattice-vector-length={max_lat}" ] if not use_cell else []),
f"--xgandalf-grad-desc-iterations={gd_iters}",
f"--tolerance={tol}",
]
if fix_profile_radius_enabled and fix_profile_radius:
cmd_parts.append(f"--fix-profile-radius={fix_profile_radius}")
# Clean up any accidental double spaces from empty flags_str
cmd = ' '.join(part for part in cmd_parts if part)
if hasattr(self, 'command_edit'):
self.command_edit.setText(cmd)
def _start_indexing(self, create_fresh_run: bool = True):
import os
# Normal Start creates a fresh run. Batch Start prepares all run
# folders up front with one shared timestamp and then starts them here.
if self.indexing_process and self.indexing_process.poll() is None:
return
start_settings = self._snapshot_index_settings()
if create_fresh_run:
rd = self._create_fresh_run_for_current_ini(settings=start_settings)
if not rd:
return
else:
rd = getattr(self, "run_dir", None)
if not rd or not os.path.isdir(rd):
QMessageBox.critical(self, "Error", "Prepared batch run directory is missing.")
return
h5_paths = self._resolve_h5_for_current_run()
if not h5_paths:
QMessageBox.critical(
self,
"Mask Required",
"Indexing requires a valid 2D mask dataset in the input HDF5 file.\n\n"
"No input HDF5 file could be resolved from input.lst or INI Paths.h5file.\n\n"
"Run Preflight Check and verify the run input list.",
)
return
mask_failures = []
for path in h5_paths:
ok_mask, reason = self._h5_has_required_mask(path)
if not ok_mask:
mask_failures.append(reason)
if mask_failures:
detail = "\n".join(f"- {msg}" for msg in mask_failures[:8])
if len(mask_failures) > 8:
detail += f"\n- ... and {len(mask_failures) - 8} more"
QMessageBox.critical(
self,
"Mask Required",
"Indexing requires a valid 2D mask dataset in every input HDF5 file.\n\n"
f"{detail}\n\n"
"Run Preflight Check and generate/apply masks before starting indexing.",
)
return
# Reset metrics/progress for the new run
self._total_frames_estimate = None
self._reset_run_metrics()
# Refresh and select the new run in the tree so the UI stays in sync
try:
self._refresh_runs_tree()
self._select_tree_item_for_run(rd)
except Exception:
pass
# Tree selection may load settings from an existing run. For the run
# we are about to launch, restore the values that were visible when
# the user pressed Start and persist them into this run folder.
try:
self._apply_index_settings(start_settings)
with open(os.path.join(rd, "index_settings.json"), "w") as jf:
json.dump(start_settings, jf, indent=2)
except Exception:
pass
self._update_command_label()
# Auto-open unit cell histograms for the new run with periodic refresh
try:
self._open_unit_cell_hist_from_run(auto_refresh=True)
except Exception:
pass
cmd = self.command_edit.text().strip()
if not cmd:
return
use_cell = hasattr(self, 'use_cell_cb') and self.use_cell_cb.isChecked()
if use_cell and not os.path.exists(getattr(self, "cell_filepath", "")):
QMessageBox.critical(
self,
"Cell file missing",
"cellfile.cell was not created. Fill all required cell fields in the Cell Editor and try again.",
)
return
if not os.path.exists(getattr(self, "geom_filepath", "")):
QMessageBox.critical(
self,
"Geometry file missing",
"geometry.geom was not created. Ensure a mask exists and regenerate geometry.",
)
return
if not self._geom_has_mask_entries(getattr(self, "geom_filepath", "")):
QMessageBox.critical(
self,
"Invalid geometry",
"The geometry file does not contain panel mask entries (e.g. p0/mask = ...).\n"
"Regenerate/save geometry after creating a mask in Preflight Check.",
)
return
try:
self._write_indexing_nxprocess(cmd)
except Exception as exc:
log_print(f"DEBUG: Failed to write NXprocess indexing: {exc}")
try:
if hasattr(self, "output_edit"):
self.output_edit.clear()
self.log_file = open(os.path.join(self.run_dir, "indexing.log"), "w")
self.indexing_process = subprocess.Popen(
cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, bufsize=1,
preexec_fn=os.setsid
)
self.process_thread = self.ProcessOutputThread(self.indexing_process)
self.process_thread.output_received.connect(self._append_output)
self.process_thread.start()
self._proc_poll_timer.start()
except Exception as e:
QMessageBox.critical(self, "Error", f"Failed to start indexing:\n{e}")
def _append_output(self, text):
# Append to output_edit and write to log file
if hasattr(self, "output_edit"):
self.output_edit.append(text.rstrip("\n"))
if hasattr(self, "log_file"):
self.log_file.write(text)
self.log_file.flush()
self._parse_run_stats_line(text)
def _stop_indexing(self):
# Use process group termination so that stop button works even if the command spawns children
if self.indexing_process and self.indexing_process.poll() is None:
try:
# first try to terminate the entire process group
os.killpg(os.getpgid(self.indexing_process.pid), signal.SIGTERM)
# give it a short grace period
self.indexing_process.wait(timeout=2)
except Exception:
try:
self.indexing_process.terminate()
except Exception as e:
QMessageBox.critical(self, "Error", f"Failed to stop indexing:\n{e}")
finally:
self.indexing_process = None
# If user stops manually, cancel any remaining batch items
if self._batch_mode:
self._proc_poll_timer.stop()
self._batch_queue = []
self._batch_run_to_ini = {}
self._batch_mode = False
self.batch_status_label.setText("")
# Close log file if open
if hasattr(self, "log_file"):
try:
self.log_file.close()
except:
pass
def _get_workspace_ini_paths(self):
paths = []
path = getattr(self, 'workspace_path', None)
if not path or not isinstance(path, str) or not os.path.exists(path):
return paths
try:
with open(path, 'r') as f:
for line in f:
s = line.strip()
if not s or s.startswith('#'):
continue
# treat as an INI path line
paths.append(s)
except Exception:
pass
return paths
def _resolve_run_root(self, ini_path: str) -> str:
"""
Decide where indexingintegration_* runs live without relying on INI Paths.outputfolder.
Preference: <ini_dir>/output, then <ini_dir>/runs, else <ini_dir>.
"""
import os
if not ini_path:
return os.getcwd()
base_dir = os.path.dirname(os.path.abspath(ini_path))
for candidate in (os.path.join(base_dir, 'output'),
os.path.join(base_dir, 'runs')):
if os.path.isdir(candidate):
return candidate
return base_dir
def _refresh_runs_tree(self):
import glob
# Pause sync timer during rebuild to avoid races
timer_active = False
if hasattr(self, 'sync_timer') and self.sync_timer.isActive():
timer_active = True
self.sync_timer.stop()
try:
self.runs_tree.clear()
ini_paths = self._get_workspace_ini_paths()
for ini in ini_paths:
try:
ini_base = os.path.basename(ini)
top = QTreeWidgetItem([ini_base])
top.setToolTip(0, ini)
self.runs_tree.addTopLevelItem(top)
# Determine base for runs
# Determine base for runs (do not rely on INI Paths.outputfolder)
base_dir = os.path.dirname(ini)
run_base = self._resolve_run_root(ini)
runs = sorted(
set(glob.glob(os.path.join(run_base, 'indexingintegration_*'))) |
set(glob.glob(os.path.join(base_dir, 'indexingintegration_*'))) |
set(glob.glob(os.path.join(run_base, 'ici_*'))) |
set(glob.glob(os.path.join(base_dir, 'ici_*')))
)
for rd in runs:
item = QTreeWidgetItem([os.path.basename(rd)])
item.setToolTip(0, rd)
top.addChild(item)
except Exception:
continue
self.runs_tree.expandAll()
finally:
if timer_active:
self.sync_timer.start()
def _on_current_ini_changed(self):
"""Refresh tab settings when the current INI changes (without creating a new run).
Prefers the latest run's cellfile; falls back to workspace cell values, then empty/new."""
if not self.ini_path or not os.path.exists(self.ini_path):
return
# Recompute context for this INI
from configparser import ConfigParser
parser = ConfigParser()
try:
parser.read(self.ini_path)
except Exception:
return
base_dir = os.path.dirname(self.ini_path)
# Detect latest previous run for this INI without relying on INI Paths
import glob, os as _os
run_base = self._resolve_run_root(self.ini_path)
prev_runs = sorted(
set(glob.glob(os.path.join(run_base, 'indexingintegration_*'))) |
set(glob.glob(os.path.join(base_dir, 'indexingintegration_*')))
)
latest_run = prev_runs[-1] if prev_runs else None
# Prefer the newest run that actually has index_settings.json for loading UI settings
latest_settings_run = None
for rd in reversed(prev_runs):
try:
if os.path.exists(os.path.join(rd, 'index_settings.json')):
latest_settings_run = rd
break
except Exception:
continue
# Update pointers used by loaders
self.prev_cell_file = None
if latest_run:
candidate = os.path.join(latest_run, 'cellfile.cell')
if os.path.exists(candidate):
self.prev_cell_file = candidate
# Prefer latest run's cellfile; fallback to workspace; then to empty/new
if getattr(self, 'prev_cell_file', None) and os.path.exists(self.prev_cell_file):
self._load_cell_file()
else:
loaded_ws = self._load_cell_from_workspace()
if not loaded_ws:
self._load_cell_file()
# Try to load geometry from latest run if present; else fall back to existing path/INI-derived defaults
if latest_run:
latest_geom = os.path.join(latest_run, 'geometry.geom')
old_geom_path = getattr(self, 'geom_filepath', None)
if os.path.exists(latest_geom):
self.geom_filepath = latest_geom
self._load_geom_file()
# Keep geom_filepath pointing at latest if it exists; otherwise restore
if not os.path.exists(latest_geom) and old_geom_path:
self.geom_filepath = old_geom_path
else:
self._load_geom_file()
# Also refresh h5 path for command building
self.h5_path = self._find_h5_path_from_ini(self.ini_path) or self.h5_path
# Load Index & Integrate settings from the most recent run that actually has them
if latest_settings_run:
self._load_index_settings_from_run(latest_settings_run)
elif getattr(self, '_last_index_settings', None):
# Apply last known good settings as a sensible default for new files without JSON yet
self._apply_index_settings(self._last_index_settings)
self._load_ici_defaults_from_ini()
# Rebuild command preview with whatever paths we have now
self._sync_ici_shared_paths()
self._update_command_label()
# Do NOT refresh the runs tree here; it destroys items and breaks selection highlighting
def _select_tree_for_ini(self, ini_path: str):
"""Highlight the tree node corresponding to ini_path, if present (safe against rebuilds)."""
if not ini_path:
return
try:
top_count = self.runs_tree.topLevelItemCount()
for i in range(top_count):
top = self.runs_tree.topLevelItem(i)
if top and top.toolTip(0) == ini_path:
self.runs_tree.setCurrentItem(top)
# Scroll safely using a model index
idx = self.runs_tree.indexFromItem(top)
if idx.isValid():
self.runs_tree.scrollTo(idx)
break
except RuntimeError:
# The tree was probably rebuilt; ignore and let next timer tick handle it
return
def _on_runs_tree_selection_changed(self, current, previous):
if current is None:
return
parent = current.parent()
if parent is None:
# Top-level INI selected -> emit to main window
ini_path = current.toolTip(0)
if ini_path and os.path.exists(ini_path):
# Update our state
self.ini_path = ini_path
self.h5_path = self._find_h5_path_from_ini(ini_path) or self.h5_path
# Emit to main window rather than calling its slots directly
try:
self.ini_selection_changed.emit(ini_path)
except Exception:
pass
# Refresh right-side tabs for the newly selected INI
self._on_current_ini_changed()
else:
# Child node = a run directory or nested run detail. For nested ICI
# details, apply the parent ICI run directory rather than the detail.
run_item = self._run_tree_item_for_tree_item(current)
run_dir = run_item.toolTip(0) if run_item is not None else current.toolTip(0)
if not run_dir or not os.path.isdir(run_dir):
return
top_ini_item = self._top_level_ini_item_for_tree_item(current)
parent_ini = top_ini_item.toolTip(0) if top_ini_item is not None else None
if parent_ini and os.path.exists(parent_ini):
# Sync selection/state to parent INI locally
self.ini_path = parent_ini
self.h5_path = self._find_h5_path_from_ini(parent_ini) or self.h5_path
try:
self.ini_selection_changed.emit(parent_ini)
except Exception:
pass
# Refresh tabs for the now-current INI
self._on_current_ini_changed()
# Finally, point Start tab to the selected run directory
self._apply_run_dir(run_dir)
def _apply_run_dir(self, run_dir: str, *, load_index_settings: bool = True):
"""Point Start tab paths to the selected run directory and update command preview."""
self.run_dir = run_dir
self.cell_filepath = os.path.join(run_dir, 'cellfile.cell')
self.geom_filepath = os.path.join(run_dir, 'geometry.geom')
# If this run has no cell file yet, write one from current UI values
if not os.path.exists(self.cell_filepath):
self._save_cell_file()
# If this run has saved index settings, load and apply them
if load_index_settings:
self._load_index_settings_from_run(self.run_dir)
self.list_filepath = os.path.join(run_dir, 'input.lst')
self.stream_filepath = self._resolve_run_stream_path(run_dir)
if hasattr(self, 'geom_path_edit'):
self.geom_path_edit.setText(self.geom_filepath)
if hasattr(self, 'cell_path_edit'):
self.cell_path_edit.setText(self.cell_filepath)
if hasattr(self, 'list_path_edit'):
self.list_path_edit.setText(self.list_filepath)
if hasattr(self, 'stream_path_edit'):
self.stream_path_edit.setText(self.stream_filepath)
# Force loader to use this run's cellfile
self.prev_cell_file = self.cell_filepath if os.path.exists(self.cell_filepath) else None
# Optionally load geom/cell into editors if files exist
self._load_cell_file()
self._load_geom_file()
# If this run has no geometry file yet, write one from current UI/INI-derived values
if not os.path.exists(self.geom_filepath):
if not self._save_geom_file():
log_print(f"DEBUG: failed to create geometry file for run {run_dir}")
# Reset metrics for selected run (will update when output arrives)
self._total_frames_estimate = None
self._reset_run_metrics()
self._sync_ici_shared_paths()
if self._is_ici_run_dir(run_dir) and hasattr(self, "_ici_apply_existing_run"):
self._ici_apply_existing_run(run_dir)
self._update_command_label()
self._notify_main_window_stream_path()
self._update_unit_cell_button_state()
def _notify_main_window_stream_path(self):
"""Inform the main window about the current run's stream file."""
stream_path = getattr(self, 'stream_filepath', None)
try:
callback = getattr(self.main_window, 'on_indexing_run_stream_changed', None)
if callable(callback):
callback(stream_path)
except Exception:
pass
self._update_unit_cell_button_state()
def _sync_from_mainwindow_selection(self):
"""Periodically reflect main window's current INI selection into this window/tree."""
mw = self.main_window
current_ini = getattr(mw, 'full_ini_file_path', None)
if not current_ini or current_ini == self._last_seen_main_ini:
return
self._last_seen_main_ini = current_ini
# Update internal references
self.ini_path = current_ini
# Try to select it in the tree, guarding against tree rebuilds
try:
self._select_tree_for_ini(current_ini)
except RuntimeError:
pass
self._on_current_ini_changed()
self._update_unit_cell_button_state()
def _find_h5_path_from_ini(self, ini_path):
import os, glob
if not ini_path:
return None
resolved = self._resolve_h5_paths_from_ini(ini_path)
if resolved:
return resolved[0]
try:
ini_path = os.path.abspath(ini_path)
base_dir = os.path.dirname(ini_path)
ini_name = os.path.basename(ini_path)
stem, _ = os.path.splitext(ini_name)
# Derive prefix: take part before '_run_' if present, else first two tokens if they look like date_time
prefix = stem
if '_run_' in stem:
prefix = stem.split('_run_')[0]
else:
parts = stem.split('_')
if len(parts) >= 2 and parts[0].isdigit() and parts[1].isdigit():
prefix = f"{parts[0]}_{parts[1]}"
elif parts:
prefix = parts[0]
# Candidate directories to search
base_parent = os.path.dirname(base_dir)
base_grand = os.path.dirname(base_parent)
candidates = []
for d in (base_dir, base_parent, base_grand):
if d and os.path.isdir(d) and d not in candidates:
candidates.append(d)
patterns = [
f"{prefix}*.h5", f"{prefix}*.hdf5", f"{prefix}*.cxi",
f"{prefix}.h5", f"{prefix}.hdf5", f"{prefix}.cxi",
]
hits = []
for root in candidates:
for pat in patterns:
hits.extend(glob.glob(os.path.join(root, pat)))
hits = sorted(h for h in hits if os.path.isfile(h))
if hits:
return hits[0]
# Shallow scan subdirs of parent that contain the prefix in their name
if os.path.isdir(base_parent):
for entry in sorted(os.listdir(base_parent)):
sub = os.path.join(base_parent, entry)
if os.path.isdir(sub) and prefix in entry:
for pat in patterns:
cand = sorted(glob.glob(os.path.join(sub, pat)))
cand = [h for h in cand if os.path.isfile(h)]
if cand:
return cand[0]
except Exception as e:
log_print(f"DEBUG: naming-scheme H5 resolve failed for {ini_path}: {e}")
return None
log_print(f"DEBUG: No H5/CXI found by naming scheme for INI: {ini_path} (prefix='{prefix}')")
return None
def _create_new_run_for_current_file(self):
# Require a current INI
if not self.ini_path or not os.path.exists(self.ini_path):
QMessageBox.warning(self, 'Run', 'No current INI selected in main window.')
return
# Compute run base
# Compute run base (independent of INI Paths)
base_dir = os.path.dirname(self.ini_path)
run_base = self._resolve_run_root(self.ini_path)
# Create new timestamped run dir
from datetime import datetime
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
new_run_dir = os.path.join(run_base, f'indexingintegration_{timestamp}')
try:
os.makedirs(new_run_dir, exist_ok=True)
except Exception as e:
QMessageBox.critical(self, 'Run', f'Failed to create run directory:\n{e}')
return
# Update internal paths to point to this new run
self.run_dir = new_run_dir
self.cell_filepath = os.path.join(self.run_dir, 'cellfile.cell')
self.geom_filepath = os.path.join(self.run_dir, 'geometry.geom')
self.list_filepath = os.path.join(self.run_dir, 'input.lst')
run_folder_name = os.path.basename(self.run_dir)
self.stream_filepath = os.path.join(self.run_dir, f'{run_folder_name}.stream')
# Write list file; if H5 is not known, prompt the user once
h5 = self.h5_path or self._find_h5_path_from_ini(self.ini_path)
if not h5:
try:
sel, _ = QFileDialog.getOpenFileName(self, 'Select HDF5/ CXI file', os.path.dirname(self.ini_path), 'Data files (*.h5 *.hdf5 *.cxi);;All files (*)')
if sel:
h5 = sel
except Exception:
pass
try:
with open(self.list_filepath, 'w') as f:
if h5:
f.write(os.path.abspath(h5))
else:
# Create an empty placeholder; user can fill it later
f.write('')
except Exception as e:
QMessageBox.warning(self, 'Run', f'Created run dir, but failed to write input.lst:\n{e}')
# Reflect new paths in the Start tab fields
if hasattr(self, 'geom_path_edit'):
self.geom_path_edit.setText(self.geom_filepath)
if hasattr(self, 'cell_path_edit'):
self.cell_path_edit.setText(self.cell_filepath)
if hasattr(self, 'list_path_edit'):
self.list_path_edit.setText(self.list_filepath)
if hasattr(self, 'stream_path_edit'):
self.stream_path_edit.setText(self.stream_filepath)
# Ensure a cell file exists for the new run (uses current UI values)
if not os.path.exists(self.cell_filepath):
self._save_cell_file()
# Ensure a geometry file exists for the new run (uses current UI/INI-derived values)
if not os.path.exists(self.geom_filepath):
if not self._save_geom_file(show_errors=True):
QMessageBox.critical(
self,
"Mask Required",
"Failed to create geometry file. Ensure this dataset has a valid mask in Preflight Check.",
)
return
# Update command string & tree
self._update_command_label()
self._refresh_runs_tree()
self._notify_main_window_stream_path()
self._update_unit_cell_button_state()
def _create_new_batch_run(self):
ini_paths = self._get_workspace_ini_paths()
if not ini_paths:
QMessageBox.warning(self, 'Batch Run', 'No INI files found in the workspace file.')
return
from datetime import datetime
from configparser import ConfigParser
import os
ts = datetime.now().strftime('%Y%m%d_%H%M%S')
created = 0
errors = 0
for ini in ini_paths:
try:
parser = ConfigParser()
parser.read(ini)
base_dir = os.path.dirname(ini)
output_folder = parser.get('Paths', 'outputfolder', fallback='')
run_base = os.path.join(base_dir, output_folder) if output_folder else base_dir
new_run_dir = os.path.join(run_base, f'indexingintegration_{ts}')
os.makedirs(new_run_dir, exist_ok=True)
# 1) Write list file using naming-based H5 resolution (placeholder if unresolved)
lst_path = os.path.join(new_run_dir, 'input.lst')
h5 = self._find_h5_path_from_ini(ini)
try:
with open(lst_path, 'w') as f:
f.write(os.path.abspath(h5) if h5 else '')
log_print(f"DEBUG: wrote input.lst for {new_run_dir}: '{os.path.abspath(h5) if h5 else ''}'")
except Exception as e:
log_print(f"DEBUG: failed to write input.lst for {new_run_dir}: {e}")
errors += 1
continue
# 2) Write a cell file for this run from current UI values, if valid
try:
cent = self.centering_combo.currentText().strip()
lt = normalize_crystfel_lattice_type(self.lattice_type_combo.currentText(), cent)
ua = self.unique_axis_combo.currentText().strip() if crystfel_unique_axis_allowed(lt) else ''
a = self.a_edit.text().strip(); b = self.b_edit.text().strip(); c = self.c_edit.text().strip()
al = self.al_edit.text().strip(); be = self.be_edit.text().strip(); ga = self.ga_edit.text().strip()
ok = True
for v in (a, b, c, al, be, ga):
try:
float(v)
except Exception:
ok = False; break
if ok and lt and cent:
with open(os.path.join(new_run_dir, 'cellfile.cell'), 'w') as fcell:
fcell.write('CrystFEL unit cell file version 1.0\n\n')
fcell.write(f'lattice_type = {lt}\n\n')
fcell.write(f'centering = {cent}\n')
if ua:
fcell.write(f'unique_axis = {ua}\n\n')
else:
fcell.write('\n')
fcell.write(f'a = {a} A\n')
fcell.write(f'b = {b} A\n')
fcell.write(f'c = {c} A\n')
fcell.write(f'al = {al} deg\n')
fcell.write(f'be = {be} deg\n')
fcell.write(f'ga = {ga} deg\n')
except Exception as e:
log_print(f"DEBUG: failed to write cellfile.cell for {new_run_dir}: {e}")
# not fatal
# 3) Write a geometry file for this run from current UI/INI-derived values
try:
geom_path = os.path.join(new_run_dir, 'geometry.geom')
if not os.path.exists(geom_path):
# Temporarily point this window's paths to the new run so _save_geom_file writes there
prev_run_dir = getattr(self, 'run_dir', None)
prev_geom = getattr(self, 'geom_filepath', '')
prev_cell = getattr(self, 'cell_filepath', '')
prev_list = getattr(self, 'list_filepath', '')
prev_stream = getattr(self, 'stream_filepath', '')
try:
self.run_dir = new_run_dir
self.geom_filepath = geom_path
self.cell_filepath = os.path.join(new_run_dir, 'cellfile.cell')
self.list_filepath = lst_path
run_folder_name = os.path.basename(new_run_dir)
self.stream_filepath = os.path.join(new_run_dir, f'{run_folder_name}.stream')
if not self._save_geom_file(show_errors=False):
raise RuntimeError("Geometry generation failed (missing/invalid mask).")
finally:
self.run_dir = prev_run_dir
self.geom_filepath = prev_geom
self.cell_filepath = prev_cell
self.list_filepath = prev_list
self.stream_filepath = prev_stream
except Exception as e:
log_print(f"DEBUG: failed to write geometry.geom for {new_run_dir}: {e}")
errors += 1
continue
created += 1
except Exception as e:
log_print(f"DEBUG: batch run creation failed for {ini}: {e}")
errors += 1
continue
# Refresh the tree so the new runs appear
self._refresh_runs_tree()
# Inform the user
if created and errors:
QMessageBox.information(self, 'Batch Run', f'Created {created} run(s) with {errors} error(s).')
elif created:
QMessageBox.information(self, 'Batch Run', f'Created {created} run(s).')
else:
QMessageBox.warning(self, 'Batch Run', 'No runs were created.')
# --- Run output metrics / progress ---------------------------------
def _reset_run_metrics(self):
self._update_metrics_ui(processed=None, hits=None, indexable=None, rate=None, progress=0)
def _update_metrics_ui(self, processed=None, hits=None, indexable=None, rate=None, progress=None):
# Hit rate (% of processed)
if processed and processed > 0 and hits is not None:
hit_rate = 100.0 * hits / processed
self.hitrate_label.setText(f"Hit rate: {hit_rate:.1f}%")
else:
self.hitrate_label.setText("Hit rate: -")
# Indexed count and rate (% of hits)
if indexable is not None:
self.indexable_label.setText(f"Indexed: {indexable}")
else:
self.indexable_label.setText("Indexed: -")
if hits and hits > 0 and indexable is not None:
idx_rate = 100.0 * indexable / hits
self.index_rate_label.setText(f"Indexing rate: {idx_rate:.1f}% of hits")
else:
self.index_rate_label.setText("Indexing rate: -")
# Throughput
self.speed_label.setText(f"Throughput: {rate if rate is not None else '-'} fps")
total = self._get_total_frames_estimate()
if total and total > 0:
self.progress_bar.setRange(0, total)
if progress is not None:
self.progress_bar.setValue(min(progress, total))
else:
self.progress_bar.setValue(0)
pct = 100.0 * (self.progress_bar.value() / float(total))
# Use literal %v/%m placeholders for Qt, inject pct ourselves
self.progress_bar.setFormat(f"%v / %m ({pct:.1f}%)")
else:
# Indeterminate when total frames unknown
self.progress_bar.setRange(0, 0)
self.progress_bar.setFormat("Processing…")
def _parse_run_stats_line(self, line: str):
"""
Parse indexamajig status lines of the form:
'500 images processed, 98 hits (19.6%), 0 indexable (0.0% of hits, 0.0% overall), 0 crystals, 54.0 images/sec.'
"""
try:
m = re.search(
r"(?P<proc>\d+)\s+images processed,\s+"
r"(?P<hits>\d+)\s+hits\s+\([\d\.]+%\),\s+"
r"(?P<idx>\d+)\s+indexable\s+\([\d\.]+% of hits, [\d\.]+% overall\),\s+"
r"(?P<crystals>\d+)\s+crystals,\s+"
r"(?P<rate>[\d\.]+)\s+images/sec",
line,
)
if not m:
return
processed = int(m.group("proc"))
hits = int(m.group("hits"))
indexable = int(m.group("idx"))
rate = float(m.group("rate"))
self._update_metrics_ui(processed=processed, hits=hits, indexable=indexable, rate=rate, progress=processed)
except Exception:
pass
def _get_total_frames_estimate(self):
if self._total_frames_estimate is not None:
return self._total_frames_estimate
# 1) Try to reuse total_frames from main window if available
try:
tf = getattr(self.main_window, "total_frames", None)
if tf is not None:
self._total_frames_estimate = int(tf)
return self._total_frames_estimate
except Exception:
pass
# 2) Try to read from HDF5 images dataset
try:
if self.h5_path and os.path.exists(self.h5_path):
import h5py
with h5py.File(self.h5_path, 'r') as f:
if 'entry/data/images' in f:
self._total_frames_estimate = int(f['entry/data/images'].shape[0])
return self._total_frames_estimate
except Exception:
pass
return None
# --- Unit cell histograms integration ---
def _update_unit_cell_button_state(self):
enabled = bool(self.stream_filepath and os.path.exists(self.stream_filepath))
try:
self.unit_cell_hist_button.setEnabled(enabled)
except Exception:
pass
def _open_unit_cell_hist_from_run(self, auto_refresh: bool = False):
"""Open the Unit Cell Histograms dialog for the current run's stream."""
stream_path = getattr(self, "stream_filepath", None)
# Inform main window about the current stream so its dialog stays in sync
try:
cb = getattr(self.main_window, "on_indexing_run_stream_changed", None)
if callable(cb):
cb(stream_path)
except Exception:
pass
# Prefer main window's handler to keep a single dialog instance
try:
opener = getattr(self.main_window, "open_unit_cell_hist_window", None)
if callable(opener):
opener(auto_refresh=auto_refresh)
return
except Exception:
pass
# Fallback: open directly from here
try:
from cosedaUI.unitcell_window import UnitCellHistogramWindow
wnd = UnitCellHistogramWindow(
parent=self,
stream_path=stream_path,
indexing_manager=getattr(self.main_window, "indexing_manager", None),
)
if auto_refresh:
wnd.start_auto_refresh()
wnd.show()
self.unit_cell_hist_window = wnd
except Exception as exc:
QMessageBox.warning(self, "Unit Cell Histograms", f"Unable to open histogram window:\n{exc}")