feat: add nexus format
This commit is contained in:
+13
-3
@@ -95,8 +95,18 @@ jobs:
|
||||
uv pip install --system -e ./bec/bec_ipython_client
|
||||
uv pip install --system -e ./bec/bec_server[dev]
|
||||
uv pip install --system -e ./bec_widgets[dev,pyside6]
|
||||
uv pip install --system -e ./xtreme_bec
|
||||
uv pip install --system -e ./xtreme_bec[dev]
|
||||
|
||||
- name: Run Pytest with Coverage
|
||||
- name: Run Tests with Coverage
|
||||
id: coverage
|
||||
run: pytest --random-order --cov=./xtreme_bec --cov-config=./xtreme_bec/pyproject.toml --cov-branch --cov-report=xml --no-cov-on-fail ./xtreme_bec/tests/ || test $? -eq 5
|
||||
shell: bash
|
||||
run: |
|
||||
set +e
|
||||
coverage run --branch --source=./xtreme_bec -m pytest --random-order ./xtreme_bec/tests/
|
||||
status=$?
|
||||
if [ "$status" -ne 0 ] && [ "$status" -ne 5 ]; then
|
||||
exit "$status"
|
||||
fi
|
||||
if [ "$status" -eq 0 ]; then
|
||||
coverage report
|
||||
fi
|
||||
|
||||
@@ -0,0 +1 @@
|
||||
from .xtreme_nexus import XtremeNeXusFormat
|
||||
@@ -0,0 +1,225 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from bec_server.file_writer.default_writer import DefaultFormat
|
||||
|
||||
|
||||
class XtremeNeXusFormat(DefaultFormat):
|
||||
"""NeXus layout for the X-Treme beamline device configuration."""
|
||||
|
||||
_MIRROR_GROUPS = {
|
||||
"cmu": ("collimating_mirror", "Collimating mirror", 16.3),
|
||||
"fmu": ("focusing_mirror", "Focusing mirror", 22.3),
|
||||
"rmu": ("refocusing_mirror", "Refocusing mirror", 29.8),
|
||||
}
|
||||
_MIRROR_AXES = ("y", "z", "rx", "ry", "rz")
|
||||
_SLIT_SIGNALS = ("sl1xi", "sl1xo", "sl1yb", "sl1yt", "sl1xc", "sl1xs", "sl1yc", "sl1ys")
|
||||
|
||||
def format(self) -> None:
|
||||
entry = self.storage.create_group("entry")
|
||||
self.write_entry_metadata(entry)
|
||||
|
||||
# The title is a combination of the sample name, the scan name and the scan number.
|
||||
# If any of these are missing, they will be omitted from the title.
|
||||
title_parts = []
|
||||
scan_number = self.info_storage["bec"].get("scan_number")
|
||||
if scan_number:
|
||||
title_parts.append(str(scan_number))
|
||||
scan_name = self.info_storage.get("scan_name") or self.info_storage["bec"].get("scan_name")
|
||||
if scan_name:
|
||||
title_parts.append(scan_name)
|
||||
sample_name = self.info_storage.get("sample_name")
|
||||
if sample_name:
|
||||
title_parts.append(sample_name)
|
||||
|
||||
if title_parts:
|
||||
title = "_".join(title_parts)
|
||||
entry.create_dataset(name="title", data=title)
|
||||
|
||||
##############################
|
||||
#### Sample ####
|
||||
##############################
|
||||
sample = entry.create_group("sample")
|
||||
sample.attrs["NX_class"] = "NXsample"
|
||||
sample.create_dataset(
|
||||
name="name",
|
||||
data=self.info_storage.get("sample_name") or self.get_entry("samplename") or "",
|
||||
)
|
||||
sample.create_dataset(
|
||||
name="description", data=self.info_storage.get("sample_description") or ""
|
||||
)
|
||||
|
||||
instrument = entry.create_group("instrument")
|
||||
instrument.attrs["NX_class"] = "NXinstrument"
|
||||
instrument.create_dataset(name="name", data="X-Treme beamline")
|
||||
|
||||
##############################
|
||||
#### Source ####
|
||||
##############################
|
||||
source = instrument.create_group("source")
|
||||
source.attrs["NX_class"] = "NXsource"
|
||||
source.create_dataset(name="type", data="Synchrotron X-ray Source")
|
||||
source.create_dataset(name="name", data="Swiss Light Source")
|
||||
source.create_dataset(name="probe", data="x-ray")
|
||||
self.safe_dataset(group=source, name="current", device="curr")
|
||||
self.safe_dataset(group=source, name="orbit_feedback_state", device="ofb")
|
||||
|
||||
insertion_device = instrument.create_group("insertion_device")
|
||||
insertion_device.attrs["NX_class"] = "NXinsertion_device"
|
||||
insertion_device.create_dataset(name="type", data="undulator")
|
||||
insertion_device.create_dataset(name="description", data="Apple II undulator")
|
||||
self.safe_dataset(group=insertion_device, name="polarization", device="uind_mode")
|
||||
self.safe_dataset(group=insertion_device, name="energy", device="uind_energy")
|
||||
self.safe_dataset(group=insertion_device, name="gap", device="idgap")
|
||||
self.safe_dataset(group=insertion_device, name="harmonic", device="uind_harmonic")
|
||||
|
||||
xbpm = instrument.create_group("xbpm_1")
|
||||
xbpm.attrs["NX_class"] = "NXpositioner"
|
||||
xbpm.create_dataset(
|
||||
name="description", data="X-ray BPM1 translation stage in the front-end"
|
||||
)
|
||||
self.safe_dataset(group=xbpm, name="x", device="xbpm1x")
|
||||
self.safe_dataset(group=xbpm, name="y", device="xbpm1y")
|
||||
|
||||
valve = instrument.create_group("valve")
|
||||
valve.attrs["NX_class"] = "NXpositioner"
|
||||
valve.create_dataset(name="description", data="Endstation vacuum valve")
|
||||
self.safe_dataset(group=valve, name="value", device="valve")
|
||||
|
||||
shutter = instrument.create_group("shutter")
|
||||
shutter.attrs["NX_class"] = "NXpositioner"
|
||||
shutter.create_dataset(name="description", data="Beamline shutter")
|
||||
self.safe_dataset(group=shutter, name="value", device="shutter")
|
||||
|
||||
##############################
|
||||
#### Front End ####
|
||||
##############################
|
||||
front_end_slit = instrument.create_group("front_end_slit")
|
||||
front_end_slit.attrs["NX_class"] = "NXslit"
|
||||
front_end_slit.create_dataset(name="description", data="Front-end slit 1")
|
||||
self.safe_dataset(group=front_end_slit, name="x_gap", device="sl1xs")
|
||||
self.safe_dataset(group=front_end_slit, name="y_gap", device="sl1ys")
|
||||
self.safe_dataset(group=front_end_slit, name="x_translation", device="sl1xc")
|
||||
self.safe_dataset(group=front_end_slit, name="y_translation", device="sl1yc")
|
||||
self.safe_dataset(group=front_end_slit, name="x_ring_blade", device="sl1xi")
|
||||
self.safe_dataset(group=front_end_slit, name="x_wall_blade", device="sl1xo")
|
||||
self.safe_dataset(group=front_end_slit, name="y_bottom_blade", device="sl1yb")
|
||||
self.safe_dataset(group=front_end_slit, name="y_top_blade", device="sl1yt")
|
||||
|
||||
##############################
|
||||
#### Optics ####
|
||||
##############################
|
||||
monochromator = instrument.create_group("monochromator")
|
||||
monochromator.attrs["NX_class"] = "NXmonochromator"
|
||||
self.safe_dataset(group=monochromator, name="energy", device="mono")
|
||||
self.safe_dataset(group=monochromator, name="beamline_energy", device="energy")
|
||||
self.safe_dataset(group=monochromator, name="mode", device="energy_mode")
|
||||
self.safe_dataset(group=monochromator, name="cff", device="mono", signal="cff")
|
||||
|
||||
plane_mirror = monochromator.create_group("plane_mirror")
|
||||
plane_mirror.attrs["NX_class"] = "NXmirror"
|
||||
plane_mirror.create_dataset(name="description", data="Plane mirror")
|
||||
plane_mirror.create_dataset(name="distance", data=21.3)
|
||||
plane_mirror._storage["distance"].attrs["units"] = "m" # pylint: disable=protected-access
|
||||
self.safe_dataset(group=plane_mirror, name="theta", device="mirror_theta")
|
||||
|
||||
grating = monochromator.create_group("grating")
|
||||
grating.attrs["NX_class"] = "NXgrating"
|
||||
grating.create_dataset(name="description", data="Plane grating")
|
||||
grating.create_dataset(name="distance", data=21.3)
|
||||
grating._storage["distance"].attrs["units"] = "m" # pylint: disable=protected-access
|
||||
self.safe_dataset(group=grating, name="beta", device="grating_beta")
|
||||
|
||||
for prefix, (group_name, description, distance_m) in self._MIRROR_GROUPS.items():
|
||||
mirror = instrument.create_group(group_name)
|
||||
mirror.attrs["NX_class"] = "NXmirror"
|
||||
mirror.create_dataset(name="description", data=description)
|
||||
mirror.create_dataset(name="distance", data=distance_m)
|
||||
mirror._storage["distance"].attrs["units"] = "m" # pylint: disable=protected-access
|
||||
for axis in self._MIRROR_AXES:
|
||||
device_name = f"{prefix}_{axis}"
|
||||
self.safe_dataset(group=mirror, name=axis, device=device_name)
|
||||
|
||||
exit_slit = instrument.create_group("exit_slit")
|
||||
exit_slit.attrs["NX_class"] = "NXslit"
|
||||
exit_slit.create_dataset(
|
||||
name="description", data="Aperture between focusing and refocusing mirror"
|
||||
)
|
||||
self.safe_dataset(group=exit_slit, name="y_gap", device="exit_slit")
|
||||
|
||||
##############################
|
||||
#### Endstation ####
|
||||
##############################
|
||||
filter_group = instrument.create_group("filter")
|
||||
filter_group.attrs["NX_class"] = "NXattenuator"
|
||||
filter_group.create_dataset(
|
||||
name="description", data="3 um-thick Al filter before the exit slit"
|
||||
)
|
||||
filter_group.create_dataset(name="type", data="Al")
|
||||
thickness = filter_group.create_dataset(name="thickness", data=3.0)
|
||||
thickness.attrs["units"] = "um"
|
||||
self.safe_dataset(group=filter_group, name="position", device="filter")
|
||||
|
||||
gold_mesh_1 = instrument.create_group("gold_mesh_1")
|
||||
gold_mesh_1.attrs["NX_class"] = "NXpositioner"
|
||||
gold_mesh_1.create_dataset(name="description", data="Gold mesh 1 insertion stage")
|
||||
self.safe_dataset(group=gold_mesh_1, name="value", device="goldmesh1")
|
||||
|
||||
gold_mesh_2 = instrument.create_group("gold_mesh_2")
|
||||
gold_mesh_2.attrs["NX_class"] = "NXpositioner"
|
||||
gold_mesh_2.create_dataset(name="description", data="Gold mesh 2 insertion stage")
|
||||
self.safe_dataset(group=gold_mesh_2, name="value", device="goldmesh2")
|
||||
|
||||
end_station = instrument.create_group("end_station")
|
||||
end_station.attrs["NX_class"] = "NXenvironment"
|
||||
self.safe_dataset(group=end_station, name="temperature", device="temperature")
|
||||
self.safe_dataset(group=end_station, name="magnetic_field_x", device="field_x")
|
||||
self.safe_dataset(group=end_station, name="magnetic_field_z", device="field_z")
|
||||
|
||||
if "signals" in self.device_manager.devices:
|
||||
adc_signals = instrument.create_group("adc_signals")
|
||||
adc_signals.attrs["NX_class"] = "NXdetector"
|
||||
self._write_signals_with_softlink(adc_signals)
|
||||
|
||||
keithley = instrument.create_group("keithley")
|
||||
for keithley_index in range(1, 4):
|
||||
tmp = keithley.create_group(f"keithley_{keithley_index}")
|
||||
tmp.attrs["NX_class"] = "NXdetector"
|
||||
self.safe_dataset(group=tmp, device=f"keithley_{keithley_index}", name="gain")
|
||||
|
||||
##############################
|
||||
#### NXdata ####
|
||||
##############################
|
||||
data = entry.create_group("data")
|
||||
data.attrs["NX_class"] = "NXdata"
|
||||
self._write_signals_with_softlink(data)
|
||||
|
||||
def _write_signals_with_softlink(self, group):
|
||||
signals_is_async = "signals" in self.info_storage["bec"].get("readout_priority", {}).get(
|
||||
"async", []
|
||||
)
|
||||
|
||||
print(f"signal is async: {signals_is_async}")
|
||||
|
||||
if signals_is_async:
|
||||
prefix = "signals_data_"
|
||||
else:
|
||||
prefix = "signals_"
|
||||
|
||||
print(self.written_async_signals)
|
||||
|
||||
self.safe_dataset(group=group, name="s1", device="signals", signal=f"{prefix}s1")
|
||||
self.safe_dataset(group=group, name="s2", device="signals", signal=f"{prefix}s2")
|
||||
self.safe_dataset(group=group, name="s3", device="signals", signal=f"{prefix}s3")
|
||||
self.safe_dataset(group=group, name="s4", device="signals", signal=f"{prefix}s4")
|
||||
self.safe_dataset(group=group, name="s5", device="signals", signal=f"{prefix}s5")
|
||||
self.safe_dataset(group=group, name="s6", device="signals", signal=f"{prefix}s6")
|
||||
self.safe_dataset(group=group, name="s7", device="signals", signal=f"{prefix}s7")
|
||||
self.safe_dataset(
|
||||
group=group, name="norm_diode", device="signals", signal=f"{prefix}norm_diode"
|
||||
)
|
||||
self.safe_dataset(
|
||||
group=group, name="norm_tey", device="signals", signal=f"{prefix}norm_tey"
|
||||
)
|
||||
self.safe_dataset(
|
||||
group=group, name="energy", device="signals", signal=f"{prefix}energy_cerbk"
|
||||
)
|
||||
Reference in New Issue
Block a user