Source code for bhf_smarthealth.utils.aggregators

"""This module contains custom aggregation functions"""

import logging
import pandas as pd
from typing import Any
from functools import partial
from typing import List

logger = logging.getLogger(__name__)




def mode_with_later_tiebreak(series: "pd.Series") -> Any:
    """Calculates the modal value of a series. Where there is a tie, the modal value that lies latest in the sorted series is returned.

    Parameters
    ----------
    series : pd.Series
        Pandas series of data to calculate a single modal value from

    Returns
    -------
    Any
        Modal value that is latest in the ordered series

    Notes
    -----
    Can be used to return the mode associated with the latest timestamp in timeseries data.
    """
    if series.empty:
        return None

    modes = series.mode()

    if len(modes) == 0:
        return pd.NA
    elif len(modes) == 1:
        return modes.iloc[0]
    else:
        # Filter the original series to only include values that are a mode
        mode_series = series[series.isin(modes)]

        # Get the last value in the series that is also a mode
        # The last value in this filtered series will be the most recent mode value.
        return mode_series.iloc[-1]





def filtered_aggregation(
    series: "pd.Series", min_value: float, max_value: float, agg_func: str
) -> object | None:
    """Perform an aggregation function on a filtered Pandas Series

    Parameters
    ----------
    series : pd.Series
        Series of data to be filtered and aggregated
    min_value : float
        The allowed minimum value (inclusive)
    max_value : float
        The maximum allowed value (inclusive)
    agg_func : str
        The aggregation function to be applied to the filtered Series

    Returns
    -------
    float | None
        The aggregated value or None if the Series, or filtered Series, is empty

    Raises
    ------
    ValueError
        Raised if an unsupported aggregation function is specified
    """

    # If Series/DataFrame contains only NaNs, it is still not considered empty. Must drop these NaN values before empty check
    if series.dropna().empty:
        logger.debug("Empty series passed to filtered_aggregation")
        return None

    filtered_series = series[series.between(min_value, max_value, inclusive="both")]

    if filtered_series.dropna().empty:
        logger.debug(
            "Series passed to filtered_aggregation is empty after filtering with parameters min_value: {min_value}, max_value: {max_value}"
        )
        return None  # return None if the filters remove everything

    try:
        # Get the attribute from the Series object
        attr = getattr(filtered_series, agg_func)

        # Differentiate between e.g. .mean() and .size (method vs. property)
        return attr() if callable(attr) else attr

    except AttributeError as e:
        e.add_note(f"Unknown aggregation function: {agg_func}")
        raise





def config_filtered_aggs(
    min_value: float, max_value: float, agg_funcs: List[str]
) -> List[partial]:
    """Configures the filtered_aggregation function(s) as a Partial so that aggregation functions passed to a Pandas groupby object can be passed additional arguments.

    The function name is manually set to provide an informative column name in the resulting DataFrame.

    See https://pandas.pydata.org/pandas-docs/stable/user_guide/groupby.html#groupby-aggregate-named for more information

    Parameters
    ----------
    min_value : float
        The minimum allowed value (inclusive)
    max_value : float
        The maximum allowed value (inclusive)
    agg_funcs : str
        The aggregation function(s) to be applied to the filtered Series

    Returns
    -------
    List[functools.partial]
        A list of partial objects, see https://docs.python.org/3/library/functools.html#partial-objects


    Notes
    -----
    When passing config_filtered_aggs to pandas.agg(), if you want to combine custom filtered aggregations with standard pandas aggregation strings you can use list expansion, i.e. {"column_to_be_aggregated":
        ["list", "of", "standard", "aggregations"]
         +
        config_filtered_aggs(min_value, max_value, ["list", "of", "filtered", "aggregations])
    }
    """
    aggs = []

    for agg_func in agg_funcs:
        # Create a partial function. The "series" will be passed when called from Pandas .agg()
        partial_func = partial(
            filtered_aggregation,
            min_value=min_value,
            max_value=max_value,
            agg_func=agg_func,
        )

        # Set the function name
        partial_func.__name__ = f"filtered_{agg_func}"  # type: ignore
        aggs.append(partial_func)

    return aggs