import os
import re
from configparser import ConfigParser
import numpy as np
from PyQt6.QtCore import QTimer
from PyQt6.QtWidgets import (
QHBoxLayout,
QLabel,
QMessageBox,
QPlainTextEdit,
QPushButton,
QVBoxLayout,
QWidget,
)
from coseda.detector_geometry import load_detector_geometry
from coseda.logging_utils import log_print
from coseda.nexus.paths import MASK_PATHS, get_first_existing_path
[docs]
class GeomfileMixin:
@staticmethod
def _format_geom_scalar(value: float) -> str:
text = f"{float(value):.8f}"
text = text.rstrip("0").rstrip(".")
return text if text else "0"
@staticmethod
def _format_geom_axis(vec_xy: np.ndarray) -> str:
vx = float(vec_xy[0])
vy = float(vec_xy[1])
terms = []
if abs(vx) > 1e-12:
terms.append(f"{vx:+.8f}x")
if abs(vy) > 1e-12:
terms.append(f"{vy:+.8f}y")
if not terms:
terms.append("+0.00000000x")
return " ".join(terms)
@staticmethod
def _extract_float(section, key, default=None):
try:
return float(section.get(key))
except Exception:
return default
@staticmethod
def _extract_optional_float(value):
if value is None:
return None
text = str(value).strip()
if not text:
return None
try:
return float(text)
except Exception:
return None
def _resolve_h5_path(self, parser: ConfigParser):
"""Resolve HDF5 path from INI Paths entries with robust fallbacks."""
try:
ini_path = os.path.abspath(getattr(self, "ini_path", "") or "")
ini_dir = os.path.dirname(ini_path) if ini_path else ""
candidates = []
seen = set()
def _add(path):
if not path:
return
path = os.path.expanduser(os.path.expandvars(path))
ap = os.path.abspath(path)
if ap in seen:
return
seen.add(ap)
candidates.append(ap)
if parser.has_section("Paths"):
h5_rel = parser.get("Paths", "h5file", fallback="").strip()
out_rel = parser.get("Paths", "outputfolder", fallback="").strip()
if h5_rel:
if os.path.isabs(h5_rel):
_add(h5_rel)
elif ini_dir:
_add(os.path.join(ini_dir, h5_rel))
if out_rel:
_add(os.path.join(ini_dir, out_rel, h5_rel))
if ini_path:
stem = os.path.splitext(ini_path)[0]
_add(stem + ".h5")
_add(stem + ".hdf5")
_add(stem + ".cxi")
for path in candidates:
if os.path.exists(path):
return path
return None
except Exception:
return None
def _get_frame_shape_from_h5(self, parser: ConfigParser):
"""Best-effort read of (size_x, size_y) from HDF5 image dataset."""
try:
h5_path = self._resolve_h5_path(parser)
if not h5_path:
return None
framepath = parser.get("Paths", "framepath", fallback="/entry/data/images").strip() or "/entry/data/images"
import h5py
with h5py.File(h5_path, "r") as h5f:
ds = h5f.get(framepath)
if ds is None:
ds = h5f.get(framepath.lstrip("/"))
if ds is None or getattr(ds, "ndim", 0) < 3:
return None
return int(ds.shape[1]), int(ds.shape[2])
except Exception as exc:
log_print(f"Unable to read frame shape from HDF5 for geometry checks: {exc}")
return None
def _get_mask_path_from_h5(self, parser: ConfigParser):
"""Return canonical existing mask dataset path or empty string when absent."""
try:
h5_path = self._resolve_h5_path(parser)
if not h5_path:
return ""
import h5py
with h5py.File(h5_path, "r") as h5f:
path = get_first_existing_path(h5f, MASK_PATHS)
return path or ""
except Exception as exc:
log_print(f"Unable to detect mask path from HDF5 for geometry export: {exc}")
return ""
@staticmethod
def _calc_electron_wavelength_angstrom(acceleration_voltage_v: float):
# Relativistic electron wavelength in Angstrom
h = 6.62607015e-34
m = 9.10938356e-31
e = 1.602176634e-19
c = 299792458
v = float(acceleration_voltage_v)
if v <= 0:
return None
lam_m = (h / np.sqrt(2.0 * m * e * v)) / np.sqrt(1.0 + (e * v) / (2.0 * m * c * c))
return lam_m * 1e10
def _collect_geom_defaults(self, parser: ConfigParser) -> dict:
acq = parser["AcquisitionDetails"] if parser.has_section("AcquisitionDetails") else {}
voltage = self._extract_float(acq, "acceleration_voltage", None)
wavelength = self._calc_electron_wavelength_angstrom(voltage) if voltage is not None else None
camera_length = self._extract_float(acq, "camera_length", None)
camera_corr = self._extract_float(acq, "camera_length_correction", 1.0)
clen = camera_length * camera_corr if camera_length is not None else None
pixels_per_meter = self._extract_float(acq, "pixels_per_meter", None)
adu_per_photon = (acq.get("adu_per_photon", "") if hasattr(acq, "get") else "")
adu_per_photon = str(adu_per_photon).strip()
if not adu_per_photon:
adu_per_photon = "1.0"
resolution_width = self._extract_float(acq, "resolution_width", None)
resolution_height = self._extract_float(acq, "resolution_height", None)
binning_width = self._extract_float(acq, "binning_width", 1.0)
binning_height = self._extract_float(acq, "binning_height", 1.0)
max_fs = None
max_ss = None
if resolution_width is not None and binning_width and binning_width > 0:
max_fs = int(round(resolution_width / binning_width - 1))
if resolution_height is not None and binning_height and binning_height > 0:
max_ss = int(round(resolution_height / binning_height - 1))
frame_shape = self._get_frame_shape_from_h5(parser)
if frame_shape is not None:
frame_size_x, frame_size_y = frame_shape
ini_size_fs = None if max_fs is None else int(max_fs + 1)
ini_size_ss = None if max_ss is None else int(max_ss + 1)
if ini_size_fs is not None and ini_size_fs != frame_size_x:
log_print(
f"Geometry FS-size mismatch: INI-derived={ini_size_fs}, HDF5={frame_size_x}. "
"Using HDF5 frame size as canonical for geometry export."
)
if ini_size_ss is not None and ini_size_ss != frame_size_y:
log_print(
f"Geometry SS-size mismatch: INI-derived={ini_size_ss}, HDF5={frame_size_y}. "
"Using HDF5 frame size as canonical for geometry export."
)
if (
ini_size_fs is not None
and ini_size_ss is not None
and ini_size_fs == frame_size_y
and ini_size_ss == frame_size_x
and (ini_size_fs != frame_size_x or ini_size_ss != frame_size_y)
):
log_print(
"Geometry sizes look transposed between INI and HDF5 (FS<->SS). "
"Verify resolution/binning metadata and panel bounds."
)
max_fs = frame_size_x - 1
max_ss = frame_size_y - 1
corner_x = None if max_fs is None else -int((max_fs + 1) // 2)
corner_y = None if max_ss is None else -int((max_ss + 1) // 2)
resolved_h5 = self._resolve_h5_path(parser)
mask_path = self._get_mask_path_from_h5(parser)
if not mask_path:
raise ValueError(
"No mask dataset found in HDF5 ('/entry/instrument/detector/mask' or '/mask'). "
f"Resolved HDF5 path: {resolved_h5 or '<unresolved>'}. "
"Run Preflight Check and generate a full mask before exporting geometry."
)
return {
"wavelength": "" if wavelength is None else self._format_geom_scalar(wavelength),
"adu_per_photon": adu_per_photon,
"clen": "" if clen is None else self._format_geom_scalar(clen),
"res": "" if pixels_per_meter is None else self._format_geom_scalar(pixels_per_meter),
"data": "/entry/data/images",
"dim0": "%",
"dim1": "ss",
"dim2": "fs",
"peak_list": "/entry/data/",
"peak_list_type": "cxi",
"detector_shift_x": "/entry/data/det_shift_x_mm mm",
"detector_shift_y": "/entry/data/det_shift_y_mm mm",
"min_ss": "0",
"max_ss": "" if max_ss is None else str(max_ss),
"min_fs": "0",
"max_fs": "" if max_fs is None else str(max_fs),
"corner_x": "" if corner_x is None else str(corner_x),
"corner_y": "" if corner_y is None else str(corner_y),
"fs": "x",
"ss": "y",
"mask": mask_path,
"mask_good": "0x01",
"mask_bad": "0x00",
}
def _panel_anchor_raw(self, defaults: dict, detector_geometry: dict) -> np.ndarray:
"""
Reference origin for panel corner export.
CrystFEL panel corner coordinates are beam-centered, while detector panel
definitions in [DetectorGeometry] are raw-pixel anchored (0,0 at raw data
origin). We align both by subtracting the same frame-center reference
used for det_shift_x/det_shift_y generation.
"""
max_fs = self._extract_optional_float(defaults.get("max_fs"))
max_ss = self._extract_optional_float(defaults.get("max_ss"))
if max_fs is not None and max_ss is not None:
return np.array([(max_fs + 1.0) / 2.0, (max_ss + 1.0) / 2.0], dtype=np.float64)
panels = detector_geometry.get("panels", [])
if not panels:
return np.array([0.0, 0.0], dtype=np.float64)
min_x = min(float(panel["raw_min_x"]) for panel in panels)
max_x = max(float(panel["raw_max_x"]) for panel in panels)
min_y = min(float(panel["raw_min_y"]) for panel in panels)
max_y = max(float(panel["raw_max_y"]) for panel in panels)
return np.array([(min_x + max_x + 1.0) / 2.0, (min_y + max_y + 1.0) / 2.0], dtype=np.float64)
def _build_geom_text(self, parser: ConfigParser, use_detector_geometry=None) -> str:
defaults = self._collect_geom_defaults(parser)
detector_geometry = load_detector_geometry(parser)
if use_detector_geometry is None:
use_panel_geometry = detector_geometry is not None and len(detector_geometry.get("panels", [])) > 0
elif use_detector_geometry:
use_panel_geometry = detector_geometry is not None and len(detector_geometry.get("panels", [])) > 0
else:
use_panel_geometry = False
lines = [";Detector file generated by COSEDA."]
lines.append(f"wavelength = {defaults['wavelength']} A")
lines.append(f"adu_per_photon = {defaults['adu_per_photon']}")
lines.append(f"clen = {defaults['clen']} m")
lines.append(f"res = {defaults['res']}")
lines.append(f"data = {defaults['data']}")
lines.append(f"dim0 = {defaults['dim0']}")
lines.append(f"dim1 = {defaults['dim1']}")
lines.append(f"dim2 = {defaults['dim2']}")
lines.append(f"peak_list = {defaults['peak_list']}")
lines.append(f"peak_list_type = {defaults['peak_list_type']}")
lines.append(f"detector_shift_x = {defaults['detector_shift_x']}")
lines.append(f"detector_shift_y = {defaults['detector_shift_y']}")
if use_panel_geometry:
lines.append("")
lines.append("; Panel layout from [DetectorGeometry]")
global_A = detector_geometry["global_A"]
global_offset = detector_geometry["global_offset"]
raw_anchor = self._panel_anchor_raw(defaults, detector_geometry)
for panel in detector_geometry["panels"]:
panel_id = panel["id"]
min_fs = int(round(panel["raw_min_x"]))
max_fs = int(round(panel["raw_max_x"]))
min_ss = int(round(panel["raw_min_y"]))
max_ss = int(round(panel["raw_max_y"]))
raw_min = np.array([panel["raw_min_x"], panel["raw_min_y"]], dtype=np.float64)
local_origin = raw_min + panel["offset"] - raw_anchor
corner_xy = global_offset + (global_A @ local_origin)
fs_vec_xy = global_A @ (panel["A"] @ np.array([1.0, 0.0], dtype=np.float64))
ss_vec_xy = global_A @ (panel["A"] @ np.array([0.0, 1.0], dtype=np.float64))
lines.append(f"{panel_id}/min_ss = {min_ss}")
lines.append(f"{panel_id}/max_ss = {max_ss}")
lines.append(f"{panel_id}/min_fs = {min_fs}")
lines.append(f"{panel_id}/max_fs = {max_fs}")
lines.append(f"{panel_id}/corner_x = {self._format_geom_scalar(corner_xy[0])}")
lines.append(f"{panel_id}/corner_y = {self._format_geom_scalar(corner_xy[1])}")
lines.append(f"{panel_id}/fs = {self._format_geom_axis(fs_vec_xy)}")
lines.append(f"{panel_id}/ss = {self._format_geom_axis(ss_vec_xy)}")
lines.append(f"{panel_id}/mask = {defaults['mask']}")
lines.append(f"{panel_id}/mask_good = {defaults['mask_good']}")
lines.append(f"{panel_id}/mask_bad = {defaults['mask_bad']}")
lines.append("")
lines.append("; End panel layout")
else:
lines.append(f"p0/min_ss = {defaults['min_ss']}")
lines.append(f"p0/max_ss = {defaults['max_ss']}")
lines.append(f"p0/min_fs = {defaults['min_fs']}")
lines.append(f"p0/max_fs = {defaults['max_fs']}")
lines.append(f"p0/corner_x = {defaults['corner_x']}")
lines.append(f"p0/corner_y = {defaults['corner_y']}")
lines.append(f"p0/fs = {defaults['fs']}")
lines.append(f"p0/ss = {defaults['ss']}")
lines.append(f"p0/mask = {defaults['mask']}")
lines.append(f"p0/mask_good = {defaults['mask_good']}")
lines.append(f"p0/mask_bad = {defaults['mask_bad']}")
return "\n".join(lines).rstrip() + "\n"
def _set_geom_text_from_generator(self, use_detector_geometry=None):
parser = ConfigParser()
parser.read(self.ini_path)
try:
text = self._build_geom_text(parser, use_detector_geometry=use_detector_geometry)
except ValueError as exc:
QMessageBox.critical(self, "Mask Required", str(exc))
return
self.geom_text_edit.blockSignals(True)
self.geom_text_edit.setPlainText(text)
self.geom_text_edit.blockSignals(False)
def _extract_geom_value(self, text: str, key: str):
pattern = re.compile(rf"^\s*{re.escape(key)}\s*=\s*(.*?)\s*$", re.MULTILINE)
match = pattern.search(text)
if not match:
return ""
return match.group(1).strip()
def _extract_adu_per_photon(self, text: str):
return self._extract_geom_value(text, "adu_per_photon")
def _validate_geom_text(self, text: str):
req = ["wavelength", "clen", "res"]
missing_or_invalid = []
number_re = re.compile(r"^[+-]?(?:\d+(?:\.\d*)?|\.\d+)(?:[eE][+-]?\d+)?")
for key in req:
value = self._extract_geom_value(text, key)
if not value:
missing_or_invalid.append(key)
continue
# Required keys must start with a numeric token; unit suffixes are allowed.
if number_re.match(value.strip()) is None:
missing_or_invalid.append(key)
if re.search(r"^\s*[^\s#;]+/mask\s*=", text, re.MULTILINE) is None:
missing_or_invalid.append("panel mask entry (e.g. p0/mask)")
return missing_or_invalid
def _init_geomfile_tab(self):
"""Build the geometry tab as a single editable text box."""
self.geom_tab = QWidget()
layout = QVBoxLayout()
hint = QLabel("Edit the full CrystFEL .geom text directly. Use generator buttons to refresh templates.")
hint.setWordWrap(True)
layout.addWidget(hint)
btn_row = QHBoxLayout()
self.geom_reload_btn = QPushButton("Load Existing")
self.geom_reload_btn.clicked.connect(self._load_geom_file)
btn_row.addWidget(self.geom_reload_btn)
self.geom_legacy_btn = QPushButton("Generate Single-Chip Geometry")
self.geom_legacy_btn.clicked.connect(lambda: self._set_geom_text_from_generator(False))
btn_row.addWidget(self.geom_legacy_btn)
self.geom_detector_btn = QPushButton("Generate From DetectorGeometry")
self.geom_detector_btn.clicked.connect(lambda: self._set_geom_text_from_generator(True))
btn_row.addWidget(self.geom_detector_btn)
self.geom_save_btn = QPushButton("Save Geometry")
self.geom_save_btn.clicked.connect(self._save_geom_file)
btn_row.addWidget(self.geom_save_btn)
layout.addLayout(btn_row)
self.geom_text_edit = QPlainTextEdit()
self.geom_text_edit.setPlaceholderText("Geometry content will appear here...")
layout.addWidget(self.geom_text_edit)
self.geom_timer = QTimer(self)
self.geom_timer.setSingleShot(True)
self.geom_timer.timeout.connect(self._save_geom_file)
self.geom_text_edit.textChanged.connect(lambda: self.geom_timer.start(500))
self.geom_tab.setLayout(layout)
self.tabs.addTab(self.geom_tab, "Geometry Editor")
def _load_geom_file(self):
if self.geom_filepath and os.path.exists(self.geom_filepath):
try:
with open(self.geom_filepath, "r") as f:
text = f.read()
self.geom_text_edit.blockSignals(True)
self.geom_text_edit.setPlainText(text)
self.geom_text_edit.blockSignals(False)
except Exception as exc:
log_print(f"Failed to load geometry file '{self.geom_filepath}': {exc}")
self._set_geom_text_from_generator(None)
else:
self._set_geom_text_from_generator(None)
self._update_command_label()
def _save_geom_file(self, *, show_errors: bool = False):
# Do not save until there is some geometry text present
text = self.geom_text_edit.toPlainText().strip()
if not text:
return False
if not self.geom_filepath:
# Try to select the latest run automatically
if not self._ensure_run_selected_or_latest():
return False
try:
if not self._write_geom_file_to_path(self.geom_filepath, show_errors=show_errors, label="geometry"):
return False
parser = ConfigParser()
parser.read(self.ini_path)
base_dir = os.path.dirname(self.ini_path)
rel_path = os.path.relpath(self.geom_filepath, base_dir)
if not parser.has_section("Paths"):
parser.add_section("Paths")
parser.set("Paths", "geomfile", rel_path)
adu = self._extract_adu_per_photon(self.geom_text_edit.toPlainText())
if not parser.has_section("AcquisitionDetails"):
parser.add_section("AcquisitionDetails")
parser.set("AcquisitionDetails", "adu_per_photon", adu)
with open(self.ini_path, "w") as ini_file:
parser.write(ini_file)
if hasattr(self, "command_edit"):
self._update_command_label()
return True
except Exception as e:
QMessageBox.critical(self, "Error", f"Failed to save geometry file: {e}")
return False
def _write_geom_file_to_path(self, path: str, *, show_errors: bool, label: str) -> bool:
text = self.geom_text_edit.toPlainText().strip()
if not text:
parser = ConfigParser()
parser.read(self.ini_path)
try:
text = self._build_geom_text(parser, use_detector_geometry=None).strip()
except ValueError as e:
if show_errors:
QMessageBox.critical(self, "Mask Required", str(e))
return False
self.geom_text_edit.blockSignals(True)
self.geom_text_edit.setPlainText(text + "\n")
self.geom_text_edit.blockSignals(False)
missing = self._validate_geom_text(text)
if missing:
if show_errors:
QMessageBox.critical(
self,
"Error",
"Fill required geometry keys before writing: " + ", ".join(missing),
)
return False
try:
with open(path, "w") as f:
f.write(text.rstrip() + "\n")
return True
except Exception as e:
if show_errors:
QMessageBox.critical(self, "Error", f"Failed to write {label} file:\n{e}")
return False
# --- Start Indexing Tab (Placeholder) ---