"""UI for configuring and running CrystFEL Partialator merges from stream files."""
from coseda.logging_utils import log_print
import os
import configparser
import multiprocessing
import datetime
import json
import re
import csv
import glob
from PyQt6.QtWidgets import (
QDialog, QVBoxLayout, QHBoxLayout, QLabel, QLineEdit,
QPushButton, QSpinBox, QDoubleSpinBox, QComboBox, QCheckBox,
QFileDialog, QMessageBox, QTableWidget, QTableWidgetItem, QHeaderView, QTabWidget,
QPlainTextEdit, QGroupBox, QAbstractItemView
)
from PyQt6.QtCore import Qt, QProcess, QTimer
from PyQt6.QtGui import QDoubleValidator, QFont, QFontDatabase
from matplotlib.backends.backend_qtagg import FigureCanvasQTAgg as FigureCanvas
from matplotlib.figure import Figure
from coseda.initialize import write_mergingsettings
from coseda.ici.run_contract import (
canonical_stream_path,
resolve_final_stream_path,
update_final_stream_alias,
)
from coseda.merge import run_merging
from coseda.nexus.process import write_nxprocess_merging
from coseda.unitcell_fit import (
workspace_has_refined_cell,
write_workspace_refined_cell_file,
)
[docs]
class MergingSettingsWindow(QDialog):
"""Dialog to select stream files, set Partialator parameters, and launch merges."""
def _latest_merge_dir_for_stream(self, stream_path: str) -> str | None:
"""Return latest merge-run* dir for a given merged stream path."""
try:
if not stream_path:
return None
folder = os.path.dirname(stream_path)
candidates = []
for name in os.listdir(folder):
full = os.path.join(folder, name)
if os.path.isdir(full) and name.startswith("merge-run"):
try:
num = int(name.replace("merge-run", ""))
except Exception:
num = None
candidates.append((num, full, os.stat(full).st_mtime))
if not candidates:
return None
numeric = [c for c in candidates if c[0] is not None]
if numeric:
return max(numeric, key=lambda t: t[0])[1]
return max(candidates, key=lambda t: t[2])[1]
except Exception:
return None
def _cell_from_current_ini(self) -> str | None:
"""Prefer cell path referenced in the currently loaded INI."""
ini = getattr(self, "ini_file_path", None)
if not ini or not os.path.isfile(ini):
return None
try:
parser = configparser.ConfigParser()
parser.read(ini, encoding="utf-8")
for key in ("cellfile", "cellfile_path"):
if parser.has_option("Paths", key):
rel = parser.get("Paths", key).strip()
if rel:
base = os.path.dirname(ini)
full = os.path.normpath(os.path.join(base, rel))
if os.path.isfile(full):
return full
except Exception:
return None
return None
def _workspace_refined_cell_path(self) -> str | None:
"""Return a materialized workspace refined cell file when available."""
workspace_path = getattr(self, "workspace_path", None)
if not workspace_path or not os.path.isfile(workspace_path):
return None
try:
if not workspace_has_refined_cell(workspace_path):
return None
return write_workspace_refined_cell_file(workspace_path)
except Exception as exc:
log_print(f"Failed to materialize workspace refined cell: {exc}")
return None
def _selected_cell_source(self) -> str:
combo = getattr(self, "cell_source_combo", None)
if combo is None or not combo.isVisible():
return "indexing"
try:
return str(combo.currentData() or "indexing")
except Exception:
return "indexing"
def _refresh_cell_source_selector(self) -> None:
combo = getattr(self, "cell_source_combo", None)
if combo is None:
return
try:
has_refined = bool(
getattr(self, "workspace_path", None)
and os.path.isfile(self.workspace_path)
and workspace_has_refined_cell(self.workspace_path)
)
except Exception:
has_refined = False
previous = combo.currentData()
combo.blockSignals(True)
combo.clear()
combo.addItem("Indexing cell", "indexing")
if has_refined:
combo.addItem("Workspace refined cell", "workspace_refined")
if previous == "workspace_refined":
combo.setCurrentIndex(1)
combo.blockSignals(False)
combo.setVisible(has_refined)
label = getattr(self, "cell_source_label", None)
if label is not None:
label.setVisible(has_refined)
def _infer_cell_for_stream(self, stream_path: str) -> tuple[str | None, dict | None]:
"""
Infer the cellfile used for the merged stream by matching timestamps to indexing runs.
Returns (cell_path, parsed_params_dict) where params may include a,b,c,al,be,ga,centering.
"""
# 0a) If the stream itself carries a unit-cell header, prefer that.
if stream_path and os.path.isfile(stream_path):
params = self._parse_stream_cell(stream_path)
if params:
return f"{stream_path} (header)", params
# 0) INI-provided cell takes precedence
ini_cell = self._cell_from_current_ini()
if ini_cell:
return ini_cell, self._parse_cell_file(ini_cell)
if not stream_path:
return None, None
merge_dir = os.path.dirname(stream_path)
parent_dir = os.path.dirname(merge_dir)
merge_ts = self._timestamp_from_merge_dir(merge_dir)
# Candidate indexing run roots: parent of merge, ini directory, and parent of ini dir
ini_dir = getattr(self, "ini_directory", parent_dir)
run_dirs_candidates = set()
for root in {parent_dir, ini_dir, os.path.dirname(ini_dir)}:
try:
for pattern in ("indexingintegration_*", "ici_*"):
for d in glob.glob(os.path.join(root, pattern)):
if os.path.isdir(d):
run_dirs_candidates.add(d)
except Exception:
continue
idx_dirs = sorted(run_dirs_candidates)
best_cell = None
best_ts = None
# Prefer latest indexing run at or before the merge timestamp
for d in idx_dirs:
ts = self._timestamp_from_run_dir(d)
if ts is None:
continue
cell_candidate = os.path.join(d, "cellfile.cell")
if not os.path.isfile(cell_candidate):
continue
if merge_ts is None or ts <= merge_ts:
if best_ts is None or ts > best_ts:
best_ts = ts
best_cell = cell_candidate
# Fallback to newest indexing run with a cell
if best_cell is None:
latest_cell = None
latest_ts = None
for d in idx_dirs:
ts = self._timestamp_from_run_dir(d)
if ts is None:
continue
cell_candidate = os.path.join(d, "cellfile.cell")
if os.path.isfile(cell_candidate):
if latest_ts is None or ts > latest_ts:
latest_ts = ts
latest_cell = cell_candidate
best_cell = latest_cell
if best_cell and os.path.isfile(best_cell):
return best_cell, self._parse_cell_file(best_cell)
return None, None
def _parse_cell_file(self, path: str) -> dict | None:
"""Lightweight parser for CrystFEL-style cell files."""
try:
data = {}
with open(path, "r", encoding="utf-8", errors="replace") as f:
for line in f:
line = line.strip()
if not line or line.startswith("#") or "=" not in line:
continue
k, v = line.split("=", 1)
data[k.strip().lower()] = v.strip()
req = ["a", "b", "c", "al", "be", "ga"]
if not all(k in data for k in req):
return data
try:
data["a"] = float(data["a"].split()[0])
data["b"] = float(data["b"].split()[0])
data["c"] = float(data["c"].split()[0])
data["al"] = float(data["al"].split()[0])
data["be"] = float(data["be"].split()[0])
data["ga"] = float(data["ga"].split()[0])
except Exception:
pass
return data
except Exception:
return None
def _parse_stream_cell(self, stream_path: str) -> dict | None:
"""
Parse the unit cell section from a CrystFEL stream header.
Looks for the block between '----- Begin unit cell -----' and '----- End unit cell -----'.
"""
try:
inside = False
data = {}
with open(stream_path, "r", encoding="utf-8", errors="replace") as f:
for line in f:
strip = line.strip()
if strip.startswith("----- Begin unit cell"):
inside = True
continue
if strip.startswith("----- End unit cell"):
break
if not inside or not strip or strip.startswith(";") or "=" not in strip:
continue
k, v = strip.split("=", 1)
data[k.strip().lower()] = v.strip()
if not data:
return None
# Try numeric parsing (robust to trailing units)
for k in ("a", "b", "c", "al", "be", "ga"):
if k in data:
try:
data[k] = float(str(data[k]).split()[0])
except Exception:
pass
return data
except Exception:
return None
def _update_inferred_cell(self, stream_path: str | None) -> None:
"""Update the UI with the cell inferred from the indexing run used for this stream."""
self._refresh_cell_source_selector()
indexing_path, indexing_params = self._infer_cell_for_stream(stream_path)
self._inferred_cell_path = indexing_path
source_label = "Indexing cell"
path, params = indexing_path, indexing_params
if self._selected_cell_source() == "workspace_refined":
refined_path = self._workspace_refined_cell_path()
if refined_path:
source_label = "Workspace refined cell"
path = refined_path
params = self._parse_cell_file(refined_path)
summary = ""
if params:
try:
a = params.get("a")
b = params.get("b")
c = params.get("c")
al = params.get("al")
be = params.get("be")
ga = params.get("ga")
cent = params.get("centering", "")
if all(isinstance(x, (float, int)) for x in (a, b, c, al, be, ga)):
summary = (
f"a = {a:.4f} Ã… b = {b:.4f} Ã… c = {c:.4f} Ã… "
f"\u03b1 = {al:.3f}° \u03b2 = {be:.3f}° \u03b3 = {ga:.3f}°"
)
elif all(x is not None for x in (a, b, c, al, be, ga)):
summary = (
f"a = {a} Ã… b = {b} Ã… c = {c} Ã… "
f"\u03b1 = {al}° \u03b2 = {be}° \u03b3 = {ga}°"
)
if cent:
summary = (summary + " " if summary else "") + f"centering: {cent}"
except Exception:
summary = ""
if summary:
summary = f"{source_label}: {summary}"
self.cell_params_display.setText(summary)
def _on_cell_source_changed(self) -> None:
stream_path = self.stream_combo.currentData() if hasattr(self, "stream_combo") else None
self._update_inferred_cell(stream_path)
latest_dir = self._latest_merge_dir_for_stream(stream_path) if stream_path else None
if latest_dir:
try:
self._run_check_hkl(latest_dir, stream_path=stream_path)
self._run_compare_hkl(latest_dir, stream_path=stream_path)
except Exception:
pass
def _stop_log_tail(self):
"""Stop tailing stdout.log from an existing run."""
if self._tail_timer is not None:
try:
self._tail_timer.stop()
except Exception:
pass
self._tail_timer = None
self._tail_path = None
self._tail_pos = 0
def _start_log_tail(self, log_path: str, start_from_end: bool = False):
"""
Start/continue tailing the given log file and feed new lines into the live parser,
useful when showing stats for a run not started from this session.
"""
try:
if not log_path or not os.path.isfile(log_path):
self._stop_log_tail()
return
self._tail_path = log_path
# Initial load
with open(log_path, 'r', encoding='utf-8', errors='replace') as f:
if start_from_end:
f.seek(0, os.SEEK_END)
self._tail_pos = f.tell()
else:
self._set_active_merge_output_dir(os.path.dirname(log_path))
self._live_reset_stats()
data = f.read()
self._tail_pos = f.tell()
if data:
self._live_parse_text(data)
self._live_commit_cycle()
self._save_merging_artifacts(self._active_merge_output_dir, self._live_stats_rows)
if self._tail_timer is None:
self._tail_timer = QTimer(self)
self._tail_timer.setInterval(500)
def _poll():
try:
if not self._tail_path:
return
with open(self._tail_path, 'r', encoding='utf-8', errors='replace') as f:
f.seek(self._tail_pos)
chunk = f.read()
self._tail_pos = f.tell()
if chunk:
self._live_parse_text(chunk)
except Exception:
pass
self._tail_timer.timeout.connect(_poll)
self._tail_timer.start()
except Exception:
self._stop_log_tail()
def _stream_dir_and_name(self, stream_path):
"""
Given a stream file path, return (directory, stream_basename)
"""
if not stream_path:
return None, None
folder = os.path.dirname(stream_path)
fname = os.path.basename(stream_path)
return folder, fname
def _next_merge_output_dir(self, stream_path):
"""
Create and return a fresh output directory inside the stream's folder.
Directory name is merge-runN where N is the next available integer.
This function *creates* the directory atomically to avoid races.
"""
folder, _ = self._stream_dir_and_name(stream_path)
if not folder:
return None
base = "merge-run"
n = 1
while True:
candidate = os.path.join(folder, f"{base}{n}")
try:
os.makedirs(candidate, exist_ok=False)
# pr-logs directory will be created later if needed; do not create here
return candidate
except FileExistsError:
n += 1
def _build_partialator_cmd(self, stream_path, output_dir, settings):
"""Build the Partialator CLI args from the selected stream and form settings."""
cmd = ["partialator"]
# Positional stream file
cmd.append(stream_path)
# Threads
if settings.get("threads"):
cmd += ["-j", str(settings["threads"])]
# Output file
out_hkl = "crystfel.hkl"
cmd += ["-o", out_hkl]
# Point group
if settings.get("pointgroup"):
pointgroup = self._crystfel_pointgroup_symbol(
settings.get("pointgroup"),
settings.get("unique_axis"),
)
cmd += ["-y", pointgroup]
# Min resolution
min_res = settings.get("min_res")
if min_res is not None:
if str(min_res) == "inf":
cmd += ["--min-res", "inf"]
else:
cmd += ["--min-res", str(min_res)]
# Iterations
if settings.get("iterations"):
cmd += ["--iterations", str(settings["iterations"])]
# Model
if settings.get("model"):
cmd += ["--model", str(settings["model"])]
# Polarisation fixed to none for electron diffraction
cmd += ["--polarisation", "none"]
# Min measurements
if settings.get("min_measurements"):
cmd += ["--min-measurements", str(settings["min_measurements"])]
# Max ADU
max_adu = settings.get("max_adu")
if max_adu is not None:
if str(max_adu) == "inf":
cmd += ["--max-adu", "inf"]
else:
cmd += ["--max-adu", str(max_adu)]
# Push resolution
push_res = settings.get("push_res")
if push_res is not None:
if str(push_res) == "inf":
cmd += ["--push-res", "inf"]
else:
cmd += ["--push-res", str(push_res)]
# no-Bscale
if settings.get("no_Bscale"):
cmd += ["--no-Bscale"]
cmd += ["--output-every-cycle"]
# Harvest file. Partialator diagnostic logs are file-heavy, so keep
# the log folder opt-in.
cmd += ["--harvest-file", "harvest.json"]
if settings.get("save_partialator_logs"):
cmd += ["--log-folder", "pr-logs"]
return cmd
@staticmethod
def _crystfel_pointgroup_symbol(pointgroup: str | None, unique_axis: str | None = None) -> str:
"""Return a CrystFEL point-group symbol, adding _uaX when requested."""
pg = str(pointgroup or "").strip().replace(" ", "")
ua = str(unique_axis or "").strip().lower()
pg_lower = pg.lower()
if ua in {"a", "b", "c"} and "_ua" not in pg_lower and not pg_lower.endswith(("_h", "_r")):
return f"{pg}_ua{ua}"
return pg
def _extract_h5_paths_from_stream(self, stream_path: str):
if not stream_path or not os.path.isfile(stream_path):
return []
base_dir = os.path.dirname(stream_path)
paths = []
seen = set()
try:
with open(stream_path, "r", encoding="utf-8", errors="replace") as handle:
for raw_line in handle:
line = raw_line.strip()
if not line:
continue
if line.startswith("Image filename:"):
token = line.split(":", 1)[1].strip()
elif line.startswith("Image filename ="):
token = line.split("=", 1)[1].strip()
else:
continue
if not token:
continue
token = token.strip("\"'")
token = token.split()[0].strip("\"'")
if token.startswith("file://"):
token = token[7:]
if "//" in token:
token = token.split("//", 1)[0]
if not os.path.isabs(token):
token = os.path.abspath(os.path.join(base_dir, token))
if token in seen:
continue
seen.add(token)
paths.append(token)
except Exception:
return []
return paths
def _write_merging_nxprocess(self, stream_path, output_dir, settings, cmd):
h5_paths = self._extract_h5_paths_from_stream(stream_path)
if not h5_paths:
return
output_hkl = os.path.join(output_dir, "crystfel.hkl") if output_dir else None
params = dict(settings or {})
params.update({
"command": " ".join(cmd) if isinstance(cmd, (list, tuple)) else str(cmd),
"stream_file": stream_path,
"output_dir": output_dir,
"output_hkl": output_hkl,
"harvest_file": "harvest.json",
})
if params.get("save_partialator_logs"):
params["log_folder"] = "pr-logs"
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_merging(
h5file,
program="cosedaUI.merging_window",
method="crystfel.partialator",
parameters=params,
input_path=stream_path,
output_path=output_hkl,
)
except Exception as exc:
log_print(f"Warning: failed to write NXprocess merging for {h5_path}: {exc}")
def _set_active_merge_output_dir(self, output_dir: str | None) -> None:
self._active_merge_output_dir = output_dir if output_dir and os.path.isdir(output_dir) else None
def _merge_artifact_dir(self, output_dir: str | None) -> str | None:
"""Place user-facing merge artifacts next to exports, above merge-run*."""
if not output_dir:
return None
try:
if os.path.basename(os.path.abspath(output_dir)).startswith("merge-run"):
parent = os.path.dirname(os.path.abspath(output_dir))
return parent if os.path.isdir(parent) else output_dir
except Exception:
pass
return output_dir
def _write_rows_csv(self, output_dir: str | None, filename: str, headers: list[str], keys: list[str], rows: list[dict]) -> None:
if not output_dir or not rows:
return
output_dir = self._merge_artifact_dir(output_dir)
try:
with open(os.path.join(output_dir, filename), "w", newline="", encoding="utf-8") as csvfile:
writer = csv.writer(csvfile)
writer.writerow(headers)
for row in rows:
writer.writerow([row.get(key, "") for key in keys])
except Exception as exc:
log_print(f"Warning: failed to write {filename}: {exc}")
def _save_figure_png(self, output_dir: str | None, filename: str, figure: Figure) -> None:
if not output_dir or figure is None:
return
output_dir = self._merge_artifact_dir(output_dir)
try:
figure.savefig(os.path.join(output_dir, filename), dpi=200, bbox_inches="tight")
except Exception as exc:
log_print(f"Warning: failed to write {filename}: {exc}")
def _save_merging_artifacts(self, output_dir: str | None, rows: list[dict]) -> None:
self._write_rows_csv(
output_dir,
"merging_table.csv",
["Cycle", *self._cc_col_headers],
["cycle", *self._cc_col_keys],
rows,
)
self._save_figure_png(output_dir, "merging_plot.png", self._cc_fig)
def _save_statistics_artifacts(self, output_dir: str | None, rows: list[dict]) -> None:
self._write_rows_csv(
output_dir,
"statistics_table.csv",
self._stats_col_headers,
self._stats_col_keys,
rows,
)
self._save_figure_png(output_dir, "statistics_plot.png", self._compl_fig)
def _save_split_half_artifacts(self, output_dir: str | None, rows: list[dict]) -> None:
self._write_rows_csv(
output_dir,
"split_half_table.csv",
["d(A)", "CC1/2", "CC*", "Rsplit (%)", "# refs", "Center 1/nm", "Min 1/nm", "Max 1/nm"],
["d_ang", "cc", "ccstar", "rsplit", "nref", "center", "min_inv_nm", "max_inv_nm"],
rows,
)
self._save_figure_png(output_dir, "split_half_plot.png", self._compare_hkl_fig)
def _parse_scaling_file(self, scaling_csv_path):
"""Parse scaling.csv into a list of row dicts keyed by column names."""
rows = []
if not scaling_csv_path or not os.path.isfile(scaling_csv_path):
return rows
try:
with open(scaling_csv_path, newline='', encoding='utf-8') as f:
reader = csv.DictReader(f)
for row in reader:
rows.append(row)
except Exception:
pass
return self._merge_scaling_rows_by_cycle(rows)
def _merge_scaling_rows_by_cycle(self, rows: list[dict]) -> list[dict]:
"""Merge duplicate sparse rows for the same Partialator cycle."""
merged = {}
order = []
for row in rows or []:
cycle = str(row.get("cycle", "")).strip()
if not cycle:
continue
if cycle not in merged:
merged[cycle] = dict(row)
order.append(cycle)
continue
for key, value in row.items():
if str(value).strip():
merged[cycle][key] = value
return [merged[cycle] for cycle in order]
def _populate_scaling_table(self, scaling_rows):
"""
Given list of scaling rows (dicts), populate self.scaling_table.
"""
table = self.scaling_table
table.setRowCount(0)
self._live_stats_rows = scaling_rows or []
if not scaling_rows:
self._update_cc_plot(scaling_rows)
return
for i, row in enumerate(scaling_rows):
table.insertRow(i)
vals = [
row.get("overall_cc_half", ""),
row.get("reflections", ""),
row.get("delta_cc_half", ""),
row.get("delta_cc_half_error", ""),
row.get("bad_crystals", ""),
row.get("ok_crystals", ""),
row.get("neg_delta_cc_half", "")
]
for j, val in enumerate(vals):
item = QTableWidgetItem(str(val))
table.setItem(i, j, item)
self._update_cc_plot(scaling_rows)
self._save_merging_artifacts(getattr(self, "_active_merge_output_dir", None), scaling_rows)
def _on_cc_header_clicked(self, col: int):
"""Toggle a merging-table column for plotting (max 2)."""
if col in self._cc_selected_cols:
self._cc_selected_cols.remove(col)
elif len(self._cc_selected_cols) < 2:
self._cc_selected_cols.append(col)
else:
# Replace the oldest selection
self._cc_selected_cols.pop(0)
self._cc_selected_cols.append(col)
self._highlight_cc_headers()
self._update_cc_plot()
def _highlight_cc_headers(self):
"""Highlight selected merging-table column headers."""
from PyQt6.QtGui import QColor, QBrush
colors = ["#1f77b4", "#d62728"]
for i, base in enumerate(self._cc_col_headers):
item = self.scaling_table.horizontalHeaderItem(i)
if item is None:
item = QTableWidgetItem(base)
self.scaling_table.setHorizontalHeaderItem(i, item)
if i in self._cc_selected_cols:
idx = self._cc_selected_cols.index(i)
item.setForeground(Qt.GlobalColor.white)
item.setBackground(QBrush(QColor(colors[idx])))
else:
item.setForeground(Qt.GlobalColor.black)
item.setBackground(QBrush())
def _update_cc_plot(self, rows=None):
"""
Update the cycle-metric plot. Plots up to 2 selected columns with
dual y-axes. If rows is None, uses cached live rows.
"""
try:
if rows is None:
rows = getattr(self, "_live_stats_rows", None)
ax = self._cc_ax
ax.clear()
for old_ax in self._cc_fig.axes[1:]:
old_ax.remove()
ax.set_xlabel("Cycle")
ax.grid(True, which="both", linestyle="--", linewidth=0.5, alpha=0.5)
colors = ["#1f77b4", "#d62728"]
if rows and self._cc_selected_cols:
cycles = []
for r in rows:
try:
cycles.append(int(str(r.get("cycle", "")).strip()))
except Exception:
cycles.append(None)
for idx, col in enumerate(self._cc_selected_cols):
key = self._cc_col_keys[col]
label = self._cc_col_headers[col]
if idx == 0:
cur_ax = ax
else:
cur_ax = ax.twinx()
xs, ys = [], []
for i, r in enumerate(rows):
if cycles[i] is None:
continue
try:
y = float(str(r.get(key, "")).strip())
except Exception:
continue
xs.append(cycles[i])
ys.append(y)
pts = sorted(zip(xs, ys), key=lambda t: t[0])
xs = [p[0] for p in pts]
ys = [p[1] for p in pts]
color = colors[idx]
if xs and ys:
cur_ax.plot(xs, ys, color=color, label=label)
cur_ax.set_ylabel(label, color=color)
cur_ax.tick_params(axis="y", labelcolor=color)
handles, labels = [], []
for a in self._cc_fig.axes:
h, l = a.get_legend_handles_labels()
handles.extend(h)
labels.extend(l)
if handles:
ax.legend(handles, labels, loc="best", fontsize="small")
self._cc_canvas.draw_idle()
except Exception:
pass
def _find_indexing_cell_file_for_stream(self, stream_path: str) -> str | None:
"""Find an actual cellfile.cell on disk for the given stream, skipping stream headers."""
# 1) INI-provided cell
ini_cell = self._cell_from_current_ini()
if ini_cell and os.path.isfile(ini_cell):
return ini_cell
# 2) Search indexing run directories for cellfile.cell
if not stream_path:
return None
stream_dir = os.path.dirname(os.path.abspath(stream_path))
direct_cell = os.path.join(stream_dir, "cellfile.cell")
if os.path.isfile(direct_cell):
return direct_cell
merge_dir = os.path.dirname(stream_path)
parent_dir = os.path.dirname(merge_dir)
merge_ts = self._timestamp_from_merge_dir(merge_dir)
ini_dir = getattr(self, "ini_directory", parent_dir)
run_dirs_candidates = set()
for root in {parent_dir, ini_dir, os.path.dirname(ini_dir)}:
try:
for pattern in ("indexingintegration_*", "ici_*"):
for d in glob.glob(os.path.join(root, pattern)):
if os.path.isdir(d):
run_dirs_candidates.add(d)
except Exception:
continue
idx_dirs = sorted(run_dirs_candidates)
best_cell = None
best_ts = None
for d in idx_dirs:
ts = self._timestamp_from_run_dir(d)
if ts is None:
continue
cell_candidate = os.path.join(d, "cellfile.cell")
if not os.path.isfile(cell_candidate):
continue
if merge_ts is None or ts <= merge_ts:
if best_ts is None or ts > best_ts:
best_ts = ts
best_cell = cell_candidate
if best_cell is None:
for d in idx_dirs:
ts = self._timestamp_from_run_dir(d)
if ts is None:
continue
cell_candidate = os.path.join(d, "cellfile.cell")
if os.path.isfile(cell_candidate):
if best_ts is None or ts > best_ts:
best_ts = ts
best_cell = cell_candidate
return best_cell
def _find_cell_file_for_stream(self, stream_path: str) -> str | None:
"""Return the selected cell file for merge analysis commands."""
if self._selected_cell_source() == "workspace_refined":
refined_cell = self._workspace_refined_cell_path()
if refined_cell and os.path.isfile(refined_cell):
return refined_cell
self.console.appendPlainText(">>> workspace refined cell unavailable; falling back to indexing cell\n")
return self._find_indexing_cell_file_for_stream(stream_path)
def _build_check_hkl_cmd(self, hkl_path: str, output_dir: str, cell_path: str, pointgroup: str) -> list[str]:
"""Build the check_hkl CLI command."""
shell_file = self._check_hkl_shell_path(output_dir)
cmd = [
"check_hkl",
"-p", cell_path,
"-y", pointgroup,
f"--shell-file={shell_file}",
hkl_path,
]
return cmd
def _cell_analysis_suffix(self) -> str:
return "_workspace_refined" if self._selected_cell_source() == "workspace_refined" else ""
def _check_hkl_shell_path(self, output_dir: str) -> str:
return os.path.join(output_dir, f"shells{self._cell_analysis_suffix()}.dat")
def _parse_check_hkl_output(self, shell_file: str) -> list[dict]:
"""Parse the whitespace-delimited shells.dat produced by check_hkl."""
rows = []
if not shell_file or not os.path.isfile(shell_file):
return rows
keys = [
"center", "refs", "possible", "compl", "meas",
"red", "snr", "mean_i", "d_ang", "min_inv_nm", "max_inv_nm",
]
try:
with open(shell_file, "r", encoding="utf-8", errors="replace") as f:
for line in f:
stripped = line.strip()
if not stripped or stripped.startswith("Center") or stripped.startswith("-"):
continue
parts = stripped.split()
if len(parts) < 11:
continue
row = {}
for i, k in enumerate(keys):
row[k] = parts[i]
rows.append(row)
except Exception:
pass
return rows
def _populate_check_hkl_table(self, rows: list[dict]) -> None:
"""Fill the check_hkl statistics table."""
table = self.check_hkl_table
table.setRowCount(0)
if not rows:
self._update_compl_plot([])
return
for i, row in enumerate(rows):
table.insertRow(i)
vals = [
row.get("center", ""), row.get("refs", ""), row.get("possible", ""),
row.get("compl", ""), row.get("meas", ""), row.get("red", ""),
row.get("snr", ""), row.get("mean_i", ""), row.get("d_ang", ""),
row.get("min_inv_nm", ""), row.get("max_inv_nm", ""),
]
for j, val in enumerate(vals):
table.setItem(i, j, QTableWidgetItem(str(val)))
self._update_compl_plot(rows)
self._save_statistics_artifacts(getattr(self, "_active_merge_output_dir", None), rows)
def _on_stats_header_clicked(self, col: int):
"""Toggle a statistics-table column for plotting (max 2, only metric columns)."""
if col not in self._stats_selectable_cols:
return
if col in self._stats_selected_cols:
self._stats_selected_cols.remove(col)
elif len(self._stats_selected_cols) < 2:
self._stats_selected_cols.append(col)
else:
self._stats_selected_cols.pop(0)
self._stats_selected_cols.append(col)
self._highlight_stats_headers()
self._update_compl_plot()
def _highlight_stats_headers(self):
"""Highlight selected statistics-table column headers."""
colors = ["#1f77b4", "#d62728"]
from PyQt6.QtGui import QColor, QBrush
for i, base in enumerate(self._stats_col_headers):
item = self.check_hkl_table.horizontalHeaderItem(i)
if item is None:
item = QTableWidgetItem(base)
self.check_hkl_table.setHorizontalHeaderItem(i, item)
if i in self._stats_selected_cols:
idx = self._stats_selected_cols.index(i)
item.setForeground(Qt.GlobalColor.white)
item.setBackground(QBrush(QColor(colors[idx])))
else:
item.setForeground(Qt.GlobalColor.black)
item.setBackground(QBrush())
def _update_compl_plot(self, rows: list[dict] | None = None) -> None:
"""Plot up to 2 selected columns vs Resolution d(A) with dual y-axes."""
try:
if rows is not None:
self._check_hkl_rows = rows
else:
rows = getattr(self, "_check_hkl_rows", None) or []
ax = self._compl_ax
ax.clear()
for child_ax in self._compl_fig.axes[1:]:
child_ax.remove()
ax.set_xlabel("Resolution d (Ã…)")
ax.grid(True, which="both", linestyle="--", linewidth=0.5, alpha=0.5)
# Extract d values and sort indices by decreasing d
ds_raw = []
for r in rows:
try:
ds_raw.append(float(r.get("d_ang", "")))
except (ValueError, TypeError):
ds_raw.append(None)
order = sorted(range(len(ds_raw)), key=lambda i: -(ds_raw[i] or 0))
ds = [ds_raw[i] for i in order if ds_raw[i] is not None]
colors = ["#1f77b4", "#d62728"]
if ds and self._stats_selected_cols:
for idx, col in enumerate(self._stats_selected_cols):
key = self._stats_col_keys[col]
label = self._stats_col_headers[col]
if idx == 0:
cur_ax = ax
else:
cur_ax = ax.twinx()
vals = []
for i in order:
if ds_raw[i] is None:
continue
try:
vals.append(float(rows[i].get(key, "")))
except (ValueError, TypeError):
vals.append(0.0)
color = colors[idx]
if vals:
cur_ax.plot(ds, vals, color=color, label=label)
cur_ax.set_ylabel(label, color=color)
cur_ax.tick_params(axis="y", labelcolor=color)
handles, labels = [], []
for a in self._compl_fig.axes:
h, l = a.get_legend_handles_labels()
handles.extend(h)
labels.extend(l)
if handles:
ax.legend(handles, labels, loc="best", fontsize="small")
if ds:
ax.invert_xaxis()
self._compl_canvas.draw_idle()
except Exception:
pass
def _run_check_hkl(self, output_dir: str, stream_path: str | None = None) -> None:
"""Run check_hkl on crystfel.hkl in output_dir via QProcess."""
self._set_active_merge_output_dir(output_dir)
hkl_path = os.path.join(output_dir, "crystfel.hkl")
if not os.path.isfile(hkl_path):
return
# If shells.dat already exists, just load it
shell_file = self._check_hkl_shell_path(output_dir)
if os.path.isfile(shell_file):
rows = self._parse_check_hkl_output(shell_file)
self._populate_check_hkl_table(rows)
return
# Find real cell file from indexing run
stream_for_lookup = stream_path or self.stream_combo.currentData()
cell_path = self._find_cell_file_for_stream(stream_for_lookup)
if not cell_path:
self.console.appendPlainText(">>> check_hkl: skipped (no cell file found)\n")
return
pointgroup = self._crystfel_pointgroup_symbol(
self.pg_combo.currentText().strip(),
self.ua_combo.currentText().strip(),
)
if not pointgroup:
self.console.appendPlainText(">>> check_hkl: skipped (no point group set)\n")
return
cmd = self._build_check_hkl_cmd(hkl_path, output_dir, cell_path, pointgroup)
self.console.appendPlainText(f"\n>>> Running: {' '.join(cmd)}\n")
proc = QProcess(self)
proc.setProcessChannelMode(QProcess.ProcessChannelMode.MergedChannels)
proc.setWorkingDirectory(output_dir)
def _on_check_hkl_output():
try:
data = proc.readAllStandardOutput()
if data:
self.console.appendPlainText(bytes(data).decode("utf-8", errors="replace"))
except Exception:
pass
def _on_check_hkl_finished(code, _status):
_on_check_hkl_output()
if code == 0:
rows = self._parse_check_hkl_output(shell_file)
self._populate_check_hkl_table(rows)
self.console.appendPlainText(f">>> check_hkl finished ({len(rows)} shells)\n")
else:
self.console.appendPlainText(f">>> check_hkl failed with exit code {code}\n")
proc.readyReadStandardOutput.connect(_on_check_hkl_output)
proc.finished.connect(_on_check_hkl_finished)
# Keep a reference so it doesn't get garbage collected
self._check_hkl_proc = proc
proc.start(cmd[0], cmd[1:])
def _compare_hkl_specs(self) -> list[tuple[str, str, str]]:
return [
("CC", "cc", "CC1/2"),
("CCstar", "ccstar", "CC*"),
("Rsplit", "rsplit", "Rsplit (%)"),
]
def _compare_hkl_shell_path(self, output_dir: str, fom: str) -> str:
return os.path.join(output_dir, f"compare_hkl_{fom.lower()}{self._cell_analysis_suffix()}.dat")
def _build_compare_hkl_cmd(
self,
hkl1_path: str,
hkl2_path: str,
output_dir: str,
cell_path: str,
pointgroup: str,
fom: str,
) -> list[str]:
return [
"compare_hkl",
hkl1_path,
hkl2_path,
"-p", cell_path,
"-y", pointgroup,
f"--fom={fom}",
f"--shell-file={self._compare_hkl_shell_path(output_dir, fom)}",
]
def _parse_compare_hkl_shell_file(self, shell_file: str, value_key: str) -> list[dict]:
rows = []
if not shell_file or not os.path.isfile(shell_file):
return rows
try:
with open(shell_file, "r", encoding="utf-8", errors="replace") as f:
for line in f:
stripped = line.strip()
if not stripped or stripped.startswith("1/d") or stripped.startswith("-"):
continue
parts = stripped.split()
if len(parts) < 6:
continue
rows.append({
"center": parts[0],
value_key: parts[1],
"nref": parts[2],
"d_ang": parts[3],
"min_inv_nm": parts[4],
"max_inv_nm": parts[5],
})
except Exception:
pass
return rows
def _parse_compare_hkl_outputs(self, output_dir: str) -> list[dict]:
merged_rows: list[dict] = []
for fom, key, _label in self._compare_hkl_specs():
rows = self._parse_compare_hkl_shell_file(self._compare_hkl_shell_path(output_dir, fom), key)
for idx, row in enumerate(rows):
while len(merged_rows) <= idx:
merged_rows.append({})
merged_rows[idx].update(row)
return merged_rows
def _populate_compare_hkl_table(self, rows: list[dict]) -> None:
table = self.compare_hkl_table
table.setRowCount(0)
self._compare_hkl_rows = rows or []
if not rows:
self._update_compare_hkl_plot([])
return
for i, row in enumerate(rows):
table.insertRow(i)
vals = [
row.get("d_ang", ""),
row.get("cc", ""),
row.get("ccstar", ""),
row.get("rsplit", ""),
row.get("nref", ""),
row.get("center", ""),
row.get("min_inv_nm", ""),
row.get("max_inv_nm", ""),
]
for j, val in enumerate(vals):
table.setItem(i, j, QTableWidgetItem(str(val)))
self._update_compare_hkl_plot(rows)
self._save_split_half_artifacts(getattr(self, "_active_merge_output_dir", None), rows)
def _update_compare_hkl_plot(self, rows: list[dict] | None = None) -> None:
try:
if rows is None:
rows = getattr(self, "_compare_hkl_rows", None) or []
ax = self._compare_hkl_ax
ax.clear()
for child_ax in self._compare_hkl_fig.axes[1:]:
child_ax.remove()
ax.set_xlabel("Resolution d (Ã…)")
ax.set_ylabel("CC")
ax.grid(True, which="both", linestyle="--", linewidth=0.5, alpha=0.5)
points = []
for row in rows or []:
try:
d = float(row.get("d_ang", ""))
except Exception:
continue
points.append((d, row))
points.sort(key=lambda item: -item[0])
if not points:
self._compare_hkl_canvas.draw_idle()
return
ds = [d for d, _row in points]
for key, label, color in (("cc", "CC1/2", "#1f77b4"), ("ccstar", "CC*", "#2ca02c")):
vals = []
for _d, row in points:
try:
vals.append(float(row.get(key, "")))
except Exception:
vals.append(float("nan"))
ax.plot(ds, vals, marker="o", markersize=3, linewidth=1.2, color=color, label=label)
rsplit_vals = []
has_rsplit = False
for _d, row in points:
try:
value = float(row.get("rsplit", ""))
has_rsplit = True
except Exception:
value = float("nan")
rsplit_vals.append(value)
if has_rsplit:
ax2 = ax.twinx()
ax2.set_ylabel("Rsplit (%)", color="#d62728")
ax2.tick_params(axis="y", labelcolor="#d62728")
ax2.plot(ds, rsplit_vals, marker="o", markersize=3, linewidth=1.2, color="#d62728", label="Rsplit")
handles, labels = [], []
for plot_ax in self._compare_hkl_fig.axes:
h, l = plot_ax.get_legend_handles_labels()
handles.extend(h)
labels.extend(l)
if handles:
ax.legend(handles, labels, loc="best", fontsize="small")
ax.invert_xaxis()
self._compare_hkl_canvas.draw_idle()
except Exception:
pass
def _run_compare_hkl(self, output_dir: str, stream_path: str | None = None) -> None:
"""Run compare_hkl on Partialator split-half HKL files and load shell metrics."""
self._set_active_merge_output_dir(output_dir)
hkl1_path = os.path.join(output_dir, "crystfel.hkl1")
hkl2_path = os.path.join(output_dir, "crystfel.hkl2")
if not (os.path.isfile(hkl1_path) and os.path.isfile(hkl2_path)):
self._populate_compare_hkl_table([])
self.console.appendPlainText(">>> compare_hkl: skipped (no crystfel.hkl1/hkl2 files found)\n")
return
shell_files = [self._compare_hkl_shell_path(output_dir, fom) for fom, _key, _label in self._compare_hkl_specs()]
if all(os.path.isfile(path) for path in shell_files):
self._populate_compare_hkl_table(self._parse_compare_hkl_outputs(output_dir))
return
stream_for_lookup = stream_path or self.stream_combo.currentData()
cell_path = self._find_cell_file_for_stream(stream_for_lookup)
if not cell_path:
self.console.appendPlainText(">>> compare_hkl: skipped (no cell file found)\n")
return
pointgroup = self._crystfel_pointgroup_symbol(
self.pg_combo.currentText().strip(),
self.ua_combo.currentText().strip(),
)
if not pointgroup:
self.console.appendPlainText(">>> compare_hkl: skipped (no point group set)\n")
return
queue = list(self._compare_hkl_specs())
self._compare_hkl_queue = queue
def _run_next():
if not queue:
rows = self._parse_compare_hkl_outputs(output_dir)
self._populate_compare_hkl_table(rows)
self.console.appendPlainText(f">>> compare_hkl finished ({len(rows)} shells)\n")
return
fom, _key, _label = queue.pop(0)
cmd = self._build_compare_hkl_cmd(hkl1_path, hkl2_path, output_dir, cell_path, pointgroup, fom)
self.console.appendPlainText(f"\n>>> Running: {' '.join(cmd)}\n")
proc = QProcess(self)
proc.setProcessChannelMode(QProcess.ProcessChannelMode.MergedChannels)
proc.setWorkingDirectory(output_dir)
def _append_output():
try:
data = proc.readAllStandardOutput()
if data:
self.console.appendPlainText(bytes(data).decode("utf-8", errors="replace"))
except Exception:
pass
def _finished(code, _status):
_append_output()
if code != 0:
self.console.appendPlainText(f">>> compare_hkl {fom} failed with exit code {code}\n")
_run_next()
proc.readyReadStandardOutput.connect(_append_output)
proc.finished.connect(_finished)
self._compare_hkl_proc = proc
proc.start(cmd[0], cmd[1:])
_run_next()
def __init__(self, parent, ini_directory, ini_file_path, workspace_path):
super().__init__(parent)
self.setWindowTitle("Merge")
self.ini_directory = ini_directory
self.ini_file_path = ini_file_path
self.workspace_path = workspace_path # path to the .cosedawsp (or similar) workspace file
# --- Live parsing state for scaling output ---
self._live_stats_rows = [] # list of dicts per cycle
self._live_current = {} # dict accumulating current cycle
# Compile regexes for live parsing
num_re = r"[-+]?(?:\d+(?:\.\d*)?|\.\d+|nan|inf)(?:[eE][-+]?\d+)?"
self._re_cycle = re.compile(r"Scaling and refinement cycle (\d+) of \d+")
self._re_cchalf = re.compile(
rf"Overall CChalf = ({num_re}) % \((\d+) reflections\)",
re.IGNORECASE,
)
self._re_delta = re.compile(
rf"deltaCChalf = ({num_re})\s*(?:±|\+/-)\s*({num_re}) %",
re.IGNORECASE,
)
self._re_bad = re.compile(r"(\d+) bad crystals:")
self._re_ok = re.compile(r"(\d+) OK")
self._re_neg = re.compile(r"(\d+) negative delta CC(?:½|1/2)")
# --- Log tailing state for existing runs ---
self._tail_timer: QTimer | None = None
self._tail_path: str | None = None
self._tail_pos: int = 0
self._active_merge_output_dir: str | None = None
# Track running merges by stream path
self._running_by_stream = {}
# Tracks whether the current run was aborted by user
self._user_aborted = False
# Cache workspace INI mapping for later lookups
self._workspace_ini_map = self._build_workspace_ini_map()
self._init_ui()
def _run_marker_path(self, out_dir: str, name: str) -> str:
return os.path.join(out_dir, f".{name}")
def _mark_run_state(self, out_dir: str, state: str) -> None:
# state in {"RUNNING", "DONE", "ABORTED"}
try:
# Clear previous markers
for m in ("RUNNING", "DONE", "ABORTED"):
p = self._run_marker_path(out_dir, m)
if os.path.exists(p):
try:
os.remove(p)
except Exception:
pass
# Write new marker
with open(self._run_marker_path(out_dir, state), "w", encoding="utf-8") as f:
f.write(state)
except Exception:
pass
def _latest_run_state(self, stream_path: str) -> tuple[str|None, str|None]:
"""
Return (latest_dir, state) where state is one of {"RUNNING","DONE","ABORTED",None} depending on marker files.
"""
latest_dir = self._latest_merge_dir_for_stream(stream_path)
if not latest_dir:
return None, None
if os.path.exists(self._run_marker_path(latest_dir, "RUNNING")):
return latest_dir, "RUNNING"
if os.path.exists(self._run_marker_path(latest_dir, "DONE")):
return latest_dir, "DONE"
if os.path.exists(self._run_marker_path(latest_dir, "ABORTED")):
return latest_dir, "ABORTED"
# Fallback: infer DONE if crystallographic output exists
if os.path.exists(os.path.join(latest_dir, "crystfel.hkl")):
return latest_dir, "DONE"
return latest_dir, None
def _clone_merge_folder_for_restart(self, stream_path: str) -> str | None:
"""
When a merged stream's folder already has a previous merge-run*, create a *new*
sibling merge folder with a fresh timestamp, copy the stream file (renamed to
the new folder basename) and copy the settings JSON if present.
Returns the full path to the new stream file, or None on failure.
"""
try:
import shutil
if not stream_path or not os.path.isfile(stream_path):
return None
parent_dir = self._workspace_root_dir()
base_name = self._workspace_stem()
ts = datetime.datetime.now().strftime('%Y%m%d_%H%M%S')
new_dir = os.path.join(parent_dir, f"{base_name}_merge_{ts}")
os.makedirs(new_dir, exist_ok=False)
# New stream path is <new_dir>/<basename(new_dir)>.stream
new_base = os.path.basename(new_dir)
new_stream_path = os.path.join(new_dir, f"{new_base}.stream")
# Copy stream file
shutil.copy2(stream_path, new_stream_path)
# Copy stream provenance manifest if present
try:
old_manifest = self._combined_stream_manifest_path(stream_path)
new_manifest = self._combined_stream_manifest_path(new_stream_path)
if old_manifest and new_manifest and os.path.isfile(old_manifest):
shutil.copy2(old_manifest, new_manifest)
try:
with open(new_manifest, "r", encoding="utf-8") as mf:
manifest = json.load(mf)
if isinstance(manifest, dict):
manifest["cloned_from_stream"] = stream_path
manifest["output_stream"] = new_stream_path
with open(new_manifest, "w", encoding="utf-8") as mf:
json.dump(manifest, mf, indent=2)
except Exception:
pass
except Exception:
pass
# Copy settings JSON if present
try:
old_json = self._settings_json_path_for_stream(stream_path)
if old_json and os.path.isfile(old_json):
new_json = self._settings_json_path_for_stream(new_stream_path)
if new_json:
shutil.copy2(old_json, new_json)
# Ensure the copied JSON points at the new stream path
try:
with open(new_json, "r", encoding="utf-8") as jf:
d = json.load(jf)
d["streamfile"] = new_stream_path
d = self._add_stream_manifest_to_settings(d, new_stream_path)
with open(new_json, "w", encoding="utf-8") as jf:
json.dump(d, jf, indent=2)
except Exception:
pass
except Exception:
pass
return new_stream_path
except Exception as e:
log_print(f"Warning: could not clone merge folder: {e}")
return None
def _live_reset_stats(self):
"""Reset live-parsing stats and clear the UI table and plot."""
self._live_stats_rows = []
self._live_current = {}
self.scaling_table.setRowCount(0)
if hasattr(self, "_cc_ax"):
self._cc_ax.clear()
for old_ax in self._cc_fig.axes[1:]:
old_ax.remove()
self._cc_ax.set_xlabel("Cycle")
self._cc_canvas.draw_idle()
def _live_commit_cycle(self):
"""If current cycle is filled, append it to rows and refresh table."""
cur = self._live_current
metric_keys = (
"overall_cc_half",
"reflections",
"delta_cc_half",
"delta_cc_half_error",
"bad_crystals",
"ok_crystals",
"neg_delta_cc_half",
)
if cur.get("cycle") and any(str(cur.get(key, "")).strip() for key in metric_keys):
# Normalize all required fields as strings (for table display)
row = {
"cycle": str(cur.get("cycle", "")),
"overall_cc_half": str(cur.get("overall_cc_half", "")),
"reflections": str(cur.get("reflections", "")),
"delta_cc_half": str(cur.get("delta_cc_half", "")),
"delta_cc_half_error": str(cur.get("delta_cc_half_error", "")),
"bad_crystals": str(cur.get("bad_crystals", "")),
"ok_crystals": str(cur.get("ok_crystals", "")),
"neg_delta_cc_half": str(cur.get("neg_delta_cc_half", "")),
}
for idx, existing in enumerate(self._live_stats_rows):
if existing.get("cycle") != row["cycle"]:
continue
merged = dict(existing)
for key, value in row.items():
if str(value).strip():
merged[key] = value
self._live_stats_rows[idx] = merged
break
else:
self._live_stats_rows.append(row)
self._live_current = {}
self._live_refresh_table()
def _live_refresh_table(self):
"""Re-populate scaling_table from live-parsed rows."""
table = self.scaling_table
table.setRowCount(0)
rows = self._live_stats_rows
if not rows:
self._update_cc_plot(rows)
return
for i, row in enumerate(rows):
table.insertRow(i)
vals = [
row.get("overall_cc_half", ""),
row.get("reflections", ""),
row.get("delta_cc_half", ""),
row.get("delta_cc_half_error", ""),
row.get("bad_crystals", ""),
row.get("ok_crystals", ""),
row.get("neg_delta_cc_half", "")
]
for j, val in enumerate(vals):
item = QTableWidgetItem(str(val))
table.setItem(i, j, item)
self._update_cc_plot(self._live_stats_rows)
def _live_parse_text(self, text: str):
"""Parse Partialator output lines as they arrive, updating stats table live."""
changed = False
for line in text.splitlines():
m = self._re_cycle.search(line)
if m:
# New cycle: commit previous, start new
self._live_commit_cycle()
self._live_current = {"cycle": m.group(1)}
changed = True
continue
m = self._re_cchalf.search(line)
if m:
self._live_current["overall_cc_half"] = m.group(1)
self._live_current["reflections"] = m.group(2)
changed = True
continue
m = self._re_delta.search(line)
if m:
self._live_current["delta_cc_half"] = m.group(1)
self._live_current["delta_cc_half_error"] = m.group(2)
changed = True
continue
m = self._re_bad.search(line)
if m:
self._live_current["bad_crystals"] = m.group(1)
changed = True
continue
m = self._re_ok.search(line)
if m:
self._live_current["ok_crystals"] = m.group(1)
changed = True
continue
m = self._re_neg.search(line)
if m:
self._live_current["neg_delta_cc_half"] = m.group(1)
changed = True
continue
if changed:
# Show live-updated cycles (append a preview row for current cycle if it has a number)
rows = self._live_stats_rows.copy()
cur = self._live_current
if cur.get("cycle"):
row = {
"cycle": str(cur.get("cycle", "")),
"overall_cc_half": str(cur.get("overall_cc_half", "")),
"reflections": str(cur.get("reflections", "")),
"delta_cc_half": str(cur.get("delta_cc_half", "")),
"delta_cc_half_error": str(cur.get("delta_cc_half_error", "")),
"bad_crystals": str(cur.get("bad_crystals", "")),
"ok_crystals": str(cur.get("ok_crystals", "")),
"neg_delta_cc_half": str(cur.get("neg_delta_cc_half", "")),
}
rows.append(row)
table = self.scaling_table
table.setRowCount(0)
for i, row in enumerate(rows):
table.insertRow(i)
vals = [
row.get("overall_cc_half", ""),
row.get("reflections", ""),
row.get("delta_cc_half", ""),
row.get("delta_cc_half_error", ""),
row.get("bad_crystals", ""),
row.get("ok_crystals", ""),
row.get("neg_delta_cc_half", "")
]
for j, val in enumerate(vals):
item = QTableWidgetItem(str(val))
table.setItem(i, j, item)
# Also update plot with preview row
self._update_cc_plot(self._live_stats_rows + ([{
"cycle": self._live_current.get("cycle",""),
"overall_cc_half": self._live_current.get("overall_cc_half","")
}] if self._live_current.get("cycle") else []))
def _workspace_stem(self) -> str:
try:
import os
return os.path.splitext(os.path.basename(self.workspace_path))[0]
except Exception:
# Fallback to INI dir name if workspace is unavailable
return os.path.basename(self.ini_directory.rstrip(os.sep))
def _workspace_root_dir(self) -> str:
"""Return the directory where workspace-level merge artifacts belong."""
path = getattr(self, "workspace_path", None)
if isinstance(path, str) and path:
abs_path = os.path.abspath(path)
if os.path.isfile(abs_path):
return os.path.dirname(abs_path)
if os.path.isdir(abs_path):
return abs_path
return os.path.abspath(self.ini_directory.rstrip(os.sep))
def _merge_output_roots(self) -> list[str]:
"""Return roots to scan for combined streams, including legacy INI-local output."""
roots = []
seen = set()
for root in (self._workspace_root_dir(), self.ini_directory.rstrip(os.sep)):
if not root:
continue
root = os.path.abspath(root)
if root in seen:
continue
seen.add(root)
roots.append(root)
return roots
def _display_path(self, path: str) -> str:
"""Display paths relative to the workspace root when possible."""
if not path or not os.path.isabs(path):
return path
try:
return os.path.relpath(path, self._workspace_root_dir())
except Exception:
return path
def _build_workspace_ini_map(self) -> dict[str, list[str]]:
"""Map base directories to the INI files they contain."""
ini_map: dict[str, list[str]] = {}
for ini_path in self._workspace_ini_paths():
base_dir = os.path.dirname(os.path.abspath(ini_path))
ini_map.setdefault(base_dir, []).append(os.path.abspath(ini_path))
for run_root in self._run_roots_for_ini(ini_path):
ini_map.setdefault(run_root, [])
abs_ini = os.path.abspath(ini_path)
if abs_ini not in ini_map[run_root]:
ini_map[run_root].append(abs_ini)
# Ensure current INI is represented even if workspace file is missing it
if self.ini_file_path:
cur_base = os.path.dirname(os.path.abspath(self.ini_file_path))
ini_map.setdefault(cur_base, [])
cur_ini = os.path.abspath(self.ini_file_path)
if cur_ini not in ini_map[cur_base]:
ini_map[cur_base].append(cur_ini)
return ini_map
def _workspace_ini_paths(self) -> list[str]:
"""Return list of INI file paths listed in the workspace file (if available)."""
paths: list[str] = []
seen: set[str] = set()
def add_path(candidate: str | None, base_dir: str | None = None) -> None:
if not candidate:
return
candidate = candidate.strip()
if not candidate or candidate.startswith("#"):
return
if not os.path.isabs(candidate) and base_dir:
candidate = os.path.abspath(os.path.join(base_dir, candidate))
else:
candidate = os.path.abspath(candidate)
if candidate.lower().endswith(".ini") and os.path.isfile(candidate) and candidate not in seen:
seen.add(candidate)
paths.append(candidate)
parent_obj = self.parent() if callable(getattr(self, "parent", None)) else None
parent_workspace = getattr(parent_obj, "workspace_ini_paths", None)
if isinstance(parent_workspace, dict):
for ini_path in parent_workspace.values():
add_path(ini_path)
add_path(getattr(self, "ini_file_path", None))
path = getattr(self, "workspace_path", None)
if not path or not isinstance(path, str):
return paths
try:
if not os.path.isfile(path):
return paths
ws_dir = os.path.dirname(os.path.abspath(path))
with open(path, "r", encoding="utf-8") as f:
for line in f:
add_path(line, ws_dir)
except Exception:
return paths
return paths
def _run_roots_for_ini(self, ini_path: str) -> list[str]:
"""Return directories that can contain indexing run folders for an INI."""
roots: list[str] = []
seen: set[str] = set()
def add_root(path: str | None) -> None:
if not path:
return
path = os.path.abspath(path)
if path not in seen:
seen.add(path)
roots.append(path)
if not ini_path:
return roots
ini_path = os.path.abspath(ini_path)
base_dir = os.path.dirname(ini_path)
add_root(base_dir)
add_root(os.path.join(base_dir, "output"))
add_root(os.path.join(base_dir, "runs"))
try:
parser = configparser.ConfigParser()
parser.read(ini_path, encoding="utf-8")
outputfolder = parser.get("Paths", "outputfolder", fallback="").strip()
if outputfolder:
if os.path.isabs(outputfolder):
add_root(outputfolder)
else:
add_root(os.path.join(base_dir, outputfolder))
except Exception:
pass
return roots
def _timestamp_from_run_dir(self, run_dir: str) -> datetime.datetime | None:
"""Parse the timestamp from a simple or ICI indexing run directory name."""
name = os.path.basename(os.path.abspath(run_dir))
for prefix in ("indexingintegration_", "ici_"):
if not name.startswith(prefix):
continue
try:
return datetime.datetime.strptime(name[len(prefix):], "%Y%m%d_%H%M%S")
except ValueError:
return None
return None
def _timestamp_from_merge_dir(self, merge_dir: str) -> datetime.datetime | None:
"""Parse the merge timestamp, allowing traceability suffixes after it."""
name = os.path.basename(os.path.abspath(merge_dir))
match = re.search(r"_merge_(\d{8}_\d{6})", name)
if not match:
return None
try:
return datetime.datetime.strptime(match.group(1), "%Y%m%d_%H%M%S")
except ValueError:
return None
@staticmethod
def _safe_name_fragment(value: str | None, fallback: str = "selection", max_len: int = 80) -> str:
"""Return a compact filesystem-safe ASCII fragment for merge folder names."""
text = str(value or "").strip() or fallback
text = re.sub(r"[^A-Za-z0-9._-]+", "-", text)
text = re.sub(r"-{2,}", "-", text).strip("-._")
if not text:
text = fallback
return text[:max_len].rstrip("-._") or fallback
def _next_subset_label(self, parent_dir: str, base_name: str) -> str:
"""Return the next subsetNNN label for this workspace merge prefix."""
prefix = f"{base_name}_merge_"
token = "_subset"
highest = 0
try:
for name in os.listdir(parent_dir):
if not name.startswith(prefix) or token not in name:
continue
suffix = name.split(token, 1)[1]
match = re.match(r"(\d{3,})", suffix)
if match:
highest = max(highest, int(match.group(1)))
except OSError:
pass
return f"subset{highest + 1:03d}"
def _combined_stream_manifest_path(self, stream_path: str | None) -> str | None:
if not stream_path:
return None
try:
folder = os.path.dirname(stream_path)
base = os.path.basename(folder)
return os.path.join(folder, f"{base}_stream_manifest.json")
except Exception:
return None
def _load_combined_stream_manifest(self, stream_path: str | None) -> dict:
manifest_path = self._combined_stream_manifest_path(stream_path)
if not manifest_path or not os.path.isfile(manifest_path):
return {}
try:
with open(manifest_path, "r", encoding="utf-8") as handle:
data = json.load(handle)
return data if isinstance(data, dict) else {}
except Exception:
return {}
def _add_stream_manifest_to_settings(self, settings: dict, stream_path: str | None) -> dict:
"""Embed lightweight provenance in merge settings for later inspection."""
data = dict(settings or {})
manifest = self._load_combined_stream_manifest(stream_path)
if manifest:
data["combined_stream_manifest"] = self._combined_stream_manifest_path(stream_path)
data["combined_stream_sources"] = {
"run_name": manifest.get("run_name"),
"selection": manifest.get("selection"),
"selected_count": manifest.get("selected_count"),
"available_count": manifest.get("available_count"),
"source_streams": manifest.get("source_streams", []),
}
return data
def _is_ici_run_dir(self, run_dir: str) -> bool:
"""Return True for top-level ICI run folders."""
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.isfile(os.path.join(run_dir, "ici_settings.json"))
or os.path.isfile(os.path.join(run_dir, "ici_run_manifest.json"))
or os.path.isdir(os.path.join(run_dir, "run_000"))
)
def _streams_for_indexing_run_dir(self, run_dir: str) -> list[str]:
"""Return mergeable stream files for one indexing run directory."""
if self._is_ici_run_dir(run_dir):
alias = update_final_stream_alias(run_dir)
if alias and os.path.isfile(alias):
return [alias]
final_stream = resolve_final_stream_path(run_dir)
if final_stream and os.path.isfile(final_stream):
return [final_stream]
canonical = canonical_stream_path(run_dir)
if os.path.isfile(canonical):
return [canonical]
return []
streams = []
try:
for fname in os.listdir(run_dir):
if fname.endswith(".stream"):
streams.append(os.path.join(run_dir, fname))
except OSError:
return []
return sorted(streams)
def _indexing_run_dirs_in_root(self, root: str) -> list[str]:
"""Return simple Indexamajig and ICI run directories directly under root."""
if not root or not os.path.isdir(root):
return []
base_name = os.path.basename(os.path.abspath(root))
if base_name.startswith("indexingintegration_") or base_name.startswith("ici_"):
return [root]
runs = []
try:
for entry in os.listdir(root):
if not (entry.startswith("indexingintegration_") or entry.startswith("ici_")):
continue
run_dir = os.path.join(root, entry)
if os.path.isdir(run_dir):
runs.append(run_dir)
except OSError:
return []
return sorted(runs)
def _init_ui(self):
outer_layout = QVBoxLayout(self)
self.tabs = QTabWidget(self)
# Tab 1: run selection
tab1 = self._build_run_select_tab()
# Tab 2: existing settings
tab2 = self._build_settings_tab()
# Tab 3: compare completed merge runs
tab3 = self._build_compare_tab()
self.tabs.addTab(tab1, "Select Run")
self.tabs.addTab(tab2, "Settings")
self.tabs.addTab(tab3, "Compare")
outer_layout.addWidget(self.tabs)
def _infer_ini_paths_for_streams(self, stream_paths: list[str]) -> list[str]:
"""Given a list of stream file paths, return the INI files that produced them."""
ini_paths: list[str] = []
seen: set[str] = set()
workspace_map = getattr(self, "_workspace_ini_map", {}) or {}
for stream in stream_paths:
current_dir = os.path.dirname(os.path.abspath(stream))
matched = False
# Walk up until we hit a known INI base directory
while True:
if current_dir in workspace_map:
for ini_path in workspace_map[current_dir]:
if ini_path not in seen:
seen.add(ini_path)
ini_paths.append(ini_path)
matched = True
break
parent = os.path.dirname(current_dir)
if not parent or parent == current_dir:
break
current_dir = parent
if not matched and self.ini_file_path:
fallback_ini = os.path.abspath(self.ini_file_path)
if fallback_ini not in seen:
seen.add(fallback_ini)
ini_paths.append(fallback_ini)
return ini_paths
def _discover_all_runs(self):
"""Scan for indexing run folders and collect mergeable stream files."""
runs = {}
seen_paths = set()
candidate_roots: set[str] = set()
# Always include the current INI directory and its parent
if self.ini_directory:
candidate_roots.add(os.path.abspath(self.ini_directory))
parent_dir = os.path.dirname(os.path.abspath(self.ini_directory))
if parent_dir:
candidate_roots.add(parent_dir)
# Include directories derived from the live workspace INI paths for broader coverage.
for ini_path in self._workspace_ini_paths():
for root in self._run_roots_for_ini(ini_path):
candidate_roots.add(root)
for base_dir in getattr(self, "_workspace_ini_map", {}).keys():
candidate_roots.add(base_dir)
for root in list(candidate_roots):
if not root or not os.path.isdir(root):
continue
try:
for run_dir in self._indexing_run_dirs_in_root(root):
run_name = os.path.basename(os.path.abspath(run_dir))
for full_path in self._streams_for_indexing_run_dir(run_dir):
if full_path in seen_paths:
continue
seen_paths.add(full_path)
runs.setdefault(run_name, []).append(full_path)
except Exception:
continue
return runs
def _build_run_select_tab(self):
from PyQt6.QtWidgets import QWidget, QVBoxLayout, QHBoxLayout, QLabel, QPushButton, QComboBox, QListWidget, QListWidgetItem, QGroupBox
w = QWidget(self)
main = QHBoxLayout(w)
# === Left column: select run ===
left_group = QGroupBox("Select Run")
left = QVBoxLayout(left_group)
# Run combo + refresh
row1 = QHBoxLayout()
self.run_combo = QComboBox()
self.refresh_runs_btn = QPushButton("Refresh")
row1.addWidget(self.run_combo, 1)
row1.addWidget(self.refresh_runs_btn, 0)
left.addLayout(row1)
# Files for selected run
self.run_files_list = QListWidget()
self.run_files_list.setSelectionMode(QListWidget.SelectionMode.NoSelection)
self.run_files_list.setEditTriggers(QListWidget.EditTrigger.NoEditTriggers)
left.addWidget(self.run_files_list)
file_btns = QHBoxLayout()
self.run_files_all_btn = QPushButton("All")
self.run_files_none_btn = QPushButton("None")
file_btns.addWidget(self.run_files_all_btn)
file_btns.addWidget(self.run_files_none_btn)
file_btns.addStretch()
left.addLayout(file_btns)
self.run_files_label = QLabel("Checked stream files will be combined")
self.run_files_label.setWordWrap(True)
left.addWidget(self.run_files_label)
# === Center column: generate button with arrow ===
center = QVBoxLayout()
center.addStretch()
self.gen_combined_btn = QPushButton("Combine \u2192")
self.gen_combined_btn.setFixedWidth(110)
center.addWidget(self.gen_combined_btn)
self.gen_latest_workspace_btn = QPushButton("Latest All \u2192")
self.gen_latest_workspace_btn.setFixedWidth(110)
self.gen_latest_workspace_btn.setToolTip(
"Combine the newest mergeable indexing stream from each INI in the workspace."
)
center.addWidget(self.gen_latest_workspace_btn)
center.addStretch()
# === Right column: existing combined streams ===
right_group = QGroupBox("Combined Streams")
right = QVBoxLayout(right_group)
self.combined_list = QListWidget()
right.addWidget(self.combined_list)
self.use_combined_btn = QPushButton("Use Selected")
right.addWidget(self.use_combined_btn)
main.addWidget(left_group, 1)
main.addLayout(center, 0)
main.addWidget(right_group, 1)
# Define helpers first, then wire signals and run initial population
def update_run_files():
"""Populate left-side file list for the selected run with its generated streams."""
self.run_files_list.clear()
run_name = self.run_combo.currentData()
if not run_name or not hasattr(self, '_runs_by_name'):
self.run_files_label.setText("Checked stream files will be combined")
return
paths = self._runs_by_name.get(run_name, [])
for stream_path in sorted(paths):
label = self._display_path(stream_path)
item = QListWidgetItem(label)
item.setData(Qt.ItemDataRole.UserRole, stream_path)
item.setToolTip(stream_path)
item.setFlags(item.flags() | Qt.ItemFlag.ItemIsUserCheckable | Qt.ItemFlag.ItemIsEnabled)
item.setCheckState(Qt.CheckState.Checked)
self.run_files_list.addItem(item)
self.run_files_label.setText(
f"Checked stream files will be combined — {len(paths)} available"
)
def set_all_run_files_checked(checked: bool):
state = Qt.CheckState.Checked if checked else Qt.CheckState.Unchecked
for row in range(self.run_files_list.count()):
self.run_files_list.item(row).setCheckState(state)
def checked_run_file_paths() -> list[str]:
paths = []
for row in range(self.run_files_list.count()):
item = self.run_files_list.item(row)
if item.checkState() != Qt.CheckState.Checked:
continue
path = item.data(Qt.ItemDataRole.UserRole)
if path:
paths.append(path)
return paths
def populate_combined():
self.combined_list.clear()
import glob, os
prefix = self._workspace_stem()
dirs = []
seen_dirs = set()
for parent_dir in self._merge_output_roots():
pattern = os.path.join(parent_dir, f"{prefix}_merge_*")
for d in glob.glob(pattern):
if not os.path.isdir(d):
continue
abs_d = os.path.abspath(d)
if abs_d in seen_dirs:
continue
seen_dirs.add(abs_d)
dirs.append(d)
for d in sorted(dirs):
label = os.path.basename(d)
# Prefer new naming: <dir>/<dir>.stream ; also support legacy: merged.stream
candidate_new = os.path.join(d, f"{label}.stream")
candidate_old = os.path.join(d, "merged.stream")
stream_path = candidate_new if os.path.isfile(candidate_new) else (candidate_old if os.path.isfile(candidate_old) else None)
if stream_path:
manifest = self._load_combined_stream_manifest(stream_path)
selection = ""
if manifest:
selected = manifest.get("selected_count")
available = manifest.get("available_count")
scope = manifest.get("selection", "")
if selected and available:
selection = f" [{scope}: {selected}/{available}]"
self.combined_list.addItem(f"{label} — {os.path.basename(stream_path)}{selection}")
self.combined_list.item(self.combined_list.count()-1).setData(Qt.ItemDataRole.UserRole, stream_path)
def populate_runs():
self.run_combo.clear()
self._workspace_ini_map = self._build_workspace_ini_map()
self._runs_by_name = self._discover_all_runs()
for run_name in sorted(self._runs_by_name.keys()):
self.run_combo.addItem(run_name, run_name)
# After filling the combo, refresh both left and right panels
update_run_files()
populate_combined()
# Wire signals AFTER helpers exist
self.run_combo.currentIndexChanged.connect(lambda _: update_run_files())
self.run_combo.currentIndexChanged.connect(lambda _: populate_combined())
self.refresh_runs_btn.clicked.connect(populate_runs)
self.run_files_all_btn.clicked.connect(lambda: set_all_run_files_checked(True))
self.run_files_none_btn.clicked.connect(lambda: set_all_run_files_checked(False))
self.gen_combined_btn.clicked.connect(self._on_generate_combined_clicked)
self.gen_latest_workspace_btn.clicked.connect(self._on_generate_latest_workspace_combined_clicked)
self._checked_run_file_paths = checked_run_file_paths
# Initial population
populate_runs()
def use_selected_combined():
item = self.combined_list.currentItem()
if not item:
return
path = item.data(Qt.ItemDataRole.UserRole)
import os
label = self._display_path(path)
idx = self.stream_combo.findData(path)
if idx == -1:
self.stream_combo.addItem(label, path)
idx = self.stream_combo.findData(path)
if idx != -1:
self.stream_combo.setCurrentIndex(idx)
# Load settings JSON if present
try:
json_path = self._settings_json_path_for_stream(path)
if json_path and os.path.isfile(json_path):
with open(json_path, "r", encoding="utf-8") as f:
d = json.load(f)
self._apply_settings_from_dict(d)
except Exception as e:
log_print(f"Warning: could not load merge settings JSON: {e}")
self.tabs.setCurrentIndex(1)
# Begin tailing latest stdout.log for this merged stream
latest_dir = self._latest_merge_dir_for_stream(path)
if latest_dir:
self._start_log_tail(os.path.join(latest_dir, "stdout.log"), start_from_end=False)
# Load or run check_hkl for existing run
try:
self._run_check_hkl(latest_dir, stream_path=path)
self._run_compare_hkl(latest_dir, stream_path=path)
except Exception:
pass
else:
self._stop_log_tail()
self.use_combined_btn.clicked.connect(use_selected_combined)
return w
def _latest_streams_for_workspace_inis(self) -> list[str]:
"""Return the newest mergeable stream for each INI in the workspace."""
ini_paths = self._workspace_ini_paths()
if not ini_paths and self.ini_file_path:
ini_paths = [self.ini_file_path]
latest_streams: list[str] = []
seen: set[str] = set()
for ini_path in ini_paths:
candidates: list[tuple[datetime.datetime, float, str]] = []
for root in self._run_roots_for_ini(ini_path):
for run_dir in self._indexing_run_dirs_in_root(root):
ts = self._timestamp_from_run_dir(run_dir)
if ts is None:
continue
for stream_path in self._streams_for_indexing_run_dir(run_dir):
if not stream_path or not os.path.isfile(stream_path):
continue
try:
mtime = os.path.getmtime(stream_path)
except OSError:
mtime = 0.0
candidates.append((ts, mtime, stream_path))
if not candidates:
continue
_, _, stream_path = max(candidates, key=lambda item: (item[0], item[1], item[2]))
if stream_path not in seen:
seen.add(stream_path)
latest_streams.append(stream_path)
return latest_streams
def _on_generate_latest_workspace_combined_clicked(self):
try:
self._workspace_ini_map = self._build_workspace_ini_map()
paths = self._latest_streams_for_workspace_inis()
if not paths:
QMessageBox.warning(
self,
"Generate Combined Stream",
"No mergeable indexing streams were found for the workspace INI files.",
)
return
combined_path = self._generate_combined_stream_for_run(
"latest_workspace",
paths,
available_count=len(paths),
)
try:
d_now = self._collect_settings_dict()
d_now["streamfile"] = combined_path
d_now = self._add_stream_manifest_to_settings(d_now, combined_path)
json_path = self._settings_json_path_for_stream(combined_path)
if json_path:
with open(json_path, "w", encoding="utf-8") as f:
json.dump(d_now, f, indent=2)
except Exception as e:
log_print(f"Warning: could not prime merge settings JSON: {e}")
except Exception as e:
QMessageBox.critical(self, "Error", f"Failed to generate combined stream:\n{e}")
return
label = self._display_path(combined_path)
idx = self.stream_combo.findData(combined_path)
if idx == -1:
self.stream_combo.addItem(label, combined_path)
idx = self.stream_combo.findData(combined_path)
if idx != -1:
self.stream_combo.setCurrentIndex(idx)
self.tabs.setCurrentIndex(1)
QMessageBox.information(
self,
"Combined Stream",
f"Combined stream created from {len(paths)} latest workspace stream(s):\n{combined_path}",
)
def _on_generate_combined_clicked(self):
run_name = self.run_combo.currentData()
if not run_name or not hasattr(self, '_runs_by_name'):
QMessageBox.warning(self, "Generate Combined Stream", "Please select a run first.")
return
available_paths = self._runs_by_name.get(run_name, [])
paths = []
try:
paths = self._checked_run_file_paths()
except Exception:
paths = list(available_paths)
if not paths:
QMessageBox.warning(self, "Generate Combined Stream", "Please check at least one stream file for the selected run.")
return
try:
combined_path = self._generate_combined_stream_for_run(
run_name,
paths,
available_count=len(available_paths),
)
# Prime settings JSON in the new merge folder
try:
d_now = self._collect_settings_dict()
d_now["streamfile"] = combined_path
d_now = self._add_stream_manifest_to_settings(d_now, combined_path)
json_path = self._settings_json_path_for_stream(combined_path)
if json_path:
with open(json_path, "w", encoding="utf-8") as f:
json.dump(d_now, f, indent=2)
except Exception as e:
log_print(f"Warning: could not prime merge settings JSON: {e}")
except Exception as e:
QMessageBox.critical(self, "Error", f"Failed to generate combined stream:\n{e}")
return
# Add to Settings tab combo and select it
label = self._display_path(combined_path)
idx = self.stream_combo.findData(combined_path)
if idx == -1:
self.stream_combo.addItem(label, combined_path)
idx = self.stream_combo.findData(combined_path)
if idx != -1:
self.stream_combo.setCurrentIndex(idx)
self.tabs.setCurrentIndex(1)
QMessageBox.information(
self,
"Combined Stream",
f"Combined stream created from {len(paths)} of {len(available_paths)} stream file(s):\n{combined_path}",
)
def _build_compare_tab(self):
from PyQt6.QtWidgets import QWidget
tab = QWidget(self)
layout = QVBoxLayout(tab)
top = QHBoxLayout()
self.compare_refresh_btn = QPushButton("Refresh")
self.compare_metric_combo = QComboBox()
self.compare_metric_combo.addItem("Completeness (%)", "compl")
self.compare_metric_combo.addItem("Redundancy", "red")
self.compare_metric_combo.addItem("SNR", "snr")
self.compare_metric_combo.addItem("Mean I", "mean_i")
top.addWidget(QLabel("Metric:"))
top.addWidget(self.compare_metric_combo)
top.addStretch()
top.addWidget(self.compare_refresh_btn)
layout.addLayout(top)
self.compare_table = QTableWidget()
self.compare_table.setColumnCount(20)
self.compare_table.setHorizontalHeaderLabels([
"Run", "Point Group", "Input Crystals", "Used Crystals",
"CC1/2 Final (%)", "Compl. ~2A (%)", "Red. ~2A",
"SNR ~2A", "CC1/2 Split ~2A", "CC* ~2A", "Rsplit ~2A",
"Min Res", "Push Res", "Min Meas.", "Model", "Iterations", "Max ADU", "B-scale",
"Partialator Logs", "Output",
])
self.compare_table.horizontalHeader().setSectionResizeMode(QHeaderView.ResizeMode.ResizeToContents)
self.compare_table.setSelectionBehavior(QAbstractItemView.SelectionBehavior.SelectRows)
self.compare_table.setSelectionMode(QAbstractItemView.SelectionMode.MultiSelection)
self.compare_table.setEditTriggers(QAbstractItemView.EditTrigger.NoEditTriggers)
layout.addWidget(self.compare_table)
self.compare_fig = Figure(figsize=(5, 3))
self.compare_ax = self.compare_fig.add_subplot(111)
self.compare_canvas = FigureCanvas(self.compare_fig)
layout.addWidget(self.compare_canvas)
self._compare_runs = []
self.compare_refresh_btn.clicked.connect(self._refresh_compare_tab)
self.compare_metric_combo.currentIndexChanged.connect(lambda _: self._update_compare_plot())
self.compare_table.itemSelectionChanged.connect(self._update_compare_plot)
self._refresh_compare_tab()
return tab
def _count_stream_crystals(self, stream_path: str | None) -> int | None:
if not stream_path or not os.path.isfile(stream_path):
return None
count = 0
try:
with open(stream_path, "r", encoding="utf-8", errors="replace") as handle:
for line in handle:
if line.startswith("--- Begin crystal"):
count += 1
return count
except Exception:
return None
def _parse_merge_stdout_summary(self, stdout_path: str | None) -> dict:
summary = {}
if not stdout_path or not os.path.isfile(stdout_path):
return summary
cc_half = None
used = None
try:
with open(stdout_path, "r", encoding="utf-8", errors="replace") as handle:
for line in handle:
m = self._re_cchalf.search(line)
if m:
try:
cc_half = float(m.group(1))
except Exception:
pass
m = re.search(r"Log residual went from .*?,\s*(\d+)\s+crystals", line)
if m:
try:
used = int(m.group(1))
except Exception:
pass
except Exception:
return summary
if cc_half is not None:
summary["cc_half"] = cc_half
if used is not None:
summary["used_crystals"] = used
return summary
def _shell_value_near_resolution(self, shell_rows: list[dict], target_d: float = 2.0) -> dict:
best = {}
best_delta = None
for row in shell_rows or []:
try:
d = float(row.get("d_ang", ""))
except Exception:
continue
delta = abs(d - target_d)
if best_delta is None or delta < best_delta:
best = row
best_delta = delta
return best
def _settings_for_merge_dir(self, merge_dir: str) -> dict:
try:
base = os.path.basename(os.path.abspath(merge_dir))
preferred = os.path.join(merge_dir, f"{base}_merge_settings.json")
paths = [preferred] if os.path.isfile(preferred) else []
paths.extend(glob.glob(os.path.join(merge_dir, "*merge_settings.json")))
for path in paths:
with open(path, "r", encoding="utf-8") as handle:
data = json.load(handle)
if isinstance(data, dict):
return data
except Exception:
pass
return {}
def _discover_completed_merge_runs(self) -> list[dict]:
prefix = self._workspace_stem()
merge_dirs = []
seen_dirs = set()
for parent_dir in self._merge_output_roots():
for pattern in (f"{prefix}_merge*", f"{prefix}-merge*"):
for d in glob.glob(os.path.join(parent_dir, pattern)):
if not os.path.isdir(d):
continue
abs_d = os.path.abspath(d)
if abs_d in seen_dirs:
continue
seen_dirs.add(abs_d)
merge_dirs.append(abs_d)
runs = []
for merge_dir in merge_dirs:
settings = self._settings_for_merge_dir(merge_dir)
stream_path = settings.get("streamfile")
if not stream_path:
base = os.path.basename(merge_dir)
candidate = os.path.join(merge_dir, f"{base}.stream")
if os.path.isfile(candidate):
stream_path = candidate
input_crystals = self._count_stream_crystals(stream_path)
try:
run_dirs = sorted(
os.path.join(merge_dir, name)
for name in os.listdir(merge_dir)
if name.startswith("merge-run") and os.path.isdir(os.path.join(merge_dir, name))
)
except OSError:
run_dirs = []
for run_dir in run_dirs:
hkl_path = os.path.join(run_dir, "crystfel.hkl")
if not os.path.isfile(hkl_path):
continue
shell_rows = self._parse_check_hkl_output(os.path.join(run_dir, "shells.dat"))
shell_2a = self._shell_value_near_resolution(shell_rows, target_d=2.0)
split_rows = self._parse_compare_hkl_outputs(run_dir)
split_2a = self._shell_value_near_resolution(split_rows, target_d=2.0)
stdout_summary = self._parse_merge_stdout_summary(os.path.join(run_dir, "stdout.log"))
try:
mtime = os.path.getmtime(hkl_path)
except OSError:
mtime = 0.0
runs.append({
"label": f"{os.path.basename(merge_dir)}/{os.path.basename(run_dir)}",
"merge_dir": merge_dir,
"run_dir": run_dir,
"hkl_path": hkl_path,
"pointgroup": settings.get("pointgroup", ""),
"min_res": settings.get("min_res", ""),
"push_res": settings.get("push_res", ""),
"min_measurements": settings.get("min_measurements", ""),
"model": settings.get("model", ""),
"iterations": settings.get("iterations", ""),
"max_adu": settings.get("max_adu", ""),
"bscale": "off" if self._as_bool(settings.get("no_Bscale"), False) else "on",
"save_partialator_logs": self._as_bool(settings.get("save_partialator_logs"), False),
"input_crystals": input_crystals,
"used_crystals": stdout_summary.get("used_crystals"),
"cc_half": stdout_summary.get("cc_half"),
"compl_2a": shell_2a.get("compl", ""),
"red_2a": shell_2a.get("red", ""),
"snr_2a": shell_2a.get("snr", ""),
"split_cc_2a": split_2a.get("cc", ""),
"split_ccstar_2a": split_2a.get("ccstar", ""),
"split_rsplit_2a": split_2a.get("rsplit", ""),
"shell_rows": shell_rows,
"split_rows": split_rows,
"mtime": mtime,
})
return sorted(runs, key=lambda row: row.get("mtime", 0.0), reverse=True)
def _refresh_compare_tab(self):
self._compare_runs = self._discover_completed_merge_runs()
table = self.compare_table
table.blockSignals(True)
table.setRowCount(0)
for row_idx, run in enumerate(self._compare_runs):
table.insertRow(row_idx)
values = [
run.get("label", ""),
run.get("pointgroup", ""),
run.get("input_crystals", ""),
run.get("used_crystals", ""),
run.get("cc_half", ""),
run.get("compl_2a", ""),
run.get("red_2a", ""),
run.get("snr_2a", ""),
run.get("split_cc_2a", ""),
run.get("split_ccstar_2a", ""),
run.get("split_rsplit_2a", ""),
run.get("min_res", ""),
run.get("push_res", ""),
run.get("min_measurements", ""),
run.get("model", ""),
run.get("iterations", ""),
run.get("max_adu", ""),
run.get("bscale", ""),
run.get("save_partialator_logs", ""),
run.get("hkl_path", ""),
]
for col, value in enumerate(values):
if isinstance(value, float):
text = f"{value:.3f}"
elif value is None:
text = ""
else:
text = str(value)
item = QTableWidgetItem(text)
if col == 0:
item.setData(Qt.ItemDataRole.UserRole, row_idx)
table.setItem(row_idx, col, item)
table.blockSignals(False)
if self._compare_runs:
table.selectRow(0)
if len(self._compare_runs) > 1:
table.selectRow(1)
self._update_compare_plot()
def _selected_compare_runs(self) -> list[dict]:
rows = sorted({idx.row() for idx in self.compare_table.selectedIndexes()})
return [self._compare_runs[i] for i in rows if 0 <= i < len(self._compare_runs)]
def _update_compare_plot(self):
try:
metric_key = self.compare_metric_combo.currentData() or "compl"
metric_name = self.compare_metric_combo.currentText()
runs = self._selected_compare_runs()
ax = self.compare_ax
ax.clear()
ax.set_xlabel("Resolution d (A)")
ax.set_ylabel(metric_name)
ax.grid(True, which="both", linestyle="--", linewidth=0.5, alpha=0.5)
for run in runs:
rows = run.get("shell_rows") or []
pts = []
for row in rows:
try:
d = float(row.get("d_ang", ""))
y = float(row.get(metric_key, ""))
except Exception:
continue
pts.append((d, y))
pts.sort(key=lambda t: -t[0])
if not pts:
continue
xs = [p[0] for p in pts]
ys = [p[1] for p in pts]
ax.plot(xs, ys, marker="o", markersize=3, linewidth=1.4, label=run.get("label", "run"))
if runs:
ax.legend(loc="best", fontsize="small")
ax.invert_xaxis()
self.compare_canvas.draw_idle()
except Exception:
pass
def _generate_combined_stream_for_run(
self,
run_name: str,
stream_paths: list[str],
available_count: int | None = None,
) -> str:
import os, datetime
# Output root: same directory as the workspace file.
parent_dir = self._workspace_root_dir()
base_name = self._workspace_stem()
ts = datetime.datetime.now().strftime('%Y%m%d_%H%M%S')
# Sort paths for stable order
uniq = []
seen = set()
for p in stream_paths:
if p not in seen:
uniq.append(p)
seen.add(p)
stream_paths = sorted(uniq)
if not stream_paths:
raise RuntimeError("No input streams provided")
selected_count = len(stream_paths)
available_count = available_count if available_count is not None else selected_count
selection = "all" if selected_count == available_count else "subset"
if selection == "all":
subset_label = None
out_dir = os.path.join(parent_dir, f"{base_name}_merge_{ts}")
else:
subset_label = self._next_subset_label(parent_dir, base_name)
out_dir = os.path.join(parent_dir, f"{base_name}_merge_{ts}_{subset_label}")
os.makedirs(out_dir, exist_ok=False)
out_basename = os.path.basename(out_dir)
out_path = os.path.join(out_dir, f"{out_basename}.stream")
def append_chunks(src_path, outfile, keep_header: bool):
with open(src_path, 'r', encoding='utf-8', errors='ignore') as f:
writing = keep_header
for line in f:
if keep_header:
outfile.write(line)
continue
if line.strip() == '----- Begin chunk -----':
writing = True
if writing:
outfile.write(line)
with open(out_path, 'w', encoding='utf-8') as out:
# First file: write full header up to first chunk, then all remaining lines
first = True
for sp in stream_paths:
if first:
append_chunks(sp, out, keep_header=True)
first = False
else:
append_chunks(sp, out, keep_header=False)
source_records = []
for idx, sp in enumerate(stream_paths, start=1):
record = {
"order": idx,
"path": os.path.abspath(sp),
"display_path": self._display_path(os.path.abspath(sp)),
"basename": os.path.basename(sp),
}
try:
stat = os.stat(sp)
record["size_bytes"] = stat.st_size
record["mtime"] = datetime.datetime.fromtimestamp(stat.st_mtime).isoformat(timespec="seconds")
except OSError:
pass
try:
real_path = os.path.realpath(sp)
if real_path != os.path.abspath(sp):
record["real_path"] = real_path
except Exception:
pass
source_records.append(record)
manifest = {
"created_at": datetime.datetime.now().isoformat(timespec="seconds"),
"workspace": getattr(self, "workspace_path", None),
"run_name": run_name,
"selection": selection,
"subset_label": subset_label,
"selected_count": selected_count,
"available_count": available_count,
"output_stream": out_path,
"source_streams": source_records,
}
manifest_path = self._combined_stream_manifest_path(out_path)
if manifest_path:
with open(manifest_path, "w", encoding="utf-8") as handle:
json.dump(manifest, handle, indent=2)
return out_path
def _build_settings_tab(self):
from PyQt6.QtWidgets import QWidget
tab = QWidget(self)
outer_layout = QVBoxLayout(tab)
# Stream file selection dropdown
stream_layout = QHBoxLayout()
stream_label = QLabel("Stream to Merge:")
self.stream_combo = QComboBox()
# Show workspace-level merged streams, plus legacy INI-local streams.
import os, glob
prefix = self._workspace_stem()
merge_dirs = []
seen_dirs = set()
for parent_dir in self._merge_output_roots():
pattern = os.path.join(parent_dir, f"{prefix}_merge_*")
for d in glob.glob(pattern):
if not os.path.isdir(d):
continue
abs_d = os.path.abspath(d)
if abs_d in seen_dirs:
continue
seen_dirs.add(abs_d)
merge_dirs.append(d)
for d in sorted(merge_dirs):
# Prefer new naming: <dir>/<dir>.stream ; also allow legacy: merged.stream
base = os.path.basename(d)
candidate_new = os.path.join(d, f"{base}.stream")
candidate_old = os.path.join(d, "merged.stream")
for full_stream_path in (candidate_new, candidate_old):
if os.path.isfile(full_stream_path):
rel_label = self._display_path(full_stream_path)
if self.stream_combo.findData(full_stream_path) == -1:
self.stream_combo.addItem(rel_label, full_stream_path)
break # stop after first existing candidate
stream_layout.addWidget(stream_label)
stream_layout.addWidget(self.stream_combo)
outer_layout.addLayout(stream_layout)
# Inferred cell parameters display
self.cell_params_display = QLabel("")
self.cell_params_display.setWordWrap(True)
outer_layout.addWidget(self.cell_params_display)
cell_source_layout = QHBoxLayout()
self.cell_source_label = QLabel("Cell for merge analysis:")
self.cell_source_combo = QComboBox()
self.cell_source_combo.addItem("Indexing cell", "indexing")
self.cell_source_combo.currentIndexChanged.connect(lambda _idx: self._on_cell_source_changed())
self.cell_source_label.setVisible(False)
self.cell_source_combo.setVisible(False)
cell_source_layout.addWidget(self.cell_source_label)
cell_source_layout.addWidget(self.cell_source_combo)
cell_source_layout.addStretch()
outer_layout.addLayout(cell_source_layout)
# Pointgroup dropdown
pg_layout = QHBoxLayout()
pg_label = QLabel("Point Group:")
self.pg_combo = QComboBox()
# Full crystallographic point-group list accepted by CrystFEL.
pg_symbols = [
"1", "-1",
"2/m", "2", "m",
"mmm", "222", "mm2",
"4/m", "4", "-4", "4/mmm", "422", "-42m", "-4m2", "4mm",
"3_R", "-3_R", "32_R", "3m_R", "-3m_R",
"3_H", "-3_H", "321_H", "312_H", "3m1_H", "31m_H", "-3m1_H", "-31m_H",
"6/m", "6", "-6", "6/mmm", "622", "-62m", "-6m2", "6mm",
"23", "m-3", "432", "-43m", "m-3m",
]
self.pg_combo.addItems(pg_symbols)
pg_layout.addWidget(pg_label)
pg_layout.addWidget(self.pg_combo)
# Unique Axis (for monoclinic point groups)
ua_layout = QHBoxLayout()
ua_label = QLabel("Unique Axis:")
self.ua_combo = QComboBox()
self.ua_combo.addItem("") # Empty default
self.ua_combo.addItems(["a", "b", "c"])
ua_layout.addWidget(ua_label)
ua_layout.addWidget(self.ua_combo)
# Threads
thr_layout = QHBoxLayout()
thr_label = QLabel("Threads:")
self.threads_spin = QSpinBox()
self.threads_spin.setRange(1, multiprocessing.cpu_count())
self.threads_spin.setValue(multiprocessing.cpu_count())
thr_layout.addWidget(thr_label)
thr_layout.addWidget(self.threads_spin)
# Min resolution
minres_layout = QHBoxLayout()
minres_label = QLabel("Min Resolution:")
self.minres_spin = QDoubleSpinBox()
self.minres_spin.setRange(0.0, 1e3)
self.minres_spin.setDecimals(3)
self.minres_spin.setValue(5.0)
self.minres_inf_chk = QCheckBox("∞")
# default unchecked; if you want default inf, set checked True and disable the spin
self.minres_inf_chk.setChecked(False)
self.minres_inf_chk.setToolTip("Include all crystals.")
self.minres_spin.setEnabled(True)
def _on_minres_inf_toggled(checked: bool):
self.minres_spin.setEnabled(not checked)
self.minres_inf_chk.toggled.connect(_on_minres_inf_toggled)
minres_layout.addWidget(minres_label)
minres_layout.addWidget(self.minres_spin)
minres_layout.addWidget(self.minres_inf_chk)
# Iterations
iter_layout = QHBoxLayout()
iter_label = QLabel("Iterations:")
self.iter_spin = QSpinBox()
self.iter_spin.setRange(1, 1000)
self.iter_spin.setValue(25)
iter_layout.addWidget(iter_label)
iter_layout.addWidget(self.iter_spin)
# Model
model_layout = QHBoxLayout()
model_label = QLabel("Model:")
self.model_combo = QComboBox()
self.model_combo.addItems(["offset", "xsphere", "unity", "ggpm"])
model_layout.addWidget(model_label)
model_layout.addWidget(self.model_combo)
# Min measurements
mm_layout = QHBoxLayout()
mm_label = QLabel("Min Measurements:")
self.mm_spin = QSpinBox()
self.mm_spin.setRange(1, 100)
self.mm_spin.setValue(5)
mm_layout.addWidget(mm_label)
mm_layout.addWidget(self.mm_spin)
# Max ADU
adu_layout = QHBoxLayout()
adu_label = QLabel("Max ADU:")
self.adu_edit = QLineEdit("inf")
self.adu_edit.setValidator(QDoubleValidator(self.adu_edit))
self.adu_inf_chk = QCheckBox("∞")
# If the default text is 'inf', start checked and disable the edit
self.adu_inf_chk.setChecked(True)
self.adu_edit.setEnabled(False)
def _on_adu_inf_toggled(checked: bool):
self.adu_edit.setEnabled(not checked)
if checked:
self.adu_edit.setText("inf")
self.adu_inf_chk.toggled.connect(_on_adu_inf_toggled)
adu_layout.addWidget(adu_label)
adu_layout.addWidget(self.adu_edit)
adu_layout.addWidget(self.adu_inf_chk)
# Push resolution
push_layout = QHBoxLayout()
push_label = QLabel("Push Resolution:")
self.push_edit = QLineEdit("0")
self.push_edit.setValidator(QDoubleValidator(self.push_edit))
self.push_inf_chk = QCheckBox("∞")
self.push_inf_chk.setChecked(False)
self.push_edit.setEnabled(True)
def _on_push_inf_toggled(checked: bool):
self.push_edit.setEnabled(not checked)
if checked:
self.push_edit.setText("inf")
self.push_inf_chk.toggled.connect(_on_push_inf_toggled)
push_layout.addWidget(push_label)
push_layout.addWidget(self.push_edit)
push_layout.addWidget(self.push_inf_chk)
# B-scaling checkbox (unchecked → partialator gets --no-Bscale)
self.no_bscale_chk = QCheckBox("Use B-scaling")
self.no_bscale_chk.setChecked(False)
# Partialator diagnostic logs checkbox
self.partialator_logs_chk = QCheckBox("Save Partialator Diagnostic Logs")
self.partialator_logs_chk.setChecked(False)
self.partialator_logs_chk.setToolTip(
"Write Partialator per-crystal diagnostic files into pr-logs. "
"Useful for debugging, but normal parsing, comparison, and export do not require them."
)
# Scaling results: Tabbed view (Table / Plot)
table_panel = QVBoxLayout()
self.scaling_tabs = QTabWidget()
self.scaling_tabs.setMinimumWidth(800)
# Merging tab: table + plot (click column headers to select up to 2 metrics)
merging_tab = QWidget()
merging_layout = QVBoxLayout(merging_tab)
merging_layout.setContentsMargins(8, 8, 8, 8)
table_tab_widget = QTableWidget()
self.scaling_table = table_tab_widget
self.scaling_table.setColumnCount(7)
self._cc_col_headers = [
"Overall CChalf (%)", "Reflections", "deltaCChalf",
"deltaCChalf Error (%)", "Bad Crystals", "OK Crystals", "Negative delta CC½"
]
self._cc_col_keys = [
"overall_cc_half", "reflections", "delta_cc_half",
"delta_cc_half_error", "bad_crystals", "ok_crystals", "neg_delta_cc_half"
]
self.scaling_table.setHorizontalHeaderLabels(self._cc_col_headers)
self.scaling_table.horizontalHeader().setSectionResizeMode(QHeaderView.ResizeMode.ResizeToContents)
self.scaling_table.horizontalHeader().sectionClicked.connect(self._on_cc_header_clicked)
self._cc_selected_cols = [0] # default: Overall CChalf
self._highlight_cc_headers()
merging_layout.addWidget(self.scaling_table)
self._cc_fig = Figure(figsize=(5, 3))
self._cc_ax = self._cc_fig.add_subplot(111)
self._cc_canvas = FigureCanvas(self._cc_fig)
merging_layout.addWidget(self._cc_canvas)
self.scaling_tabs.addTab(merging_tab, "Merging")
# Statistics tab: check_hkl resolution shell table + plot (click headers)
stats_tab = QWidget()
stats_layout = QVBoxLayout(stats_tab)
stats_layout.setContentsMargins(8, 8, 8, 8)
self.check_hkl_table = QTableWidget()
self.check_hkl_table.setColumnCount(11)
self._stats_col_headers = [
"Center 1/nm", "# refs", "Possible", "Compl (%)", "Meas",
"Red", "SNR", "Mean I", "d(A)", "Min 1/nm", "Max 1/nm"
]
self._stats_col_keys = [
"center", "refs", "possible", "compl", "meas",
"red", "snr", "mean_i", "d_ang", "min_inv_nm", "max_inv_nm"
]
# Columns that can be selected as metrics (exclude x-axis / metadata)
self._stats_selectable_cols = {1, 2, 3, 4, 5, 6, 7} # refs..mean_i
self.check_hkl_table.setHorizontalHeaderLabels(self._stats_col_headers)
self.check_hkl_table.horizontalHeader().setSectionResizeMode(QHeaderView.ResizeMode.ResizeToContents)
self.check_hkl_table.horizontalHeader().sectionClicked.connect(self._on_stats_header_clicked)
self._stats_selected_cols = [3] # default: Compl (%)
self._highlight_stats_headers()
stats_layout.addWidget(self.check_hkl_table)
self._compl_fig = Figure(figsize=(5, 3))
self._compl_ax = self._compl_fig.add_subplot(111)
self._compl_canvas = FigureCanvas(self._compl_fig)
stats_layout.addWidget(self._compl_canvas)
self.scaling_tabs.addTab(stats_tab, "Statistics")
# Split-half tab: compare_hkl statistics from crystfel.hkl1/hkl2
split_tab = QWidget()
split_layout = QVBoxLayout(split_tab)
split_layout.setContentsMargins(8, 8, 8, 8)
self.compare_hkl_table = QTableWidget()
self.compare_hkl_table.setColumnCount(8)
self.compare_hkl_table.setHorizontalHeaderLabels([
"d(A)", "CC1/2", "CC*", "Rsplit (%)", "# refs",
"Center 1/nm", "Min 1/nm", "Max 1/nm",
])
self.compare_hkl_table.horizontalHeader().setSectionResizeMode(QHeaderView.ResizeMode.ResizeToContents)
split_layout.addWidget(self.compare_hkl_table)
self._compare_hkl_fig = Figure(figsize=(5, 3))
self._compare_hkl_ax = self._compare_hkl_fig.add_subplot(111)
self._compare_hkl_canvas = FigureCanvas(self._compare_hkl_fig)
split_layout.addWidget(self._compare_hkl_canvas)
self._compare_hkl_rows = []
self.scaling_tabs.addTab(split_tab, "Split Half")
table_panel.addWidget(self.scaling_tabs)
# Save, Run, and Abort buttons
btn_layout = QVBoxLayout()
self.save_btn = QPushButton("Save Settings")
self.save_btn.clicked.connect(self._on_save_clicked)
btn_layout.addWidget(self.save_btn)
self.run_btn = QPushButton("Merge")
self.run_btn.clicked.connect(self._on_run_clicked)
btn_layout.addWidget(self.run_btn)
self.abort_btn = QPushButton("Abort")
self.abort_btn.setEnabled(False)
self.abort_btn.clicked.connect(self._on_abort_clicked)
btn_layout.addWidget(self.abort_btn)
# Right panel: Form controls in a fixed-width widget
right_widget = QWidget()
right_widget.setFixedWidth(230)
right_panel = QVBoxLayout(right_widget)
right_panel.setContentsMargins(0, 0, 0, 0)
# Move all GUI element layouts here
right_panel.addLayout(pg_layout)
right_panel.addLayout(ua_layout)
right_panel.addLayout(thr_layout)
right_panel.addLayout(minres_layout)
right_panel.addLayout(iter_layout)
right_panel.addLayout(model_layout)
right_panel.addLayout(mm_layout)
right_panel.addLayout(adu_layout)
right_panel.addLayout(push_layout)
right_panel.addWidget(self.no_bscale_chk)
right_panel.addWidget(self.partialator_logs_chk)
right_panel.addStretch()
right_panel.addLayout(btn_layout)
# Add both panels to main layout using an inner horizontal layout
inner_layout = QHBoxLayout()
inner_layout.addLayout(table_panel, 1)
inner_layout.addWidget(right_widget, 0)
outer_layout.addLayout(inner_layout)
# Console output panel (merged stdout/stderr)
console_group = QGroupBox("Console Output")
console_v = QVBoxLayout(console_group)
self.console = QPlainTextEdit()
self.console.setReadOnly(True)
self.console.setLineWrapMode(QPlainTextEdit.LineWrapMode.NoWrap)
# Monospace font
mono = QFontDatabase.systemFont(QFontDatabase.SystemFont.FixedFont)
self.console.setFont(mono)
console_btns = QHBoxLayout()
self.btn_clear_console = QPushButton("Clear")
self.btn_save_console = QPushButton("Save Log…")
console_btns.addStretch(1)
console_btns.addWidget(self.btn_clear_console)
console_btns.addWidget(self.btn_save_console)
console_v.addWidget(self.console)
console_v.addLayout(console_btns)
outer_layout.addWidget(console_group)
def _on_clear_console():
self.console.clear()
self.btn_clear_console.clicked.connect(_on_clear_console)
def _on_save_console():
path, _ = QFileDialog.getSaveFileName(self, "Save Console Log", os.path.join(self.ini_directory, "merge_console.log"), "Log files (*.log);;Text files (*.txt);;All files (*)")
if path:
try:
with open(path, "w", encoding="utf-8") as f:
f.write(self.console.toPlainText())
except Exception as e:
QMessageBox.warning(self, "Save Log", f"Failed to save log: {e}")
self.btn_save_console.clicked.connect(_on_save_console)
# QProcess for non-blocking merging
self._merge_proc = None
def _on_stream_changed(_idx):
path = self.stream_combo.currentData()
try:
json_path = self._settings_json_path_for_stream(path)
if json_path and os.path.isfile(json_path):
with open(json_path, "r", encoding="utf-8") as f:
d = json.load(f)
self._apply_settings_from_dict(d)
except Exception as e:
log_print(f"Warning: could not load merge settings JSON on change: {e}")
# Tail latest stdout.log for this stream so table updates even for existing runs
latest_dir = self._latest_merge_dir_for_stream(path) if path else None
if latest_dir:
self._start_log_tail(os.path.join(latest_dir, "stdout.log"), start_from_end=False)
# Load or run check_hkl for existing run
try:
self._run_check_hkl(latest_dir, stream_path=path)
self._run_compare_hkl(latest_dir, stream_path=path)
except Exception:
pass
else:
self._stop_log_tail()
self._update_inferred_cell(path)
self.stream_combo.currentIndexChanged.connect(_on_stream_changed)
# Populate inferred cell for the initial selection (if any)
try:
self._update_inferred_cell(self.stream_combo.currentData())
except Exception:
pass
# IMPORTANT: do not call self.setLayout here, we already return tab
return tab
def _collect_settings_dict(self):
"""Collect current GUI values into a dict for JSON dump and INI write."""
raw_pg = self.pg_combo.currentText().strip()
pointgroup = raw_pg
ua = self.ua_combo.currentText().strip()
unique_axis = ua if ua else None
return {
"pointgroup": pointgroup,
"unique_axis": unique_axis,
"threads": self.threads_spin.value(),
"min_res": "inf" if self.minres_inf_chk.isChecked() else self.minres_spin.value(),
"iterations": self.iter_spin.value(),
"model": self.model_combo.currentText(),
"polarisation": "none",
"min_measurements": self.mm_spin.value(),
"max_adu": "inf" if self.adu_inf_chk.isChecked() else self.adu_edit.text().strip().replace(",", "."),
"push_res": "inf" if self.push_inf_chk.isChecked() else self.push_edit.text().strip().replace(",", "."),
"no_Bscale": not self.no_bscale_chk.isChecked(),
"output_every_cycle": True,
"save_partialator_logs": self.partialator_logs_chk.isChecked(),
"cell_source": self._selected_cell_source(),
"streamfile": self.stream_combo.currentData(),
}
def _apply_settings_from_dict(self, d: dict):
if not d:
return
# Point group + unique axis
pg = d.get("pointgroup") or ""
idx = self.pg_combo.findText(pg)
if idx == -1:
idx = self.pg_combo.findText(pg.replace(" ", ""))
if idx != -1:
self.pg_combo.setCurrentIndex(idx)
ua = d.get("unique_axis") or ""
idx = self.ua_combo.findText(ua)
if idx != -1:
self.ua_combo.setCurrentIndex(idx)
# Numerics / toggles
try:
self.threads_spin.setValue(int(d.get("threads", self.threads_spin.value())))
except Exception:
pass
mr = d.get("min_res", "inf")
if str(mr) == "inf":
self.minres_inf_chk.setChecked(True)
self.minres_spin.setEnabled(False)
else:
self.minres_inf_chk.setChecked(False)
self.minres_spin.setEnabled(True)
try:
self.minres_spin.setValue(float(mr))
except Exception:
pass
try:
self.iter_spin.setValue(int(d.get("iterations", self.iter_spin.value())))
except Exception:
pass
m = d.get("model")
if m is not None:
idx = self.model_combo.findText(m)
if idx != -1:
self.model_combo.setCurrentIndex(idx)
try:
self.mm_spin.setValue(int(d.get("min_measurements", self.mm_spin.value())))
except Exception:
pass
ma = d.get("max_adu", "inf")
if str(ma) == "inf":
self.adu_inf_chk.setChecked(True)
self.adu_edit.setEnabled(False)
self.adu_edit.setText("inf")
else:
self.adu_inf_chk.setChecked(False)
self.adu_edit.setEnabled(True)
self.adu_edit.setText(str(ma))
pr = d.get("push_res", "inf")
if str(pr) == "inf":
self.push_inf_chk.setChecked(True)
self.push_edit.setEnabled(False)
self.push_edit.setText("inf")
else:
self.push_inf_chk.setChecked(False)
self.push_edit.setEnabled(True)
self.push_edit.setText(str(pr))
self.no_bscale_chk.setChecked(not self._as_bool(d.get("no_Bscale"), not self.no_bscale_chk.isChecked()))
d["output_every_cycle"] = True
self.partialator_logs_chk.setChecked(self._as_bool(d.get("save_partialator_logs"), False))
try:
cell_source = d.get("cell_source")
if cell_source and hasattr(self, "cell_source_combo"):
idx = self.cell_source_combo.findData(cell_source)
if idx != -1:
self.cell_source_combo.setCurrentIndex(idx)
except Exception:
pass
def _settings_json_path_for_stream(self, stream_path: str) -> str | None:
if not stream_path:
return None
try:
folder = os.path.dirname(stream_path)
base = os.path.basename(folder)
return os.path.join(folder, f"{base}_merge_settings.json")
except Exception:
return None
@staticmethod
def _as_bool(value, default: bool = False) -> bool:
if value is None:
return default
if isinstance(value, bool):
return value
if isinstance(value, (int, float)):
return bool(value)
text = str(value).strip().lower()
if text in {"1", "true", "yes", "on"}:
return True
if text in {"0", "false", "no", "off", ""}:
return False
return default
def _on_save_clicked(self):
"""
Save current merging settings to INI and JSON next to the selected merged stream.
"""
try:
d = self._collect_settings_dict()
d = self._add_stream_manifest_to_settings(d, d.get("streamfile"))
base_name = self._workspace_stem()
timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
hkl_filename = f"{base_name}_merge_{timestamp}.hkl"
write_mergingsettings(
input_path=self.ini_file_path,
hkl_filename=hkl_filename,
pointgroup=d["pointgroup"],
min_res=d["min_res"],
iterations=d["iterations"],
threads=d["threads"],
model=d["model"],
polarisation="none",
min_measurements=d["min_measurements"],
max_adu=d["max_adu"],
push_res=d["push_res"],
no_Bscale=d["no_Bscale"],
output_every_cycle=d["output_every_cycle"],
save_partialator_logs=d["save_partialator_logs"],
unique_axis=d["unique_axis"],
streamfile=d["streamfile"],
)
# JSON dump next to selected stream (if any)
try:
json_path = self._settings_json_path_for_stream(d.get("streamfile"))
if json_path:
with open(json_path, "w", encoding="utf-8") as f:
json.dump(d, f, indent=2)
except Exception as e:
log_print(f"Warning: could not write merge settings JSON: {e}")
QMessageBox.information(self, "Settings Saved", "Merging settings have been written to the INI file.")
except Exception as e:
QMessageBox.warning(self, "Save Settings", f"Failed to save settings: {e}")
def _on_abort_clicked(self):
"""
Abort the currently running Partialator process for the selected stream, if any.
Tries a graceful terminate() first, then force-kill after a short timeout.
"""
try:
proc = getattr(self, "_merge_proc", None)
if not proc or proc.state() == QProcess.ProcessState.NotRunning:
self.console.appendPlainText(">>> No running merge to abort.\n")
return
# Mark that user initiated an abort
self._user_aborted = True
self.console.appendPlainText(">>> Aborting merge... sending terminate()\n")
try:
proc.terminate()
except Exception:
pass
# After 5 seconds, if still running, kill it
def _force_kill():
try:
if proc.state() != QProcess.ProcessState.NotRunning:
self.console.appendPlainText(">>> Merge still running; forcing kill()\n")
proc.kill()
except Exception:
pass
timer = QTimer(self)
timer.setSingleShot(True)
timer.timeout.connect(_force_kill)
timer.start(5000)
except Exception as e:
self.console.appendPlainText(f">>> Abort failed: {e}\n")
def _on_run_clicked(self):
"""
Collect settings, choose output folder inside merged stream folder (merge-runN),
build Partialator command, start QProcess, stream stdout/stderr to console and log file,
and on finish, parse scaling.csv and populate self.scaling_table.
"""
selected_stream = self.stream_combo.currentData()
self.selected_stream = selected_stream
if not selected_stream or not os.path.isfile(selected_stream):
QMessageBox.warning(self, "Merging", "Please select a valid merged stream file.")
return
# Prevent starting another merge concurrently for the same stream
existing_proc = self._running_by_stream.get(selected_stream)
if existing_proc is not None and existing_proc.state() != QProcess.ProcessState.NotRunning:
QMessageBox.information(self, "Merging", "This stream is already merging right now. Wait for it to finish before starting another.")
return
# If this stream has been merged before (any merge-run* present), create a NEW
# timestamped merge folder and run there. This guarantees that pressing "Run"
# on a stream that already has results never reuses the same parent folder.
latest_dir, latest_state = self._latest_run_state(selected_stream)
if latest_state == "RUNNING":
QMessageBox.information(self, "Merging", f"Latest run ({os.path.basename(latest_dir)}) is still running. Please wait for it to finish.")
return
# If there is any existing run directory, clone to a fresh folder
if latest_dir is not None:
self.console.appendPlainText(f">>> Detected previous merge folder: {latest_dir}\n>>> Cloning selected stream to a fresh timestamped folder...\n")
cloned_stream = self._clone_merge_folder_for_restart(selected_stream)
if cloned_stream:
self.console.appendPlainText(f">>> New stream path: {cloned_stream}\n")
# Switch selection to the cloned stream
selected_stream = cloned_stream
self.selected_stream = selected_stream
# Ensure the cloned stream is present in the dropdown and selected
rel_label = self._display_path(selected_stream)
idx = self.stream_combo.findData(selected_stream)
if idx == -1:
self.stream_combo.addItem(rel_label, selected_stream)
idx = self.stream_combo.findData(selected_stream)
if idx != -1:
self.stream_combo.setCurrentIndex(idx)
# Reset live tail; it will be re-started for the new run's stdout.log
self._stop_log_tail()
else:
# Fall back to running in the *existing* folder but make that explicit
self.console.appendPlainText(">>> Warning: could not create a fresh merge folder; will continue in the current one.\n")
# 1. Collect settings
settings = self._collect_settings_dict()
settings = self._add_stream_manifest_to_settings(settings, selected_stream)
# 2. Choose output folder inside the merged stream folder (merge-runN)
output_dir = self._next_merge_output_dir(selected_stream)
if not output_dir:
QMessageBox.critical(self, "Merging", "Could not determine output directory for merging.")
return
try:
json_path = self._settings_json_path_for_stream(selected_stream)
if json_path:
with open(json_path, "w", encoding="utf-8") as f:
json.dump(settings, f, indent=2)
except Exception as e:
log_print(f"Warning: could not write merge settings JSON before run: {e}")
# Mark this run as RUNNING
self._mark_run_state(output_dir, "RUNNING")
# 3. Build Partialator command
cmd = self._build_partialator_cmd(selected_stream, output_dir, settings)
try:
self._write_merging_nxprocess(selected_stream, output_dir, settings, cmd)
except Exception as exc:
self.console.appendPlainText(f">>> Warning: failed to write NXprocess merging: {exc}\n")
# Ensure we don't double-parse while a live process is running
self._stop_log_tail()
# Reset live parsing state before starting process
self._live_reset_stats()
# 4. Start QProcess, stream stdout/stderr to console and log file
self.console.appendPlainText(f"\n>>> CWD: {output_dir}\n>>> Running: {' '.join(cmd)}\n")
self._merge_proc = QProcess(self)
self._merge_proc.setProcessChannelMode(QProcess.ProcessChannelMode.MergedChannels)
# Set working directory to output_dir
self._merge_proc.setWorkingDirectory(output_dir)
# Redirect output to log files
stdout_log = os.path.join(output_dir, "stdout.log")
try:
self._stdout_log_f = open(stdout_log, "w", encoding="utf-8")
except Exception:
self._stdout_log_f = None
# Register this process for the stream
self._running_by_stream[selected_stream] = self._merge_proc
def _append_ready_output():
try:
data = self._merge_proc.readAllStandardOutput()
if data:
text = bytes(data).decode("utf-8", errors="replace")
self.console.appendPlainText(text)
if self._stdout_log_f:
self._stdout_log_f.write(text)
self._stdout_log_f.flush()
# Live parse output for scaling stats table
self._live_parse_text(text)
except Exception:
pass
self._merge_proc.readyReadStandardOutput.connect(_append_ready_output)
def _on_started():
self.run_btn.setEnabled(False)
self.save_btn.setEnabled(False)
self.gen_combined_btn.setEnabled(False)
self.refresh_runs_btn.setEnabled(False)
# enable abort during run
if hasattr(self, "abort_btn"):
self.abort_btn.setEnabled(True)
# reset abort flag
self._user_aborted = False
self._merge_proc.started.connect(_on_started)
def _on_finished(code, status):
_append_ready_output()
self._set_active_merge_output_dir(output_dir)
self.run_btn.setEnabled(True)
self.save_btn.setEnabled(True)
self.gen_combined_btn.setEnabled(True)
self.refresh_runs_btn.setEnabled(True)
if hasattr(self, "abort_btn"):
self.abort_btn.setEnabled(False)
if self._stdout_log_f:
self._stdout_log_f.close()
# On finish, commit the last live cycle before parsing CSV
self._live_commit_cycle()
# Try to load scaling.csv; if missing, dump our live stats to scaling.csv
scaling_csv_path = os.path.join(output_dir, "scaling.csv")
scaling_rows = self._parse_scaling_file(scaling_csv_path)
if not scaling_rows and self._live_stats_rows:
try:
with open(scaling_csv_path, "w", newline="", encoding="utf-8") as csvfile:
w = csv.writer(csvfile)
w.writerow([
"cycle","overall_cc_half","reflections","delta_cc_half",
"delta_cc_half_error","bad_crystals","ok_crystals","neg_delta_cc_half"
])
for r in self._live_stats_rows:
w.writerow([
r.get("cycle",""),
r.get("overall_cc_half",""),
r.get("reflections",""),
r.get("delta_cc_half",""),
r.get("delta_cc_half_error",""),
r.get("bad_crystals",""),
r.get("ok_crystals",""),
r.get("neg_delta_cc_half",""),
])
scaling_rows = self._parse_scaling_file(scaling_csv_path)
except Exception as e:
log_print(f"Warning: failed to write scaling.csv: {e}")
if scaling_rows:
self._populate_scaling_table(scaling_rows)
self._update_cc_plot(scaling_rows)
# After finish, tail the just-written stdout.log from EOF (to catch any late writes)
try:
self._start_log_tail(os.path.join(output_dir, "stdout.log"), start_from_end=True)
except Exception:
pass
# Update run markers
try:
# New logic for DONE/ABORTED/RUNNING markers and abort messaging
if code == 0 and not getattr(self, "_user_aborted", False):
self._mark_run_state(output_dir, "DONE")
else:
if getattr(self, "_user_aborted", False):
self.console.appendPlainText(">>> Merge aborted by user.\n")
try:
self._mark_run_state(output_dir, "ABORTED")
except Exception:
pass
QMessageBox.information(self, "Merging", "Merge aborted by user.")
else:
QMessageBox.critical(self, "Merging", f"Merging failed with exit code {code}.")
# Clear RUNNING marker on failure/abort
try:
p = self._run_marker_path(output_dir, "RUNNING")
if os.path.exists(p):
os.remove(p)
except Exception:
pass
except Exception:
pass
# Clear running registry for this stream
try:
self._running_by_stream.pop(self.selected_stream, None)
except Exception:
pass
# Run check_hkl on the final output if crystfel.hkl exists
try:
self._run_check_hkl(output_dir, stream_path=selected_stream)
self._run_compare_hkl(output_dir, stream_path=selected_stream)
except Exception as exc:
self.console.appendPlainText(f">>> check_hkl error: {exc}\n")
if code == 0 and not getattr(self, "_user_aborted", False):
QMessageBox.information(self, "Merging", "Partialator merging finished successfully.")
self._merge_proc.finished.connect(_on_finished)
# 5. Start the process
self._user_aborted = False
self._merge_proc.start(cmd[0], cmd[1:])