From d484934b5969964faeea705d3d30895b2fa6728e Mon Sep 17 00:00:00 2001 From: wakonig_k Date: Mon, 17 Aug 2026 09:50:32 +0200 Subject: [PATCH] feat: add nexus format --- .gitea/workflows/ci.yml | 16 +- xtreme_bec/file_writer/__init__.py | 1 + xtreme_bec/file_writer/xtreme_nexus.py | 225 +++++++++++++++++++++++++ 3 files changed, 239 insertions(+), 3 deletions(-) create mode 100644 xtreme_bec/file_writer/xtreme_nexus.py diff --git a/.gitea/workflows/ci.yml b/.gitea/workflows/ci.yml index 9a263e3..dd7b436 100644 --- a/.gitea/workflows/ci.yml +++ b/.gitea/workflows/ci.yml @@ -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 diff --git a/xtreme_bec/file_writer/__init__.py b/xtreme_bec/file_writer/__init__.py index e69de29..d376bc9 100644 --- a/xtreme_bec/file_writer/__init__.py +++ b/xtreme_bec/file_writer/__init__.py @@ -0,0 +1 @@ +from .xtreme_nexus import XtremeNeXusFormat \ No newline at end of file diff --git a/xtreme_bec/file_writer/xtreme_nexus.py b/xtreme_bec/file_writer/xtreme_nexus.py new file mode 100644 index 0000000..74bd361 --- /dev/null +++ b/xtreme_bec/file_writer/xtreme_nexus.py @@ -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" + )