Source code for csaxs_bec.file_writer.csaxs_nexus
from __future__ import annotations
import numpy as np
from bec_server.file_writer.default_writer import DefaultFormat
[docs]
class cSAXSNeXusFormat(DefaultFormat):
"""
NeXus file format for the cSAXS beamline (BEC era).
Mirrors the old SPEC layout.xml hierarchy and adds the flOMNI instrument
group for the nano-positioning stage used in ptychography.
Device resilience
-----------------
Every device read (self.get_entry / device call) is wrapped in try/except.
If a device is removed from the BEC config file between sessions it simply
disappears from the device_manager — the corresponding dataset or link is
silently omitted from the HDF5 file without raising an error. This means
the file structure is additive: re-add the device to the config and the
field reappears automatically on the next scan.
Top-level HDF5 structure
────────────────────────
/entry NXentry (definition = NXptycho)
/sample NXsample ← primary sample group
/entry_ptycho NXentry ← generic ptycho entry
/data_soft NXentry ← convenience Eiger frame links
/control NXmonitor
/instrument NXinstrument
/source
/insertion_device
/monochromator
/XBPM3
/slit_3 … slit_5
/filter_set
/beam_stop_1 … beam_stop_2
/eiger_1_5 NXdetector
/mcs NXdetector
/flOMNI NXpositioner
Device name mapping (old SPEC → current BEC)
────────────────────────────────────────────
samx / samy → samx / samy (generic; kept for non-flOMNI configs)
sl3wh/wv/ch/cv → sl3trxi/o/b/t (individual blade motors; gap/centre TODO)
sl4wh/wv/ch/cv → sl4trxi/o/b/t
sl5wh/wv/ch/cv → sl5trxi/o/b/t
bs1x / bs1y → bs1x / bs1y
bs2x / bs2y → bs2x / bs2y
dettrx → dettrx
eiger_4 → eiger_1_5
mcs → mcs
filter_array → filter_array_1_x … filter_array_4_x
xbpm3 → xbpm3x / xbpm3y (stage positions; signal readouts TODO)
energy → ccm_energy
TODO (devices not yet in BEC list)
───────────────────────────────────
curr, idgap ring current, undulator gap
moth1, mobd monochromator crystal angles
mith, mibd, mirror_coating mirror
bpm3s/x/y/z XBPM3 signal readouts
sl0 / sl1 / sl2 upstream optics-hutch slits
slit gap / centre derived from blade pairs + calibration offset
"""
# -------------------------------------------------------------------------
# Helpers
# -------------------------------------------------------------------------
def _safe_dataset(self, group, name: str, device: str,
units: str | None = None,
description: str | None = None) -> None:
"""
Write a dataset from the BEC scan data dictionary.
Silently skips if the device was not recorded in this scan
(e.g. removed from config, readoutPriority=on_request and not triggered,
or the scan finished before the device responded).
"""
try:
value = self.get_entry(device)
ds = group.create_dataset(name, data=value)
if units:
ds.attrs["units"] = units
if description:
ds.attrs["description"] = description
except Exception:
pass
def _safe_soft_link(self, group, name: str, target: str) -> None:
"""Create a soft link; silently skip on any error."""
try:
group.create_soft_link(name, target)
except Exception:
pass
def _slit_blades(self, group, prefix: str) -> None:
"""
Store individual blade motor positions for a 4-blade slit set.
Derived quantities (gap, centre) require a per-slit calibration offset
and will be added in a later update.
"""
for blade, motor in [
("inner_x", f"{prefix}trxi"),
("outer_x", f"{prefix}trxo"),
("bottom_y", f"{prefix}trxb"),
("top_y", f"{prefix}trxt"),
]:
self._safe_dataset(group, blade, motor, units="mm")
# -------------------------------------------------------------------------
# Main format method
# -------------------------------------------------------------------------
[docs]
def format(self) -> None:
"""Build the NeXus/HDF5 layout for a cSAXS scan."""
# Canonical paths referenced by multiple groups
RT_POS_PATH = "/entry/instrument/flOMNI/rt_positions"
EIGER_COLL = "/entry/collection/file_references/eiger_1_5"
# ── Root entry ────────────────────────────────────────────────────────
entry = self.storage.create_group("entry")
entry.attrs["NX_class"] = "NXentry"
entry.attrs["definition"] = "NXptycho"
# ── /entry/sample ─────────────────────────────────────────────────────
# Primary sample group. Contains the name of the mounted sample and a
# link to the real-time scan positions. Generic samx/samy are recorded
# here so the group is meaningful for non-flOMNI configurations too.
sample = entry.create_group("sample")
sample.attrs["NX_class"] = "NXsample"
# Soft-link name directly to the value BEC recorded in the collection.
# Only written when flomni_samples is present; other configs leave name absent.
if "flomni_samples" in self.device_manager.devices:
self._safe_soft_link(
sample, "name",
"/entry/collection/devices/flomni_samples"
"/flomni_samples_sample_names_sample0/value",
)
# Generic coarse stage positions (meaningful in non-flOMNI setups)
self._safe_dataset(sample, "x_translation", "samx", units="mm")
self._safe_dataset(sample, "y_translation", "samy", units="mm")
# Real-time encoder positions — the primary scan coordinate
self._safe_soft_link(sample, "positions", RT_POS_PATH)
# ── /entry/entry_ptycho ───────────────────────────────────────────────
# Generic ptychography entry. Detector data and scan positions are
# linked in from the instrument groups so this entry is self-contained
# for downstream reconstruction codes.
entry_ptycho = entry.create_group("entry_ptycho")
entry_ptycho.attrs["NX_class"] = "NXentry"
entry_ptycho.attrs["definition"] = "NXptycho"
nxdata = entry_ptycho.create_group("data")
nxdata.attrs["NX_class"] = "NXdata"
nxdata.attrs["signal"] = "data"
# Detector frames
try:
for k in self.file_references["eiger_1_5"].hinted_h5_entries.keys():
self._safe_soft_link(nxdata, k, f"{EIGER_COLL}/{k}")
except Exception:
pass
# Scan positions
self._safe_soft_link(nxdata, "positions", RT_POS_PATH)
# Link to the primary sample group
self._safe_soft_link(entry_ptycho, "sample", "/entry/sample")
# ── /entry/data_soft ──────────────────────────────────────────────────
# Convenience group mirroring the old /entry/data hardlink from layout.xml.
data_soft = entry.create_group("data_soft")
data_soft.attrs["NX_class"] = "NXentry"
try:
for k in self.file_references["eiger_1_5"].hinted_h5_entries.keys():
self._safe_soft_link(data_soft, k, f"{EIGER_COLL}/{k}")
except Exception:
pass
# ── /entry/control ────────────────────────────────────────────────────
control = entry.create_group("control")
control.attrs["NX_class"] = "NXmonitor"
control.create_dataset("mode", data="monitor")
# TODO: beam intensity integral — add device when available
# self._safe_dataset(control, "integral", "bpm_sum", units="NX_DIMENSIONLESS")
# ── /entry/instrument ─────────────────────────────────────────────────
instrument = entry.create_group("instrument")
instrument.attrs["NX_class"] = "NXinstrument"
instrument.create_dataset("name", data="cSAXS beamline")
# ── Source ────────────────────────────────────────────────────────────
# Numerical values are currently unknown and stored as 0.
# Will be updated once the corresponding devices are in BEC.
source = instrument.create_group("source")
source.attrs["NX_class"] = "NXsource"
source.create_dataset("type", data="Synchrotron X-ray Source")
source.create_dataset("name", data="Swiss Light Source")
source.create_dataset("probe", data="x-ray")
source.create_dataset("sigma_x", data=0.0).attrs["units"] = "mm"
source.create_dataset("sigma_y", data=0.0).attrs["units"] = "mm"
source.create_dataset("divergence_x", data=0.0).attrs["units"] = "radians"
source.create_dataset("divergence_y", data=0.0).attrs["units"] = "radians"
# TODO: current — add device when available
# self._safe_dataset(source, "current", "curr", units="mA")
# ── Insertion device ──────────────────────────────────────────────────
insertion_device = instrument.create_group("insertion_device")
insertion_device.attrs["NX_class"] = "NXinsertion_device"
insertion_device.create_dataset("type", data="undulator")
insertion_device.create_dataset("k", data=0.0)
insertion_device.create_dataset("length", data=0.0).attrs["units"] = "mm"
# TODO: gap — add device when available
# self._safe_dataset(insertion_device, "gap", "idgap", units="mm")
# ── Monochromator ─────────────────────────────────────────────────────
# ccm_energy is a baseline device and is recorded in the scan data.
mono = instrument.create_group("monochromator")
mono.attrs["NX_class"] = "NXmonochromator"
mono.create_dataset("type", data="Double crystal fixed exit monochromator.")
try:
energy_kev = self.get_entry("ccm_energy")
energy_arr = np.asarray(energy_kev, dtype=float)
en_ds = mono.create_dataset("energy", data=energy_arr)
en_ds.attrs["units"] = "keV"
with np.errstate(divide="ignore", invalid="ignore"):
wavelength = np.where(energy_arr != 0, 12.3984193 / energy_arr, 0.0)
wl_ds = mono.create_dataset("wavelength", data=wavelength)
wl_ds.attrs["units"] = "Angstrom"
except Exception:
pass
# TODO: crystal angles — add moth1 / mobd when available
# crystal_1 = mono.create_group("crystal_1")
# crystal_1.attrs["NX_class"] = "NXcrystal"
# crystal_1.create_dataset("usage", data="Bragg")
# crystal_1.create_dataset("type", data="Si")
# crystal_1.create_dataset("order_no", data=1.0)
# crystal_1.create_dataset("reflection", data="[1 1 1]")
# self._safe_dataset(crystal_1, "bragg_angle", "moth1", units="degrees")
# crystal_2 = mono.create_group("crystal_2")
# crystal_2.attrs["NX_class"] = "NXcrystal"
# crystal_2.create_dataset("usage", data="Bragg")
# crystal_2.create_dataset("type", data="Si")
# crystal_2.create_dataset("order_no", data=2.0)
# crystal_2.create_dataset("reflection", data="[1 1 1]")
# self._safe_dataset(crystal_2, "bragg_angle", "moth1", units="degrees")
# self._safe_dataset(crystal_2, "bend_x", "mobd", units="degrees")
# ── Mirror ────────────────────────────────────────────────────────────
# TODO: mith, mibd, mirror_coating not yet in device list
# mirror = instrument.create_group("mirror")
# mirror.attrs["NX_class"] = "NXmirror"
# mirror.create_dataset("type", data="single")
# mirror.create_dataset(
# "description",
# data=(
# "Grazing incidence mirror to reject high-harmonic wavelengths. "
# "Three coating options: no coating (SiO2), rhodium (Rh), platinum (Pt)."
# ),
# )
# mirror.create_dataset("substrate_material", data="SiO2")
# self._safe_dataset(mirror, "incident_angle", "mith", units="degrees")
# self._safe_dataset(mirror, "coating_material", "mirror_coating", units="NX_CHAR")
# self._safe_dataset(mirror, "bend_y", "mibd", units="NX_DIMENSIONLESS")
# ── Upstream slits (optics hutch) ─────────────────────────────────────
# TODO: slit_0 / slit_1 / slit_2 motors not yet in BEC device list
# slit_0 = instrument.create_group("slit_0")
# ...
# slit_1 = instrument.create_group("slit_1")
# ...
# slit_2 = instrument.create_group("slit_2")
# ...
# ── XBPM3 ─────────────────────────────────────────────────────────────
# xbpm3x/xbpm3y are stage motor positions for aligning the monitor.
# Signal readouts (sum/x/y/skew) are TODO once MCS channels are mapped.
xbpm3 = instrument.create_group("XBPM3")
xbpm3.attrs["NX_class"] = "NXdetector"
xbpm3.attrs["description"] = "X-ray beam position monitor 3, experimental hutch"
self._safe_dataset(xbpm3, "x_stage", "xbpm3x", units="mm",
description="XBPM3 stage x-translation")
self._safe_dataset(xbpm3, "y_stage", "xbpm3y", units="mm",
description="XBPM3 stage y-translation")
# TODO: signal readout sub-groups once MCS channels are configured
# for suffix, entry_name, desc in [
# ("sum", "bpm3s", "Sum of counts for the four quadrants."),
# ("x", "bpm3x", "Normalized diff, left vs right quadrants."),
# ("y", "bpm3y", "Normalized diff, high vs low quadrants."),
# ("skew", "bpm3z", "Normalized diff, diagonal quadrants."),
# ]:
# g = xbpm3.create_group(f"XBPM3_{suffix}")
# self._safe_dataset(g, "data", entry_name, units="NX_DIMENSIONLESS")
# g.create_dataset("description", data=desc)
# ── Slit 3 (experimental hutch, exposure box) ─────────────────────────
slit_3 = instrument.create_group("slit_3")
slit_3.attrs["NX_class"] = "NXslit"
slit_3.create_dataset("material", data="Si")
slit_3.create_dataset("description", data="Slit 3, experimental hutch, exposure box")
# TODO: gap / centre require per-slit calibration offset — add later
self._slit_blades(slit_3, "sl3")
# ── Filter set ────────────────────────────────────────────────────────
filter_set = instrument.create_group("filter_set")
filter_set.attrs["NX_class"] = "NXattenuator"
filter_set.create_dataset("material", data="Si")
filter_set.create_dataset(
"description",
data=(
"Four linear filter stages (filter_array_1_x … filter_array_4_x). "
"Each stage has five filter positions plus an 'out' position."
),
)
for i in range(1, 5):
self._safe_dataset(filter_set, f"stage_{i}_x",
f"filter_array_{i}_x", units="mm")
# TODO: attenuator_transmission = 10^(ftrans) once device is available
# ── Slit 4 (experimental hutch, exposure box) ─────────────────────────
slit_4 = instrument.create_group("slit_4")
slit_4.attrs["NX_class"] = "NXslit"
slit_4.create_dataset("material", data="Ge")
slit_4.create_dataset("description", data="Slit 4, experimental hutch, exposure box")
self._slit_blades(slit_4, "sl4")
# ── Slit 5 (experimental hutch, exposure box) ─────────────────────────
slit_5 = instrument.create_group("slit_5")
slit_5.attrs["NX_class"] = "NXslit"
slit_5.create_dataset("material", data="Si")
slit_5.create_dataset("description", data="Slit 5, experimental hutch, exposure box")
self._slit_blades(slit_5, "sl5")
# ── Beam stop 1 ────────────────────────────────────────────────────────
beam_stop_1 = instrument.create_group("beam_stop_1")
beam_stop_1.attrs["NX_class"] = "NXbeam_stop"
beam_stop_1.create_dataset("description", data="circular")
beam_stop_1.create_dataset("size", data=3.0).attrs["units"] = "mm"
self._safe_dataset(beam_stop_1, "x", "bs1x", units="mm")
self._safe_dataset(beam_stop_1, "y", "bs1y", units="mm")
# TODO: diode signal behind beam stop 1 when device is available
# ── Beam stop 2 ────────────────────────────────────────────────────────
beam_stop_2 = instrument.create_group("beam_stop_2")
beam_stop_2.attrs["NX_class"] = "NXbeam_stop"
beam_stop_2.create_dataset("description", data="rectangular")
beam_stop_2.create_dataset("size_x", data=5.0).attrs["units"] = "mm"
beam_stop_2.create_dataset("size_y", data=2.25).attrs["units"] = "mm"
self._safe_dataset(beam_stop_2, "x", "bs2x", units="mm")
self._safe_dataset(beam_stop_2, "y", "bs2y", units="mm")
# TODO: diode (transmitted signal) when device is available
# ── Detector translation ───────────────────────────────────────────────
self._safe_dataset(
instrument, "detector_translation_x", "dettrx",
units="mm", description="Detector x-translation stage",
)
# ── Eiger 1.5M detector ───────────────────────────────────────────────
if (
"eiger_1_5" in self.device_manager.devices
and self.device_manager.devices.eiger_1_5.enabled
and "eiger_1_5" in self.file_references
):
eiger = instrument.create_group("eiger_1_5")
eiger.attrs["NX_class"] = "NXdetector"
eiger.create_dataset("x_pixel_size", data=75.0).attrs["units"] = "um"
eiger.create_dataset("y_pixel_size", data=75.0).attrs["units"] = "um"
eiger.create_dataset("polar_angle", data=0.0).attrs["units"] = "degrees"
eiger.create_dataset("azimuthal_angle", data=0.0).attrs["units"] = "degrees"
eiger.create_dataset("rotation_angle", data=0.0).attrs["units"] = "degrees"
eiger.create_dataset(
"description",
data="Eiger 1.5M detector, in-house developed, Paul Scherrer Institute",
)
eiger.create_dataset(
"type",
data="Single-photon counting detector, 320 micron-thick Si chip",
)
orientation = eiger.create_group("orientation")
orientation.attrs["description"] = (
"Orientation defines the number of counterclockwise rotations by 90 deg "
"followed by a transposition to reach the 'cameraman orientation', "
"looking towards the beam."
)
orientation.create_dataset("transpose", data=1)
orientation.create_dataset("rot90", data=3)
# Soft-link recorded frame data from the BEC collection
try:
for k in self.file_references["eiger_1_5"].hinted_h5_entries.keys():
self._safe_soft_link(eiger, k, f"{EIGER_COLL}/{k}")
except Exception:
pass
# External link to pixel mask in the Eiger master file
try:
eiger.create_ext_link(
"pixel_mask",
self.file_references["eiger_1_5"].file_path,
"/entry/instrument/detector/pixel_mask",
)
except Exception:
pass
# ── MCS (multi-channel scaler) ─────────────────────────────────────────
if (
"mcs" in self.device_manager.devices
and self.device_manager.devices.mcs.enabled
):
mcs_group = instrument.create_group("mcs")
mcs_group.attrs["NX_class"] = "NXdetector"
mcs_group.attrs["description"] = "MCS card cSAXS — multi-channel scaler"
self._safe_soft_link(mcs_group, "data", "/entry/collection/devices/mcs")
# ── flOMNI ────────────────────────────────────────────────────────────
# flomni_samples is used as the sentinel for the entire flOMNI setup.
# If it is absent from the device_manager (removed from config) the
# whole group is omitted. Individual datasets inside are still each
# guarded by _safe_dataset / _safe_soft_link in case a specific motor
# is temporarily disabled without removing the full setup.
if "flomni_samples" in self.device_manager.devices:
flomni = instrument.create_group("flOMNI")
flomni.attrs["NX_class"] = "NXpositioner"
flomni.attrs["description"] = "flOMNI flexible tOMography Nano Imaging"
# Galil motors — coarse sample stage
self._safe_dataset(flomni, "fsamx", "fsamx", units="mm", description="Sample coarse X")
self._safe_dataset(flomni, "fsamy", "fsamy", units="mm", description="Sample coarse Y")
self._safe_dataset(flomni, "fsamroy", "fsamroy", units="degrees", description="Sample rotation")
# Galil motors — sample transfer / tray
self._safe_dataset(flomni, "ftransx", "ftransx", units="mm", description="Sample transfer X")
self._safe_dataset(flomni, "ftransy", "ftransy", units="mm", description="Sample transfer Y")
self._safe_dataset(flomni, "ftransz", "ftransz", units="mm", description="Sample transfer Z")
self._safe_dataset(flomni, "ftray", "ftray", units="mm", description="Sample transfer tray")
# Galil motors — laser tracker
self._safe_dataset(flomni, "ftracky", "ftracky", units="mm", description="Laser tracker coarse Y")
self._safe_dataset(flomni, "ftrackz", "ftrackz", units="mm", description="Laser tracker coarse Z")
# Galil motors — X-ray eye
self._safe_dataset(flomni, "feyex", "feyex", units="mm", description="X-ray eye X")
self._safe_dataset(flomni, "feyey", "feyey", units="mm", description="X-ray eye Y")
# Galil motors — optics (zone plate)
self._safe_dataset(flomni, "foptx", "foptx", units="mm", description="Optics X")
self._safe_dataset(flomni, "fopty", "fopty", units="mm", description="Optics Y")
self._safe_dataset(flomni, "foptz", "foptz", units="mm", description="Optics Z")
# Galil motor — heater
self._safe_dataset(flomni, "fheater", "fheater", units="mm", description="Heater Y")
# Smaract motors — OSA (order-sorting aperture)
self._safe_dataset(flomni, "fosax", "fosax", units="mm", description="OSA X")
self._safe_dataset(flomni, "fosay", "fosay", units="mm", description="OSA Y")
self._safe_dataset(flomni, "fosaz", "fosaz", units="mm", description="OSA Z")
# Temperature and humidity sensor (soft link to BEC collection entry)
self._safe_soft_link(
flomni, "flomni_temphum",
"/entry/collection/devices/flomni_temphum",
)
# Real-time encoder positions (RtFlomniFlyer)
# Single soft link to the entire rt_positions folder in the BEC
# collection. This is the primary scan coordinate for ptychography.
self._safe_soft_link(
flomni, "rt_positions",
"/entry/collection/devices/rt_positions",
)