<commit_before>"""
Intrinsic Mutual Informations
"""

from __future__ import division

from abc import ABCMeta, abstractmethod

import numpy as np

from .. import Distribution, insert_rvf, modify_outcomes
from ..algorithms import channel_capacity
from ..exceptions import ditException
from ..helpers import flatten, normalize_rvs, parse_rvs
from ..shannon import entropy_pmf as h
from ..utils import partitions
from ..utils.optimization import BasinHoppingCallBack, Uniquifier, accept_test

__all__ = [
    'intrinsic_total_correlation',
    'intrinsic_dual_total_correlation',
    'intrinsic_caekl_mutual_information',
]

class BaseIntrinsicMutualInformation(object):
    """
    Compute a generalized intrinsic mutual information:

        IMI[X:Y|Z] = min_{p(z_bar|z)} I[X:Y|Z]
    """
    __metaclass__ = ABCMeta

    name = ""

    def __init__(self, dist, rvs=None, crvs=None, rv_mode=None):
        """
        Initialize the optimizer.

        Parameters
        ----------
        dist : Distribution
            The distribution to compute the intrinsic mutual information of.
        rvs : list, None
            A list of lists. Each inner list specifies the indexes of the random
            variables used to calculate the intrinsic mutual information. If None,
            then it is calculated over all random variables, which is equivalent
            to passing `rvs=dist.rvs`.
        crvs : list
            A single list of indexes specifying the random variables to
            condition on.
        rv_mode : str, None
            Specifies how to interpret `rvs` and `crvs`. Valid options are:
            {'indices', 'names'}. If equal to 'indices', then the elements of
            `crvs` and `rvs` are interpreted as random variable indices. If
            equal to 'names', the the elements are interpreted as random
            variable names. If `None`, then the value of `dist._rv_mode` is
            consulted, which defaults to 'indices'.
        """
        self._dist = dist.copy()
        self._alphabet = self._dist.alphabet
        rvs, crvs, self._rv_mode = normalize_rvs(self._dist, rvs, crvs, rv_mode)
        self._rvs = [ parse_rvs(self._dist, rv, rv_mode)[1] for rv in rvs ]
        self._crvs = parse_rvs(self._dist, crvs, rv_mode)[1]
        self._dist = modify_outcomes(self._dist, tuple)
        if not self._crvs:
            msg = "Intrinsic mutual informations require a conditional variable."
            raise ditException(msg)

        # compress all conditional random variables down to a single var
        self._unq = Uniquifier()
        self._dist = insert_rvf(self._dist, lambda x: (self._unq(tuple(x[i] for i in self._crvs)),))

        sizes = list(map(len, self._dist.alphabet))
        self._dist.make_dense()
        self._pmf = self._dist.pmf.reshape(sizes)

        self._crv = self._dist.outcome_length() - 1
        self._crv_size = sizes[-1]

        self._default_hops = self._crv_size * len(self._crvs)

        all_vars = set(range(len(sizes)))
        keepers = set(flatten(self._rvs)) | {self._crv+1}
        self._others = tuple(all_vars - keepers)

    def construct_random_initial(self):
        """
        Construct a random optimization vector.

        Returns
        -------
        x : ndarray
            A random optimization vector.
        """
        n = self._crv_size
        x = np.random.random((n, n))
        return x

    def construct_joint(self, x):
        """
        Construct the joint distribution.

        Parameters
        ----------
        x : ndarray
            An optimization vector.

        Returns
        -------
        joint : float
            The joint distribution resulting from the distribution passed
            in and the optimization vector.
        """
        n = self._crv_size
        channel = x.copy().reshape((n, n))
        channel /= channel.sum(axis=1, keepdims=True)
        channel[np.isnan(channel)] = 0
        slc = (len(self._pmf.shape) - 1)*[np.newaxis] + 2*[slice(None, None)]<commit_msg>hopefully fix issue<commit_after>"""
Intrinsic Mutual Informations
"""

from __future__ import division

from abc import ABCMeta, abstractmethod

import numpy as np

from .. import Distribution, insert_rvf, modify_outcomes
from ..algorithms import channel_capacity
from ..exceptions import ditException
from ..helpers import flatten, normalize_rvs, parse_rvs
from ..shannon import entropy_pmf as h
from ..utils import partitions
from ..utils.optimization import BasinHoppingCallBack, Uniquifier, accept_test

__all__ = [
    'intrinsic_total_correlation',
    'intrinsic_dual_total_correlation',
    'intrinsic_caekl_mutual_information',
]

class BaseIntrinsicMutualInformation(object):
    """
    Compute a generalized intrinsic mutual information:

        IMI[X:Y|Z] = min_{p(z_bar|z)} I[X:Y|Z]
    """
    __metaclass__ = ABCMeta

    name = ""

    def __init__(self, dist, rvs=None, crvs=None, rv_mode=None):
        """
        Initialize the optimizer.

        Parameters
        ----------
        dist : Distribution
            The distribution to compute the intrinsic mutual information of.
        rvs : list, None
            A list of lists. Each inner list specifies the indexes of the random
            variables used to calculate the intrinsic mutual information. If None,
            then it is calculated over all random variables, which is equivalent
            to passing `rvs=dist.rvs`.
        crvs : list
            A single list of indexes specifying the random variables to
            condition on.
        rv_mode : str, None
            Specifies how to interpret `rvs` and `crvs`. Valid options are:
            {'indices', 'names'}. If equal to 'indices', then the elements of
            `crvs` and `rvs` are interpreted as random variable indices. If
            equal to 'names', the the elements are interpreted as random
            variable names. If `None`, then the value of `dist._rv_mode` is
            consulted, which defaults to 'indices'.
        """
        self._dist = dist.copy()
        self._alphabet = self._dist.alphabet
        rvs, crvs, self._rv_mode = normalize_rvs(self._dist, rvs, crvs, rv_mode)
        self._rvs = [ parse_rvs(self._dist, rv, rv_mode)[1] for rv in rvs ]
        self._crvs = parse_rvs(self._dist, crvs, rv_mode)[1]
        self._dist = modify_outcomes(self._dist, tuple)
        if not self._crvs:
            msg = "Intrinsic mutual informations require a conditional variable."
            raise ditException(msg)

        # compress all conditional random variables down to a single var
        self._unq = Uniquifier()
        self._dist = insert_rvf(self._dist, lambda x: (self._unq(tuple(x[i] for i in self._crvs)),))

        sizes = list(map(len, self._dist.alphabet))
        self._dist.make_dense()
        self._pmf = self._dist.pmf.reshape(sizes)

        self._crv = self._dist.outcome_length() - 1
        self._crv_size = sizes[-1]

        self._default_hops = self._crv_size * len(self._crvs)

        all_vars = set(range(len(sizes)))
        keepers = set(flatten(self._rvs)) | {self._crv+1}
        self._others = tuple(all_vars - keepers)

    def construct_random_initial(self):
        """
        Construct a random optimization vector.

        Returns
        -------
        x : ndarray
            A random optimization vector.
        """
        n = self._crv_size
        x = np.random.random((n, n))
        return x

    def construct_joint(self, x):
        """
        Construct the joint distribution.

        Parameters
        ----------
        x : ndarray
            An optimization vector.

        Returns
        -------
        joint : float
            The joint distribution resulting from the distribution passed
            in and the optimization vector.
        """
        n = self._crv_size
        channel = x.reshape((n, n))
        channel /= channel.sum(axis=1, keepdims=True)
        mask = np.ones_like(channel)
        mask /= mask.sum(axis=1, keepdims=True)
        channel[np.isnan(channel)] = mask[np.isnan(channel)]
        slc = (len(self._pmf.shape) - 1)*[np.newaxis] + 2*[slice(None, None)]
{'fraction_non_alphanumeric': 0.0980440097799511, 'fraction_numerical': 0.002200488997555012, 'mean_word_length': 2.2632628639808536, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, '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': '6030814', 'n_tokens_mistral': 2372, 'n_tokens_neox': 2296, 'n_words': 706}