[ENH]: Add scalar table storage support for HDF5FeatureStorage #343
5 changed files with 208 additions and 33 deletions
1
docs/changes/newsfragments/343.feature
Normal file
1
docs/changes/newsfragments/343.feature
Normal file
|
|
@ -0,0 +1 @@
|
|||
Introduce new storage type ``scalar_table`` and adapt :class:`.HDF5FeatureStorage` to support it by `Synchon Mandal`_
|
||||
|
|
@ -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 :ref:`Markers <marker>` are responsible for defining what *storage kind*
|
||||
(``matrix``, ``vector``, ``timeseries``) they support for which
|
||||
:ref:`data type <data_types>` by overriding its ``get_output_type`` method. The
|
||||
storage object in turn declares and provides implementation for specific
|
||||
*storage kind*. For example, :class:`.SQLiteFeatureStorage` supports saving
|
||||
``matrix``, ``vector`` and ``timeseries`` via ``store_matrix``, ``store_vector``
|
||||
and ``store_timeseries`` methods respectively.
|
||||
(``matrix``, ``vector``, ``timeseries``, ``scalar_table``) they support for
|
||||
which :ref:`data type <data_types>` by overriding its ``get_output_type``
|
||||
method. The storage object in turn declares and provides implementation for
|
||||
specific *storage kind*. For example, :class:`.SQLiteFeatureStorage` supports
|
||||
saving ``matrix``, ``vector`` and ``timeseries`` via ``store_matrix``,
|
||||
``store_vector`` and ``store_timeseries`` methods respectively.
|
||||
|
||||
For storage interfaces not supported by ``junifer`` yet, you can either make
|
||||
your own ``Storage`` by providing a concrete implementation of
|
||||
|
|
@ -44,17 +44,24 @@ Storage Types
|
|||
- Options
|
||||
- Reference
|
||||
* - ``matrix``
|
||||
- A 2D matrix with row and column names
|
||||
- ``col_names``, ``row_names``, ``matrix_kind``, ``diagonal``
|
||||
- A 2D square matrix with row and column names
|
||||
- | ``col_names``, ``row_names``, ``matrix_kind``, ``diagonal``
|
||||
| ``row_header_col_name``
|
||||
| (only for :meth:`.HDF5FeatureStorage.store_matrix`)
|
||||
- :meth:`.BaseFeatureStorage.store_matrix`
|
||||
* - ``vector``
|
||||
- A 1D row vector of values with column names
|
||||
- ``col_names``
|
||||
- :meth:`.BaseFeatureStorage.store_vector`
|
||||
* - ``timeseries``
|
||||
- A 2D matrix of values with column names
|
||||
- A 2D square or non-square matrix of scalar values with column names
|
||||
- ``col_names``
|
||||
- :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:
|
||||
|
||||
|
|
@ -76,4 +83,4 @@ Storage Interfaces
|
|||
* - :class:`.HDF5FeatureStorage`
|
||||
- ``.hdf5``
|
||||
- HDF5
|
||||
- ``matrix``, ``vector``, ``timeseries``
|
||||
- ``matrix``, ``vector``, ``timeseries``, ``scalar_table``
|
||||
|
|
|
|||
|
|
@ -189,7 +189,7 @@ class BaseFeatureStorage(ABC):
|
|||
|
||||
Parameters
|
||||
----------
|
||||
kind : {"matrix", "timeseries", "vector"}
|
||||
kind : {"matrix", "timeseries", "vector", "scalar_table"}
|
||||
The storage kind.
|
||||
**kwargs
|
||||
The keyword arguments.
|
||||
|
|
@ -218,6 +218,10 @@ class BaseFeatureStorage(ABC):
|
|||
)
|
||||
elif kind == "vector":
|
||||
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(
|
||||
self,
|
||||
|
|
@ -313,6 +317,38 @@ class BaseFeatureStorage(ABC):
|
|||
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
|
||||
def collect(self) -> None:
|
||||
"""Collect data."""
|
||||
|
|
|
|||
|
|
@ -56,7 +56,8 @@ def _create_chunk(
|
|||
Raises
|
||||
------
|
||||
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"]:
|
||||
|
|
@ -77,7 +78,7 @@ def _create_chunk(
|
|||
chunk_size=tuple(array_chunk_size),
|
||||
n_chunk=i_chunk,
|
||||
)
|
||||
elif kind == "timeseries":
|
||||
elif kind in ["timeseries", "scalar_table"]:
|
||||
out = ChunkedList(
|
||||
data=chunk_data,
|
||||
size=element_count,
|
||||
|
|
@ -86,7 +87,8 @@ def _create_chunk(
|
|||
else:
|
||||
raise_error(
|
||||
f"Invalid kind: {kind}. "
|
||||
"Must be one of ['vector', 'matrix', 'timeseries']."
|
||||
"Must be one of ['vector', 'matrix', 'timeseries',"
|
||||
"'scalar_table']."
|
||||
)
|
||||
return out
|
||||
|
||||
|
|
@ -146,7 +148,7 @@ class HDF5FeatureStorage(BaseFeatureStorage):
|
|||
uri.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# Available storage kinds
|
||||
storage_types = ["vector", "timeseries", "matrix"]
|
||||
storage_types = ["vector", "timeseries", "matrix", "scalar_table"]
|
||||
|
||||
super().__init__(
|
||||
uri=uri,
|
||||
|
|
@ -169,7 +171,7 @@ class HDF5FeatureStorage(BaseFeatureStorage):
|
|||
storage.
|
||||
|
||||
"""
|
||||
return ["matrix", "vector", "timeseries"]
|
||||
return ["matrix", "vector", "timeseries", "scalar_table"]
|
||||
|
||||
def _fetch_correct_uri_for_io(self, element: Optional[Dict]) -> str:
|
||||
"""Return proper URI for I/O based on `element`.
|
||||
|
|
@ -508,6 +510,26 @@ class HDF5FeatureStorage(BaseFeatureStorage):
|
|||
columns = hdf_data["column_headers"]
|
||||
# Convert data from 3D to 2D
|
||||
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
|
||||
idx_df = pd.DataFrame(data=element_idx) # type: ignore
|
||||
|
|
@ -643,7 +665,7 @@ class HDF5FeatureStorage(BaseFeatureStorage):
|
|||
|
||||
Parameters
|
||||
----------
|
||||
kind : {"matrix", "vector", "timeseries"}
|
||||
kind : {"matrix", "vector", "timeseries", "scalar_table"}
|
||||
The storage kind.
|
||||
meta_md5 : str
|
||||
The metadata MD5 hash.
|
||||
|
|
@ -739,8 +761,8 @@ class HDF5FeatureStorage(BaseFeatureStorage):
|
|||
)
|
||||
|
||||
t_data = stored_data["data"]
|
||||
if kind == "timeseries":
|
||||
t_data.append(data)
|
||||
if kind in ["timeseries", "scalar_table"]:
|
||||
t_data += data
|
||||
else:
|
||||
t_data = np.concatenate((t_data, data), axis=-1)
|
||||
# Existing entry; append to existing
|
||||
|
|
@ -921,6 +943,43 @@ class HDF5FeatureStorage(BaseFeatureStorage):
|
|||
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:
|
||||
"""Implement data collection.
|
||||
|
||||
|
|
@ -1029,7 +1088,7 @@ class HDF5FeatureStorage(BaseFeatureStorage):
|
|||
kind = static_data["kind"]
|
||||
|
||||
# Append the "dynamic" data
|
||||
if kind == "timeseries":
|
||||
if kind in ["timeseries", "scalar_table"]:
|
||||
chunk_data.extend(t_data["data"])
|
||||
else:
|
||||
chunk_data.append(t_data["data"])
|
||||
|
|
|
|||
|
|
@ -25,7 +25,12 @@ from junifer.storage.utils import (
|
|||
def test_get_valid_inputs() -> None:
|
||||
"""Test valid inputs."""
|
||||
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:
|
||||
|
|
@ -808,7 +813,7 @@ def test_store_timeseries(tmp_path: Path) -> None:
|
|||
data = np.array([[10], [20], [30], [40], [50]])
|
||||
col_names = ["signal"]
|
||||
|
||||
# Store vector
|
||||
# Store timeseries
|
||||
storage.store_timeseries(
|
||||
meta_md5=meta_md5,
|
||||
element=element_to_store,
|
||||
|
|
@ -822,6 +827,53 @@ def test_store_timeseries(tmp_path: Path) -> None:
|
|||
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]:
|
||||
"""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)],
|
||||
"matrix_kind": "full",
|
||||
}
|
||||
elif kind == "timeseries":
|
||||
elif kind in "timeseries":
|
||||
data_to_store = {
|
||||
"data": np.arange(20).reshape(2, 10),
|
||||
"col_names": [f"col-{i}" for i in range(10)],
|
||||
}
|
||||
else:
|
||||
raise ValueError(f"Unknown kind {kind}.")
|
||||
elif kind in "scalar_table":
|
||||
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):
|
||||
element = {"subject": f"sub-{i // 2}", "session": f"ses-{i % 2}"}
|
||||
meta = {
|
||||
|
|
@ -903,6 +961,7 @@ def _create_data_to_store(n_elements: int, kind: str) -> Tuple[str, Dict]:
|
|||
(10, 3, "matrix"),
|
||||
(10, 5, "matrix"),
|
||||
(10, 5, "timeseries"),
|
||||
(10, 5, "scalar_table"),
|
||||
],
|
||||
)
|
||||
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)
|
||||
|
||||
for t_data in all_data:
|
||||
# Store metadata for tables
|
||||
# Store metadata
|
||||
storage.store_metadata(
|
||||
meta_md5=meta_md5,
|
||||
element=t_data["element"],
|
||||
meta=t_data["meta"],
|
||||
)
|
||||
# Store data
|
||||
if kind == "vector":
|
||||
# Store tables
|
||||
storage.store_vector(
|
||||
meta_md5=meta_md5,
|
||||
element=t_data["element"],
|
||||
**t_data["data"],
|
||||
)
|
||||
elif kind == "matrix":
|
||||
# Store tables
|
||||
storage.store_matrix(
|
||||
meta_md5=meta_md5,
|
||||
element=t_data["element"],
|
||||
|
|
@ -956,11 +1014,17 @@ def test_multi_output_store_and_collect(
|
|||
element=t_data["element"],
|
||||
**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
|
||||
assert not uri.exists()
|
||||
|
||||
for t_data in all_data:
|
||||
# Convert element to preifx
|
||||
# Convert element to prefix
|
||||
prefix = element_to_prefix(t_data["element"])
|
||||
# URIs for data storage
|
||||
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
|
||||
assert uri.exists()
|
||||
|
||||
# # Read unified metadata
|
||||
# Read unified metadata
|
||||
read_unified_meta = storage.list_features()
|
||||
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])
|
||||
assert len(all_df) == data_size
|
||||
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:
|
||||
assert len(all_df) == len(all_data)
|
||||
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"])
|
||||
series_names = t_series.columns.values.tolist()
|
||||
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:
|
||||
|
|
|
|||
Loading…
Reference in a new issue