[ENH]: Add scalar table storage support for HDF5FeatureStorage #343

Merged
synchon merged 6 commits from feat/scalar-table-storage-type into main 2024-05-22 13:23:37 +00:00
5 changed files with 208 additions and 33 deletions

View file

@ -0,0 +1 @@
Introduce new storage type ``scalar_table`` and adapt :class:`.HDF5FeatureStorage` to support it by `Synchon Mandal`_

View file

@ -18,12 +18,12 @@ on them outside the context as long as the processed data is in the memory and
the Python runtime has not garbage-collected it. the Python runtime has not garbage-collected it.
The :ref:`Markers <marker>` are responsible for defining what *storage kind* The :ref:`Markers <marker>` are responsible for defining what *storage kind*
(``matrix``, ``vector``, ``timeseries``) they support for which (``matrix``, ``vector``, ``timeseries``, ``scalar_table``) they support for
:ref:`data type <data_types>` by overriding its ``get_output_type`` method. The which :ref:`data type <data_types>` by overriding its ``get_output_type``
storage object in turn declares and provides implementation for specific method. The storage object in turn declares and provides implementation for
*storage kind*. For example, :class:`.SQLiteFeatureStorage` supports saving specific *storage kind*. For example, :class:`.SQLiteFeatureStorage` supports
``matrix``, ``vector`` and ``timeseries`` via ``store_matrix``, ``store_vector`` saving ``matrix``, ``vector`` and ``timeseries`` via ``store_matrix``,
and ``store_timeseries`` methods respectively. ``store_vector`` and ``store_timeseries`` methods respectively.
For storage interfaces not supported by ``junifer`` yet, you can either make For storage interfaces not supported by ``junifer`` yet, you can either make
your own ``Storage`` by providing a concrete implementation of your own ``Storage`` by providing a concrete implementation of
@ -44,17 +44,24 @@ Storage Types
- Options - Options
- Reference - Reference
* - ``matrix`` * - ``matrix``
- A 2D matrix with row and column names - A 2D square matrix with row and column names
- ``col_names``, ``row_names``, ``matrix_kind``, ``diagonal`` - | ``col_names``, ``row_names``, ``matrix_kind``, ``diagonal``
| ``row_header_col_name``
| (only for :meth:`.HDF5FeatureStorage.store_matrix`)
- :meth:`.BaseFeatureStorage.store_matrix` - :meth:`.BaseFeatureStorage.store_matrix`
* - ``vector`` * - ``vector``
- A 1D row vector of values with column names - A 1D row vector of values with column names
- ``col_names`` - ``col_names``
- :meth:`.BaseFeatureStorage.store_vector` - :meth:`.BaseFeatureStorage.store_vector`
* - ``timeseries`` * - ``timeseries``
- A 2D matrix of values with column names - A 2D square or non-square matrix of scalar values with column names
- ``col_names`` - ``col_names``
- :meth:`.BaseFeatureStorage.store_timeseries` - :meth:`.BaseFeatureStorage.store_timeseries`
* - ``scalar_table``
- | A 2D square or non-square matrix of scalar values with row name, column
| name and row header column name
- ``col_names``, ``row_names``, ``row_header_col_name``
- :meth:`.BaseFeatureStorage.store_scalar_table`
.. _storage_interfaces: .. _storage_interfaces:
@ -76,4 +83,4 @@ Storage Interfaces
* - :class:`.HDF5FeatureStorage` * - :class:`.HDF5FeatureStorage`
- ``.hdf5`` - ``.hdf5``
- HDF5 - HDF5
- ``matrix``, ``vector``, ``timeseries`` - ``matrix``, ``vector``, ``timeseries``, ``scalar_table``

View file

@ -189,7 +189,7 @@ class BaseFeatureStorage(ABC):
Parameters Parameters
---------- ----------
kind : {"matrix", "timeseries", "vector"} kind : {"matrix", "timeseries", "vector", "scalar_table"}
The storage kind. The storage kind.
**kwargs **kwargs
The keyword arguments. The keyword arguments.
@ -218,6 +218,10 @@ class BaseFeatureStorage(ABC):
) )
elif kind == "vector": elif kind == "vector":
self.store_vector(meta_md5=meta_md5, element=t_element, **kwargs) self.store_vector(meta_md5=meta_md5, element=t_element, **kwargs)
elif kind == "scalar_table":
self.store_scalar_table(
meta_md5=meta_md5, element=t_element, **kwargs
)
def store_matrix( def store_matrix(
self, self,
@ -313,6 +317,38 @@ class BaseFeatureStorage(ABC):
klass=NotImplementedError, klass=NotImplementedError,
) )
def store_scalar_table(
self,
meta_md5: str,
element: Dict,
data: np.ndarray,
col_names: Optional[Iterable[str]] = None,
row_names: Optional[Iterable[str]] = None,
row_header_col_name: Optional[str] = "feature",
) -> None:
"""Store table with scalar values.
Parameters
----------
meta_md5 : str
The metadata MD5 hash.
element : dict
The element as a dictionary.
data : numpy.ndarray
The timeseries data to store.
col_names : list or tuple of str, optional
The column labels (default None).
row_names : str, optional
The row labels (default None).
row_header_col_name : str, optional
The column name for the row header column (default "feature").
"""
raise_error(
msg="Concrete classes need to implement store_scalar_table().",
klass=NotImplementedError,
)
@abstractmethod @abstractmethod
def collect(self) -> None: def collect(self) -> None:
"""Collect data.""" """Collect data."""

View file

@ -56,7 +56,8 @@ def _create_chunk(
Raises Raises
------ ------
ValueError ValueError
If `kind` is not one of ['vector', 'matrix', 'timeseries']. If `kind` is not one of ['vector', 'matrix', 'timeseries',
'scalar_table'].
""" """
if kind in ["vector", "matrix"]: if kind in ["vector", "matrix"]:
@ -77,7 +78,7 @@ def _create_chunk(
chunk_size=tuple(array_chunk_size), chunk_size=tuple(array_chunk_size),
n_chunk=i_chunk, n_chunk=i_chunk,
) )
elif kind == "timeseries": elif kind in ["timeseries", "scalar_table"]:
out = ChunkedList( out = ChunkedList(
data=chunk_data, data=chunk_data,
size=element_count, size=element_count,
@ -86,7 +87,8 @@ def _create_chunk(
else: else:
raise_error( raise_error(
f"Invalid kind: {kind}. " f"Invalid kind: {kind}. "
"Must be one of ['vector', 'matrix', 'timeseries']." "Must be one of ['vector', 'matrix', 'timeseries',"
"'scalar_table']."
) )
return out return out
@ -146,7 +148,7 @@ class HDF5FeatureStorage(BaseFeatureStorage):
uri.parent.mkdir(parents=True, exist_ok=True) uri.parent.mkdir(parents=True, exist_ok=True)
# Available storage kinds # Available storage kinds
storage_types = ["vector", "timeseries", "matrix"] storage_types = ["vector", "timeseries", "matrix", "scalar_table"]
super().__init__( super().__init__(
uri=uri, uri=uri,
@ -169,7 +171,7 @@ class HDF5FeatureStorage(BaseFeatureStorage):
storage. storage.
""" """
return ["matrix", "vector", "timeseries"] return ["matrix", "vector", "timeseries", "scalar_table"]
def _fetch_correct_uri_for_io(self, element: Optional[Dict]) -> str: def _fetch_correct_uri_for_io(self, element: Optional[Dict]) -> str:
"""Return proper URI for I/O based on `element`. """Return proper URI for I/O based on `element`.
@ -508,6 +510,26 @@ class HDF5FeatureStorage(BaseFeatureStorage):
columns = hdf_data["column_headers"] columns = hdf_data["column_headers"]
# Convert data from 3D to 2D # Convert data from 3D to 2D
reshaped_data = np.concatenate(all_data, axis=0) reshaped_data = np.concatenate(all_data, axis=0)
elif hdf_data["kind"] == "scalar_table":
# Create dictionary for aggregating index data
element_idx = defaultdict(list)
all_data = []
for idx, element in enumerate(hdf_data["element"]):
# Get row count for the element
t_data = hdf_data["data"][idx]
all_data.append(t_data)
n_rows = len(hdf_data["row_headers"])
# Set rows for the index
for key, val in element.items():
element_idx[key].extend([val] * n_rows)
# Add extra column for row header column name
element_idx[hdf_data["row_header_column_name"]].extend(
hdf_data["row_headers"]
)
# Set column headers for dataframe
columns = hdf_data["column_headers"]
# Convert data from 3D to 2D
reshaped_data = np.concatenate(all_data, axis=0)
# Create dataframe for index # Create dataframe for index
idx_df = pd.DataFrame(data=element_idx) # type: ignore idx_df = pd.DataFrame(data=element_idx) # type: ignore
@ -643,7 +665,7 @@ class HDF5FeatureStorage(BaseFeatureStorage):
Parameters Parameters
---------- ----------
kind : {"matrix", "vector", "timeseries"} kind : {"matrix", "vector", "timeseries", "scalar_table"}
The storage kind. The storage kind.
meta_md5 : str meta_md5 : str
The metadata MD5 hash. The metadata MD5 hash.
@ -739,8 +761,8 @@ class HDF5FeatureStorage(BaseFeatureStorage):
) )
t_data = stored_data["data"] t_data = stored_data["data"]
if kind == "timeseries": if kind in ["timeseries", "scalar_table"]:
t_data.append(data) t_data += data
else: else:
t_data = np.concatenate((t_data, data), axis=-1) t_data = np.concatenate((t_data, data), axis=-1)
# Existing entry; append to existing # Existing entry; append to existing
@ -921,6 +943,43 @@ class HDF5FeatureStorage(BaseFeatureStorage):
row_header_column_name="timepoint", row_header_column_name="timepoint",
) )
def store_scalar_table(
self,
meta_md5: str,
element: Dict,
data: np.ndarray,
col_names: Optional[Iterable[str]] = None,
row_names: Optional[Iterable[str]] = None,
row_header_col_name: Optional[str] = "feature",
) -> None:
"""Store table with scalar values.
Parameters
----------
meta_md5 : str
The metadata MD5 hash.
element : dict
The element as a dictionary.
data : numpy.ndarray
The scalar table data to store.
col_names : list or tuple of str, optional
The column labels (default None).
row_names : str, optional
The row labels (default None).
row_header_col_name : str, optional
The column name for the row header column (default "feature").
"""
self._store_data(
kind="scalar_table",
meta_md5=meta_md5,
element=[element], # convert to list
data=[data], # convert to list
column_headers=col_names,
row_headers=row_names,
row_header_column_name=row_header_col_name,
)
def collect(self) -> None: def collect(self) -> None:
"""Implement data collection. """Implement data collection.
@ -1029,7 +1088,7 @@ class HDF5FeatureStorage(BaseFeatureStorage):
kind = static_data["kind"] kind = static_data["kind"]
# Append the "dynamic" data # Append the "dynamic" data
if kind == "timeseries": if kind in ["timeseries", "scalar_table"]:
chunk_data.extend(t_data["data"]) chunk_data.extend(t_data["data"])
else: else:
chunk_data.append(t_data["data"]) chunk_data.append(t_data["data"])

View file

@ -25,7 +25,12 @@ from junifer.storage.utils import (
def test_get_valid_inputs() -> None: def test_get_valid_inputs() -> None:
"""Test valid inputs.""" """Test valid inputs."""
storage = HDF5FeatureStorage(uri="/tmp") storage = HDF5FeatureStorage(uri="/tmp")
assert storage.get_valid_inputs() == ["matrix", "vector", "timeseries"] assert storage.get_valid_inputs() == [
"matrix",
"vector",
"timeseries",
"scalar_table",
]
def test_single_output(tmp_path: Path) -> None: def test_single_output(tmp_path: Path) -> None:
@ -808,7 +813,7 @@ def test_store_timeseries(tmp_path: Path) -> None:
data = np.array([[10], [20], [30], [40], [50]]) data = np.array([[10], [20], [30], [40], [50]])
col_names = ["signal"] col_names = ["signal"]
# Store vector # Store timeseries
storage.store_timeseries( storage.store_timeseries(
meta_md5=meta_md5, meta_md5=meta_md5,
element=element_to_store, element=element_to_store,
@ -822,6 +827,53 @@ def test_store_timeseries(tmp_path: Path) -> None:
assert_array_equal(read_df.values, data) assert_array_equal(read_df.values, data)
def test_store_scalar_table(tmp_path: Path) -> None:
"""Test scalar table store.
Parameters
----------
tmp_path : pathlib.Path
The path to the test directory.
"""
uri = tmp_path / "test_store_scalar_table.hdf5"
storage = HDF5FeatureStorage(uri=uri)
# Metadata to store
element = {"subject": "test"}
meta = {
"element": element,
"dependencies": ["numpy"],
"marker": {"name": "brainprint"},
"type": "FreeSurfer",
}
# Process the metadata
meta_md5, meta_to_store, element_to_store = process_meta(meta)
# Store metadata
storage.store_metadata(
meta_md5=meta_md5, element=element_to_store, meta=meta_to_store
)
# Data to store
data = np.array([[10, 20], [30, 40], [50, 60]])
col_names = ["roi1", "roi2"]
row_names = ["ev1", "ev2", "ev3"]
# Store timeseries
storage.store_scalar_table(
meta_md5=meta_md5,
element=element_to_store,
data=data,
col_names=col_names,
row_names=row_names,
row_header_col_name="eigenvalue",
)
# Read into dataframe
read_df = storage.read_df(feature_md5=meta_md5)
# Check if data are equal
assert_array_equal(read_df.values, data)
def _create_data_to_store(n_elements: int, kind: str) -> Tuple[str, Dict]: def _create_data_to_store(n_elements: int, kind: str) -> Tuple[str, Dict]:
"""Create data to store. """Create data to store.
@ -854,13 +906,19 @@ def _create_data_to_store(n_elements: int, kind: str) -> Tuple[str, Dict]:
"col_names": [f"col-{i}" for i in range(10)], "col_names": [f"col-{i}" for i in range(10)],
"matrix_kind": "full", "matrix_kind": "full",
} }
elif kind == "timeseries": elif kind in "timeseries":
data_to_store = { data_to_store = {
"data": np.arange(20).reshape(2, 10), "data": np.arange(20).reshape(2, 10),
"col_names": [f"col-{i}" for i in range(10)], "col_names": [f"col-{i}" for i in range(10)],
} }
else: elif kind in "scalar_table":
raise ValueError(f"Unknown kind {kind}.") data_to_store = {
"data": np.arange(50).reshape(5, 10),
"row_names": [f"row-{i}" for i in range(5)],
"col_names": [f"col-{i}" for i in range(10)],
"row_header_col_name": "row",
}
for i in range(n_elements): for i in range(n_elements):
element = {"subject": f"sub-{i // 2}", "session": f"ses-{i % 2}"} element = {"subject": f"sub-{i // 2}", "session": f"ses-{i % 2}"}
meta = { meta = {
@ -903,6 +961,7 @@ def _create_data_to_store(n_elements: int, kind: str) -> Tuple[str, Dict]:
(10, 3, "matrix"), (10, 3, "matrix"),
(10, 5, "matrix"), (10, 5, "matrix"),
(10, 5, "timeseries"), (10, 5, "timeseries"),
(10, 5, "scalar_table"),
], ],
) )
def test_multi_output_store_and_collect( def test_multi_output_store_and_collect(
@ -930,21 +989,20 @@ def test_multi_output_store_and_collect(
meta_md5, all_data = _create_data_to_store(n_elements, kind) meta_md5, all_data = _create_data_to_store(n_elements, kind)
for t_data in all_data: for t_data in all_data:
# Store metadata for tables # Store metadata
storage.store_metadata( storage.store_metadata(
meta_md5=meta_md5, meta_md5=meta_md5,
element=t_data["element"], element=t_data["element"],
meta=t_data["meta"], meta=t_data["meta"],
) )
# Store data
if kind == "vector": if kind == "vector":
# Store tables
storage.store_vector( storage.store_vector(
meta_md5=meta_md5, meta_md5=meta_md5,
element=t_data["element"], element=t_data["element"],
**t_data["data"], **t_data["data"],
) )
elif kind == "matrix": elif kind == "matrix":
# Store tables
storage.store_matrix( storage.store_matrix(
meta_md5=meta_md5, meta_md5=meta_md5,
element=t_data["element"], element=t_data["element"],
@ -956,11 +1014,17 @@ def test_multi_output_store_and_collect(
element=t_data["element"], element=t_data["element"],
**t_data["data"], **t_data["data"],
) )
elif kind == "scalar_table":
storage.store_scalar_table(
meta_md5=meta_md5,
element=t_data["element"],
**t_data["data"],
)
# Check that base URI does not exist yet # Check that base URI does not exist yet
assert not uri.exists() assert not uri.exists()
for t_data in all_data: for t_data in all_data:
# Convert element to preifx # Convert element to prefix
prefix = element_to_prefix(t_data["element"]) prefix = element_to_prefix(t_data["element"])
# URIs for data storage # URIs for data storage
elem_uri = uri.parent / f"{prefix}{uri.name}" elem_uri = uri.parent / f"{prefix}{uri.name}"
@ -977,7 +1041,7 @@ def test_multi_output_store_and_collect(
# Check that base URI exists now # Check that base URI exists now
assert uri.exists() assert uri.exists()
# # Read unified metadata # Read unified metadata
read_unified_meta = storage.list_features() read_unified_meta = storage.list_features()
assert meta_md5 in read_unified_meta assert meta_md5 in read_unified_meta
@ -989,6 +1053,10 @@ def test_multi_output_store_and_collect(
data_size = np.sum([x["data"]["data"].shape[0] for x in all_data]) data_size = np.sum([x["data"]["data"].shape[0] for x in all_data])
assert len(all_df) == data_size assert len(all_df) == data_size
idx_names = [x for x in all_df.index.names if x != "timepoint"] idx_names = [x for x in all_df.index.names if x != "timepoint"]
elif kind == "scalar_table":
data_size = np.sum([x["data"]["data"].shape[0] for x in all_data])
assert len(all_df) == data_size
idx_names = [x for x in all_df.index.names if x != "row"]
else: else:
assert len(all_df) == len(all_data) assert len(all_df) == len(all_data)
idx_names = all_df.index.names idx_names = all_df.index.names
@ -1013,6 +1081,10 @@ def test_multi_output_store_and_collect(
assert_array_equal(t_series.values, t_data["data"]["data"]) assert_array_equal(t_series.values, t_data["data"]["data"])
series_names = t_series.columns.values.tolist() series_names = t_series.columns.values.tolist()
assert series_names == t_data["data"]["col_names"] assert series_names == t_data["data"]["col_names"]
elif kind == "scalar_table":
assert_array_equal(t_series.values, t_data["data"]["data"])
series_names = t_series.columns.values.tolist()
assert series_names == t_data["data"]["col_names"]
def test_collect_error_single_output() -> None: def test_collect_error_single_output() -> None: