diff --git a/docs/changes/newsfragments/472.enh b/docs/changes/newsfragments/472.enh new file mode 100644 index 000000000..52ce4c2c5 --- /dev/null +++ b/docs/changes/newsfragments/472.enh @@ -0,0 +1 @@ +Simplify ``Storage`` interface and implementations by `Synchon Mandal`_ diff --git a/docs/understanding/storage.rst b/docs/understanding/storage.rst index 714a02d5f..9f820479b 100644 --- a/docs/understanding/storage.rst +++ b/docs/understanding/storage.rst @@ -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 ` are responsible for defining what *storage kind* -(``matrix``, ``vector``, ``timeseries``, ``scalar_table``) they support for -which :ref:`data type ` 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 ` are responsible for mapping the input +:ref:`data type ` to its output :ref:`storage type ` +as shown :ref:`here `. +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`` diff --git a/junifer/storage/base.py b/junifer/storage/base.py index fd6c879a7..13f4448b7 100644 --- a/junifer/storage/base.py +++ b/junifer/storage/base.py @@ -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. diff --git a/junifer/storage/hdf5.py b/junifer/storage/hdf5.py index 99c2f406e..13ade4af5 100644 --- a/junifer/storage/hdf5.py +++ b/junifer/storage/hdf5.py @@ -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", diff --git a/junifer/storage/pandas_base.py b/junifer/storage/pandas_base.py index fbc66b9f4..3a37952a7 100644 --- a/junifer/storage/pandas_base.py +++ b/junifer/storage/pandas_base.py @@ -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). """ diff --git a/junifer/storage/sqlite.py b/junifer/storage/sqlite.py index 6c0fb03d3..63a8ced91 100644 --- a/junifer/storage/sqlite.py +++ b/junifer/storage/sqlite.py @@ -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) diff --git a/junifer/storage/tests/test_hdf5.py b/junifer/storage/tests/test_hdf5.py index 0685d410e..9cb4c0d76 100644 --- a/junifer/storage/tests/test_hdf5.py +++ b/junifer/storage/tests/test_hdf5.py @@ -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") diff --git a/junifer/storage/tests/test_sqlite.py b/junifer/storage/tests/test_sqlite.py index 5f496733d..ce205c08c 100644 --- a/junifer/storage/tests/test_sqlite.py +++ b/junifer/storage/tests/test_sqlite.py @@ -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() diff --git a/junifer/storage/tests/test_storage_base.py b/junifer/storage/tests/test_storage_base.py index 5463909bc..2fbe1b3b0 100644 --- a/junifer/storage/tests/test_storage_base.py +++ b/junifer/storage/tests/test_storage_base.py @@ -4,6 +4,9 @@ # Synchon Mandal # 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" diff --git a/junifer/storage/utils.py b/junifer/storage/utils.py index 743a39d56..0ab8be595 100644 --- a/junifer/storage/utils.py +++ b/junifer/storage/utils.py @@ -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