VDS: add tests for new vds mapping

src/nexgen/beamlines/i19_2/eiger.py

@@ -177,8 +177,7 @@ def eiger_writer(
             "Not using meta file to update metadata, only the external links will be set up."
         )
         vds_dtype = define_vds_dtype_from_bit_depth(eiger_settings.bit_depth)
-        # wavelength = parameters.wavelength
-        # beam_center = parameters.beam_center
+
         # Update axes
         # Goniometer
         for gax in parameters.axes_pos:

src/nexgen/beamlines/i19_2/serial.py

@@ -36,6 +36,28 @@ def serial_nexus_writer(
     vds_mapping: VdsMapping = VdsMapping.BLOCKED,
     notes: dict[str, Any] | None = None,
 ):
+    """Wrapper function to gather all parameters from the beamline and kick off the nexus writer for a
+    serial experiment on I19-2.
+
+    Args:
+        params (dict[str, Any]): Dictionary representation of CollectionParams.
+        master_file (Path): Full path to the nexus file to be written.
+        use_meta (bool, optional): Eiger option only, if True use metadata from meta.h5 file. Otherwise
+            all parameters will need to be passed manually. Defaults to False.
+        vds_offset (int, optional): Start index for the vds writer. Defaults to 0.
+        n_frames (int | None, optional): Number of images for the nexus file. Only needed if different
+            from the tot_num_images in the collection params. If passed, the VDS will only contain the
+            number of frames specified here. Defaults to None.
+        bit_depth(int, optional): Default bit depth for eiger collections, used to define dtype of vds data. \
+            Defaults to 32.
+        data_entry_key (str, optional): Where to find the dataset. Defaults to "data".
+        eiger_stream_format (EigerStreamFormat, optional): Stream format setting on the new fastcs eiger.
+            The metafile in the new cbor format is slightly different. Defaults to "legacy".
+        vds_mapping (VdsMapping, optional): How to map the frames when building the VDS.
+        notes (dict[str, Any] | None, optional): Any additional information to be written as NXnote,
+            passed as a dictionary of (key, value) pairs where key represents the dataset name and
+            value its data. Defaults to None.
+    """
     _setup_logging(master_file.parent)
 
     collection_params = CollectionParams(**params)

src/nexgen/tools/vds_tools/strided_mapping.py

@@ -17,7 +17,7 @@
 class SingleDataset:
     name: str
     # Full size of the source dataset
-    src_shape = Sequence[int]
+    src_shape: Sequence[int]
     # Index to start the mapping from, usually 0 or 1 for this
     start_index: int = 0
     # Step for slicing the dataset. Defaults to every other image

tests/tools/conftest.py

@@ -0,0 +1,23 @@
+import tempfile
+
+import h5py
+import pytest
+
+
+@pytest.fixture
+def nexus_file_with_single_dataset():
+    test_hdf_file = tempfile.TemporaryFile()
+    test_nexus_file = h5py.File(test_hdf_file, "w")
+    test_nexus_file["/entry/data/data_0001"] = h5py.ExternalLink("filename", "data")
+    yield test_nexus_file
+
+
+@pytest.fixture(scope="session")
+def nexus_file_with_multiple_datasets():
+    test_hdf_file = tempfile.TemporaryFile()
+    test_nexus_file = h5py.File(test_hdf_file, "r+")
+    test_nexus_file.require_group("/entry/data")
+    test_nexus_file["/entry/data/data_0001"] = h5py.ExternalLink("file_1", "data")
+    test_nexus_file["/entry/data/data_0002"] = h5py.ExternalLink("file_2", "data")
+
+    yield test_nexus_file

tests/tools/test_VDS_tools.py

@@ -9,7 +9,6 @@
     Dataset,
     create_virtual_layout,
     define_vds_dtype_from_bit_depth,
-    find_datasets_in_file,
     image_vds_writer,
     jungfrau_vds_writer,
     split_datasets,
@@ -91,21 +90,6 @@ def test_when_start_idx_negative_then_exception_raised():
         split_datasets(["test1"], (1100, 10, 10), -100)
 
 
-@pytest.fixture
-def nexus_file_with_single_dataset():
-    test_hdf_file = tempfile.TemporaryFile()
-    test_nexus_file = h5py.File(test_hdf_file, "w")
-    test_nexus_file["/entry/data/data_0001"] = h5py.ExternalLink("filename", "path")
-    yield test_nexus_file
-
-
-def test_find_datasets_int_file(nexus_file_with_single_dataset):
-    nxdata = nexus_file_with_single_dataset["/entry/data"]
-    dsets = find_datasets_in_file(nxdata)
-    assert len(dsets) == 1
-    assert dsets[0] == "data_0001"
-
-
 def test_when_float_shape_passed_to_vds_writer_then_no_exception(
     nexus_file_with_single_dataset,
 ):

tests/tools/vds/test_strided_vds.py

@@ -0,0 +1,43 @@
+from unittest.mock import patch
+
+import numpy as np
+import pytest
+
+from nexgen.tools.vds_tools.strided_mapping import (
+    SingleDataset,
+    create_dataset_list,
+    create_vds_layout,
+    write_strided_vds,
+)
+
+
+@patch("nexgen.tools.vds_tools.strided_mapping.find_datasets_in_file")
+def test_create_dataset_list_fails_if_no_external_links_found(mock_find):
+    mock_find.return_value = []
+    with pytest.raises(ValueError):
+        create_dataset_list("", 0)
+
+
+def test_create_vds_layout():
+    dset_list = [
+        SingleDataset(name="data_01", src_shape=(6, 2, 2), start_index=1, stride=2)
+    ]
+    expected_shape = (3, 2, 2)
+
+    layout = create_vds_layout(dset_list, expected_shape, np.uint16)
+
+    assert layout.shape == expected_shape
+    assert list(layout._src_filenames)[0] == b"."
+
+
+@patch("nexgen.tools.vds_tools.strided_mapping.create_dataset_list")
+@patch("nexgen.tools.vds_tools.strided_mapping.create_vds_layout")
+def test_write_strided_vds(mock_dset_list, mock_layout, nexus_file_with_single_dataset):
+    with patch(
+        "nexgen.tools.vds_tools.strided_mapping.h5py.Group.create_virtual_dataset"
+    ) as mock_create:
+        write_strided_vds(nexus_file_with_single_dataset, (10, 2, 3), 0)
+
+        mock_dset_list.assert_called_once()
+        mock_layout.assert_called_once()
+        mock_create.assert_called_once()

tests/tools/vds/test_utils.py

@@ -0,0 +1,28 @@
+import tempfile
+
+import h5py
+import pytest
+
+from nexgen.tools.vds_tools.utils import find_datasets_in_file
+
+
+def test_find_datasets_in_file(nexus_file_with_single_dataset):
+    nxdata = nexus_file_with_single_dataset["/entry/data"]
+    dsets = find_datasets_in_file(nxdata)
+    assert len(dsets) == 1
+    assert dsets[0] == "data_0001"
+
+
+def test_find_multiple_datasets_in_file(nexus_file_with_multiple_datasets):
+    nxdata = nexus_file_with_multiple_datasets["/entry/data"]
+    dsets = find_datasets_in_file(nxdata)
+    assert len(dsets) == 2
+    assert dsets[0] == "data_0001" and dsets[1] == "data_0002"
+
+
+def test_find_datasets_fails_if_no_links_in_file():
+    test_hdf_file = tempfile.TemporaryFile()
+    test_nexus_file = h5py.File(test_hdf_file, "w")
+    test_nexus_file["/entry/data/data_0001"] = [0, 0, 0]
+    with pytest.raises(KeyError):
+        find_datasets_in_file(test_nexus_file)