[ENH]: Simplify Storage interface #472

Merged
synchon merged 24 commits from refactor/storage into main 2025-11-07 11:57:58 +00:00
10 changed files with 230 additions and 251 deletions

View file

@ -0,0 +1 @@
Simplify ``Storage`` interface and implementations by `Synchon Mandal`_

View file

@ -17,11 +17,11 @@ Storage is meant to be used inside the DataGrabber context but you can operate
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``, ``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
The :ref:`Markers <marker>` are responsible for mapping the input
:ref:`data type <data_types>` to its output :ref:`storage type <storage_types>`
as shown :ref:`here <extending_markers_input_output>`.
The storage object in turn declares and provides implementation for
specific *storage type*. For example, :class:`.SQLiteFeatureStorage` supports
saving ``matrix``, ``vector`` and ``timeseries`` via ``store_matrix``,
``store_vector`` and ``store_timeseries`` methods respectively.
@ -57,6 +57,10 @@ Storage Types
- A 2D square or non-square matrix of scalar values with column names
- ``col_names``
- :meth:`.BaseFeatureStorage.store_timeseries`
* - ``timeseries_2d``
- A 3D(2D+1D) square or non-square matrix of scalar values with column names across sessions
- ``col_names``
- :meth:`.BaseFeatureStorage.store_timeseries_2d`
* - ``scalar_table``
- | A 2D square or non-square matrix of scalar values with row name, column
| name and row header column name
@ -83,4 +87,4 @@ Storage Interfaces
* - :class:`.HDF5FeatureStorage`
- ``.hdf5``
- HDF5
- ``matrix``, ``vector``, ``timeseries``, ``scalar_table``
- ``matrix``, ``vector``, ``timeseries``, ``timeseries_2d``, ``scalar_table``

View file

@ -5,14 +5,14 @@
# License: AGPL
from abc import ABC, abstractmethod
from collections.abc import Iterable
from collections.abc import Sequence
from pathlib import Path
from typing import Any, Optional, Union
from typing import Any, ClassVar, Optional, Union
import numpy as np
import pandas as pd
from ..utils import raise_error
from ..utils import logger, raise_error
from .utils import process_meta
@ -22,44 +22,47 @@ __all__ = ["BaseFeatureStorage"]
class BaseFeatureStorage(ABC):
"""Abstract base class for feature storage.
For every interface that is required, one needs to provide a concrete
For every storage, one needs to provide a concrete
implementation of this abstract class.
Parameters
----------
uri : str or pathlib.Path
The path to the storage.
storage_types : str or list of str
The available storage types for the class.
single_output : bool, optional
Whether to have single output (default True).
Raises
------
ValueError
If required storage type(s) is(are) missing from ``storage_types``.
AttributeError
If the storage does not have `_STORAGE_TYPES` attribute.
"""
_STORAGE_TYPES: ClassVar[Sequence[str]]
def __init__(
self,
uri: Union[str, Path],
storage_types: Union[list[str], str],
single_output: bool = True,
) -> None:
self.uri = uri
# Convert storage_types to list
if not isinstance(storage_types, list):
storage_types = [storage_types]
# Check if required inputs are found
if any(x not in self.get_valid_inputs() for x in storage_types):
wrong_storage_types = [
x for x in storage_types if x not in self.get_valid_inputs()
]
# Check for missing storage types attribute
if not hasattr(self, "_STORAGE_TYPES"):
raise_error(
f"{self.__class__.__name__} cannot store {wrong_storage_types}"
msg="Missing `_STORAGE_TYPES` for the storage",
klass=AttributeError,
)
self._valid_inputs = storage_types
# Convert str to Path
if not isinstance(uri, Path):
uri = Path(uri)
self.uri = uri
# Create parent directories if not present
if not self.uri.parent.exists():
logger.info(
f"Output directory: '{self.uri.parent.resolve()}' "
"does not exist, creating now"
)
self.uri.parent.mkdir(parents=True, exist_ok=True)
self.single_output = single_output
def get_valid_inputs(self) -> list[str]:
@ -69,13 +72,10 @@ class BaseFeatureStorage(ABC):
-------
list of str
The list of storage types that can be used as input for this
storage interface.
storage.
"""
raise_error(
msg="Concrete classes need to implement get_valid_inputs().",
klass=NotImplementedError,
)
return list(self._STORAGE_TYPES)
def validate(self, input_: list[str]) -> None:
"""Validate the input to the pipeline step.
@ -91,11 +91,12 @@ class BaseFeatureStorage(ABC):
If the ``input_`` is invalid.
"""
if not any(x in input_ for x in self._valid_inputs):
# self._STORAGE_TYPES should be there already
if not any(x in input_ for x in self._STORAGE_TYPES):
raise_error(
"Input does not have the required data."
f"\t Input: {input}"
f"\t Required (any of): {self._valid_inputs}"
f"\t Required (any of): {self._STORAGE_TYPES}"
)
@abstractmethod
@ -113,7 +114,7 @@ class BaseFeatureStorage(ABC):
raise_error(
msg="Concrete classes need to implement list_features().",
klass=NotImplementedError,
)
) # pragma: no cover
@abstractmethod
def read(
@ -141,7 +142,7 @@ class BaseFeatureStorage(ABC):
raise_error(
msg="Concrete classes need to implement read().",
klass=NotImplementedError,
)
) # pragma: no cover
@abstractmethod
def read_df(
@ -167,7 +168,7 @@ class BaseFeatureStorage(ABC):
raise_error(
msg="Concrete classes need to implement read_df().",
klass=NotImplementedError,
)
) # pragma: no cover
@abstractmethod
def store_metadata(self, meta_md5: str, element: dict, meta: dict) -> None:
@ -186,7 +187,7 @@ class BaseFeatureStorage(ABC):
raise_error(
msg="Concrete classes need to implement store_metadata().",
klass=NotImplementedError,
)
) # pragma: no cover
def store(self, kind: str, **kwargs) -> None:
"""Store extracted features data.
@ -206,14 +207,16 @@ class BaseFeatureStorage(ABC):
"""
# Do the check before calling the abstract methods, otherwise the
# meta might be stored even if the data is not stored.
if kind not in self._valid_inputs:
if kind not in self._STORAGE_TYPES:
raise_error(
msg=f"I don't know how to store {kind}.",
klass=ValueError,
)
# Process and store metadata
t_meta = kwargs.pop("meta")
meta_md5, t_meta, t_element = process_meta(t_meta)
self.store_metadata(meta_md5=meta_md5, element=t_element, meta=t_meta)
# Store data
if kind == "matrix":
self.store_matrix(meta_md5=meta_md5, element=t_element, **kwargs)
elif kind == "timeseries":
@ -236,8 +239,8 @@ class BaseFeatureStorage(ABC):
meta_md5: str,
element: dict,
data: np.ndarray,
col_names: Optional[Iterable[str]] = None,
row_names: Optional[Iterable[str]] = None,
col_names: Optional[Sequence[str]] = None,
row_names: Optional[Sequence[str]] = None,
matrix_kind: str = "full",
diagonal: bool = True,
) -> None:
@ -251,9 +254,9 @@ class BaseFeatureStorage(ABC):
The element as a dictionary.
data : numpy.ndarray
The matrix data to store.
col_names : list or tuple of str, optional
col_names : list-like of str, optional
The column labels (default None).
row_names : str, optional
row_names : list-like of str, optional
The row labels (default None).
matrix_kind : str, optional
The kind of matrix:
@ -278,7 +281,7 @@ class BaseFeatureStorage(ABC):
meta_md5: str,
element: dict,
data: Union[np.ndarray, list],
col_names: Optional[Iterable[str]] = None,
col_names: Optional[Sequence[str]] = None,
) -> None:
"""Store vector.
@ -290,7 +293,7 @@ class BaseFeatureStorage(ABC):
The element as a dictionary.
data : numpy.ndarray or list
The vector data to store.
col_names : list or tuple of str, optional
col_names : list-like of str, optional
The column labels (default None).
"""
@ -304,7 +307,7 @@ class BaseFeatureStorage(ABC):
meta_md5: str,
element: dict,
data: np.ndarray,
col_names: Optional[Iterable[str]] = None,
col_names: Optional[Sequence[str]] = None,
) -> None:
"""Store timeseries.
@ -316,7 +319,7 @@ class BaseFeatureStorage(ABC):
The element as a dictionary.
data : numpy.ndarray
The timeseries data to store.
col_names : list or tuple of str, optional
col_names : list-like of str, optional
The column labels (default None).
"""
@ -330,8 +333,8 @@ class BaseFeatureStorage(ABC):
meta_md5: str,
element: dict,
data: np.ndarray,
col_names: Optional[Iterable[str]] = None,
row_names: Optional[Iterable[str]] = None,
col_names: Optional[Sequence[str]] = None,
row_names: Optional[Sequence[str]] = None,
) -> None:
"""Store 2D timeseries.
@ -343,9 +346,9 @@ class BaseFeatureStorage(ABC):
The element as a dictionary.
data : numpy.ndarray
The timeseries data to store.
col_names : list or tuple of str, optional
col_names : list-like of str, optional
The column labels (default None).
row_names : list or tuple of str, optional
row_names : list-like of str, optional
The row labels (default None).
"""
@ -359,8 +362,8 @@ class BaseFeatureStorage(ABC):
meta_md5: str,
element: dict,
data: np.ndarray,
col_names: Optional[Iterable[str]] = None,
row_names: Optional[Iterable[str]] = None,
col_names: Optional[Sequence[str]] = None,
row_names: Optional[Sequence[str]] = None,
row_header_col_name: Optional[str] = "feature",
) -> None:
"""Store table with scalar values.
@ -373,9 +376,9 @@ class BaseFeatureStorage(ABC):
The element as a dictionary.
data : numpy.ndarray
The timeseries data to store.
col_names : list or tuple of str, optional
col_names : list-like of str, optional
The column labels (default None).
row_names : str, optional
row_names : list-like of str, optional
The row labels (default None).
row_header_col_name : str, optional
The column name for the row header column (default "feature").
@ -392,7 +395,7 @@ class BaseFeatureStorage(ABC):
raise_error(
msg="Concrete classes need to implement collect().",
klass=NotImplementedError,
)
) # pragma: no cover
def __str__(self) -> str:
"""Represent object as string.

View file

@ -5,9 +5,9 @@
# License: AGPL
from collections import defaultdict
from collections.abc import Iterable
from collections.abc import Sequence
from pathlib import Path
from typing import Any, Optional, Union
from typing import Any, ClassVar, Optional, Union
import numpy as np
import pandas as pd
@ -135,6 +135,14 @@ class HDF5FeatureStorage(BaseFeatureStorage):
"""
_STORAGE_TYPES: ClassVar[Sequence[str]] = [
"vector",
"timeseries",
"matrix",
"scalar_table",
"timeseries_2d",
]
def __init__(
self,
uri: Union[str, Path],
@ -144,60 +152,19 @@ class HDF5FeatureStorage(BaseFeatureStorage):
force_float32: bool = True,
chunk_size: int = 100,
) -> None:
# Convert str to Path
if not isinstance(uri, Path):
uri = Path(uri)
# Create parent directories if not present
if not uri.parent.exists():
logger.info(
f"Output directory: '{uri.parent.resolve()}' "
"does not exist, creating now"
)
uri.parent.mkdir(parents=True, exist_ok=True)
# Available storage kinds
storage_types = [
"vector",
"timeseries",
"matrix",
"scalar_table",
"timeseries_2d",
]
super().__init__(
uri=uri,
storage_types=storage_types,
single_output=single_output,
)
self.overwrite = overwrite
self.compression = compression
self.force_float32 = force_float32
self.chunk_size = chunk_size
def get_valid_inputs(self) -> list[str]:
"""Get valid storage types for input.
Returns
-------
list of str
The list of storage types that can be used as input for this
storage.
"""
return [
"matrix",
"vector",
"timeseries",
"scalar_table",
"timeseries_2d",
]
super().__init__(
uri=uri,
single_output=single_output,
)
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``.
If `element` is None, will return `self.uri`.
If ``element`` is None, will return ``self.uri``.
Parameters
----------
@ -209,22 +176,25 @@ class HDF5FeatureStorage(BaseFeatureStorage):
str
Formatted URI for accessing metadata and data.
Raises
------
RuntimeError
If ``element=None`` when ``single_output=False``.
"""
if not self.single_output and not element:
if not self.single_output and element is None:
raise_error(
msg=(
"`element` must be provided when `single_output` is False"
),
msg="`element` must be provided when `single_output=False`",
klass=RuntimeError,
)
elif not self.single_output and element:
elif not self.single_output and element is not None:
# element access for multi output only
prefix = element_to_prefix(element=element)
else:
# parent access for single output, ignore element
prefix = ""
# Format URI based on prefix
return f"{self.uri.parent}/{prefix}{self.uri.name}" # type: ignore
return f"{self.uri.parent}/{prefix}{self.uri.name}"
def _read_metadata(
self, element: Optional[dict[str, str]] = None
@ -291,7 +261,7 @@ class HDF5FeatureStorage(BaseFeatureStorage):
"""
# Read metadata
metadata = read_hdf5(
fname=str(self.uri.resolve()), # type: ignore
fname=str(self.uri.resolve()),
title="meta",
slash="ignore",
)
@ -374,8 +344,14 @@ class HDF5FeatureStorage(BaseFeatureStorage):
Raises
------
ValueError
If both ``feature_md5`` and ``feature_name`` are provided or
if none of ``feature_md5`` or ``feature_name`` is provided.
IOError
If HDF5 file does not exist.
RuntimeError
If feature is not found or
if duplicate feature is found with the same name.
"""
# Parameter conflict
@ -396,7 +372,7 @@ class HDF5FeatureStorage(BaseFeatureStorage):
)
# Parameter check pass; read metadata
metadata = read_hdf5(
fname=str(self.uri.resolve()), # type: ignore
fname=str(self.uri.resolve()),
title="meta",
slash="ignore",
)
@ -411,15 +387,18 @@ class HDF5FeatureStorage(BaseFeatureStorage):
)
# Validate MD5
if feature_md5 in metadata:
md5 = feature_md5 # type: ignore
md5 = feature_md5
else:
raise_error(msg=f"Feature MD5 '{feature_md5}' not found")
raise_error(
msg=f"Feature MD5 '{feature_md5}' not found",
klass=RuntimeError,
)
# Consider feature_name
elif feature_name:
logger.debug(
f"Validating feature name '{feature_name}' in metadata "
f"for: {self.uri.resolve()} ..." # type: ignore
f"for: {self.uri.resolve()} ..."
)
# Retrieve MD5 for feature_name
# Implicit counter for duplicate feature_name with different
@ -445,7 +424,7 @@ class HDF5FeatureStorage(BaseFeatureStorage):
# Read data from HDF5
hdf_data = read_hdf5(
fname=str(self.uri.resolve()), # type: ignore
fname=str(self.uri.resolve()),
title=md5,
slash="ignore",
)
@ -573,7 +552,7 @@ class HDF5FeatureStorage(BaseFeatureStorage):
reshaped_data = np.concatenate(all_data, axis=0)
# Create dataframe for index
idx_df = pd.DataFrame(data=element_idx) # type: ignore
idx_df = pd.DataFrame(data=element_idx)
# Create multiindex from dataframe
hdf_data_idx = pd.MultiIndex.from_frame(df=idx_df)
logger.debug(
@ -589,7 +568,7 @@ class HDF5FeatureStorage(BaseFeatureStorage):
df = pd.DataFrame(
data=reshaped_data,
index=hdf_data_idx,
columns=columns, # type: ignore
columns=columns,
dtype=reshaped_data.dtype,
)
logger.debug(
@ -621,7 +600,7 @@ class HDF5FeatureStorage(BaseFeatureStorage):
write_hdf5(
fname=fname,
data=processed_data,
overwrite=self.overwrite, # type: ignore
overwrite=self.overwrite,
compression=self.compression,
title=title,
slash="error",
@ -837,8 +816,8 @@ class HDF5FeatureStorage(BaseFeatureStorage):
meta_md5: str,
element: dict[str, str],
data: np.ndarray,
col_names: Optional[Iterable[str]] = None,
row_names: Optional[Iterable[str]] = None,
col_names: Optional[Sequence[str]] = None,
row_names: Optional[Sequence[str]] = None,
matrix_kind: str = "full",
diagonal: bool = True,
row_header_col_name: str = "ROI",
@ -856,9 +835,9 @@ class HDF5FeatureStorage(BaseFeatureStorage):
The element as dictionary.
data : numpy.ndarray
The matrix data to store.
col_names : list or tuple of str, optional
col_names : list-like of str, optional
The column labels (default None).
row_names : list or tuple of str, optional
row_names : list-like of str, optional
The row labels (default None).
matrix_kind : str, optional
The kind of matrix:
@ -895,8 +874,8 @@ class HDF5FeatureStorage(BaseFeatureStorage):
matrix_kind=matrix_kind,
diagonal=diagonal,
data_shape=data.shape,
row_names_len=len(row_names), # type: ignore
col_names_len=len(col_names), # type: ignore
row_names_len=len(row_names),
col_names_len=len(col_names),
)
# Store
self._store_data(
@ -914,9 +893,9 @@ class HDF5FeatureStorage(BaseFeatureStorage):
def store_vector(
self,
meta_md5: str,
element: dict[str, str],
element: dict,
data: Union[np.ndarray, list],
col_names: Optional[Iterable[str]] = None,
col_names: Optional[Sequence[str]] = None,
) -> None:
"""Store vector.
@ -928,7 +907,7 @@ class HDF5FeatureStorage(BaseFeatureStorage):
The element as dictionary.
data : numpy.ndarray or list
The vector data to store.
col_names : list or tuple of str, optional
col_names : list-like of str, optional
The column labels (default None).
"""
@ -957,9 +936,9 @@ class HDF5FeatureStorage(BaseFeatureStorage):
def store_timeseries(
self,
meta_md5: str,
element: dict[str, str],
element: dict,
data: np.ndarray,
col_names: Optional[Iterable[str]] = None,
col_names: Optional[Sequence[str]] = None,
) -> None:
"""Store timeseries.
@ -971,7 +950,7 @@ class HDF5FeatureStorage(BaseFeatureStorage):
The element as dictionary.
data : numpy.ndarray
The timeseries data to store.
col_names : list or tuple of str, optional
col_names : list-like of str, optional
The column labels (default None).
"""
@ -987,31 +966,31 @@ class HDF5FeatureStorage(BaseFeatureStorage):
def store_timeseries_2d(
self,
meta_md5: str,
element: dict[str, str],
element: dict,
data: np.ndarray,
col_names: Optional[Iterable[str]] = None,
row_names: Optional[Iterable[str]] = None,
col_names: Optional[Sequence[str]] = None,
row_names: Optional[Sequence[str]] = None,
) -> None:
"""Store a 2D timeseries.
"""Store 2D timeseries.
Parameters
----------
meta_md5 : str
The metadata MD5 hash.
element : dict
The element as dictionary.
The element as a dictionary.
data : numpy.ndarray
The 2D timeseries data to store.
col_names : list or tuple of str, optional
col_names : list-like of str, optional
The column labels (default None).
row_names : list or tuple of str, optional
row_names : list-like of str, optional
The row labels (default None).
"""
store_timeseries_2d_checks(
data_shape=data.shape,
row_names_len=len(row_names), # type: ignore
col_names_len=len(col_names), # type: ignore
row_names_len=len(row_names) if row_names is not None else 0,
col_names_len=len(col_names) if col_names is not None else 0,
)
self._store_data(
kind="timeseries_2d",
@ -1027,8 +1006,8 @@ class HDF5FeatureStorage(BaseFeatureStorage):
meta_md5: str,
element: dict,
data: np.ndarray,
col_names: Optional[Iterable[str]] = None,
row_names: Optional[Iterable[str]] = None,
col_names: Optional[Sequence[str]] = None,
row_names: Optional[Sequence[str]] = None,
row_header_col_name: Optional[str] = "feature",
) -> None:
"""Store table with scalar values.
@ -1041,9 +1020,9 @@ class HDF5FeatureStorage(BaseFeatureStorage):
The element as a dictionary.
data : numpy.ndarray
The scalar table data to store.
col_names : list or tuple of str, optional
col_names : list-like of str, optional
The column labels (default None).
row_names : str, optional
row_names : list-like of str, optional
The row labels (default None).
row_header_col_name : str, optional
The column name for the row header column (default "feature").
@ -1079,22 +1058,19 @@ class HDF5FeatureStorage(BaseFeatureStorage):
klass=NotImplementedError,
)
# Glob files
globbed_files = list(
self.uri.parent.glob(f"*_{self.uri.name}") # type: ignore
)
# Create new storage instance
out_storage = HDF5FeatureStorage(uri=self.uri, overwrite="update")
# Run loop to collect metadata
logger.info(
f"Collecting metadata from {self.uri.parent}/*_{self.uri.name}" # type: ignore
f"Collecting metadata from {self.uri.parent}/*_{self.uri.name}"
)
# Collect element files per feature MD5
elements_per_feature_md5 = defaultdict(list)
out_metadata = {}
for file_ in tqdm(globbed_files, desc="file-metadata"):
for file_ in tqdm(
self.uri.parent.glob(f"*_{self.uri.name}"), desc="file-metadata"
):
logger.debug(f"Reading HDF5 file: {file_} ...")
# Create new storage instance to load metadata
in_storage = HDF5FeatureStorage(uri=file_)
@ -1113,7 +1089,7 @@ class HDF5FeatureStorage(BaseFeatureStorage):
logger.info("Writing metadata to HDF5 file ...")
# Save metadata out metadata
out_storage._write_processed_data(
fname=str(self.uri.resolve()), # type: ignore
fname=str(self.uri.resolve()),
processed_data=out_metadata,
title="meta",
)
@ -1194,9 +1170,9 @@ class HDF5FeatureStorage(BaseFeatureStorage):
# Write to HDF5
write_hdf5(
fname=str(self.uri.resolve()), # type: ignore
fname=str(self.uri.resolve()),
data=to_write,
overwrite="update", # type: ignore
overwrite="update",
compression=0,
title=feature_md5,
slash="error",

View file

@ -5,9 +5,9 @@
# License: AGPL
import json
from collections.abc import Iterable
from collections.abc import Sequence
from pathlib import Path
from typing import Optional, Union
from typing import ClassVar, Optional, Union
import numpy as np
import pandas as pd
@ -31,8 +31,6 @@ class PandasBaseFeatureStorage(BaseFeatureStorage):
The path to the storage.
single_output : bool, optional
Whether to have single output (default True).
**kwargs
Keyword arguments passed to superclass.
See Also
--------
@ -40,22 +38,18 @@ class PandasBaseFeatureStorage(BaseFeatureStorage):
"""
_STORAGE_TYPES: ClassVar[Sequence[str]] = [
"vector",
"timeseries",
"matrix",
]
def __init__(
self, uri: Union[str, Path], single_output: bool = True, **kwargs
self,
uri: Union[str, Path],
single_output: bool = True,
) -> None:
super().__init__(uri=uri, single_output=single_output, **kwargs)
def get_valid_inputs(self) -> list[str]:
"""Get valid storage types for input.
Returns
-------
list of str
The list of storage types that can be used as input for this
storage interface.
"""
return ["matrix", "vector", "timeseries"]
super().__init__(uri=uri, single_output=single_output)
def _meta_row(self, meta: dict, meta_md5: str) -> pd.DataFrame:
"""Convert the metadata to a pandas DataFrame.
@ -102,7 +96,7 @@ class PandasBaseFeatureStorage(BaseFeatureStorage):
"""
# Make mapping between element access keys and values
elem_idx: dict[str, Iterable[str]] = {
elem_idx: dict[str, Sequence[str]] = {
k: [v] * n_rows for k, v in element.items()
}
@ -160,7 +154,7 @@ class PandasBaseFeatureStorage(BaseFeatureStorage):
meta_md5: str,
element: dict,
data: Union[np.ndarray, list],
col_names: Optional[Iterable[str]] = None,
col_names: Optional[Sequence[str]] = None,
rows_col_name: Optional[str] = None,
) -> None:
"""Store 2D data.
@ -173,7 +167,7 @@ class PandasBaseFeatureStorage(BaseFeatureStorage):
The element as a dictionary.
data : numpy.ndarray or list
The data to store.
col_names : list or tuple of str, optional
col_names : list-like of str, optional
The column labels (default None).
rows_col_name : str, optional
The column name to use in case number of rows greater than 1.
@ -199,7 +193,7 @@ class PandasBaseFeatureStorage(BaseFeatureStorage):
meta_md5: str,
element: dict,
data: Union[np.ndarray, list],
col_names: Optional[Iterable[str]] = None,
col_names: Optional[Sequence[str]] = None,
) -> None:
"""Store vector.
@ -211,7 +205,7 @@ class PandasBaseFeatureStorage(BaseFeatureStorage):
The element as a dictionary.
data : numpy.ndarray or list
The vector data to store.
col_names : list or tuple of str, optional
col_names : list-like of str, optional
The column labels (default None).
"""
@ -237,7 +231,7 @@ class PandasBaseFeatureStorage(BaseFeatureStorage):
meta_md5: str,
element: dict,
data: np.ndarray,
col_names: Optional[Iterable[str]] = None,
col_names: Optional[Sequence[str]] = None,
) -> None:
"""Store timeseries.
@ -249,7 +243,7 @@ class PandasBaseFeatureStorage(BaseFeatureStorage):
The element as a dictionary.
data : numpy.ndarray
The timeseries data to store.
col_names : list or tuple of str, optional
col_names : list-like of str, optional
The column labels (default None).
"""

View file

@ -5,13 +5,14 @@
# License: AGPL
import json
from collections.abc import Sequence
from pathlib import Path
from typing import TYPE_CHECKING, Any, Optional, Union
import numpy as np
import pandas as pd
from pandas.core.base import NoNewAttributesMixin
from pandas.io.sql import pandasSQL_builder # type: ignore
from pandas.io.sql import pandasSQL_builder
from sqlalchemy import create_engine, inspect
from tqdm import tqdm
@ -46,8 +47,6 @@ class SQLiteFeatureStorage(PandasBaseFeatureStorage):
upsert : {"ignore", "update"}, optional
Upsert mode. If "ignore" is used, the existing elements are ignored.
If "update", the existing elements are updated (default "update").
**kwargs : dict
The keyword arguments passed to the superclass.
See Also
--------
@ -61,9 +60,8 @@ class SQLiteFeatureStorage(PandasBaseFeatureStorage):
uri: Union[str, Path],
single_output: bool = True,
upsert: str = "update",
**kwargs: str,
) -> None:
# Check upsert argument value
# Check and set upsert argument value
if upsert not in ["update", "ignore"]:
raise_error(
msg=(
@ -71,26 +69,11 @@ class SQLiteFeatureStorage(PandasBaseFeatureStorage):
"Must be either 'update' or 'ignore'."
)
)
# Convert str to Path
if not isinstance(uri, Path):
uri = Path(uri)
# Create parent directories if not present
if not uri.parent.exists():
logger.info(
f"Output directory ({uri.parent.absolute()!s}) "
"does not exist, creating now."
)
uri.parent.mkdir(parents=True, exist_ok=True)
# Available storage kinds
storage_types = ["vector", "timeseries", "matrix"]
self.upsert = upsert
super().__init__(
uri=uri,
storage_types=storage_types,
single_output=single_output,
**kwargs,
)
# Set upsert
self._upsert = upsert
def get_engine(self, element: Optional[dict] = None) -> "Engine":
"""Get engine.
@ -105,19 +88,23 @@ class SQLiteFeatureStorage(PandasBaseFeatureStorage):
sqlalchemy.engine.Engine
The sqlalchemy engine.
Raises
------
ValueError
If ``element=None`` when ``single_output=False``.
"""
# Prefixed elements
prefix = ""
if self.single_output is False:
if element is None:
msg = "element must be specified when single_output is False."
raise_error(msg)
prefix = element_to_prefix(element)
raise_error(
"`element` cannot be None when `single_output=False`"
)
else:
prefix = element_to_prefix(element)
# Format URI for engine creation
uri = (
f"sqlite:///{self.uri.parent}/" # type: ignore
f"{prefix}{self.uri.name}" # type: ignore
)
uri = f"sqlite:///{self.uri.parent}/{prefix}{self.uri.name}"
return create_engine(uri, echo=False)
def _save_upsert(
@ -147,8 +134,8 @@ class SQLiteFeatureStorage(PandasBaseFeatureStorage):
Raises
------
ValueError
If the table exists and if_exists is "fail" or if invalid option is
passed to `if_exists`.
If the table exists and ``if_exists="fail"`` or
if invalid option is passed to ``if_exists``.
"""
# Get index names
@ -187,12 +174,12 @@ class SQLiteFeatureStorage(PandasBaseFeatureStorage):
warn_with_log(
f"Some rows (n={len(existing)}) are already "
"present in the database. The storage is "
f"configured to {self._upsert} the existing "
f"configured to {self.upsert} the existing "
f"elements. The new rows (n={len(new)}) will be "
"appended. This warning is shown because normally "
"all of the elements should be updated."
)
if self._upsert == "update":
if self.upsert == "update":
update_stmts = _generate_update_statements(
table, index_col, existing
)
@ -257,10 +244,14 @@ class SQLiteFeatureStorage(PandasBaseFeatureStorage):
dict
The stored feature as a dictionary.
Raises
------
NotImplementedError
"""
raise_error(
msg=(
"read() for SQLiteFeatureStorage is not currently planned to "
"SQLiteFeatureStorage.read() is not currently planned to "
"be implemented in the near future. If you need the "
"functionality, contact the junifer developers "
"(https://juaml.github.io/junifer/main/help.html)."
@ -434,8 +425,8 @@ class SQLiteFeatureStorage(PandasBaseFeatureStorage):
meta_md5: str,
element: dict,
data: np.ndarray,
col_names: Optional[list[str]] = None,
row_names: Optional[list[str]] = None,
col_names: Optional[Sequence[str]] = None,
row_names: Optional[Sequence[str]] = None,
matrix_kind: str = "full",
diagonal: bool = True,
) -> None:
@ -451,9 +442,9 @@ class SQLiteFeatureStorage(PandasBaseFeatureStorage):
The matrix data to store.
meta : dict
The metadata as a dictionary.
col_names : list or tuple of str, optional
col_names : list-like of str, optional
The column labels (default None).
row_names : str, optional
row_names : list-like of str, optional
The row labels (optional None).
matrix_kind : str, optional
The kind of matrix:
@ -479,8 +470,8 @@ class SQLiteFeatureStorage(PandasBaseFeatureStorage):
matrix_kind=matrix_kind,
diagonal=diagonal,
data_shape=data.shape,
row_names_len=len(row_names), # type: ignore
col_names_len=len(col_names), # type: ignore
row_names_len=len(row_names),
col_names_len=len(col_names),
)
# Matrix to vector conversion
flat_data, columns = matrix_to_vector(
@ -533,13 +524,11 @@ class SQLiteFeatureStorage(PandasBaseFeatureStorage):
msg="collect() is not implemented for single output.",
klass=IOError,
)
logger.info(
f"Collecting data from {self.uri.parent}/*{self.uri.name}" # type: ignore
)
logger.info(f"Collecting data from {self.uri.parent}/*{self.uri.name}")
# Create new instance
out_storage = SQLiteFeatureStorage(uri=self.uri, upsert="ignore")
# Glob files
files = self.uri.parent.glob(f"*{self.uri.name}") # type: ignore
files = self.uri.parent.glob(f"*{self.uri.name}")
for elem in tqdm(files, desc="file"):
logger.debug(f"Reading from {elem.absolute()!s}")
in_storage = SQLiteFeatureStorage(uri=elem)

View file

@ -24,13 +24,13 @@ from junifer.storage.utils import (
def test_get_valid_inputs() -> None:
"""Test valid inputs."""
storage = HDF5FeatureStorage(uri="/tmp")
assert storage.get_valid_inputs() == [
assert set(storage.get_valid_inputs()) == {
"matrix",
"vector",
"timeseries",
"scalar_table",
"timeseries_2d",
]
}
def test_single_output(tmp_path: Path) -> None:
@ -285,7 +285,7 @@ def test_read_df_params_error(tmp_path: Path) -> None:
with pytest.raises(ValueError, match="At least one of"):
storage.read_df()
with pytest.raises(ValueError, match="Feature MD5"):
with pytest.raises(RuntimeError, match="Feature MD5"):
storage.read_df(feature_md5="md5")

View file

@ -111,7 +111,7 @@ def test_get_engine_multi_output(tmp_path: Path) -> None:
storage = SQLiteFeatureStorage(
uri=uri, single_output=False, upsert="ignore"
)
with pytest.raises(ValueError, match="element must be specified"):
with pytest.raises(ValueError, match="`element` cannot be None"):
storage.get_engine()

View file

@ -4,6 +4,9 @@
# Synchon Mandal <s.mandal@fz-juelich.de>
# License: AGPL
from collections.abc import Sequence
from typing import ClassVar
import pytest
from junifer.storage.base import BaseFeatureStorage
@ -12,10 +15,7 @@ from junifer.storage.base import BaseFeatureStorage
def test_BaseFeatureStorage_abstractness() -> None:
"""Test BaseFeatureStorage is abstract base class."""
with pytest.raises(TypeError, match=r"abstract"):
BaseFeatureStorage(
uri="/tmp",
storage_types=["matrix"], # type: ignore
)
BaseFeatureStorage(uri="/tmp")
def test_BaseFeatureStorage() -> None:
@ -25,17 +25,19 @@ def test_BaseFeatureStorage() -> None:
class MyFeatureStorage(BaseFeatureStorage):
"""Implement concrete class."""
_STORAGE_TYPES: ClassVar[Sequence[str]] = [
"matrix",
"vector",
"timeseries",
"timeseries_2d",
]
def __init__(self, uri, single_output=True):
storage_types = ["matrix", "vector", "timeseries", "timeseries_2d"]
super().__init__(
uri=uri,
storage_types=storage_types,
single_output=single_output,
)
def get_valid_inputs(self):
return ["matrix", "vector", "timeseries", "timeseries_2d"]
def list_features(self):
super().list_features()
@ -107,4 +109,4 @@ def test_BaseFeatureStorage() -> None:
with pytest.raises(ValueError):
st.store(kind="lego", meta=meta)
assert st.uri == "/tmp"
assert str(st.uri) == "/tmp"

View file

@ -6,7 +6,7 @@
import hashlib
import json
from collections.abc import Iterable
from collections.abc import Sequence
from importlib.metadata import PackageNotFoundError, version
import numpy as np
@ -36,6 +36,11 @@ def get_dependency_version(dependency: str) -> str:
str
The version of the dependency.
Raises
------
PackageNotFoundError
If ``dependency`` is not found.
"""
dep_version = ""
try:
@ -141,6 +146,11 @@ def element_to_prefix(element: dict) -> str:
str
The element converted to prefix.
Raises
------
ValueError
If ``element`` is not a dict.
"""
logger.debug(f"Converting element {element} to prefix.")
prefix = "element"
@ -208,13 +218,13 @@ def store_matrix_checks(
klass=ValueError,
)
# Row label validation
if row_names_len != data_shape[0]: # type: ignore
if row_names_len != data_shape[0]:
raise_error(
msg="Number of row names does not match number of rows",
klass=ValueError,
)
# Column label validation
if col_names_len != data_shape[1]: # type: ignore
if col_names_len != data_shape[1]:
raise_error(
msg="Number of column names does not match number of columns",
klass=ValueError,
@ -245,7 +255,7 @@ def store_timeseries_2d_checks(
If the number of column names does not match the number of columns
"""
# data validation
# Data validation
if len(data_shape) != 3:
raise_error(
msg="Data must be a 3D array",
@ -253,13 +263,13 @@ def store_timeseries_2d_checks(
)
# Row label validation
if row_names_len != data_shape[1]: # type: ignore
if row_names_len != data_shape[1]:
raise_error(
msg="Number of row names does not match number of rows",
klass=ValueError,
)
# Column label validation
if col_names_len != data_shape[2]: # type: ignore
if col_names_len != data_shape[2]:
raise_error(
msg="Number of column names does not match number of columns",
klass=ValueError,
@ -268,8 +278,8 @@ def store_timeseries_2d_checks(
def matrix_to_vector(
data: np.ndarray,
col_names: Iterable[str],
row_names: Iterable[str],
col_names: Sequence[str],
row_names: Sequence[str],
matrix_kind: str,
diagonal: bool,
) -> tuple[np.ndarray, list[str]]:
@ -279,9 +289,9 @@ def matrix_to_vector(
----------
data : 2D / 3D numpy.ndarray
The matrix / tensor data to store / read.
col_names : list or tuple of str
col_names : list-like of str
The column labels.
row_names : list or tuple of str
row_names : list-like of str
The row labels.
matrix_kind : str
The kind of matrix:
@ -317,7 +327,7 @@ def matrix_to_vector(
flat_data = data[data_idx]
# Generate flat 1D row X column names
columns = [
f"{row_names[i]}~{col_names[j]}" # type: ignore
f"{row_names[i]}~{col_names[j]}"
for i, j in zip(data_idx[0], data_idx[1])
]
@ -326,8 +336,8 @@ def matrix_to_vector(
def timeseries2d_to_vector(
data: np.ndarray,
col_names: Iterable[str],
row_names: Iterable[str],
col_names: Sequence[str],
row_names: Sequence[str],
) -> tuple[np.ndarray, list[str]]:
"""Convert matrix to vector based on parameters.
@ -335,9 +345,9 @@ def timeseries2d_to_vector(
----------
data : 2D / 3D numpy.ndarray
The matrix / tensor data to store / read.
col_names : list or tuple of str
col_names : list-like of str
The column labels.
row_names : list or tuple of str
row_names : list-like of str
The row labels.
Returns