docs/changes/latest.inc

@@ -61,6 +61,8 @@ Enhancements
 
 - Allow for empty parcels in :class:`junifer.markers.ParcelAggregation`, that will result in NaNs (:gh:`194` by `Fede Raimondo`_).
 
+- Improve metadata and data I/O for :class:`junifer.storage.HDF5FeatureStorage` (:gh:`196` by `Synchon Mandal`_).
+
 Bugs
 ~~~~
 

junifer/external/h5io

@@ -1 +1 @@
-Subproject commit 1d2ba9925c12f5e3c191eca888b829e43c206658
+Subproject commit 4413a9fc3db2ca8fd0982f86e28b326eee6e0b28

junifer/storage/hdf5.py

@@ -15,7 +15,7 @@ import pandas as pd
 from tqdm import tqdm
 
 from ..api.decorators import register_storage
-from ..external.h5io.h5io import ChunkedArray, read_hdf5, write_hdf5
+from ..external.h5io.h5io import ChunkedArray, has_hdf5, read_hdf5, write_hdf5
 from ..utils import logger, raise_error
 from .base import BaseFeatureStorage
 from .utils import element_to_prefix, matrix_to_vector, store_matrix_checks
@@ -151,34 +151,40 @@ class HDF5FeatureStorage(BaseFeatureStorage):
 
         Raises
         ------
-        IOError
+        FileNotFoundError
-            If HDF5 file or `meta` does not exist.
+            If HDF5 file does not exist.
+        RuntimeError
+            If ``meta`` does not exist in the file.
 
         """
         # Get correct URI for element;
         # is different from uri if single_output is False
         uri = self._fetch_correct_uri_for_io(element=element)
 
-        try:
+        # Check if file exists
-            logger.info(f"Loading HDF5 metadata from: {uri}")
+        if not Path(uri).exists():
-            metadata = read_hdf5(
-                fname=uri,
-                title="meta",
-                slash="ignore",
-            )
-        except IOError:
             raise_error(
-                msg=f"HDF5 file not found at: {uri}",
+                f"HDF5 file not found at: {uri}",
-                klass=IOError,
+                klass=FileNotFoundError,
             )
-        except ValueError:
+
+        # Check if group is found in the storage
+        if not has_hdf5(fname=uri, title="meta"):
             raise_error(
-                msg=f"`meta` not found in: {uri}",
+                f"Invalid junifer HDF5 file at: {uri}",
-                klass=IOError,
+                klass=RuntimeError,
             )
-        else:
+
-            logger.info(f"Loaded HDF5 metadata from: {uri}")
+        # Read metadata
-            return metadata
+        logger.info(f"Loading HDF5 metadata from: {uri}")
+        metadata = read_hdf5(
+            fname=uri,
+            title="meta",
+            slash="ignore",
+        )
+        logger.info(f"Loaded HDF5 metadata from: {uri}")
+
+        return metadata
 
     def list_features(self) -> Dict[str, Dict[str, Any]]:
         """List the features in the storage.
@@ -218,34 +224,40 @@ class HDF5FeatureStorage(BaseFeatureStorage):
 
         Raises
         ------
-        IOError
+        FileNotFoundError
-            If HDF5 file or data does not exist.
+            If HDF5 file does not exist.
+        RuntimeError
+            If the specified ``md5`` does not exist in the file.
 
         """
         # Get correct URI for element;
         # is different from uri if single_output is False
         uri = self._fetch_correct_uri_for_io(element=element)
 
-        try:
+        # Check if file exists
-            logger.info(f"Loading HDF5 data for {md5} from: {uri}")
+        if not Path(uri).exists():
-            data = read_hdf5(
-                fname=uri,
-                title=md5,
-                slash="ignore",
-            )
-        except IOError:
             raise_error(
-                msg=f"HDF5 file not found at: {uri}",
+                f"HDF5 file not found at: {uri}",
-                klass=IOError,
+                klass=FileNotFoundError,
             )
-        except ValueError:
+
+        # Check if group is found in the storage
+        if not has_hdf5(fname=uri, title=md5):
             raise_error(
-                msg=f"`{md5}` not found in: {uri}",
+                f"{md5} not found in HDF5 file at: {uri}",
-                klass=IOError,
+                klass=RuntimeError,
             )
-        else:
+
-            logger.info(f"Loaded HDF5 data for {md5} from: {uri}")
+        # Read data
-            return data
+        logger.info(f"Loading HDF5 data for {md5} from: {uri}")
+        data = read_hdf5(
+            fname=uri,
+            title=md5,
+            slash="ignore",
+        )
+        logger.info(f"Loaded HDF5 data for {md5} from: {uri}")
+
+        return data
 
     def read_df(
         self,
@@ -462,11 +474,15 @@ class HDF5FeatureStorage(BaseFeatureStorage):
             The metadata as a dictionary.
 
         """
-        # Read metadata; if no file found, create an empty dictionary
+        # Get correct URI for element;
-        try:
+        # is different from uri if single_output is False
+        uri = self._fetch_correct_uri_for_io(element=element)
+
+        # Check if file exists, then read metadata else create empty dictionary
+        if Path(uri).exists():
             metadata = self._read_metadata(element=element)
-        except IOError:
+        else:
-            logger.debug(f"Creating new metadata map for {meta_md5} ...")
+            logger.debug(f"Creating new file at {uri} ...")
             metadata = {}
 
         # Only add entry if MD5 is not present
@@ -475,10 +491,6 @@ class HDF5FeatureStorage(BaseFeatureStorage):
             # Update metadata
             metadata[meta_md5] = meta
 
-            # Get correct URI for element;
-            # is different from uri if single_output is False
-            uri = self._fetch_correct_uri_for_io(element=element)
-
             logger.info(f"Writing HDF5 metadata for {meta_md5} to: {uri}")
             logger.debug(f"HDF5 overwrite is set to: {self.overwrite} ...")
             logger.debug(
@@ -528,10 +540,15 @@ class HDF5FeatureStorage(BaseFeatureStorage):
             Keyword arguments passed from the calling method.
 
         """
-        # Read existing data; if no file found, create an empty dictionary
+        # Get correct URI for element;
-        try:
+        # is different from uri if single_output is False
+        uri = self._fetch_correct_uri_for_io(element=element[0])
+
+        # Check if MD5 exists, then read data else create empty dictionary
+        # File should be present here already
+        if has_hdf5(fname=uri, title=meta_md5):
             stored_data = self._read_data(md5=meta_md5, element=element[0])
-        except IOError:
+        else:
             logger.debug(f"Creating new data map for {meta_md5} ...")
             stored_data = {}
 
@@ -604,10 +621,6 @@ class HDF5FeatureStorage(BaseFeatureStorage):
                 }
             )
 
-        # Get correct URI for element;
-        # is different from uri if single_output is False
-        uri = self._fetch_correct_uri_for_io(element=element[0])
-
         logger.info(f"Writing HDF5 data for {meta_md5} to: {uri}")
         logger.debug(f"HDF5 overwrite is set to: {self.overwrite} ...")
         logger.debug(

junifer/storage/tests/test_hdf5.py

@@ -6,6 +6,7 @@
 
 from pathlib import Path
 
+import h5py
 import numpy as np
 import pytest
 from numpy.testing import assert_array_equal
@@ -36,45 +37,6 @@ def test_single_output(tmp_path: Path) -> None:
     assert storage.single_output is True
 
 
-def test_single_output_meta_not_found_error(tmp_path: Path) -> None:
-    """Test single output metadata not found error.
-
-    Parameters
-    ----------
-    tmp_path : pathlib.Path
-        The path to the test directory.
-
-    """
-    uri = tmp_path / "test_single_output_no_meta.hdf5"
-    storage = HDF5FeatureStorage(uri=uri, single_output=True)
-    # Store data to create the file
-    storage._store_data(
-        kind="vector",
-        meta_md5="md5",
-        element=[{"sub": "001"}],
-        data=np.empty((1, 1)),
-    )
-    # Check metadata error
-    with pytest.raises(IOError, match="`meta` not found in:"):
-        storage._read_metadata()
-
-
-def test_single_output_file_not_found_error(tmp_path: Path) -> None:
-    """Test single output file not found error.
-
-    Parameters
-    ----------
-    tmp_path : pathlib.Path
-        The path to the test directory.
-
-    """
-    uri = tmp_path / "test_single_output_no_file.hdf5"
-    storage = HDF5FeatureStorage(uri=uri, single_output=True)
-    # Check file error
-    with pytest.raises(IOError, match="HDF5 file not found at:"):
-        storage._read_data(md5="md5")
-
-
 def test_multi_output_error(tmp_path: Path) -> None:
     """Test error for multi output.
 
@@ -109,6 +71,76 @@ def test_single_output_parent_path_creation(tmp_path: Path) -> None:
     assert to_create_hdf5.exists()
 
 
+def test_read_metadata_file_not_found_error(tmp_path: Path) -> None:
+    """Test file not found error when reading metadata.
+
+    Parameters
+    ----------
+    tmp_path : pathlib.Path
+        The path to the test directory.
+
+    """
+    uri = tmp_path / "test_read_metadata_no_file.hdf5"
+    storage = HDF5FeatureStorage(uri=uri, single_output=True)
+    # Check file not found error
+    with pytest.raises(FileNotFoundError, match="HDF5 file not found at:"):
+        storage._read_metadata()
+
+
+def test_read_metadata_meta_not_found_error(tmp_path: Path) -> None:
+    """Test meta not found error when reading metadata.
+
+    Parameters
+    ----------
+    tmp_path : pathlib.Path
+        The path to the test directory.
+
+    """
+    uri = tmp_path / "test_read_metadata_no_meta.hdf5"
+    storage = HDF5FeatureStorage(uri=uri, single_output=True)
+    # Create file
+    with h5py.File(uri, "w") as f:
+        f.create_dataset("mydataset", (100,), dtype="i")
+    # Check meta not found error
+    with pytest.raises(RuntimeError, match="Invalid junifer HDF5 file at:"):
+        storage._read_metadata()
+
+
+def test_read_data_file_not_found_error(tmp_path: Path) -> None:
+    """Test file not found error when reading data.
+
+    Parameters
+    ----------
+    tmp_path : pathlib.Path
+        The path to the test directory.
+
+    """
+    uri = tmp_path / "test_read_data_no_file.hdf5"
+    storage = HDF5FeatureStorage(uri=uri, single_output=True)
+    # Check file not found error
+    with pytest.raises(FileNotFoundError, match="HDF5 file not found at:"):
+        storage._read_data(md5="md5")
+
+
+def test_read_data_md5_not_found_error(tmp_path: Path) -> None:
+    """Test meta not found error when reading data.
+
+    Parameters
+    ----------
+    tmp_path : pathlib.Path
+        The path to the test directory.
+
+    """
+    uri = tmp_path / "test_read_data_no_md5.hdf5"
+    storage = HDF5FeatureStorage(uri=uri, single_output=True)
+    # Create file
+    with h5py.File(uri, "w") as f:
+        f.create_dataset("mydataset", (100,), dtype="i")
+    # Check MD5 not found error
+    with pytest.raises(RuntimeError, match="not found in HDF5 file at:"):
+        storage._read_data(md5="md5")
+
+
 def test_store_metadata_and_list_features(tmp_path: Path) -> None:
     """Test metadata store and features listing.
 
@@ -272,64 +304,6 @@ def test_read_df(tmp_path: Path) -> None:
     )
 
 
-def test_store_data_ignore_duplicate(tmp_path: Path) -> None:
-    """Test duplicate ignore for data store.
-
-    Parameters
-    ----------
-    tmp_path : pathlib.Path
-        The path to the test directory.
-
-    """
-    uri = tmp_path / "test_duplicate_data_store.hdf5"
-    # Single storage, must be the uri
-    storage = HDF5FeatureStorage(uri=uri, single_output=True)
-    # Store data first time
-    storage._store_data(
-        kind="vector",
-        meta_md5="md5",
-        element=[{"sub": "001"}],
-        data=np.empty((1, 1)),
-    )
-    # Store data second time, should be ignored
-    storage._store_data(
-        kind="vector",
-        meta_md5="md5",
-        element=[{"sub": "001"}],
-        data=np.empty((1, 1)),
-    )
-
-
-def test_store_data_incorrect_kwargs(tmp_path: Path) -> None:
-    """Test incorrect kwargs for data store.
-
-    Parameters
-    ----------
-    tmp_path : pathlib.Path
-        The path to the test directory.
-
-    """
-    uri = tmp_path / "test_incorrect_kwargs_data_store.hdf5"
-    # Single storage, must be the uri
-    storage = HDF5FeatureStorage(uri=uri, single_output=True)
-    # Store data first time
-    storage._store_data(
-        kind="vector",
-        meta_md5="md5",
-        element=[{"sub": "001"}],
-        data=np.empty((1, 1)),
-    )
-    # Store data second time, should be ignored
-    with pytest.raises(RuntimeError, match="The additional data for"):
-        storage._store_data(
-            kind="vector",
-            meta_md5="md5",
-            element=[{"sub": "001"}],
-            data=np.empty((1, 1)),
-            col_names="col",
-        )
-
-
 @pytest.mark.parametrize(
     "force, dtype",
     [