<commit_before>"""
=========================
The Value Pipeline System
=========================

The value pipeline system is a vital part of the :mod:`vivarium`
infrastructure. It allows for values that determine the behavior of individual
:term:`simulants <Simulant>` to be constructed across across multiple
:ref:`components <components_concept>`.

For more information about when and how you should use pipelines in your
simulations, see the value system :ref:`concept note <values_concept>`.

"""
from collections import defaultdict
from numbers import Number
from typing import Callable, List, Any, TypeVar, Union, Iterable, Tuple

from loguru import logger
import numpy as np
import pandas as pd

from vivarium.exceptions import VivariumError
from vivarium.framework.utilities import from_yearly

T = TypeVar('T')
# Supports standard algebraic operations with scalar values.
NumberLike = Union[np.ndarray, pd.Series, pd.DataFrame, Number]


class DynamicValueError(VivariumError):
    """Indicates an improperly configured value was invoked."""
    pass


def replace_combiner(value: T, mutator: Callable[..., T], *args: Any, **kwargs: Any) -> T:
    """Replace the previous pipeline output with the output of the mutator.

    This is the default combiner.

    Parameters
    ----------
    value
        The value from the previous step in the pipeline.
    mutator
        A callable that takes in all arguments that the pipeline source takes
        in plus an additional last positional argument for the value from
        the previous stage in the pipeline.
    args, kwargs
        The same args and kwargs provided during the invocation of the
        pipeline.

    Returns
    -------
        A modified version of the input value.

    """
    args = list(args) + [value]
    return mutator(*args, **kwargs)


def list_combiner(value: List[T], mutator: Callable[..., T], *args: Any, **kwargs: Any) -> List[T]:
    """Aggregates source and mutator output into a list.

    This combiner is meant to be used with a post-processor that does some
    kind of reduce operation like summing all values in the list.

    Parameters
    ----------
    value
        A list of all values provided by the source and prior mutators in the
        pipeline.
    mutator
        A callable that returns some portion of this pipeline's final value.
    args, kwargs
        The same args and kwargs provided during the invocation of the
        pipeline.

    Returns
    -------
        The input list with new mutator portion of the pipeline value
        appended to it.

    """
    value.append(mutator(*args, **kwargs))
    return value


def rescale_post_processor(value: NumberLike, time_step: pd.Timedelta):
    """Rescales annual rates to time-step appropriate rates.

    This should only be used with a simulation using a
    :class:`~vivarium.framework.time.DateTimeClock` or another implementation
    of a clock that traffics in pandas date-time objects.

    Parameters
    ----------
    value
        Annual rates, either as a number or something we can broadcast
        multiplication over like a :mod:`numpy` array or :mod:`pandas`
        data frame.
    time_step
        A pandas time delta representing the size of the upcoming time
        step.

    Returns
    -------
        The annual rates rescaled to the size of the current time step size.

    """
    return from_yearly(value, time_step)


def union_post_processor(values: List[NumberLike], _) -> NumberLike:
    """Computes a probability on the union of the sample spaces in the values.

    Given a list of values where each value is a probability of an independent
    event, this post processor computes the probability of the union of the
    events.

    .. list-table::
       :width: 100%
       :widths: 1 3

       * - :math:`p_x`
         - Probability of event x
       * - :math:`1 - p_x`
         - Probability of not event x
       * - :math:`\prod_x(1 - p_x)`
         - Probability of not any events x
       * - :math:`1 - \prod_x(1 - p_x)`
         - Probability of any event x

    Parameters
    ----------
    values
        A list of independent proportions or probabilities, either
        as numbers or as a something we can broadcast addition and
        multiplication over.

    Returns
    -------
        The probability over the union of the sample spaces represented
        by the original probabilities.

    """
    return 1 - np.prod([1 - v for v in values], axis=0)


class Pipeline:
    """A tool for building up values across several components.
<commit_msg>Revert logic changes to union post processor<commit_after>"""
=========================
The Value Pipeline System
=========================

The value pipeline system is a vital part of the :mod:`vivarium`
infrastructure. It allows for values that determine the behavior of individual
:term:`simulants <Simulant>` to be constructed across across multiple
:ref:`components <components_concept>`.

For more information about when and how you should use pipelines in your
simulations, see the value system :ref:`concept note <values_concept>`.

"""
from collections import defaultdict
from numbers import Number
from typing import Callable, List, Any, TypeVar, Union, Iterable, Tuple

from loguru import logger
import numpy as np
import pandas as pd

from vivarium.exceptions import VivariumError
from vivarium.framework.utilities import from_yearly

T = TypeVar('T')
# Supports standard algebraic operations with scalar values.
NumberLike = Union[np.ndarray, pd.Series, pd.DataFrame, Number]


class DynamicValueError(VivariumError):
    """Indicates an improperly configured value was invoked."""
    pass


def replace_combiner(value: T, mutator: Callable[..., T], *args: Any, **kwargs: Any) -> T:
    """Replace the previous pipeline output with the output of the mutator.

    This is the default combiner.

    Parameters
    ----------
    value
        The value from the previous step in the pipeline.
    mutator
        A callable that takes in all arguments that the pipeline source takes
        in plus an additional last positional argument for the value from
        the previous stage in the pipeline.
    args, kwargs
        The same args and kwargs provided during the invocation of the
        pipeline.

    Returns
    -------
        A modified version of the input value.

    """
    args = list(args) + [value]
    return mutator(*args, **kwargs)


def list_combiner(value: List[T], mutator: Callable[..., T], *args: Any, **kwargs: Any) -> List[T]:
    """Aggregates source and mutator output into a list.

    This combiner is meant to be used with a post-processor that does some
    kind of reduce operation like summing all values in the list.

    Parameters
    ----------
    value
        A list of all values provided by the source and prior mutators in the
        pipeline.
    mutator
        A callable that returns some portion of this pipeline's final value.
    args, kwargs
        The same args and kwargs provided during the invocation of the
        pipeline.

    Returns
    -------
        The input list with new mutator portion of the pipeline value
        appended to it.

    """
    value.append(mutator(*args, **kwargs))
    return value


def rescale_post_processor(value: NumberLike, time_step: pd.Timedelta):
    """Rescales annual rates to time-step appropriate rates.

    This should only be used with a simulation using a
    :class:`~vivarium.framework.time.DateTimeClock` or another implementation
    of a clock that traffics in pandas date-time objects.

    Parameters
    ----------
    value
        Annual rates, either as a number or something we can broadcast
        multiplication over like a :mod:`numpy` array or :mod:`pandas`
        data frame.
    time_step
        A pandas time delta representing the size of the upcoming time
        step.

    Returns
    -------
        The annual rates rescaled to the size of the current time step size.

    """
    return from_yearly(value, time_step)


def union_post_processor(values: List[NumberLike], _) -> NumberLike:
    """Computes a probability on the union of the sample spaces in the values.

    Given a list of values where each value is a probability of an independent
    event, this post processor computes the probability of the union of the
    events.

    .. list-table::
       :width: 100%
       :widths: 1 3

       * - :math:`p_x`
         - Probability of event x
       * - :math:`1 - p_x`
         - Probability of not event x
       * - :math:`\prod_x(1 - p_x)`
         - Probability of not any events x
       * - :math:`1 - \prod_x(1 - p_x)`
         - Probability of any event x

    Parameters
    ----------
    values
        A list of independent proportions or probabilities, either
        as numbers or as a something we can broadcast addition and
        multiplication over.

    Returns
    -------
        The probability over the union of the sample spaces represented
        by the original probabilities.

    """
    # if there is only one value, return the value
    if len(values) == 1:
        return values[0]

    # if there are multiple values, calculate the joint value
    product = 1
    for v in values:
        new_value = (1-v)
        product = product * new_value
    joint_value = 1 - product
    return joint_value


class Pipeline:
    """A tool for building up values across several components.

{'fraction_non_alphanumeric': 0.0870024284658431, 'fraction_numerical': 0.0027452222574173794, 'mean_word_length': 2.659969088098918, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 15, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4101419', 'n_tokens_mistral': 2507, 'n_tokens_neox': 2401, 'n_words': 1126}