"""UI for exporting SHELX HKLF4 files and generating INS/P4P from merged streams (cell parsed from stream header)."""
from __future__ import annotations
import os
import sys
import glob
import json
import re
import tempfile
import csv
import datetime
import subprocess
from typing import Optional
from PyQt6.QtCore import Qt
from PyQt6.QtWidgets import (
QDialog, QVBoxLayout, QHBoxLayout, QLabel, QPushButton,
QListWidget, QListWidgetItem, QLineEdit, QCheckBox,
QPlainTextEdit, QMessageBox, QGroupBox, QFormLayout,
QWidget, QGridLayout
)
from coseda.crystfel_cell import (
crystfel_unique_axis_allowed,
normalize_crystfel_lattice_type,
)
def _human_rel(path: str, base: str) -> str:
"""Return a path relative to base; fall back to absolute on error."""
try:
return os.path.relpath(path, base)
except Exception:
return path
def _open_in_explorer(path: str) -> None:
"""Open a file or directory in the OS file explorer (cross-platform)."""
try:
if sys.platform.startswith("darwin"):
os.system(f'open "{path}"')
elif os.name == "nt":
# On Windows, open the parent folder and select the file if it's a file
if os.path.isfile(path):
os.system(f'explorer /select,"{path}"')
else:
os.startfile(path) # type: ignore
else:
os.system(f'xdg-open "{path}"')
except Exception:
pass
[docs]
class ExportMtzWindow(QDialog):
"""UI to list CrystFEL HKL files under merge folders and export HKLF4."""
def __init__(self, workspace_path: str, parent: Optional[QWidget] = None) -> None:
super().__init__(parent)
self.setWindowTitle("Export HKLF4 / MTZ / INS / P4P")
self.resize(900, 600)
self.workspace_path = workspace_path # path to workspace file or folder
# If a file was passed (e.g., .ws), use its folder as root
if os.path.isfile(self.workspace_path):
self.root_dir = os.path.dirname(self.workspace_path)
else:
self.root_dir = self.workspace_path
self.current_cell_params: Optional[dict] = None
self.current_stream_path: Optional[str] = None
self.current_pointgroup: Optional[str] = None
self.current_noncentro: bool = False
self.current_wavelength: Optional[float] = None
self._build_ui()
self._refresh_list()
# ---------------------------- UI ----------------------------
def _build_ui(self) -> None:
"""Construct the dialog layout and wire signals."""
layout = QVBoxLayout(self)
# Top: List HKL files
list_box = QGroupBox("Merge Runs")
list_layout = QVBoxLayout(list_box)
self.hkl_list = QListWidget()
self.hkl_list.itemSelectionChanged.connect(self._on_hkl_selected)
list_layout.addWidget(self.hkl_list)
top_row = QHBoxLayout()
self.refresh_btn = QPushButton("Refresh")
self.refresh_btn.clicked.connect(self._refresh_list)
self.open_loc_btn = QPushButton("Open Location")
self.open_loc_btn.clicked.connect(self._open_selected_location)
self.open_loc_btn.setEnabled(False)
top_row.addWidget(self.refresh_btn)
top_row.addWidget(self.open_loc_btn)
top_row.addStretch(1)
list_layout.addLayout(top_row)
layout.addWidget(list_box)
# Middle: Cell (parsed from stream) -> RES/P4P
cell_box = QGroupBox("Cell")
cell_form = QFormLayout(cell_box)
self.cell_source_label = QLabel("Select an HKL to infer stream & cell")
self.cell_source_label.setTextInteractionFlags(Qt.TextInteractionFlag.TextSelectableByMouse)
cell_form.addRow(QLabel("Cell source:"), self.cell_source_label)
# Cell parameter editors (prefilled, locked by default)
self.lattice_edit = QLineEdit()
self.centering_edit = QLineEdit()
self.pg_edit = QLineEdit()
self.a_edit = QLineEdit()
self.b_edit = QLineEdit()
self.c_edit = QLineEdit()
self.al_edit = QLineEdit()
self.be_edit = QLineEdit()
self.ga_edit = QLineEdit()
for w in (
self.lattice_edit,
self.centering_edit,
self.pg_edit,
self.a_edit,
self.b_edit,
self.c_edit,
self.al_edit,
self.be_edit,
self.ga_edit,
):
w.setReadOnly(True)
# Arrange fields in 3 columns using a grid
grid_widget = QWidget()
grid = QGridLayout(grid_widget)
grid.setHorizontalSpacing(12)
grid.setVerticalSpacing(6)
def add(row: int, col: int, label: str, widget: QLineEdit):
lbl = QLabel(label)
lbl.setAlignment(Qt.AlignmentFlag.AlignRight | Qt.AlignmentFlag.AlignVCenter)
grid.addWidget(lbl, row, col * 2)
grid.addWidget(widget, row, col * 2 + 1)
add(0, 0, "Lattice type:", self.lattice_edit)
add(0, 1, "Centering:", self.centering_edit)
add(0, 2, "Point group:", self.pg_edit)
add(1, 0, "a (Å):", self.a_edit)
add(1, 1, "b (Å):", self.b_edit)
add(1, 2, "c (Å):", self.c_edit)
add(2, 0, "\u03b1 (°):", self.al_edit)
add(2, 1, "\u03b2 (°):", self.be_edit)
add(2, 2, "\u03b3 (°):", self.ga_edit)
# Space group and composition (optional, for INS export)
self.spacegroup_edit = QLineEdit()
self.spacegroup_edit.setPlaceholderText("e.g. P 21 21 21")
add(3, 0, "Space group:", self.spacegroup_edit)
self.composition_edit = QLineEdit()
self.composition_edit.setPlaceholderText("e.g. C6 H12 N2 O")
add(3, 1, "Composition:", self.composition_edit)
self.dmin_edit = QLineEdit()
self.dmin_edit.setPlaceholderText("e.g. 0.8")
self.dmin_edit.setToolTip("Resolution cutoff in Å for HKLF4 export (leave empty for no cutoff)")
add(3, 2, "d_min (Å):", self.dmin_edit)
cell_form.addRow(grid_widget)
self.edit_cell_btn = QPushButton("Enable edits…")
self.edit_cell_btn.clicked.connect(self._toggle_cell_editing)
cell_form.addRow(QLabel(""), self.edit_cell_btn)
layout.addWidget(cell_box)
# Bottom: actions in their own frame
actions_box = QGroupBox("Write files")
btn_row = QHBoxLayout(actions_box)
self.export_btn = QPushButton("Export HKLF4")
self.export_btn.clicked.connect(self._on_export)
self.export_btn.setEnabled(False)
btn_row.addWidget(self.export_btn)
self.export_res_btn = QPushButton("Write INS")
self.export_res_btn.clicked.connect(self._on_export_res)
btn_row.addWidget(self.export_res_btn)
self.export_p4p_btn = QPushButton("Write P4P")
self.export_p4p_btn.clicked.connect(self._on_export_p4p)
btn_row.addWidget(self.export_p4p_btn)
self.export_mtz_btn = QPushButton("Export MTZ")
self.export_mtz_btn.clicked.connect(self._on_export_mtz)
btn_row.addWidget(self.export_mtz_btn)
self.export_txt_btn = QPushButton("Write Summary TXT")
self.export_txt_btn.clicked.connect(self._on_export_txt)
self.export_txt_btn.setEnabled(False)
btn_row.addWidget(self.export_txt_btn)
self.anomalous_cb = QCheckBox("Anomalous")
self.anomalous_cb.setChecked(False)
self.anomalous_cb.setToolTip("Keep Friedel pairs separate (for anomalous data)")
btn_row.addWidget(self.anomalous_cb)
self.overwrite_cb = QCheckBox("Overwrite existing files")
self.overwrite_cb.setChecked(False)
btn_row.addWidget(self.overwrite_cb)
btn_row.addStretch(1)
layout.addWidget(actions_box)
# ---------------------------- listing ----------------------------
def _refresh_list(self) -> None:
"""Populate the HKL list by globbing merge folders adjacent to the workspace."""
self.hkl_list.clear()
root = self.root_dir.rstrip(os.sep)
# Common layouts:
# <root>/<workspace>_merge_YYYYMMDD_HHMMSS/merge-run*/crystfel.hkl
# <root>/<ini-data-folder>/<workspace>_merge_YYYYMMDD_HHMMSS/merge-run*/crystfel.hkl
# Older merge runs were written next to the selected INI/data folder.
# Recursing keeps those legacy runs discoverable from the workspace root.
patterns = [
os.path.join(root, "*_merge_*", "merge-run*", "crystfel.hkl"),
os.path.join(root, "**", "*_merge_*", "merge-run*", "crystfel.hkl"),
]
paths = []
seen = set()
for pattern in patterns:
for p in glob.glob(pattern, recursive=True):
if p in seen:
continue
seen.add(p)
paths.append(p)
paths.sort()
for p in paths:
lbl = _human_rel(p, root)
item = QListWidgetItem(lbl)
item.setData(Qt.ItemDataRole.UserRole, p)
self.hkl_list.addItem(item)
self.export_btn.setEnabled(False)
self.export_txt_btn.setEnabled(False)
self.open_loc_btn.setEnabled(False)
def _on_hkl_selected(self) -> None:
"""Handle selection changes; enable buttons."""
items = self.hkl_list.selectedItems()
has = bool(items)
self.export_btn.setEnabled(has)
self.export_txt_btn.setEnabled(has)
self.open_loc_btn.setEnabled(has)
if has:
hkl = items[0].data(Qt.ItemDataRole.UserRole)
self._update_cell_from_hkl(hkl)
def _open_selected_location(self) -> None:
"""Open the folder containing the selected HKL."""
items = self.hkl_list.selectedItems()
if not items:
return
hkl = items[0].data(Qt.ItemDataRole.UserRole)
_open_in_explorer(hkl)
# ---------------------------- inference helpers ----------------------------
# ---------------------------- export ----------------------------
def _on_export(self) -> None:
"""Convert selected CrystFEL HKL to HKLF4 using local converter."""
from coseda.exporters import crystfel_to_hklf4
items = self.hkl_list.selectedItems()
if not items:
return
hkl_path = items[0].data(Qt.ItemDataRole.UserRole)
run_dir = os.path.dirname(hkl_path)
merge_root = os.path.dirname(run_dir) # place at top level of merge folder
merge_base = os.path.basename(merge_root)
workspace_name = merge_base.split("_merge_")[0] if "_merge_" in merge_base else merge_base
out_hkl = f"{workspace_name}.hkl"
out_hkl_path = os.path.join(merge_root, out_hkl)
if os.path.exists(out_hkl_path) and not self.overwrite_cb.isChecked():
QMessageBox.warning(self, "Export HKLF4", f"{os.path.basename(out_hkl_path)} already exists. Enable overwrite to replace.")
return
try:
d_min = None
cell = None
dmin_text = self.dmin_edit.text().strip().replace(",", ".")
if dmin_text:
d_min = float(dmin_text)
params = self._collect_cell_from_edits()
if params:
cell = (params["a"], params["b"], params["c"],
params["al"], params["be"], params["ga"])
else:
QMessageBox.warning(self, "Export HKLF4",
"Cell parameters are required for resolution cutoff.")
return
count = crystfel_to_hklf4(hkl_path, out_hkl_path, d_min=d_min, cell=cell)
except Exception as exc:
QMessageBox.critical(self, "Export HKLF4", f"Export failed:\n{exc}")
return
msg = f"HKLF4 written ({count} reflections)"
if d_min is not None:
msg += f" [d > {d_min:.2f} Å]"
self._set_status(msg)
def _on_export_res(self) -> None:
"""Generate an INS instruction file from the selected cell file."""
from coseda.exporters import crystfel_cell_to_ins
hkl_item = self.hkl_list.selectedItems()
if not hkl_item:
QMessageBox.warning(self, "Write INS", "Select an HKL/merge-run first.")
return
hkl_path = hkl_item[0].data(Qt.ItemDataRole.UserRole)
params = self._collect_cell_from_edits()
if not params:
QMessageBox.warning(self, "Write INS", "No unit cell could be parsed from the associated stream.")
return
merge_root = os.path.dirname(os.path.dirname(hkl_path))
merge_base = os.path.basename(merge_root)
workspace_name = merge_base.split("_merge_")[0] if "_merge_" in merge_base else merge_base
out_ins = os.path.join(merge_root, f"{workspace_name}.ins")
if os.path.exists(out_ins) and not self.overwrite_cb.isChecked():
QMessageBox.warning(self, "Write INS", f"{os.path.basename(out_ins)} already exists. Enable overwrite to replace.")
return
# Write a temporary .cell from the parsed stream cell
tmp_cell = None
try:
tmp_cell = self._write_temp_cell(params)
wl = self.current_wavelength if self.current_wavelength is not None else 1.0
sg = self.spacegroup_edit.text().strip()
comp = self.composition_edit.text().strip()
crystfel_cell_to_ins(
tmp_cell, out_ins, overwrite=True,
noncentrosymmetric=self.current_noncentro,
wavelength=wl, spacegroup=sg, composition=comp,
)
except Exception as exc:
QMessageBox.critical(self, "Write INS", f"INS generation failed:\n{exc}")
return
finally:
if tmp_cell and os.path.exists(tmp_cell):
try:
os.remove(tmp_cell)
except Exception:
pass
self._set_status("INS written")
def _on_export_p4p(self) -> None:
"""Generate a P4P file from the selected cell file."""
from coseda.exporters import crystfel_cell_to_p4p
hkl_item = self.hkl_list.selectedItems()
if not hkl_item:
QMessageBox.warning(self, "Write P4P", "Select an HKL/merge-run first.")
return
hkl_path = hkl_item[0].data(Qt.ItemDataRole.UserRole)
params = self._collect_cell_from_edits()
if not params:
QMessageBox.warning(self, "Write P4P", "No unit cell could be parsed from the associated stream.")
return
merge_root = os.path.dirname(os.path.dirname(hkl_path))
merge_base = os.path.basename(merge_root)
workspace_name = merge_base.split("_merge_")[0] if "_merge_" in merge_base else merge_base
out_p4p = os.path.join(merge_root, f"{workspace_name}.p4p")
if os.path.exists(out_p4p) and not self.overwrite_cb.isChecked():
QMessageBox.warning(self, "Write P4P", f"{os.path.basename(out_p4p)} already exists. Enable overwrite to replace.")
return
tmp_cell = None
try:
tmp_cell = self._write_temp_cell(params)
crystfel_cell_to_p4p(tmp_cell, out_p4p, overwrite=True, noncentrosymmetric=self.current_noncentro)
except Exception as exc:
QMessageBox.critical(self, "Write P4P", f"P4P generation failed:\n{exc}")
return
finally:
if tmp_cell and os.path.exists(tmp_cell):
try:
os.remove(tmp_cell)
except Exception:
pass
self._set_status("P4P written")
def _on_export_mtz(self) -> None:
"""Convert selected CrystFEL HKL to MTZ format."""
from coseda.exporters.mtz import hklf4_to_mtz
hkl_item = self.hkl_list.selectedItems()
if not hkl_item:
QMessageBox.warning(self, "Export MTZ", "Select an HKL/merge-run first.")
return
hkl_path = hkl_item[0].data(Qt.ItemDataRole.UserRole)
params = self._collect_cell_from_edits()
if not params:
QMessageBox.warning(self, "Export MTZ", "No unit cell could be parsed from the associated stream.")
return
if self.current_wavelength is None:
QMessageBox.warning(self, "Export MTZ", "No wavelength found in stream; cannot export MTZ.")
return
# Build unit cell tuple
unit_cell = (
params["a"],
params["b"],
params["c"],
params["al"],
params["be"],
params["ga"],
)
# Build output path
merge_root = os.path.dirname(os.path.dirname(hkl_path))
merge_base = os.path.basename(merge_root)
workspace_name = merge_base.split("_merge_")[0] if "_merge_" in merge_base else merge_base
out_mtz = os.path.join(merge_root, f"{workspace_name}.mtz")
if os.path.exists(out_mtz) and not self.overwrite_cb.isChecked():
QMessageBox.warning(self, "Export MTZ", f"{os.path.basename(out_mtz)} already exists. Enable overwrite to replace.")
return
spacegroup = self.spacegroup_edit.text().strip()
if not spacegroup:
QMessageBox.warning(
self,
"Export MTZ",
"Enter the crystal space group before exporting MTZ. Use P 1 only if the data should really be exported as P1.",
)
return
d_min = None
dmin_text = self.dmin_edit.text().strip().replace(",", ".")
if dmin_text:
try:
d_min = float(dmin_text)
except ValueError:
QMessageBox.warning(self, "Export MTZ", f"Invalid d_min value: {dmin_text!r}")
return
try:
count = hklf4_to_mtz(
hkl_path,
out_mtz,
unit_cell=unit_cell,
spacegroup=spacegroup,
wavelength=self.current_wavelength,
anomalous=self.anomalous_cb.isChecked(),
d_min=d_min,
overwrite=True,
)
except Exception as exc:
QMessageBox.critical(self, "Export MTZ", f"MTZ export failed:\n{exc}")
return
msg = f"MTZ written ({count} reflections, {spacegroup})"
if d_min is not None:
msg += f" [d > {d_min:.2f} Å]"
self._set_status(msg)
def _on_export_txt(self) -> None:
"""Write a human-readable summary for the selected merge run."""
hkl_item = self.hkl_list.selectedItems()
if not hkl_item:
QMessageBox.warning(self, "Write Summary TXT", "Select an HKL/merge-run first.")
return
hkl_path = hkl_item[0].data(Qt.ItemDataRole.UserRole)
merge_root = os.path.dirname(os.path.dirname(hkl_path))
merge_base = os.path.basename(merge_root)
workspace_name = merge_base.split("_merge_")[0] if "_merge_" in merge_base else merge_base
out_txt = os.path.join(merge_root, f"{workspace_name}.txt")
if os.path.exists(out_txt) and not self.overwrite_cb.isChecked():
QMessageBox.warning(
self,
"Write Summary TXT",
f"{os.path.basename(out_txt)} already exists. Enable overwrite to replace.",
)
return
try:
with open(out_txt, "w", encoding="utf-8") as handle:
handle.write(self._build_text_summary(hkl_path))
except Exception as exc:
QMessageBox.critical(self, "Write Summary TXT", f"TXT export failed:\n{exc}")
return
self._set_status(f"Summary TXT written: {os.path.basename(out_txt)}")
def _build_text_summary(self, hkl_path: str) -> str:
"""Build a plain-text summary of the selected merge and export settings."""
merge_run = os.path.dirname(hkl_path)
merge_root = os.path.dirname(merge_run)
stream_path = self._stream_for_hkl(hkl_path)
settings = self._settings_for_stream(stream_path)
lines = [
"COSEDA export summary",
f"Generated: {datetime.datetime.now().isoformat(timespec='seconds')}",
"",
"[Selected files]",
f"HKL: {hkl_path}",
f"Merge run: {merge_run}",
f"Merge folder: {merge_root}",
f"Stream: {stream_path or ''}",
"",
"[Paper table values]",
*self._paper_table_lines(hkl_path),
"",
"[Cell and symmetry]",
f"Lattice type: {self.lattice_edit.text().strip()}",
f"Centering: {self.centering_edit.text().strip()}",
f"Point group: {self.pg_edit.text().strip()}",
f"Space group: {self.spacegroup_edit.text().strip()}",
f"a (A): {self.a_edit.text().strip()}",
f"b (A): {self.b_edit.text().strip()}",
f"c (A): {self.c_edit.text().strip()}",
f"alpha (deg): {self.al_edit.text().strip()}",
f"beta (deg): {self.be_edit.text().strip()}",
f"gamma (deg): {self.ga_edit.text().strip()}",
f"Wavelength: {'' if self.current_wavelength is None else self.current_wavelength}",
f"Composition: {self.composition_edit.text().strip()}",
"",
"[Export options]",
f"d_min (A): {self.dmin_edit.text().strip()}",
f"Anomalous: {self.anomalous_cb.isChecked()}",
"",
]
if settings:
lines.extend(["[Merge settings]"])
for key in sorted(settings):
lines.append(f"{key}: {settings[key]}")
lines.append("")
for filename, title in (
("merging_table.csv", "Merging table"),
("statistics_table.csv", "Resolution statistics"),
("split_half_table.csv", "Split-half statistics"),
):
csv_path = os.path.join(merge_root, filename)
if os.path.isfile(csv_path):
lines.extend([f"[{title}]", self._csv_as_tsv(csv_path), ""])
return "\n".join(lines).rstrip() + "\n"
def _paper_table_lines(self, hkl_path: str) -> list[str]:
"""Return manuscript-oriented overall/outer-shell values from saved merge artifacts."""
exact_lines = self._crystfel_paper_table_lines(hkl_path)
if exact_lines:
return exact_lines
merge_run = os.path.dirname(hkl_path)
merge_root = os.path.dirname(merge_run)
stats_rows = self._read_csv_dicts(os.path.join(merge_root, "statistics_table.csv"))
split_rows = self._read_csv_dicts(os.path.join(merge_root, "split_half_table.csv"))
merge_rows = self._read_csv_dicts(os.path.join(merge_root, "merging_table.csv"))
lines = [
"Values are formatted as overall (outer shell) where both are available.",
"Fallback values from saved shell CSV artifacts; run with CrystFEL available for exact cutoff values.",
]
final_merge = merge_rows[-1] if merge_rows else {}
used_crystals = final_merge.get("OK Crystals") or final_merge.get("ok_crystals") or ""
final_cc_half = final_merge.get("Overall CChalf (%)") or final_merge.get("overall_cc_half") or ""
if used_crystals:
lines.append(f"Number of crystals: {used_crystals}")
if final_cc_half:
lines.append(f"Final partialator CC1/2 (%): {final_cc_half}")
if stats_rows:
stats_summary = self._summarize_statistics_rows(stats_rows)
if stats_summary:
lines.extend([
f"Resolution range (A): {stats_summary['resolution']}",
f"Reflections total / unique: {stats_summary['measurements']} / {stats_summary['unique']}",
f"Multiplicity: {stats_summary['multiplicity']}",
f"Completeness (%): {stats_summary['completeness']}",
f"I/sigma(I): {stats_summary['snr']}",
])
else:
lines.append("Resolution statistics: not available; run merge statistics first.")
if split_rows:
split_summary = self._summarize_split_rows(split_rows)
if split_summary:
lines.extend([
f"CC1/2 (%): {split_summary['cc']}",
f"CC* (%): {split_summary['ccstar']}",
f"Rsplit (%): {split_summary['rsplit']}",
])
else:
lines.append("Split-half statistics: not available; run split-half analysis first.")
return lines
def _crystfel_paper_table_lines(self, hkl_path: str) -> list[str]:
"""Compute exact manuscript statistics with check_hkl/compare_hkl when available."""
params = self._collect_cell_from_edits()
pointgroup = (self.pg_edit.text().strip() or self.current_pointgroup or "").strip()
if not params or not pointgroup:
return []
merge_run = os.path.dirname(hkl_path)
hkl1_path = os.path.join(merge_run, "crystfel.hkl1")
hkl2_path = os.path.join(merge_run, "crystfel.hkl2")
if not os.path.isfile(hkl_path) or not os.path.isfile(hkl1_path) or not os.path.isfile(hkl2_path):
return []
d_min = self._requested_d_min()
try:
tmp_cell = self._write_temp_cell(params)
with tempfile.TemporaryDirectory() as tmpdir:
check_result = self._run_check_hkl_summary(hkl_path, tmp_cell, pointgroup, tmpdir, d_min)
if not check_result:
return []
compare_result = self._run_compare_hkl_summary(hkl1_path, hkl2_path, tmp_cell, pointgroup, tmpdir, d_min)
except Exception as exc:
self._set_status(f"Summary TXT statistics fallback: {exc}")
return []
finally:
try:
if "tmp_cell" in locals() and tmp_cell and os.path.exists(tmp_cell):
os.remove(tmp_cell)
except Exception:
pass
merge_root = os.path.dirname(merge_run)
merge_rows = self._read_csv_dicts(os.path.join(merge_root, "merging_table.csv"))
final_merge = merge_rows[-1] if merge_rows else {}
used_crystals = final_merge.get("OK Crystals") or final_merge.get("ok_crystals") or ""
final_cc_half = final_merge.get("Overall CChalf (%)") or final_merge.get("overall_cc_half") or ""
lines = [
"Values are formatted as overall (outer shell).",
"Computed with check_hkl/compare_hkl from the selected HKL and split-half HKL files.",
]
if d_min is not None:
lines.append(f"High-resolution cutoff used for statistics: {d_min:.3f} A")
if used_crystals:
lines.append(f"Number of crystals: {used_crystals}")
if final_cc_half:
lines.append(f"Final partialator CC1/2 (%): {final_cc_half}")
lines.extend([
f"Resolution range (A): {check_result['resolution']}",
f"Reflections total / unique: {check_result['reflections']}",
f"Multiplicity: {check_result['multiplicity']}",
f"Completeness (%): {check_result['completeness']}",
f"I/sigma(I): {check_result['snr']}",
])
if compare_result:
lines.extend([
f"CC1/2 (%): {compare_result.get('cc', '')}",
f"CC* (%): {compare_result.get('ccstar', '')}",
f"Rsplit (%): {compare_result.get('rsplit', '')}",
])
return lines
def _requested_d_min(self) -> float | None:
text = self.dmin_edit.text().strip().replace(",", ".")
if not text:
return None
try:
value = float(text)
except ValueError:
return None
return value if value > 0 else None
def _run_check_hkl_summary(
self,
hkl_path: str,
cell_path: str,
pointgroup: str,
tmpdir: str,
d_min: float | None,
) -> dict:
shell_file = os.path.join(tmpdir, "check_hkl_shells.dat")
cmd = [
"check_hkl",
"-p", cell_path,
"-y", pointgroup,
"--nshells=1",
f"--shell-file={shell_file}",
]
if d_min is not None:
cmd.append(f"--highres={d_min}")
cmd.append(hkl_path)
proc = subprocess.run(cmd, capture_output=True, text=True, timeout=120)
if proc.returncode != 0:
raise RuntimeError((proc.stderr or proc.stdout or "check_hkl failed").strip())
stdout = "\n".join(part for part in (proc.stdout, proc.stderr) if part)
measurements = self._regex_first_float(stdout, r"(\d+)\s+measurements in total")
unique = self._regex_first_float(stdout, r"(\d+)\s+reflections in total")
possible = self._regex_first_float(stdout, r"(\d+)\s+reflections possible")
snr = self._regex_first_float(stdout, r"Overall <snr> =\s*([0-9.eE+-]+)")
redundancy = self._regex_first_float(stdout, r"Overall redundancy =\s*([0-9.eE+-]+)")
completeness = self._regex_first_float(stdout, r"Overall completeness =\s*([0-9.eE+-]+)\s*%")
inv_range = re.search(r"1/d goes from\s*([0-9.eE+-]+)\s+to\s+([0-9.eE+-]+)\s+nm\^-1", stdout)
low_res = self._resolution_from_inv_nm(float(inv_range.group(1))) if inv_range else None
high_res = self._resolution_from_inv_nm(float(inv_range.group(2))) if inv_range else d_min
outer_red = outer_snr = outer_compl = outer_low = outer_high = None
outer_measurements = outer_unique = None
if d_min is not None:
outer_low_cutoff = self._outer_shell_lowres(d_min)
outer_shell_file = os.path.join(tmpdir, "check_hkl_outer_shell.dat")
outer_cmd = [
"check_hkl",
"-p", cell_path,
"-y", pointgroup,
"--nshells=1",
f"--shell-file={outer_shell_file}",
f"--lowres={outer_low_cutoff}",
f"--highres={d_min}",
hkl_path,
]
outer_proc = subprocess.run(outer_cmd, capture_output=True, text=True, timeout=120)
if outer_proc.returncode == 0:
outer_output = "\n".join(part for part in (outer_proc.stdout, outer_proc.stderr) if part)
outer_measurements = self._regex_first_float(outer_output, r"(\d+)\s+measurements in total")
outer_unique = self._regex_first_float(outer_output, r"(\d+)\s+reflections in total")
outer_snr = self._regex_first_float(outer_output, r"Overall <snr> =\s*([0-9.eE+-]+)")
outer_red = self._regex_first_float(outer_output, r"Overall redundancy =\s*([0-9.eE+-]+)")
outer_compl = self._regex_first_float(outer_output, r"Overall completeness =\s*([0-9.eE+-]+)\s*%")
outer_inv_range = re.search(r"1/d goes from\s*([0-9.eE+-]+)\s+to\s+([0-9.eE+-]+)\s+nm\^-1", outer_output)
outer_low = self._resolution_from_inv_nm(float(outer_inv_range.group(1))) if outer_inv_range else outer_low_cutoff
outer_high = self._resolution_from_inv_nm(float(outer_inv_range.group(2))) if outer_inv_range else d_min
else:
rows = self._parse_check_hkl_shell_rows(shell_file)
outer = rows[-1] if rows else {}
outer_refs = self._float_from_row(outer, "refs")
outer_possible = self._float_from_row(outer, "possible")
outer_meas = self._float_from_row(outer, "meas")
outer_measurements = outer_meas
outer_unique = outer_refs
outer_compl = (
100.0 * outer_refs / outer_possible
if outer_refs is not None and outer_possible
else self._float_from_row(outer, "compl")
)
outer_red = outer_meas / outer_refs if outer_meas is not None and outer_refs else self._float_from_row(outer, "red")
outer_snr = self._float_from_row(outer, "snr")
outer_low = self._resolution_from_inv_nm(self._float_from_row(outer, "min_inv_nm"))
outer_high = self._resolution_from_inv_nm(self._float_from_row(outer, "max_inv_nm"))
return {
"resolution": self._format_resolution_range(low_res, high_res, outer_low, outer_high),
"measurements": self._format_int(measurements),
"unique": self._format_int(unique),
"reflections": self._format_reflection_counts(measurements, unique, outer_measurements, outer_unique),
"possible": self._format_int(possible),
"multiplicity": self._format_overall_outer(redundancy, outer_red, 1),
"completeness": self._format_overall_outer(completeness, outer_compl, 1),
"snr": self._format_overall_outer(snr, outer_snr, 2),
}
def _run_compare_hkl_summary(
self,
hkl1_path: str,
hkl2_path: str,
cell_path: str,
pointgroup: str,
tmpdir: str,
d_min: float | None,
) -> dict:
results = {}
for fom, key, pattern, as_percent in (
("CC", "cc", r"Overall CC =\s*([0-9.eE+-]+)", True),
("CCstar", "ccstar", r"Overall CC\* =\s*([0-9.eE+-]+)", True),
("Rsplit", "rsplit", r"Overall Rsplit =\s*([0-9.eE+-]+)\s*%", False),
):
shell_file = os.path.join(tmpdir, f"compare_{fom.lower()}.dat")
cmd = [
"compare_hkl",
"-p", cell_path,
"-y", pointgroup,
f"--fom={fom}",
"--nshells=10",
f"--shell-file={shell_file}",
]
if d_min is not None:
cmd.append(f"--highres={d_min}")
cmd.extend([hkl1_path, hkl2_path])
proc = subprocess.run(cmd, capture_output=True, text=True, timeout=120)
if proc.returncode != 0:
continue
output = "\n".join(part for part in (proc.stdout, proc.stderr) if part)
overall = self._regex_first_float(output, pattern)
outer = None
if d_min is not None:
outer_shell_file = os.path.join(tmpdir, f"compare_{fom.lower()}_outer.dat")
outer_cmd = [
"compare_hkl",
"-p", cell_path,
"-y", pointgroup,
f"--fom={fom}",
"--nshells=1",
f"--shell-file={outer_shell_file}",
f"--lowres={self._outer_shell_lowres(d_min)}",
f"--highres={d_min}",
hkl1_path,
hkl2_path,
]
outer_proc = subprocess.run(outer_cmd, capture_output=True, text=True, timeout=120)
if outer_proc.returncode == 0:
outer_output = "\n".join(part for part in (outer_proc.stdout, outer_proc.stderr) if part)
outer = self._regex_first_float(outer_output, pattern)
else:
shell_rows = self._parse_compare_hkl_shell_rows(shell_file, key)
if shell_rows:
outer = self._float_from_row(shell_rows[-1], key)
if as_percent:
overall = self._fraction_to_percent(overall)
outer = self._fraction_to_percent(outer)
results[key] = self._format_overall_outer(overall, outer, 1)
return results
def _outer_shell_lowres(self, d_min: float) -> float:
"""Choose a manuscript-style final shell close to the high-resolution cutoff."""
return round(d_min * 1.04, 2)
def _regex_first_float(self, text: str, pattern: str) -> float | None:
match = re.search(pattern, text)
if not match:
return None
try:
return float(match.group(1))
except Exception:
return None
def _parse_check_hkl_shell_rows(self, shell_file: str) -> list[dict]:
keys = [
"center", "refs", "possible", "compl", "meas",
"red", "snr", "mean_i", "d_ang", "min_inv_nm", "max_inv_nm",
]
rows = []
if not os.path.isfile(shell_file):
return rows
with open(shell_file, "r", encoding="utf-8", errors="replace") as handle:
for line in handle:
stripped = line.strip()
if not stripped or stripped.startswith("Center") or stripped.startswith("-"):
continue
parts = stripped.split()
if len(parts) >= len(keys):
rows.append({key: parts[idx] for idx, key in enumerate(keys)})
return rows
def _parse_compare_hkl_shell_rows(self, shell_file: str, value_key: str) -> list[dict]:
rows = []
if not os.path.isfile(shell_file):
return rows
with open(shell_file, "r", encoding="utf-8", errors="replace") as handle:
for line in handle:
stripped = line.strip()
if not stripped or stripped.startswith("1/d") or stripped.startswith("-"):
continue
parts = stripped.split()
if len(parts) >= 6:
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],
})
return rows
def _summarize_statistics_rows(self, rows: list[dict]) -> dict:
refs_total = 0.0
possible_total = 0.0
measurements_total = 0.0
snr_weighted = 0.0
snr_weight = 0.0
parsed = []
for row in rows:
refs = self._float_from_row(row, "# refs", "refs")
possible = self._float_from_row(row, "Possible", "possible")
measurements = self._float_from_row(row, "Meas", "meas")
snr = self._float_from_row(row, "SNR", "snr")
min_inv = self._float_from_row(row, "Min 1/nm", "min_inv_nm")
max_inv = self._float_from_row(row, "Max 1/nm", "max_inv_nm")
if refs is not None:
refs_total += refs
if snr is not None:
snr_weighted += snr * refs
snr_weight += refs
if possible is not None:
possible_total += possible
if measurements is not None:
measurements_total += measurements
parsed.append((row, refs, possible, measurements, snr, min_inv, max_inv))
outer = parsed[-1] if parsed else None
completeness = 100.0 * refs_total / possible_total if possible_total else None
multiplicity = measurements_total / refs_total if refs_total else None
snr_overall = snr_weighted / snr_weight if snr_weight else None
low_res = self._resolution_from_inv_nm(parsed[0][5]) if parsed and parsed[0][5] else None
high_res = self._resolution_from_inv_nm(parsed[-1][6]) if parsed and parsed[-1][6] else None
outer_refs = outer[1] if outer else None
outer_possible = outer[2] if outer else None
outer_measurements = outer[3] if outer else None
outer_snr = outer[4] if outer else None
outer_completeness = (
100.0 * outer_refs / outer_possible
if outer_refs is not None and outer_possible
else self._float_from_row(outer[0], "Compl (%)", "compl") if outer else None
)
outer_multiplicity = (
outer_measurements / outer_refs
if outer_measurements is not None and outer_refs
else self._float_from_row(outer[0], "Red", "red") if outer else None
)
outer_low = self._resolution_from_inv_nm(outer[5]) if outer and outer[5] else None
outer_high = self._resolution_from_inv_nm(outer[6]) if outer and outer[6] else None
return {
"resolution": self._format_resolution_range(low_res, high_res, outer_low, outer_high),
"measurements": self._format_int(measurements_total),
"unique": self._format_int(refs_total),
"multiplicity": self._format_overall_outer(multiplicity, outer_multiplicity, 1),
"completeness": self._format_overall_outer(completeness, outer_completeness, 1),
"snr": self._format_overall_outer(snr_overall, outer_snr, 2),
}
def _summarize_split_rows(self, rows: list[dict]) -> dict:
def weighted_average(key: str) -> float | None:
total = 0.0
weight = 0.0
for row in rows:
value = self._float_from_row(row, key)
nref = self._float_from_row(row, "# refs", "nref")
if value is None:
continue
if nref is None:
nref = 1.0
total += value * nref
weight += nref
return total / weight if weight else None
outer = rows[-1] if rows else {}
return {
"cc": self._format_overall_outer(
self._fraction_to_percent(weighted_average("CC1/2")),
self._fraction_to_percent(self._float_from_row(outer, "CC1/2")),
1,
),
"ccstar": self._format_overall_outer(
self._fraction_to_percent(weighted_average("CC*")),
self._fraction_to_percent(self._float_from_row(outer, "CC*")),
1,
),
"rsplit": self._format_overall_outer(
weighted_average("Rsplit (%)"),
self._float_from_row(outer, "Rsplit (%)"),
1,
),
}
def _read_csv_dicts(self, path: str) -> list[dict]:
if not os.path.isfile(path):
return []
try:
with open(path, newline="", encoding="utf-8") as handle:
return list(csv.DictReader(handle))
except Exception:
return []
def _float_from_row(self, row: dict | None, *keys: str) -> float | None:
if not row:
return None
for key in keys:
value = row.get(key)
if value is None:
continue
text = str(value).strip().replace(",", ".")
if not text:
continue
try:
return float(text)
except ValueError:
continue
return None
def _fraction_to_percent(self, value: float | None) -> float | None:
if value is None:
return None
return value * 100.0 if abs(value) <= 1.5 else value
def _resolution_from_inv_nm(self, value: float | None) -> float | None:
if not value:
return None
return 10.0 / value
def _format_resolution_range(
self,
low_res: float | None,
high_res: float | None,
outer_low: float | None,
outer_high: float | None,
) -> str:
if low_res is None or high_res is None:
return ""
text = f"{low_res:.2f}-{high_res:.2f}"
if outer_low is not None and outer_high is not None:
text += f" ({outer_low:.2f}-{outer_high:.2f})"
return text
def _format_overall_outer(self, overall: float | None, outer: float | None, decimals: int) -> str:
if overall is None:
return ""
text = f"{overall:.{decimals}f}"
if outer is not None:
text += f" ({outer:.{decimals}f})"
return text
def _format_int(self, value: float | None) -> str:
if value is None:
return ""
return str(int(round(value)))
def _format_reflection_counts(
self,
measurements: float | None,
unique: float | None,
outer_measurements: float | None,
outer_unique: float | None,
) -> str:
text = f"{self._format_int(measurements)} / {self._format_int(unique)}"
if outer_measurements is not None and outer_unique is not None:
text += f" ({self._format_int(outer_measurements)} / {self._format_int(outer_unique)})"
return text
def _settings_for_stream(self, stream_path: Optional[str]) -> dict:
"""Load merge settings JSON associated with a combined stream."""
if not stream_path:
return {}
try:
folder = os.path.dirname(stream_path)
base = os.path.basename(folder)
json_path = os.path.join(folder, f"{base}_merge_settings.json")
if not os.path.isfile(json_path):
return {}
with open(json_path, "r", encoding="utf-8") as handle:
data = json.load(handle)
return data if isinstance(data, dict) else {}
except Exception:
return {}
def _csv_as_tsv(self, path: str) -> str:
"""Render a CSV artifact as tab-separated text for the summary file."""
rows = []
with open(path, newline="", encoding="utf-8") as handle:
for row in csv.reader(handle):
rows.append("\t".join(str(cell).replace("\t", " ").replace("\n", " ") for cell in row))
return "\n".join(rows)
# ---------------------------- inference helpers ----------------------------
def _update_cell_from_hkl(self, hkl_path: str | None) -> None:
"""Locate combined stream for HKL and parse its unit-cell header."""
self.current_cell_params = None
self.current_stream_path = None
self.current_pointgroup = None
self.current_noncentro = False
self.current_wavelength = None
self.cell_source_label.setText("Select an HKL to infer stream & cell")
for w in (
self.lattice_edit,
self.centering_edit,
self.pg_edit,
self.a_edit,
self.b_edit,
self.c_edit,
self.al_edit,
self.be_edit,
self.ga_edit,
):
w.setText("")
w.setReadOnly(True)
self.edit_cell_btn.setText("Enable edits…")
if not hkl_path or not os.path.isfile(hkl_path):
return
stream_path = self._stream_for_hkl(hkl_path)
if stream_path:
params = self._parse_stream_cell(stream_path)
pg, noncentro = self._pointgroup_from_settings(stream_path)
wl = self._parse_stream_wavelength(stream_path)
if params:
self.current_cell_params = params
self.current_stream_path = stream_path
self.current_pointgroup = pg
self.current_noncentro = noncentro
self.current_wavelength = wl
if wl is None:
self.cell_source_label.setText(f"{stream_path} (no wavelength found)")
else:
self.cell_source_label.setText(stream_path)
self._populate_cell_fields(params, pg)
else:
self.cell_source_label.setText(f"{stream_path} (no unit cell found)")
def _stream_for_hkl(self, hkl_path: str) -> Optional[str]:
"""Given a merge-run*/crystfel.hkl path, find its combined stream in the parent merge folder."""
merge_run = os.path.dirname(hkl_path)
merge_root = os.path.dirname(merge_run)
base = os.path.basename(merge_root)
candidates = [
os.path.join(merge_root, f"{base}.stream"),
os.path.join(merge_root, "merged.stream"),
]
for c in candidates:
if os.path.isfile(c):
return c
return None
def _parse_stream_cell(self, stream_path: str) -> Optional[dict]:
"""Parse unit-cell block from a CrystFEL stream."""
try:
inside = False
data = {}
with open(stream_path, "r", encoding="utf-8", errors="replace") as f:
for line in f:
s = line.strip()
if s.startswith("----- Begin unit cell"):
inside = True
continue
if s.startswith("----- End unit cell"):
break
if not inside or not s or s.startswith(";") or "=" not in s:
continue
k, v = s.split("=", 1)
data[k.strip().lower()] = v.strip()
if not data:
return None
# numeric parsing
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 _parse_stream_wavelength(self, stream_path: str) -> Optional[float]:
"""Parse wavelength from the geometry block in a CrystFEL stream."""
try:
inside = False
with open(stream_path, "r", encoding="utf-8", errors="replace") as f:
for line in f:
s = line.strip()
if s.startswith("----- Begin geometry file"):
inside = True
continue
if s.startswith("----- End geometry file"):
break
if not inside:
continue
if s.lower().startswith("wavelength"):
match = re.search(r"[-+]?(?:\d+(?:\.\d*)?|\.\d+)(?:[eE][-+]?\d+)?", s)
if match:
return float(match.group(0))
return None
except Exception:
return None
def _pointgroup_from_settings(self, stream_path: str) -> tuple[Optional[str], bool]:
"""Load point group from merge settings JSON next to the stream."""
try:
folder = os.path.dirname(stream_path)
base = os.path.basename(folder)
json_path = os.path.join(folder, f"{base}_merge_settings.json")
if not os.path.isfile(json_path):
return None, False
with open(json_path, "r", encoding="utf-8") as f:
d = json.load(f)
pg = d.get("pointgroup") or None
noncentro = not self._is_centrosymmetric(pg) if pg else False
return pg, noncentro
except Exception:
return None, False
def _is_centrosymmetric(self, pg: Optional[str]) -> bool:
"""Return True if the point group is centrosymmetric."""
if not pg:
return False
pg_norm = pg.replace(" ", "").lower()
centrosym = {
"-1", "2/m", "mmm",
"4/m", "4/mmm",
"-3", "-3m",
"6/m", "6/mmm",
"m-3", "m-3m",
}
return pg_norm in centrosym
def _populate_cell_fields(self, params: dict, pointgroup: Optional[str]) -> None:
"""Fill the read-only cell edits with parsed values."""
self.lattice_edit.setText(str(params.get("lattice_type", "")))
self.centering_edit.setText(str(params.get("centering", "")))
self.pg_edit.setText(pointgroup or "")
self.a_edit.setText(str(params.get("a", "")))
self.b_edit.setText(str(params.get("b", "")))
self.c_edit.setText(str(params.get("c", "")))
self.al_edit.setText(str(params.get("al", "")))
self.be_edit.setText(str(params.get("be", "")))
self.ga_edit.setText(str(params.get("ga", "")))
def _write_temp_cell(self, params: dict) -> str:
"""Emit a temporary CrystFEL-style cell file from parsed params."""
lines = ["CrystFEL unit cell file version 1.0", ""]
cent = params.get("centering") or "P"
lt = normalize_crystfel_lattice_type(params.get("lattice_type") or "", cent)
if lt:
lines.append(f"lattice_type = {lt}")
lines.append(f"centering = {cent}")
unique_axis = params.get("unique_axis")
if unique_axis and crystfel_unique_axis_allowed(lt):
lines.append(f"unique_axis = {unique_axis}")
def fmt(key, unit):
val = params.get(key)
if val is None:
return None
try:
fval = float(val)
return f"{key} = {fval:.6f} {unit}"
except Exception:
return f"{key} = {val}"
for key, unit in (("a", "A"), ("b", "A"), ("c", "A"), ("al", "deg"), ("be", "deg"), ("ga", "deg")):
entry = fmt(key, unit)
if entry:
lines.append(entry)
tmp = tempfile.NamedTemporaryFile(delete=False, suffix=".cell")
tmp.write("\n".join(lines).encode("utf-8"))
tmp.close()
return tmp.name
def _set_status(self, message: str) -> None:
"""Send a short status message to the main window's status label, if available."""
try:
mw = self.parent()
if mw and hasattr(mw, "status_label"):
mw.status_label.setText(message)
mw.status_label.setVisible(True)
except Exception:
pass
def _collect_cell_from_edits(self) -> Optional[dict]:
"""Read current values from the editable fields; validate numerics."""
try:
a = float(self.a_edit.text().strip())
b = float(self.b_edit.text().strip())
c = float(self.c_edit.text().strip())
al = float(self.al_edit.text().strip())
be = float(self.be_edit.text().strip())
ga = float(self.ga_edit.text().strip())
except Exception:
return None
cent = self.centering_edit.text().strip() or "P"
lt = normalize_crystfel_lattice_type(self.lattice_edit.text().strip(), cent)
unique_axis = ""
if self.current_cell_params:
unique_axis = str(self.current_cell_params.get("unique_axis", "")).strip()
lt_norm = lt.lower()
if not unique_axis and lt_norm in {"tetragonal", "hexagonal"}:
unique_axis = "c"
elif not unique_axis and lt_norm == "monoclinic":
unique_axis = "b"
if not crystfel_unique_axis_allowed(lt_norm):
unique_axis = ""
return {
"a": a,
"b": b,
"c": c,
"al": al,
"be": be,
"ga": ga,
"centering": cent,
"lattice_type": lt,
"unique_axis": unique_axis,
}
def _toggle_cell_editing(self) -> None:
"""Allow the user to edit the parsed cell after confirmation."""
editable = self.a_edit.isReadOnly()
if editable:
resp = QMessageBox.question(
self,
"Edit cell values?",
"You are about to edit the unit cell parsed from the stream. Continue?",
QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No,
)
if resp != QMessageBox.StandardButton.Yes:
return
for w in (
self.lattice_edit,
self.centering_edit,
self.a_edit,
self.b_edit,
self.c_edit,
self.al_edit,
self.be_edit,
self.ga_edit,
):
w.setReadOnly(False)
self.edit_cell_btn.setText("Lock edits")
else:
for w in (
self.lattice_edit,
self.centering_edit,
self.a_edit,
self.b_edit,
self.c_edit,
self.al_edit,
self.be_edit,
self.ga_edit,
):
w.setReadOnly(True)
self.edit_cell_btn.setText("Enable edits…")