From f50d38d4836407f64f52a3d211b4d57e5765f1f9 Mon Sep 17 00:00:00 2001 From: wakonig_k Date: Sun, 28 Jun 2026 15:55:06 +0200 Subject: [PATCH 1/2] feat(default_format): enhance get_entry method and add safe_dataset for robust data handling --- .../bec_server/file_writer/default_writer.py | 97 +++++++++++++++++-- .../bec_server/file_writer_plugins/cSAXS.py | 14 +-- .../tests_file_writer/test_file_writer.py | 89 ++++++++++++++++- 3 files changed, 185 insertions(+), 15 deletions(-) diff --git a/bec_server/bec_server/file_writer/default_writer.py b/bec_server/bec_server/file_writer/default_writer.py index f17d22b36..83b60af3e 100644 --- a/bec_server/bec_server/file_writer/default_writer.py +++ b/bec_server/bec_server/file_writer/default_writer.py @@ -47,21 +47,23 @@ def get_storage_format(self) -> dict: # pylint: disable=protected-access return self.storage._storage - def get_entry(self, name: str, default=None) -> Any: + def get_entry(self, name: str, signal: str | None = None, default=None) -> Any: """ - Get an entry from the scan data assuming a ..value structure. + Get an entry from the scan data assuming a ..value structure. This method is a helper to extract the device data from the scan data, irrespective of the data structure (list of entries or single entry). Args: name (str): Entry name + signal (str, optional): Signal name. Defaults to None. default (Any, optional): Default value. Defaults to None. """ + signal = signal or name if isinstance(self.data.get(name), list) and isinstance(self.data[name][0], dict): - return [sub_data.get(name, {}).get("value", default) for sub_data in self.data[name]] + return [sub_data.get(signal, {}).get("value", default) for sub_data in self.data[name]] - return self.data.get(name, {}).get(name, {}).get("value", default) + return self.data.get(name, {}).get(signal, {}).get("value", default) def write_bec_entries(self) -> None: """ @@ -133,6 +135,90 @@ def write_bec_entries(self) -> None: state_group = beamline_states_group.create_dataset(name=state_name, data=state_values) state_group.attrs["NX_class"] = "NXcollection" + def safe_dataset( + self, + group: HDF5Storage, + name: str, + device: str, + signal: str | None = None, + units: str | None = None, + description: str | None = None, + attributes: dict | None = None, + softlink: bool = True, + ) -> 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). + + Args: + group (HDF5Storage): The HDF5 group to write the dataset to. + name (str): The name of the dataset. + device (str): The device name to retrieve the data from. + attributes (dict, optional): Additional attributes to set on the dataset. Defaults to None. + units (str, optional): The units of the dataset. Defaults to None. + description (str, optional): The description of the dataset. Defaults to None. + softlink (bool, optional): Create a soft link into /entry/collection/devices instead of + copying the value into a new dataset. Defaults to True. + """ + signal = signal or device + value = self.get_entry(device, signal=signal) + if value is None: + return + if softlink: + group.create_soft_link( + name=name, target=f"/entry/collection/devices/{device}/{signal}/value" + ) + return + ds = group.create_dataset(name, data=value) + if attributes: + for key, val in attributes.items(): + ds.attrs[key] = val + if units: + ds.attrs["units"] = units + if description: + ds.attrs["description"] = description + + def _device_shape_matches(self, reference_device: str, candidate_device: str) -> bool: + """ + Check whether two scan report devices have compatible value shapes for NXdata. + """ + reference_value = self.get_entry(reference_device) + candidate_value = self.get_entry(candidate_device) + if reference_value is None or candidate_value is None: + return False + try: + return np.asarray(reference_value).shape == np.asarray(candidate_value).shape + except Exception: + return False + + def _write_scan_report_data(self, entry: HDF5Storage) -> None: + """ + Write an NXdata group containing soft links to the scan report devices. + """ + data_group = entry.create_group("data") + data_group.attrs["NX_class"] = "NXdata" + + scan_report_devices = self.info_storage.get("bec", {}).get("scan_report_devices") or [] + if scan_report_devices: + data_group.attrs["signal"] = scan_report_devices[0] + if not scan_report_devices: + return + + primary_device = scan_report_devices[0] + compatible_devices = [primary_device] + auxiliary_signals = [] + for device in scan_report_devices[1:]: + if self._device_shape_matches(primary_device, device): + compatible_devices.append(device) + auxiliary_signals.append(device) + if auxiliary_signals: + data_group.attrs["auxiliary_signals"] = auxiliary_signals + + for device in compatible_devices: + self.safe_dataset(data_group, name=device, device=device, softlink=True) + def format(self) -> None: """ Prepare the NeXus file format. @@ -156,8 +242,7 @@ def format(self) -> None: control.create_dataset(name="mode", data="monitor") # /entry/data - if "eiger_4" in self.device_manager.devices: - entry.create_soft_link(name="data", target="/entry/instrument/eiger_4") + self._write_scan_report_data(entry) # /entry/sample control = entry.create_group("sample") diff --git a/bec_server/bec_server/file_writer_plugins/cSAXS.py b/bec_server/bec_server/file_writer_plugins/cSAXS.py index 2a1a2ef70..731201bf8 100644 --- a/bec_server/bec_server/file_writer_plugins/cSAXS.py +++ b/bec_server/bec_server/file_writer_plugins/cSAXS.py @@ -75,7 +75,7 @@ def format(self) -> None: source.create_dataset(name="name", data="Swiss Light Source") source.create_dataset(name="probe", data="x-ray") distance = source.create_dataset( - name="distance", data=-33800 - np.asarray(self.get_entry("samz", 0)) + name="distance", data=-33800 - np.asarray(self.get_entry("samz", default=0)) ) distance.attrs["units"] = "mm" sigma_x = source.create_dataset(name="sigma_x", data=0.202) @@ -108,7 +108,7 @@ def format(self) -> None: x_translation = source.create_dataset(name="x_translation", data=self.get_entry("sl0ch")) x_translation.attrs["units"] = "mm" distance = source.create_dataset( - name="distance", data=-21700 - np.asarray(self.get_entry("samz", 0)) + name="distance", data=-21700 - np.asarray(self.get_entry("samz", default=0)) ) distance.attrs["units"] = "mm" @@ -125,7 +125,7 @@ def format(self) -> None: height = source.create_dataset(name="x_translation", data=self.get_entry("sl1ch")) height.attrs["units"] = "mm" distance = source.create_dataset( - name="distance", data=-7800 - np.asarray(self.get_entry("samz", 0)) + name="distance", data=-7800 - np.asarray(self.get_entry("samz", default=0)) ) distance.attrs["units"] = "mm" @@ -143,7 +143,7 @@ def format(self) -> None: energy.attrs["units"] = "keV" mono.create_dataset(name="type", data="Double crystal fixed exit monochromator.") distance = mono.create_dataset( - name="distance", data=-5220 - np.asarray(self.get_entry("samz", 0)) + name="distance", data=-5220 - np.asarray(self.get_entry("samz", default=0)) ) distance.attrs["units"] = "mm" @@ -208,7 +208,7 @@ def format(self) -> None: bend_y = mirror.create_dataset(name="bend_y", data="mibd") bend_y.attrs["units"] = "NX_DIMENSIONLESS" distance = mirror.create_dataset( - name="distance", data=-4370 - np.asarray(self.get_entry("samz", 0)) + name="distance", data=-4370 - np.asarray(self.get_entry("samz", default=0)) ) distance.attrs["units"] = "mm" @@ -252,7 +252,7 @@ def format(self) -> None: height = source.create_dataset(name="x_translation", data=self.get_entry("sl2cv")) height.attrs["units"] = "mm" distance = source.create_dataset( - name="distance", data=-3140 - np.asarray(self.get_entry("samz", 0)) + name="distance", data=-3140 - np.asarray(self.get_entry("samz", default=0)) ) distance.attrs["units"] = "mm" @@ -279,7 +279,7 @@ def format(self) -> None: data="The filter set consists of 4 linear stages, each with five filter positions. Additionally, each one allows for an out position to allow 'no filtering'.", ) attenuator_transmission = filter_set.create_dataset( - name="attenuator_transmission", data=10 ** self.get_entry("ftrans", 0) + name="attenuator_transmission", data=10 ** self.get_entry("ftrans", default=0) ) attenuator_transmission.attrs["units"] = "NX_DIMENSIONLESS" diff --git a/bec_server/tests/tests_file_writer/test_file_writer.py b/bec_server/tests/tests_file_writer/test_file_writer.py index f9c305d64..74eed797c 100644 --- a/bec_server/tests/tests_file_writer/test_file_writer.py +++ b/bec_server/tests/tests_file_writer/test_file_writer.py @@ -10,6 +10,7 @@ from bec_lib import messages from bec_server import file_writer from bec_server.file_writer import HDF5FileWriter +from bec_server.file_writer.default_writer import DefaultFormat from bec_server.file_writer.file_writer import HDF5Storage from bec_server.file_writer.file_writer_manager import ScanStorage from bec_server.file_writer_plugins.cSAXS import cSAXSFormat @@ -62,6 +63,19 @@ def scan_storage_mock(tmp_path): yield storage +@pytest.fixture +def default_format(file_writer_manager_mock_with_dm): + yield DefaultFormat( + storage=HDF5Storage(), + data={}, + file_references={}, + info_storage={"bec": {"readout_priority": {}, "scan_report_devices": []}}, + configuration={}, + device_manager=file_writer_manager_mock_with_dm.device_manager, + beamline_states={}, + ) + + def test_csaxs_nexus_format(file_writer_manager_mock_with_dm): file_manager = file_writer_manager_mock_with_dm writer_storage = cSAXSFormat( @@ -79,6 +93,71 @@ def test_csaxs_nexus_format(file_writer_manager_mock_with_dm): assert writer_storage["entry"]._storage["sample"]._storage["x_translation"]._data == [0, 1, 2] +def test_get_entry_returns_values_for_scalar_and_list_data(default_format): + default_format.data = { + "samx": [{"samx": {"value": 1}}, {"samx": {"value": 2}}], + "temperature": {"readback": {"value": 273.15}}, + } + + assert default_format.get_entry("samx") == [1, 2] + assert default_format.get_entry("temperature", signal="readback") == 273.15 + + +def test_get_entry_returns_default_for_missing_signal(default_format): + default_format.data = {"samx": {"samx": {"value": 1}}} + + assert default_format.get_entry("samx", signal="missing", default="fallback") == "fallback" + assert default_format.get_entry("missing", default="fallback") == "fallback" + + +def test_safe_dataset_skips_missing_device(default_format): + group = default_format.storage.create_group("group") + + default_format.safe_dataset(group, name="samx", device="samx") + + assert "samx" not in group._storage + + +def test_safe_dataset_writes_dataset_attrs_and_softlink(default_format): + default_format.data = {"samx": {"samx": {"value": [0, 1, 2]}}} + group = default_format.storage.create_group("group") + + default_format.safe_dataset( + group, + name="x_translation", + device="samx", + units="mm", + description="sample x position", + attributes={"long_name": "sample_x"}, + softlink=False, + ) + default_format.safe_dataset(group, name="samx_link", device="samx", softlink=True) + + dataset = group._storage["x_translation"] + assert dataset._data == [0, 1, 2] + assert dataset.attrs["units"] == "mm" + assert dataset.attrs["description"] == "sample x position" + assert dataset.attrs["long_name"] == "sample_x" + assert group._storage["samx_link"]._storage_type == "softlink" + assert group._storage["samx_link"]._data == "/entry/collection/devices/samx/samx/value" + + +def test_scan_report_data_only_includes_shape_compatible_auxiliary_signals(default_format): + default_format.data = { + "samx": {"samx": {"value": [0, 1, 2]}}, + "samy": {"samy": {"value": [3, 4, 5]}}, + "mokev": {"mokev": {"value": 12.456}}, + } + default_format.info_storage["bec"]["scan_report_devices"] = ["samx", "samy", "mokev"] + + writer_storage = default_format.get_storage_format() + data_group = writer_storage["entry"]._storage["data"] + + assert data_group.attrs["signal"] == "samx" + assert data_group.attrs["auxiliary_signals"] == ["samy"] + assert set(data_group._storage) == {"samx", "samy"} + + def test_nexus_file_writer(hdf5_file_writer, scan_storage_mock, tmp_path): file_writer = hdf5_file_writer with mock.patch.object( @@ -97,11 +176,12 @@ def test_nexus_file_writer(hdf5_file_writer, scan_storage_mock, tmp_path): file_writer.write(f"{tmp_path}/test.h5", scan_storage_mock, configuration_data={}) with h5py.File(f"{tmp_path}/test.tmp", "r") as test_file: assert list(test_file) == ["entry"] - assert list(test_file["entry"]) == ["collection", "control", "instrument", "sample"] + assert list(test_file["entry"]) == ["collection", "control", "data", "instrument", "sample"] assert np.allclose( test_file["entry/collection/devices/samx/samx/value"][...], [0, 1, 2, 3, 4] ) assert test_file["entry/collection/file_references/eiger"] is not None + assert test_file["entry/data"].attrs["NX_class"] == "NXdata" # assert list(test_file["entry"]["sample"]) == ["x_translation"] # assert test_file["entry"]["sample"].attrs["NX_class"] == "NXsample" # assert test_file["entry"]["sample"]["x_translation"].attrs["units"] == "mm" @@ -168,7 +248,7 @@ def test_create_device_data_storage(hdf5_file_writer, scan_storage_mock): "scan_number": 88, "dataset_number": 88, "exp_time": 0.1, - "scan_report_devices": ["samx"], + "scan_report_devices": ["samx", "samy"], "scan_msgs": [ "ScanQueueMessage(({'scan_type': 'monitor_scan', 'parameter': {'args': {'samx':" " [-100, 100]}, 'kwargs': {'relative': False}}, 'queue': 'primary'}, {'RID':" @@ -216,6 +296,11 @@ def test_write_data_storage(segments, baseline, metadata, hdf5_file_writer, tmp_ == datetime.datetime.fromtimestamp(1679226971.580867).isoformat() ) assert "non_existing_file" not in test_file["entry/collection/file_references"].keys() + assert test_file["entry/data"].attrs["NX_class"] == "NXdata" + assert test_file["entry/data"].attrs["signal"] == "samx" + assert list(test_file["entry/data"].attrs["auxiliary_signals"]) == ["samy"] + assert np.allclose(test_file["entry/data/samx"][...], [0.11, 0.21]) + assert np.allclose(test_file["entry/data/samy"][...], [1.1, 1.2]) def test_load_format_from_plugin(tmp_path, hdf5_file_writer): From 8a8ce4e45f8a29076455e55531ef243e7f6190f3 Mon Sep 17 00:00:00 2001 From: wakonig_k Date: Mon, 29 Jun 2026 17:49:11 +0200 Subject: [PATCH 2/2] feat(async): add has_async_signal method and update get_entry for async signal handling --- .../bec_server/file_writer/default_writer.py | 32 ++++++++++-- .../tests_file_writer/test_file_writer.py | 51 +++++++++++++++++++ 2 files changed, 80 insertions(+), 3 deletions(-) diff --git a/bec_server/bec_server/file_writer/default_writer.py b/bec_server/bec_server/file_writer/default_writer.py index 83b60af3e..19c47f3b5 100644 --- a/bec_server/bec_server/file_writer/default_writer.py +++ b/bec_server/bec_server/file_writer/default_writer.py @@ -47,13 +47,36 @@ def get_storage_format(self) -> dict: # pylint: disable=protected-access return self.storage._storage + def has_async_signal(self, device_name: str, signal_name: str) -> bool: + """ + Check if a device has an async signal. + + Args: + device_name (str): The name of the device. + signal_name (str): The name of the signal. + + Returns: + bool: True if the device has an async signal, False otherwise. + """ + signals = self.device_manager.get_bec_signals( + ["AsyncMultiSignal", "AsyncSignal", "DynamicSignal"] + ) + for device_name_, _, signal_info in signals: + obj_name = signal_info.get("object_name", "") + obj_name_without_prefix = obj_name.removeprefix("devicename") + if device_name_ == device_name and (signal_name in [obj_name, obj_name_without_prefix]): + return True + return False + def get_entry(self, name: str, signal: str | None = None, default=None) -> Any: """ - Get an entry from the scan data assuming a ..value structure. + Get an entry from the scan data (monitored or baseline) assuming a ..value structure. This method is a helper to extract the device data from the scan data, irrespective of the data structure (list of entries or single entry). + Note: This method does not handle async signals. Use `has_async_signal` to check for async signals. + Args: name (str): Entry name signal (str, optional): Signal name. Defaults to None. @@ -160,12 +183,15 @@ def safe_dataset( units (str, optional): The units of the dataset. Defaults to None. description (str, optional): The description of the dataset. Defaults to None. softlink (bool, optional): Create a soft link into /entry/collection/devices instead of - copying the value into a new dataset. Defaults to True. + copying the value into a new dataset. Defaults to True. For async signals, this is always True. """ signal = signal or device value = self.get_entry(device, signal=signal) - if value is None: + if self.has_async_signal(device, signal): + softlink = True + elif value is None: return + if softlink: group.create_soft_link( name=name, target=f"/entry/collection/devices/{device}/{signal}/value" diff --git a/bec_server/tests/tests_file_writer/test_file_writer.py b/bec_server/tests/tests_file_writer/test_file_writer.py index 74eed797c..06b2a1044 100644 --- a/bec_server/tests/tests_file_writer/test_file_writer.py +++ b/bec_server/tests/tests_file_writer/test_file_writer.py @@ -110,6 +110,20 @@ def test_get_entry_returns_default_for_missing_signal(default_format): assert default_format.get_entry("missing", default="fallback") == "fallback" +def test_has_async_signal_matches_prefixed_and_unprefixed_object_names(default_format): + with mock.patch.object( + default_format.device_manager, "get_bec_signals" + ) as mock_get_bec_signals: + mock_get_bec_signals.return_value = [ + ("samx", None, {"object_name": "devicenamesamx"}), + ("waveform", None, {"object_name": "waveform"}), + ] + + assert default_format.has_async_signal("samx", "samx") is True + assert default_format.has_async_signal("waveform", "waveform") is True + assert default_format.has_async_signal("samx", "other") is False + + def test_safe_dataset_skips_missing_device(default_format): group = default_format.storage.create_group("group") @@ -142,6 +156,31 @@ def test_safe_dataset_writes_dataset_attrs_and_softlink(default_format): assert group._storage["samx_link"]._data == "/entry/collection/devices/samx/samx/value" +def test_safe_dataset_forces_softlink_for_async_signal(default_format): + default_format.data = {"samx": {"samx": {"value": [0, 1, 2]}}} + group = default_format.storage.create_group("group") + + with mock.patch.object(default_format, "has_async_signal", return_value=True): + default_format.safe_dataset(group, name="samx", device="samx", softlink=False) + + assert group._storage["samx"]._storage_type == "softlink" + assert group._storage["samx"]._data == "/entry/collection/devices/samx/samx/value" + + +def test_device_shape_matches_returns_false_for_missing_or_unshapeable_values(default_format): + default_format.data = { + "samx": {"samx": {"value": [0, 1, 2]}}, + "broken": {"broken": {"value": mock.Mock(side_effect=TypeError("boom"))}}, + } + + assert default_format._device_shape_matches("samx", "missing") is False + + with mock.patch( + "bec_server.file_writer.default_writer.np.asarray", side_effect=TypeError("boom") + ): + assert default_format._device_shape_matches("samx", "broken") is False + + def test_scan_report_data_only_includes_shape_compatible_auxiliary_signals(default_format): default_format.data = { "samx": {"samx": {"value": [0, 1, 2]}}, @@ -158,6 +197,18 @@ def test_scan_report_data_only_includes_shape_compatible_auxiliary_signals(defau assert set(data_group._storage) == {"samx", "samy"} +def test_write_scan_report_data_skips_attrs_when_no_devices(default_format): + entry = default_format.storage.create_group("entry") + + default_format._write_scan_report_data(entry) + + data_group = entry._storage["data"] + assert data_group.attrs["NX_class"] == "NXdata" + assert "signal" not in data_group.attrs + assert "auxiliary_signals" not in data_group.attrs + assert data_group._storage == {} + + def test_nexus_file_writer(hdf5_file_writer, scan_storage_mock, tmp_path): file_writer = hdf5_file_writer with mock.patch.object(