Repack Nwb Files

src/neuroconv/tools/nwb_helpers/_configuration_models/_hdf5_dataset_io.py

@@ -3,9 +3,11 @@
 from typing import Any, Dict, Literal, Union
 
 import h5py
+from hdmf import Container
 from pydantic import Field, InstanceOf
+from typing_extensions import Self
 
-from ._base_dataset_io import DatasetIOConfiguration
+from ._base_dataset_io import DatasetIOConfiguration, _find_location_in_memory_nwbfile
 from ...importing import is_package_installed
 
 _base_hdf5_filters = set(h5py.filters.decode)
@@ -78,3 +80,28 @@ def get_data_io_kwargs(self) -> Dict[str, Any]:
             compression_bundle = dict(compression=self.compression_method, compression_opts=compression_opts)
 
         return dict(chunks=self.chunk_shape, **compression_bundle)
+
+    @classmethod
+    def from_existing_neurodata_object(
+        cls, neurodata_object: Container, dataset_name: Literal["data", "timestamps"]
+    ) -> Self:
+        location_in_file = _find_location_in_memory_nwbfile(neurodata_object=neurodata_object, field_name=dataset_name)
+        full_shape = getattr(neurodata_object, dataset_name).shape
+        dtype = getattr(neurodata_object, dataset_name).dtype
+        chunk_shape = getattr(neurodata_object, dataset_name).chunks
+        buffer_shape = getattr(neurodata_object, dataset_name).maxshape
+        compression_method = getattr(neurodata_object, dataset_name).compression
+        compression_opts = getattr(neurodata_object, dataset_name).compression_opts
+        compression_options = dict(compression_opts=compression_opts)
+        return cls(
+            object_id=neurodata_object.object_id,
+            object_name=neurodata_object.name,
+            location_in_file=location_in_file,
+            dataset_name=dataset_name,
+            full_shape=full_shape,
+            dtype=dtype,
+            chunk_shape=chunk_shape,
+            buffer_shape=buffer_shape,
+            compression_method=compression_method,
+            compression_options=compression_options,
+        )

src/neuroconv/tools/nwb_helpers/_dataset_configuration.py

@@ -172,3 +172,63 @@ def get_default_dataset_io_configurations(
                 )
 
                 yield dataset_io_configuration
+
+
+def get_existing_dataset_io_configurations(
+    nwbfile: NWBFile,
+    backend: Literal["hdf5", "zarr"],
+) -> Generator[DatasetIOConfiguration, None, None]:
+
+    DatasetIOConfigurationClass = DATASET_IO_CONFIGURATIONS[backend]
+
+    known_dataset_fields = ("data", "timestamps")
+    for neurodata_object in nwbfile.objects.values():
+        if isinstance(neurodata_object, DynamicTable):
+            dynamic_table = neurodata_object  # For readability
+
+            for column in dynamic_table.columns:
+                candidate_dataset = column.data  # VectorData object
+
+                # Skip over columns whose values are links, such as the 'group' of an ElectrodesTable
+                if any(isinstance(value, Container) for value in candidate_dataset):
+                    continue  # Skip
+
+                # Skip when columns whose values are a reference type
+                if isinstance(column, TimeSeriesReferenceVectorData):
+                    continue
+
+                # Skip datasets with any zero-length axes
+                dataset_name = "data"
+                candidate_dataset = getattr(column, dataset_name)
+                full_shape = get_data_shape(data=candidate_dataset)
+                if any(axis_length == 0 for axis_length in full_shape):
+                    continue
+
+                dataset_io_configuration = DatasetIOConfigurationClass.from_existing_neurodata_object(
+                    neurodata_object=column, dataset_name=dataset_name
+                )
+
+                yield dataset_io_configuration
+        elif isinstance(neurodata_object, NWBContainer):
+            for known_dataset_field in known_dataset_fields:
+                # Skip optional fields that aren't present
+                if known_dataset_field not in neurodata_object.fields:
+                    continue
+
+                candidate_dataset = getattr(neurodata_object, known_dataset_field)
+
+                # Skip edge case of in-memory ImageSeries with external mode; data is in fields and is empty array
+                if isinstance(candidate_dataset, np.ndarray) and candidate_dataset.size == 0:
+                    continue
+
+                # Skip datasets with any zero-length axes
+                candidate_dataset = getattr(neurodata_object, known_dataset_field)
+                full_shape = get_data_shape(data=candidate_dataset)
+                if any(axis_length == 0 for axis_length in full_shape):
+                    continue
+
+                dataset_io_configuration = DatasetIOConfigurationClass.from_existing_neurodata_object(
+                    neurodata_object=neurodata_object, dataset_name=known_dataset_field
+                )
+
+                yield dataset_io_configuration

temp_test.py

@@ -0,0 +1,44 @@
+import os
+from pathlib import Path
+
+import numpy as np
+from pynwb import NWBHDF5IO, H5DataIO, TimeSeries
+from pynwb.testing.mock.file import mock_NWBFile
+
+from neuroconv.tools.nwb_helpers._dataset_configuration import (
+    get_existing_dataset_io_configurations,
+)
+
+
+def write_nwbfile(nwbfile_path: Path):
+    if nwbfile_path.exists():
+        os.remove(nwbfile_path)
+    nwbfile = mock_NWBFile()
+    timestamps = np.arange(10.0)
+    data = np.arange(100, 200, 10)
+    time_series_with_timestamps = TimeSeries(
+        name="test_timeseries",
+        description="an example time series",
+        data=H5DataIO(data=data, compression="gzip", chunks=(1,), compression_opts=2),
+        unit="m",
+        timestamps=H5DataIO(
+            timestamps, compression="gzip", chunks=(1,), compression_opts=2
+        ),  # TODO: add support for uncompressed timestamps
+    )
+    nwbfile.add_acquisition(time_series_with_timestamps)
+    with NWBHDF5IO(nwbfile_path, mode="w") as io:
+        io.write(nwbfile)
+
+
+def main():
+    nwbfile_path = Path("/Volumes/T7/CatalystNeuro/temp.nwb")
+    write_nwbfile(nwbfile_path)
+    with NWBHDF5IO(nwbfile_path, mode="r") as io:
+        nwbfile = io.read()
+        existing_dataset_io_configurations = get_existing_dataset_io_configurations(nwbfile, backend="hdf5")
+        for dataset_io_configuration in existing_dataset_io_configurations:
+            print(dataset_io_configuration)
+
+
+if __name__ == "__main__":
+    main()