ltnjax.fuzzy_ops ¶

class AggregBoltzmann(AggregationOperator):
    r"""[Boltzmann fuzzy aggregation operator](https://en.wikipedia.org/wiki/Smooth_maximum).
    This is intended to be used as an disjunction operator as it approximates
    the maximum aggregation.

    $S_\alpha(x_1,...,x_n)= \sum_{i = 1}^n x_i * \exp(\alpha x_i) / \sum_{i = 1}^n exp(\alpha x_i)$

    Attributes:
        alpha: $\alpha$ parameter for
            $S_\alpha(x_1,...,x_n)$.

    Notes:
    - $S_\alpha \rightarrow \max$ as $\alpha \rightarrow \infty$.
    - $S_0$ is the arithmetic mean of its inputs.
    - $S_\alpha \rightarrow \min$ as $\alpha \rightarrow -\infty$.
    - With zero-only mask, the function will return <b>`nan`</b> as we divide
    by a sum, that will be $0$.
    """  # noqa: E501

    def __init__(self, alpha: float):
        r"""Constructor.

        Args:
            alpha: $\alpha$ parameter for
                $S_\alpha(x_1,...,x_n)$.
        """
        self.alpha = alpha

    def __call__(
        self,
        xs: ArrayLike,
        axis: Axis = None,
        keepdims: bool = False,
        mask: ArrayLike | None = None,
    ) -> Array:
        """It applies the Boltzmann aggregation operator to the given
        expression's grounding on the selected dimensions.

        Args:
            xs: Grounding of expression on which the aggregation has to be
                performed.
            axis: (default=None) Axis along which the aggregation to be
                computed. If None, the aggregation is computed along all the
                axes.
            keepdims: (default=False) Flag indicating whether the output has to
                keep the same dimensions as the input after the aggregation.
            mask: (default=None) Boolean mask for excluding values of 'xs'
                from the aggregation. It is internally used for guarded
                quantification. The mask must have the same shape of 'xs'.
                `False` means exclusion, `True` means inclusion.

        Returns:
            Boltzmann aggregation applied to the expression.
        """
        # ArrayLike to Array
        xs = jnp.asarray(xs)

        exp_expr = jnp.exp(jnp.multiply(self.alpha, xs))
        return jnp.divide(
            jnp.sum(
                jnp.multiply(xs, exp_expr),
                axis=axis,
                keepdims=keepdims,
                where=mask,
            ),
            jnp.sum(exp_expr, axis=axis, keepdims=keepdims, where=mask),
        )

class AggregCMean(AggregationOperator):
    r"""Cubic mean or root mean square aggregation operator. Intended for
    disjunction as it approximates the maximum operator.

    Cubic mean:
    $\left( \sum_{i = 1}^n x_i^3 / n \right)^{\frac{1}{3}}$

    Weighted cubic mean:
    $\left( \frac{\sum_{i=1}^n w_i x_i^3}{\sum_{i=1}^n w_i} \right)^{1/3}$
    """  # noqa: E501

    def __init__(self, stable: bool = True):
        """Constructor.

        Args:
            stable: (default=True) Flag indicating whether to use the
                [stable](../../stable.md) version of the operator or not.
        """
        self.stable = stable

    def __call__(
        self,
        xs: ArrayLike,
        axis: Axis = None,
        weights: ArrayLike | None = None,
        keepdims: bool = False,
        mask: ArrayLike | None = None,
        stable: bool | None = None,
    ) -> Array:
        """It applies the quadratic mean aggregation operator to the given
        expression's grounding on the selected dimensions.

        Args:
            xs: Grounding of expression on which the aggregation has to be
                performed.
            axis: (default=None) Axis along which the aggregation to be
                computed. If None, the aggregation is computed along all the
                axes.
            weights: (default=None) The weights for the aggregation operator.
                If `axis=None`, weights must have the same shape as `xs`.
                If there is a shape defined, it must be an `int` or `list[int]`
                containing exactly one `int` that must be the same as
                `xs.shape[axis]`.
            keepdims: (default=False) Flag indicating whether the output has to
                keep the same dimensions as the input after the aggregation.
            mask: (default=None) Boolean mask for excluding values of 'xs'
                from the aggregation. It is internally used for guarded
                quantification. The mask must have the same shape of 'xs'.
                `False` means exclusion, `True` means inclusion.
            stable: (default=None) Flag indicating whether to use the
                [stable](../../stable.md) version of the operator or not.

        Returns:
            Cubic mean aggregation applied to the expression.
        """
        # ArrayLike to Array
        xs = jnp.asarray(xs)
        # Preparing input
        stable = self.stable if stable is None else stable
        if stable:
            xs = not_zeros(xs)

        if weights is None:
            return jnp.cbrt(
                jnp.mean(
                    jnp.pow(xs, 3), axis=axis, keepdims=keepdims, where=mask
                )
            )
        elif mask is None:
            return jnp.cbrt(
                jnp.average(
                    jnp.pow(xs, 3),
                    axis=axis,
                    weights=weights,
                    keepdims=keepdims,
                )
            )
        else:  # weights is not None and mask is not None:
            weights = jnp.asarray(weights)
            x_p = jnp.pow(xs, 3)  # may contain nan values if x_i == 0.

            # Suppress UserWarnings from fuzzy_ops.AggregSum().
            warnings.filterwarnings("ignore", category=UserWarning)
            Sum = AggregSum()

            upper = Sum(
                x_p, axis=axis, weights=weights, keepdims=keepdims, mask=mask
            )
            if len(weights.shape) == 1:
                weights = jnp.broadcast_to(weights, xs.shape)
            lower = jnp.sum(weights, axis=axis, keepdims=keepdims, where=mask)
            return jnp.cbrt(jnp.divide(upper, lower))

class AggregGMean(AggregationOperator):
    r"""[Geometric Mean fuzzy aggregation operator](https://en.wikipedia.org/wiki/Geometric_mean).
    Intended for conjunction as it approximates the minimum aggregation
    operator.

    Geometric mean:
    $(\prod_{i = 1}^n x_i)^{1/n} = \exp(\sum_{i = 1}^n \ln(x_i) / n)$

    Weighted geometric mean:
    $(\prod_{i = 1}^n x_i^{w_i})^{1/(\sum_{i=1}^n w_i)} = \exp(\sum_{i = 1}^n \ln(x_i) / n)$

    Attributes:
        stable: (default=True) Flag indicating whether to use the
            [stable](../../stable.md) version of the operator or not.
    """  # noqa: E501

    def __init__(self, stable: bool = True):
        """Constructor.

        Args:
            stable: (default=True) Flag indicating whether to use the
                [stable](../../stable.md) version of the operator or not.
        """
        self.stable = stable

    def __call__(
        self,
        xs: ArrayLike,
        axis: Axis = None,
        weights: ArrayLike | None = None,
        keepdims: bool = False,
        mask: ArrayLike | None = None,
        stable: bool | None = None,
    ) -> Array:
        """It applies the geometric mean aggregation operator to the given
        expression's grounding on the selected dimensions.

        Args:
            xs: Grounding of expression on which the aggregation has to be
                performed.
            axis: (default=None) Axis along which the aggregation to be
                computed. If None, the aggregation is computed along all the
                axes.
            weights: (default=None) The weights for the aggregation operator.
                If `axis=None`, weights must have the same shape as `xs`.
                If there is a shape defined, it must be an `int` or `list[int]`
                containing exactly one `int` that must be the same as
                `xs.shape[axis]`.
            keepdims: (default=False) Flag indicating whether the output has to
                keep the same dimensions as the input after the aggregation.
            mask: (default=None) Boolean mask for excluding values of 'xs'
                from the aggregation. It is internally used for guarded
                quantification. The mask must have the same shape of 'xs'.
                `False` means exclusion, `True` means inclusion.
            stable: (default=None) Flag indicating whether to use the
                [stable](../../stable.md) version of the operator or not.

        Returns:
            Geometric mean aggregation applied to the expression.

        Note:
        -jax.numpy.log computes the <b>natural logarithm</b>.
        """
        # ArrayLike to Array
        xs = jnp.asarray(xs)
        # preparing input
        stable = self.stable if stable is None else stable
        if stable:
            xs = not_zeros(xs)

        if weights is None:
            return jnp.exp(
                jnp.mean(jnp.log(xs), axis=axis, keepdims=keepdims, where=mask)
            )
        elif mask is None:
            return jnp.exp(
                jnp.average(
                    jnp.log(xs), axis=axis, weights=weights, keepdims=keepdims
                )
            )
        else:  # weights is not None and mask is not None:
            weights = jnp.asarray(weights)
            x_p = jnp.log(xs)  # may contain nan values if x_i == 0.

            # Suppress UserWarnings from fuzzy_ops.AggregSum().
            warnings.filterwarnings("ignore", category=UserWarning)
            Sum = AggregSum()

            upper = Sum(
                x_p, axis=axis, weights=weights, keepdims=keepdims, mask=mask
            )
            if len(weights.shape) == 1:
                weights = jnp.broadcast_to(weights, xs.shape)
            lower = jnp.sum(weights, axis=axis, keepdims=keepdims, where=mask)
            return jnp.exp(jnp.divide(upper, lower))

class AggregHMean(AggregationOperator):
    r"""[Harmonic mean fuzzy aggregation operator](https://en.wikipedia.org/wiki/Harmonic_mean).

    The harmonic mean is the special case of the Power Mean with $p=-1$.

    Harmonic mean:
    $n / (\sum_{i = 1}^n \frac{1}{x_i}$

    Weighted harmonic mean:
    $\frac{\sum_{i=1}^n w_i}{\sum_{i=1}^n \frac{w_i}{x_i}}$
    $= \left( \frac{\sum_{i=1}^n w_i x_i^{-1}}{\sum_{i=1}^n w_i} \right)^{-1}$

    Attributes:
        stable: (default=True) Flag indicating whether to use the
            [stable](../../stable.md) version of the operator or not.

    Notes:
    - As we divide by $(\sum_{i = 1}^n \frac{1}{x_i})$, the values
    $x_i$ <b>must not be $0$</b>.
    """  # noqa: E501

    def __init__(self, stable: bool = True):
        """Constructor.

        Args:
            stable: (default=True) Flag indicating whether to use the
                [stable](../../stable.md) version of the operator or not.
        """
        self.stable = stable

    def __call__(
        self,
        xs: ArrayLike,
        axis: Axis = None,
        weights: ArrayLike | None = None,
        keepdims: bool = False,
        mask: ArrayLike | None = None,
        stable: bool | None = None,
    ) -> Array:
        """It applies the harmonic mean aggregation operator to the given
        expression's grounding on the selected dimensions.

        Args:
            xs: Grounding of expression on which the aggregation has to be
                performed.
            axis: (default=None) Axis along which the aggregation to be
                computed. If None, the aggregation is computed along all the
                axes.
            weights: (default=None) The weights for the aggregation operator.
                If `axis=None`, weights must have the same shape as `xs`.
                If there is a shape defined, it must be an `int` or `list[int]`
                containing exactly one `int` that must be the same as
                `xs.shape[axis]`.
            keepdims: (default=False) Flag indicating whether the output has to
                keep the same dimensions as the input after the aggregation.
            mask: (default=None) Boolean mask for excluding values of 'xs'
                from the aggregation. It is internally used for guarded
                quantification. The mask must have the same shape of 'xs'.
                `False` means exclusion, `True` means inclusion.
            stable: (default=None) Flag indicating whether to use the
                [stable](../../stable.md) version of the operator or not.

        Returns:
            harmonic mean aggregation applied to the expression.
        """
        # ArrayLike to Array
        xs = jnp.asarray(xs)
        # Preparing input
        stable = self.stable if stable is None else stable
        if stable:
            xs = not_zeros(xs)

        if weights is None:
            return jnp.reciprocal(
                jnp.mean(
                    jnp.reciprocal(xs),
                    axis=axis,
                    keepdims=keepdims,
                    where=mask,
                )
            )
        elif mask is None:
            return jnp.reciprocal(
                jnp.average(
                    jnp.reciprocal(xs),
                    axis=axis,
                    weights=weights,
                    keepdims=keepdims,
                )
            )
        else:  # weights is not None and mask is not None:
            weights = jnp.asarray(weights)
            x_p = jnp.reciprocal(xs)  # may contain nan values if x_i == 0.

            # Suppress UserWarnings from fuzzy_ops.AggregSum().
            warnings.filterwarnings("ignore", category=UserWarning)
            Sum = AggregSum()

            upper = Sum(
                x_p, axis=axis, weights=weights, keepdims=keepdims, mask=mask
            )
            if len(weights.shape) == 1:
                weights = jnp.broadcast_to(weights, xs.shape)
            lower = jnp.sum(weights, axis=axis, keepdims=keepdims, where=mask)
            return jnp.reciprocal(jnp.divide(upper, lower))

class AggregInverted(AggregationOperator):
    r"""Inversion operator, that inverts conjunction operators to disjunction
    operators and vise verca. The given `aggreg_op` is inverted.

    For that, we make use of the fact
    $\bigvee(x1,x2,x3) = \neg(\bigwedge(\neg(x1), \neg(x2), \neg(x3)))$

    For example we have
    `Aggreg_Inverted(Aggreg_pMeanError(p=2))) == Aggreg_pMean(p=2))`

    Attributes:
        aggreg_op: The aggregation operator that is inverted.
    """  # noqa: E501

    def __init__(self, aggreg_op: AggregationOperator):
        """Constructor.

        Args:
            aggreg_op: The aggregation operator that is inverted.
        """
        self.aggreg_op = aggreg_op

    def __call__(
        self,
        xs: ArrayLike,
        axis: Axis = None,
        keepdims: bool = False,
        mask: ArrayLike | None = None,
        **kwargs: Any,
    ) -> Array:
        """It applies the inverted aggregation operator `aggreg_op` to the
        given expression's grounding on the selected dimensions.

        Args:
            xs: Grounding of expression on which the aggregation has to be
                performed.
            axis: (default=None) Axis along which the aggregation to be
                computed. If None, the aggregation is computed along all the
                axes.
            keepdims: (default=False) Flag indicating whether the output has to
                keep the same dimensions as the input after the aggregation.
            mask: (default=None) Boolean mask for excluding values of 'xs'
                from the aggregation. It is internally used for guarded
                quantification. The mask must have the same shape of 'xs'.
                `False` means exclusion, `True` means inclusion.
            kwargs: Further arguments to pass to `connective_op`.

        Returns:
            Inverted aggregation applied to the expression.
        """
        # ArrayLike to Array
        xs = jnp.asarray(xs)

        return 1.0 - self.aggreg_op(
            1.0 - xs, axis=axis, keepdims=keepdims, mask=mask, **kwargs
        )