"""Helpers for writing NeXus NXprocess metadata."""
from __future__ import annotations
import datetime
from typing import Any, Dict, Iterable, Optional
import h5py
import numpy as np
from coseda.nexus.groups import ensure_nexus_parents
from coseda.nexus.paths import BEAM_DIRECTION_PATH, BEAM_GROUP_PATH
def _string_dtype() -> h5py.Datatype:
return h5py.string_dtype(encoding="utf-8")
def _safe_dataset(group: h5py.Group, name: str, value: Any) -> None:
if name in group:
del group[name]
if isinstance(value, str):
group.create_dataset(name, data=value, dtype=_string_dtype())
elif isinstance(value, (int, float, np.integer, np.floating, np.bool_)):
group.create_dataset(name, data=value)
else:
group.create_dataset(name, data=str(value), dtype=_string_dtype())
def _unique_group(parent: h5py.Group, base: str) -> h5py.Group:
if base not in parent:
return parent.create_group(base)
suffix = 1
while f"{base}_{suffix}" in parent:
suffix += 1
return parent.create_group(f"{base}_{suffix}")
[docs]
def write_nxprocess_peakfinding(
h5file: h5py.File,
parameters: Dict[str, Any],
program: str,
inputs: Optional[Iterable[str]] = None,
outputs: Optional[Iterable[str]] = None,
) -> None:
"""Write an NXprocess record for peakfinding (non-destructive to data)."""
process_root = h5file.require_group("entry/process")
timestamp = datetime.datetime.now().strftime("%Y%m%dT%H%M%S%f")
group = _unique_group(process_root, f"peakfinding_{timestamp}")
group.attrs["NX_class"] = "NXprocess"
_safe_dataset(group, "program", program)
_safe_dataset(group, "date", datetime.datetime.now().isoformat(timespec="seconds"))
if inputs:
_safe_dataset(group, "input", ", ".join(inputs))
if outputs:
_safe_dataset(group, "output", ", ".join(outputs))
params_group = group.require_group("parameters")
params_group.attrs["NX_class"] = "NXparameters"
for key, value in parameters.items():
_safe_dataset(params_group, key, value)
[docs]
def write_nxprocess_stripping(
h5file: h5py.File,
threshold: Optional[float],
force: Optional[bool],
input_path: Optional[str] = None,
output_path: Optional[str] = None,
kept_frames: Optional[int] = None,
stripped_frames: Optional[int] = None,
) -> None:
"""Write an NXprocess record for stripping."""
process_root = h5file.require_group("entry/process")
timestamp = datetime.datetime.now().strftime("%Y%m%dT%H%M%S%f")
group = _unique_group(process_root, f"strip_{timestamp}")
group.attrs["NX_class"] = "NXprocess"
_safe_dataset(group, "program", "coseda.pipeline.strip_h5")
_safe_dataset(group, "date", datetime.datetime.now().isoformat(timespec="seconds"))
if input_path:
_safe_dataset(group, "input", input_path)
if output_path:
_safe_dataset(group, "output", output_path)
params_group = group.require_group("parameters")
params_group.attrs["NX_class"] = "NXparameters"
if threshold is not None:
_safe_dataset(params_group, "threshold", threshold)
if force is not None:
_safe_dataset(params_group, "force", force)
if kept_frames is not None:
_safe_dataset(params_group, "kept_frames", kept_frames)
if stripped_frames is not None:
_safe_dataset(params_group, "stripped_frames", stripped_frames)
[docs]
def write_nxprocess_indexing(
h5file: h5py.File,
program: str,
method: Optional[str] = None,
parameters: Optional[Dict[str, Any]] = None,
input_path: Optional[str] = None,
output_path: Optional[str] = None,
) -> None:
"""Write an NXprocess record for indexing/integration runs."""
process_root = h5file.require_group("entry/process")
timestamp = datetime.datetime.now().strftime("%Y%m%dT%H%M%S%f")
group = _unique_group(process_root, f"indexing_{timestamp}")
group.attrs["NX_class"] = "NXprocess"
_safe_dataset(group, "program", program)
_safe_dataset(group, "date", datetime.datetime.now().isoformat(timespec="seconds"))
if method:
_safe_dataset(group, "method", method)
if input_path:
_safe_dataset(group, "input", input_path)
if output_path:
_safe_dataset(group, "output", output_path)
params_group = group.require_group("parameters")
params_group.attrs["NX_class"] = "NXparameters"
if parameters:
for key, value in parameters.items():
_safe_dataset(params_group, key, value)
[docs]
def write_nxprocess_merging(
h5file: h5py.File,
program: str,
method: Optional[str] = None,
parameters: Optional[Dict[str, Any]] = None,
input_path: Optional[str] = None,
output_path: Optional[str] = None,
) -> None:
"""Write an NXprocess record for merging runs."""
process_root = h5file.require_group("entry/process")
timestamp = datetime.datetime.now().strftime("%Y%m%dT%H%M%S%f")
group = _unique_group(process_root, f"merging_{timestamp}")
group.attrs["NX_class"] = "NXprocess"
_safe_dataset(group, "program", program)
_safe_dataset(group, "date", datetime.datetime.now().isoformat(timespec="seconds"))
if method:
_safe_dataset(group, "method", method)
if input_path:
_safe_dataset(group, "input", input_path)
if output_path:
_safe_dataset(group, "output", output_path)
params_group = group.require_group("parameters")
params_group.attrs["NX_class"] = "NXparameters"
if parameters:
for key, value in parameters.items():
_safe_dataset(params_group, key, value)
[docs]
def write_beam_incident_energy(h5file: h5py.File, energy_ev: float) -> None:
"""Ensure NXbeam exists and write incident_energy in eV."""
if energy_ev is None or energy_ev <= 0:
return
ensure_nexus_parents(h5file)
beam_group = h5file.require_group(BEAM_GROUP_PATH.lstrip("/"))
beam_group.attrs["NX_class"] = "NXbeam"
if "incident_energy" in beam_group:
del beam_group["incident_energy"]
ds = beam_group.create_dataset("incident_energy", data=energy_ev)
ds.attrs["units"] = "eV"
ds.attrs["long_name"] = "Incident beam energy"
if BEAM_DIRECTION_PATH.rsplit("/", 1)[-1] in beam_group:
del beam_group[BEAM_DIRECTION_PATH.rsplit("/", 1)[-1]]
direction = np.array([0.0, 0.0, 1.0], dtype=np.float32)
ds = beam_group.create_dataset("beam_direction", data=direction)
ds.attrs["units"] = "1"
ds.attrs["long_name"] = "Beam direction in detector coordinates (+Z along beam)"
[docs]
def write_nxprocess_import(
h5file: h5py.File,
program: str,
parameters: Optional[Dict[str, Any]] = None,
input_path: Optional[str] = None,
output_path: Optional[str] = None,
) -> None:
"""Write an NXprocess record for import/conversion steps."""
process_root = h5file.require_group("entry/process")
timestamp = datetime.datetime.now().strftime("%Y%m%dT%H%M%S%f")
group = _unique_group(process_root, f"import_{timestamp}")
group.attrs["NX_class"] = "NXprocess"
_safe_dataset(group, "program", program)
_safe_dataset(group, "date", datetime.datetime.now().isoformat(timespec="seconds"))
if input_path:
_safe_dataset(group, "input", input_path)
if output_path:
_safe_dataset(group, "output", output_path)
params_group = group.require_group("parameters")
params_group.attrs["NX_class"] = "NXparameters"
if parameters:
for key, value in parameters.items():
_safe_dataset(params_group, key, value)
[docs]
def write_nxprocess_centerfinding(
h5file: h5py.File,
program: str,
method: Optional[str] = None,
parameters: Optional[Dict[str, Any]] = None,
input_path: Optional[str] = None,
output_path: Optional[str] = None,
) -> None:
"""Write an NXprocess record for beam center finding."""
process_root = h5file.require_group("entry/process")
timestamp = datetime.datetime.now().strftime("%Y%m%dT%H%M%S%f")
group = _unique_group(process_root, f"centerfinding_{timestamp}")
group.attrs["NX_class"] = "NXprocess"
_safe_dataset(group, "program", program)
_safe_dataset(group, "date", datetime.datetime.now().isoformat(timespec="seconds"))
if method:
_safe_dataset(group, "method", method)
if input_path:
_safe_dataset(group, "input", input_path)
if output_path:
_safe_dataset(group, "output", output_path)
params_group = group.require_group("parameters")
params_group.attrs["NX_class"] = "NXparameters"
if parameters:
for key, value in parameters.items():
_safe_dataset(params_group, key, value)
[docs]
def write_nxprocess_centerrefinement(
h5file: h5py.File,
program: str,
method: Optional[str] = None,
parameters: Optional[Dict[str, Any]] = None,
input_path: Optional[str] = None,
output_path: Optional[str] = None,
) -> None:
"""Write an NXprocess record for beam center refinement."""
process_root = h5file.require_group("entry/process")
timestamp = datetime.datetime.now().strftime("%Y%m%dT%H%M%S%f")
group = _unique_group(process_root, f"centerrefinement_{timestamp}")
group.attrs["NX_class"] = "NXprocess"
_safe_dataset(group, "program", program)
_safe_dataset(group, "date", datetime.datetime.now().isoformat(timespec="seconds"))
if method:
_safe_dataset(group, "method", method)
if input_path:
_safe_dataset(group, "input", input_path)
if output_path:
_safe_dataset(group, "output", output_path)
params_group = group.require_group("parameters")
params_group.attrs["NX_class"] = "NXparameters"
if parameters:
for key, value in parameters.items():
_safe_dataset(params_group, key, value)