bgc_data_processing.core.storers

Data storing objects.

Storer(data, category, providers, variables)

Storing data class, to keep track of metadata.

Parameters:

Name	Type	Description	Default
data	DataFrame	Dataframe to store.	required
category	str	Data category.	required
providers	list	Names of the data providers.	required
variables	StoringVariablesSet	Variables storer of object to keep track of the variables in the Dataframe.	required

Source code in src/bgc_data_processing/core/storers.py

def __init__(
    self,
    data: pd.DataFrame,
    category: str,
    providers: list,
    variables: "StoringVariablesSet",
) -> None:
    self._data = data
    self._category = category
    self._providers = providers
    self._variables = deepcopy(variables)

data: pd.DataFrame property

Getter for self._data.

Returns:

Type	Description
DataFrame	Dataframe.

category: str property

Returns the category of the provider.

Returns:

Type	Description
str	Category provider belongs to.

providers: list property

Getter for self._providers.

Returns:

Type	Description
list	List of providers.

variables: StoringVariablesSet property

Getter for self._variables.

Returns:

Type	Description
StoringVariablesSet	Variables storer.

__repr__()

Representation of self.

Returns:

Type	Description
str	Representation of self.data.

Source code in src/bgc_data_processing/core/storers.py

def __repr__(self) -> str:
    """Representation of self.

    Returns
    -------
    str
        Representation of self.data.
    """
    return repr(self.data)

__eq__(__o)

Test equality with other object.

Parameters:

Name	Type	Description	Default
__o	object	Object to test equality with.	required

Returns:

Type	Description
bool	True if is same object only.

Source code in src/bgc_data_processing/core/storers.py

def __eq__(self, __o: object) -> bool:
    """Test equality with other object.

    Parameters
    ----------
    __o : object
        Object to test equality with.

    Returns
    -------
    bool
        True if is same object only.
    """
    return self is __o

__radd__(other)

Perform right addition.

Parameters:

Name	Type	Description	Default
other	Any	Object to add.	required

Returns:

Type	Description
Storer	Concatenation of both storer's dataframes.

Source code in src/bgc_data_processing/core/storers.py

def __radd__(self, other: Any) -> "Storer":
    """Perform right addition.

    Parameters
    ----------
    other : Any
        Object to add.

    Returns
    -------
    Storer
        Concatenation of both storer's dataframes.
    """
    if other == 0:
        return self
    return self.__add__(other)

__add__(other)

Perform left addition.

Parameters:

Name	Type	Description	Default
other	Any	Object to add.	required

Returns:

Type	Description
Storer	Concatenation of both storer's dataframes.

Raises:

Type	Description
TypeError	If other is not a storer.
IncompatibleVariableSetsError	If both storers have a different variable set.
IncompatibleCategoriesError	If both storers have different categories.

Source code in src/bgc_data_processing/core/storers.py

def __add__(self, other: object) -> "Storer":
    """Perform left addition.

    Parameters
    ----------
    other : Any
        Object to add.

    Returns
    -------
    Storer
        Concatenation of both storer's dataframes.

    Raises
    ------
    TypeError
        If other is not a storer.
    IncompatibleVariableSetsError
        If both storers have a different variable set.
    IncompatibleCategoriesError
        If both storers have different categories.
    """
    if not isinstance(other, Storer):
        error_msg = f"Can't add CSVStorer object to {type(other)}"
        raise TypeError(error_msg)
    # Assert variables are the same
    if self.variables != other.variables:
        error_msg = "Variables or categories are not compatible"
        raise IncompatibleVariableSetsError(error_msg)
    # Assert categories are the same
    if self.category != other.category:
        error_msg = "Categories are not compatible"
        raise IncompatibleCategoriesError(error_msg)

    concat_data = pd.concat([self._data, other.data], ignore_index=True)
    concat_providers = list(set(self.providers + other.providers))
    # Return Storer with similar variables
    return Storer(
        data=concat_data,
        category=self.category,
        providers=concat_providers,
        variables=self.variables,
    )

remove_duplicates(priority_list=None)

Update self._data to remove duplicates in data.

Parameters:

Name	Type	Description	Default
priority_list	list	Providers priority order, first has priority over others and so on. , by default None	None

Source code in src/bgc_data_processing/core/storers.py

def remove_duplicates(self, priority_list: list | None = None) -> None:
    """Update self._data to remove duplicates in data.

    Parameters
    ----------
    priority_list : list, optional
        Providers priority order, first has priority over others and so on.
        , by default None
    """
    df = self._data
    df = self._remove_duplicates_among_providers(df)
    df = self._remove_duplicates_between_providers(df, priority_list=priority_list)
    self._data = df

slice_verbose(_start_date, _end_date) staticmethod

Invoke Verbose for slicing.

Parameters:

Name	Type	Description	Default
_start_date	date	Start date.	required
_end_date	date	End date.	required

Source code in src/bgc_data_processing/core/storers.py

@staticmethod
@with_verbose(
    trigger_threshold=1,
    message="Slicing data for date range: [_start_date]-[_end_date].",
)
def slice_verbose(_start_date: dt.date, _end_date: dt.date) -> None:
    """Invoke Verbose for slicing.

    Parameters
    ----------
    _start_date : dt.date
        Start date.
    _end_date : dt.date
        End date.
    """
    return

slice_on_dates(drng)

Slice the Dataframe using the date column.

Only returns indexes to use for slicing.

Parameters:

Name	Type	Description	Default
drng	Series	Two values Series, "start_date" and "end_date".	required

Returns:

Type	Description
list	Indexes to use for slicing.

Source code in src/bgc_data_processing/core/storers.py

def slice_on_dates(
    self,
    drng: pd.Series,
) -> "Slice":
    """Slice the Dataframe using the date column.

    Only returns indexes to use for slicing.

    Parameters
    ----------
    drng : pd.Series
        Two values Series, "start_date" and "end_date".

    Returns
    -------
    list
        Indexes to use for slicing.
    """
    # Params
    start_date: dt.datetime = drng["start_date"]
    end_date: dt.datetime = drng["end_date"]
    self.slice_verbose(_start_date=start_date.date(), _end_date=end_date.date())
    dates_col = self._data[self._variables.get(self._variables.date_var_name).label]
    # slice
    after_start = dates_col >= start_date
    before_end = dates_col <= end_date
    slice_index = dates_col.loc[after_start & before_end].index.values.tolist()
    return Slice(
        storer=self,
        slice_index=slice_index,
    )

add_feature(variable, data)

Add a new feature to the storer.

Parameters:

Name	Type	Description	Default
variable	NotExistingVar	Variable corresponding to the feature.	required
data	Series	Feature data.	required

Source code in src/bgc_data_processing/core/storers.py

def add_feature(
    self,
    variable: "NotExistingVar",
    data: pd.Series,
) -> None:
    """Add a new feature to the storer.

    Parameters
    ----------
    variable : NotExistingVar
        Variable corresponding to the feature.
    data : pd.Series
        Feature data.
    """
    self.variables.add_var(variable)
    self._data[variable.name] = data

pop(variable_name)

Remove and return the data for a given variable.

Parameters:

Name	Type	Description	Default
variable_name	str	Name of the variable to remove.	required

Returns:

Type	Description
Series	Data of the corresponding variable.

Source code in src/bgc_data_processing/core/storers.py

def pop(self, variable_name: str) -> pd.Series:
    """Remove and return the data for a given variable.

    Parameters
    ----------
    variable_name : str
        Name of the variable to remove.

    Returns
    -------
    pd.Series
        Data of the corresponding variable.
    """
    var = self.variables.pop(variable_name)
    return self._data.pop(var.name)

from_constraints(storer, constraints) classmethod

Create a new storer object from an existing storer and constraints.

Parameters:

Name	Type	Description	Default
storer	Storer	Storer to modify with constraints.	required
constraints	Constraints	Constraints to use to modify the storer.	required

Returns:

Type	Description
Storer	New storer respecting the constraints.

Source code in src/bgc_data_processing/core/storers.py

@classmethod
def from_constraints(
    cls,
    storer: "Storer",
    constraints: "Constraints",
) -> "Storer":
    """Create a new storer object from an existing storer and constraints.

    Parameters
    ----------
    storer : Storer
        Storer to modify with constraints.
    constraints : Constraints
        Constraints to use to modify the storer.

    Returns
    -------
    Storer
        New storer respecting the constraints.
    """
    data = constraints.apply_constraints_to_dataframe(dataframe=storer.data)
    return Storer(
        data=data,
        category=storer.category,
        providers=storer.providers,
        variables=storer.variables,
    )

slice_using_index(index)

Slice Storer using.

Parameters:

Name	Type	Description	Default
index	Index	Index values to keep.	required

Returns:

Type	Description
Storer	Corresponding storer.

Source code in src/bgc_data_processing/core/storers.py

def slice_using_index(self, index: pd.Index) -> "Storer":
    """Slice Storer using.

    Parameters
    ----------
    index : pd.Index
        Index values to keep.

    Returns
    -------
    Storer
        Corresponding storer.
    """
    return Storer(
        data=self._data.loc[index, :],
        category=self._category,
        providers=self._providers,
        variables=self._variables,
    )

Slice(storer, slice_index)

Bases: Storer

Slice storing object, instance of Storer to inherit of the saving method.

Parameters:

Name	Type	Description	Default
storer	Storer	Storer to slice.	required
slice_index	list	Indexes to keep from the Storer dataframe.	required

Slice storing object, instance of Storer to inherit of the saving method.

Parameters:

Name	Type	Description	Default
storer	Storer	Storer to slice.	required
slice_index	list	Indexes to keep from the Storer dataframe.	required

Source code in src/bgc_data_processing/core/storers.py

def __init__(
    self,
    storer: Storer,
    slice_index: list,
) -> None:
    """Slice storing object, instance of Storer to inherit of the saving method.

    Parameters
    ----------
    storer : Storer
        Storer to slice.
    slice_index : list
        Indexes to keep from the Storer dataframe.
    """
    self.slice_index = slice_index
    self.storer = storer
    super().__init__(
        data=storer.data,
        category=storer.category,
        providers=storer.providers,
        variables=storer.variables,
    )

slice_index = slice_index instance-attribute

storer = storer instance-attribute

providers: list property

Getter for self.storer._providers.

Returns:

Type	Description
list	Providers of the dataframe which the slice comes from.

variables: list property

Getter for self.storer._variables.

Returns:

Type	Description
list	Variables of the dataframe which the slice comes from.

data: pd.DataFrame property

Getter for self.storer._data.

Returns:

Type	Description
DataFrame	The dataframe which the slice comes from.

__repr__()

Represent self as a string.

Returns:

Type	Description
str	str(slice.index)

Source code in src/bgc_data_processing/core/storers.py

def __repr__(self) -> str:
    """Represent self as a string.

    Returns
    -------
    str
        str(slice.index)
    """
    return str(self.slice_index)

__add__(__o)

Perform left addition.

Parameters:

Name	Type	Description	Default
__o	object	Object to add.	required

Returns:

Type	Description
Slice	Concatenation of both slices.

Raises:

Type	Description
DifferentSliceOriginError	If the slices don't originate from same storer.

Source code in src/bgc_data_processing/core/storers.py

def __add__(self, __o: object) -> "Slice":
    """Perform left addition.

    Parameters
    ----------
    __o : object
        Object to add.

    Returns
    -------
    Slice
        Concatenation of both slices.

    Raises
    ------
    DifferentSliceOriginError
        If the slices don't originate from same storer.
    """
    if self.storer != __o.storer:
        error_msg = "Addition can only be performed with slice from same CSVStorer"
        raise DifferentSliceOriginError(error_msg)
    new_index = list(set(self.slice_index).union(set(__o.slice_index)))
    return Slice(self.storer, new_index)