juaml/junifer
Watch
3
0
Fork
You've already forked junifer
 0
Code Issues 14 Pull requests 2 Packages Activity Actions

[ENH]: Simplify Storage interface #472

Merged
synchon merged 24 commits from refactor/storage into main 2025-11-07 11:57:58 +00:00
Conversation 8 Commits 24 Files changed 10 +230 -251
10 changed files with 230 additions and 251 deletions
Download patch file Download diff file Expand all files Collapse all files

1
docs/changes/newsfragments/472.enh Normal file
View file

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

16
docs/understanding/storage.rst
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 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 mapping the input
(``matrix``, ``vector``, ``timeseries``, ``scalar_table``) they support for :ref:`data type <data_types>` to its output :ref:`storage type <storage_types>`
which :ref:`data type <data_types>` by overriding its ``get_output_type`` as shown :ref:`here <extending_markers_input_output>`.
method. The storage object in turn declares and provides implementation for The storage object in turn declares and provides implementation for
specific *storage kind*. For example, :class:`.SQLiteFeatureStorage` supports specific *storage type*. For example, :class:`.SQLiteFeatureStorage` supports
saving ``matrix``, ``vector`` and ``timeseries`` via ``store_matrix``, saving ``matrix``, ``vector`` and ``timeseries`` via ``store_matrix``,
``store_vector`` and ``store_timeseries`` methods respectively. ``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 - 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`
* - ``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`` * - ``scalar_table``
- | A 2D square or non-square matrix of scalar values with row name, column - | A 2D square or non-square matrix of scalar values with row name, column
| name and row header column name | name and row header column name
@ -83,4 +87,4 @@ Storage Interfaces
* - :class:`.HDF5FeatureStorage` * - :class:`.HDF5FeatureStorage`
- ``.hdf5`` - ``.hdf5``
- HDF5 - HDF5
- ``matrix``, ``vector``, ``timeseries``, ``scalar_table`` - ``matrix``, ``vector``, ``timeseries``, ``timeseries_2d``, ``scalar_table``

101
junifer/storage/base.py
View file

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

174
junifer/storage/hdf5.py
View file

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

44
junifer/storage/pandas_base.py
View file

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

75
junifer/storage/sqlite.py
View file

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

6
junifer/storage/tests/test_hdf5.py
View file

@ -24,13 +24,13 @@ 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() == [ assert set(storage.get_valid_inputs()) == {
"matrix", "matrix",
"vector", "vector",
"timeseries", "timeseries",
"scalar_table", "scalar_table",
"timeseries_2d", "timeseries_2d",
] }
def test_single_output(tmp_path: Path) -> None: 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"): with pytest.raises(ValueError, match="At least one of"):
storage.read_df() storage.read_df()
with pytest.raises(ValueError, match="Feature MD5"): with pytest.raises(RuntimeError, match="Feature MD5"):
storage.read_df(feature_md5="md5") storage.read_df(feature_md5="md5")

2
junifer/storage/tests/test_sqlite.py
View file

@ -111,7 +111,7 @@ def test_get_engine_multi_output(tmp_path: Path) -> None:
storage = SQLiteFeatureStorage( storage = SQLiteFeatureStorage(
uri=uri, single_output=False, upsert="ignore" 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() storage.get_engine()

22
junifer/storage/tests/test_storage_base.py
View file

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

40
junifer/storage/utils.py
View file

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