Source code for dit.rate_distortion.gray_wyner.region

"""
Coordinate systems on the Gray-Wyner region.

The Gray-Wyner rate region, the *region of tension* of Prabhakaran &
Prabhakaran :cite:`prabhakaran2014assisted`, the *extension profile* of
Matveev & Romashchenko :cite:`matveev2026beyond`, and the *mutual information
region* of Li & El Gamal :cite:`li2018extended` are all images of one another
under invertible affine maps. Every one of them is generated by the same
auxiliary variable ``W``; they differ only in which affine functions of the
entropies of ``(X_0, ..., X_{n-1}, W)`` are recorded.

Writing ``R_0 = I[X_{0:n} : W]`` and ``R_i = H[X_i | W]`` for the Gray-Wyner
rates, the maps are

.. math::

    \\tau_i     &= R_0 + R_i - H[X_i]           &&= I[X_{-i} : W | X_i] \\\\
    \\tau_{res} &= R_0 + \\sum_i R_i - H[X_{0:n}] &&= T[X_{0:n} | W] \\\\
    v_i        &= H[X_i] - R_i                 &&= I[X_i : W] \\\\
    v_{joint}  &= R_0                          &&= I[X_{0:n} : W]

The tension identities are the chain rule ``I[X_{0:n}:W] = I[X_i:W] +
I[X_{-i}:W|X_i]`` and the definition of the conditional total correlation
``T[X_{0:n}|W] = sum_i H[X_i|W] - H[X_{0:n}|W]``. The matrix carrying
``(R_0, ..., R_n)`` to ``(tau_0, ..., tau_{n-1}, tau_res)`` is unimodular, so
nothing is lost in either direction.

At ``n = 2`` the residual coordinate is ``I[X:Y|W]`` and these reduce to the
map ``g_{X,Y}`` of :cite:`prabhakaran2014assisted` (their Theorem 5.2).

Conventions
-----------
``dit`` follows the coordinate ordering of :cite:`prabhakaran2014assisted` and
:cite:`li2018extended`: the tension coordinate indexed by ``i`` is
``I[X_{-i} : W | X_i]``, so for a pair it is ``(I[Y:W|X], I[X:W|Y],
I[X:Y|W])``. Csirmaz :cite:`csirmaz2023short` swaps the first two coordinates
and omits the increasing hull (and correspondingly uses a cardinality bound of
``|X||Y| + 3`` rather than ``+ 2``). Formulas transported between these papers
must be checked against the convention in force.

The tension region proper is the *increasing hull* of the attainable tension
points -- every point dominating an attainable one is in the region. The
routines here report attainable points on the lower boundary; the hull is
implicit.

Besides the coordinate maps, this module holds the three quantities that
:cite:`li2018extended` identify as extreme points of the mutual information
region: the maximal interaction information and the two private interaction
informations. They are the reason `GrayWynerOptimizer` accepts signed
weights.
"""

from collections import namedtuple

import numpy as np

from ...exceptions import ditException
from ...utils import unitful
from .optimizer import GrayWynerPoint

__all__ = (
    "ExtensionPoint",
    "MutualInformationPoint",
    "TensionPoint",
    "asymmetric_private_interaction_information",
    "extension_to_rates",
    "maximal_interaction_information",
    "mutual_information_to_rates",
    "rates_to_extension",
    "rates_to_mutual_information",
    "rates_to_tension",
    "symmetric_private_interaction_information",
    "tension_to_rates",
)


TensionPoint = namedtuple("TensionPoint", ["tensions", "residual"])
TensionPoint.__doc__ = """
A point of the region of tension.

Attributes
----------
tensions : tuple of float
    The per-source tensions ``tau_i = I[X_{-i} : W | X_i]``.
residual : float
    The residual dependence ``tau_res = T[X_{0:n} | W]``, which for a pair is
    the conditional mutual information ``I[X:Y|W]``.
"""


ExtensionPoint = namedtuple("ExtensionPoint", ["conditional_entropies", "residual"])
ExtensionPoint.__doc__ = """
A point of the extension profile.

Attributes
----------
conditional_entropies : tuple of float
    The conditional entropies ``H[X_i | W]``.
residual : float
    The residual dependence ``T[X_{0:n} | W]``.
"""


MutualInformationPoint = namedtuple("MutualInformationPoint", ["marginals", "joint"])
MutualInformationPoint.__doc__ = """
A point of the mutual information region of :cite:`li2018extended`.

Attributes
----------
marginals : tuple of float
    The per-source informations ``I[X_i : W]``.
joint : float
    The joint information ``I[X_{0:n} : W]``.
"""


def _check_lengths(values, marginal_entropies, what):
    """
    Validate that a coordinate tuple matches the number of sources.

    Parameters
    ----------
    values : iterable of float
        The per-source coordinates.
    marginal_entropies : iterable of float
        The marginal entropies ``H[X_i]``.
    what : str
        A name for the coordinate tuple, used in the error message.

    Returns
    -------
    values, marginal_entropies : np.ndarray, np.ndarray
        The validated inputs as arrays.
    """
    values = np.asarray(values, dtype=float)
    marginal_entropies = np.asarray(marginal_entropies, dtype=float)
    if values.size != marginal_entropies.size:
        msg = f"`{what}` has length {values.size} but there are {marginal_entropies.size} sources."
        raise ditException(msg)
    return values, marginal_entropies


[docs] def rates_to_tension(point, marginal_entropies, joint_entropy): """ Convert a Gray-Wyner rate point to tension coordinates. Parameters ---------- point : GrayWynerPoint The rate point ``(R_0, (R_1, ..., R_n))``. marginal_entropies : iterable of float The marginal entropies ``H[X_i]``. joint_entropy : float The joint entropy ``H[X_{0:n}]``. Returns ------- tension : TensionPoint The corresponding point of the region of tension. """ private, marginals = _check_lengths(point.private, marginal_entropies, "point.private") tensions = point.common + private - marginals residual = point.common + private.sum() - joint_entropy return TensionPoint(tensions=tuple(tensions), residual=float(residual))
[docs] def tension_to_rates(tension, marginal_entropies, joint_entropy): """ Convert tension coordinates back to a Gray-Wyner rate point. Inverts :func:`rates_to_tension`. Summing the per-source tensions and subtracting the residual eliminates the private rates and leaves ``(n - 1) R_0``, which is why at least two sources are required. Parameters ---------- tension : TensionPoint The point of the region of tension. marginal_entropies : iterable of float The marginal entropies ``H[X_i]``. joint_entropy : float The joint entropy ``H[X_{0:n}]``. Returns ------- point : GrayWynerPoint The corresponding Gray-Wyner rate point. """ tensions, marginals = _check_lengths(tension.tensions, marginal_entropies, "tension.tensions") n = tensions.size if n < 2: msg = "The region of tension requires at least two sources." raise ditException(msg) common = (tensions.sum() - tension.residual + marginals.sum() - joint_entropy) / (n - 1) private = tensions + marginals - common return GrayWynerPoint(common=float(common), private=tuple(private))
[docs] def rates_to_extension(point, joint_entropy): """ Convert a Gray-Wyner rate point to extension-profile coordinates. The private rates *are* the conditional entropies ``H[X_i | W]``, so this map only computes the residual coordinate. Parameters ---------- point : GrayWynerPoint The rate point ``(R_0, (R_1, ..., R_n))``. joint_entropy : float The joint entropy ``H[X_{0:n}]``. Returns ------- extension : ExtensionPoint The corresponding point of the extension profile. """ private = np.asarray(point.private, dtype=float) residual = point.common + private.sum() - joint_entropy return ExtensionPoint(conditional_entropies=tuple(private), residual=float(residual))
[docs] def extension_to_rates(extension, joint_entropy): """ Convert extension-profile coordinates back to a Gray-Wyner rate point. Inverts :func:`rates_to_extension`. Parameters ---------- extension : ExtensionPoint The point of the extension profile. joint_entropy : float The joint entropy ``H[X_{0:n}]``. Returns ------- point : GrayWynerPoint The corresponding Gray-Wyner rate point. """ private = np.asarray(extension.conditional_entropies, dtype=float) common = extension.residual + joint_entropy - private.sum() return GrayWynerPoint(common=float(common), private=tuple(private))
[docs] def rates_to_mutual_information(point, marginal_entropies): """ Convert a Gray-Wyner rate point to mutual information coordinates. Parameters ---------- point : GrayWynerPoint The rate point ``(R_0, (R_1, ..., R_n))``. marginal_entropies : iterable of float The marginal entropies ``H[X_i]``. Returns ------- info : MutualInformationPoint The corresponding point of the mutual information region. """ private, marginals = _check_lengths(point.private, marginal_entropies, "point.private") return MutualInformationPoint(marginals=tuple(marginals - private), joint=float(point.common))
[docs] def mutual_information_to_rates(info, marginal_entropies): """ Convert mutual information coordinates back to a Gray-Wyner rate point. Inverts :func:`rates_to_mutual_information`. Parameters ---------- info : MutualInformationPoint The point of the mutual information region. marginal_entropies : iterable of float The marginal entropies ``H[X_i]``. Returns ------- point : GrayWynerPoint The corresponding Gray-Wyner rate point. """ values, marginals = _check_lengths(info.marginals, marginal_entropies, "info.marginals") return GrayWynerPoint(common=float(info.joint), private=tuple(marginals - values))
def _pair_network(dist, rvs, crvs, bound): """ Build the Gray-Wyner network backing the interaction informations. Parameters ---------- dist : Distribution The distribution of interest. rvs : list of lists, None The two sources. crvs : list, None Variables to condition on. bound : int, None Optional cap on the cardinality of the probe. Returns ------- network : GrayWynerNetwork The network, guaranteed to have exactly two sources. """ from .network import GrayWynerNetwork network = GrayWynerNetwork(dist, rvs=rvs, crvs=crvs, bound=bound) if network.n != 2: msg = f"The interaction informations are defined for 2 sources, not {network.n}." raise ditException(msg) return network def _no_path_of_length_three(network): """ Whether every outcome has ``p(x|y) = 1`` or ``p(y|x) = 1``. This is exactly when the interaction informations vanish :cite:`li2018extended`. Checked only without conditioning variables. """ if network.crvs: return False d = network.dist.coalesce(network.rvs) d = d.copy(base="linear") d.make_dense() p = d.pmf.reshape([len(a) for a in d.alphabet]) with np.errstate(divide="ignore", invalid="ignore"): x_given_y = p / p.sum(axis=0, keepdims=True) y_given_x = p / p.sum(axis=1, keepdims=True) determined = np.isclose(x_given_y, 1.0) | np.isclose(y_given_x, 1.0) return bool(np.all(determined[p > 0]))
[docs] @unitful def maximal_interaction_information(dist, rvs=None, crvs=None, niter=None, maxiter=1000, bound=None): """ Compute the maximal interaction information. .. math:: G_{NNI}(X : Y) = \\max_U I[X{:}Y|U] - I[X{:}Y] the largest amount by which a probe can *increase* the dependence between ``X`` and ``Y`` :cite:`li2018extended`. It vanishes exactly when the support graph contains no path of length three, equivalently when ``p(x|y) = 1`` or ``p(y|x) = 1`` at every outcome of positive probability, and is bounded above by ``min{H[X|Y], H[Y|X]}``. Parameters ---------- dist : Distribution The distribution of interest. rvs : list of lists, None The two sources. If None, all variables of `dist` are used. crvs : list, None Variables to condition on. niter : int, None Number of basin hops to perform during the optimization. maxiter : int The number of iterations of the optimization subroutine to perform. bound : int, None Bound the size of the probe variable. Returns ------- G : float The maximal interaction information. """ network = _pair_network(dist, rvs, crvs, bound) if _no_path_of_length_three(network): return 0.0 marginals = network._marginal_entropies # max (v_XY - v_X - v_Y) = max (R_0 + R_1 + R_2) - H[X] - H[Y]. point = network.rate_point([-1.0, -1.0, -1.0], niter=niter, maxiter=maxiter, allow_signed=True) value = point.common + sum(point.private) - sum(marginals) ceiling = min(network._joint_entropy - h for h in marginals) return min(max(value, 0.0), ceiling)
[docs] @unitful def asymmetric_private_interaction_information(dist, rvs=None, crvs=None, niter=None, maxiter=1000, bound=None): """ Compute the asymmetric private interaction information. .. math:: G_{PNI}(X \\to Y) = \\max_{U \\perp X} I[X{:}Y|U] - I[X{:}Y] the maximal interaction information achievable by a probe independent of the *first* source :cite:`li2018extended`. Operationally, if Alice holds ``X`` and must generate ``Y`` using only public randomness ``W``, the least leakage ``I[Y:W]`` she can achieve is ``H[Y|X] - G_PNI(X -> Y)``. It vanishes exactly when the maximal interaction information does. This quantity is not symmetric; swap `rvs` for the other direction. Parameters ---------- dist : Distribution The distribution of interest. rvs : list of lists, None The two sources, in the order ``[X, Y]``. If None, all variables of `dist` are used. crvs : list, None Variables to condition on. niter : int, None Number of basin hops to perform during the optimization. maxiter : int The number of iterations of the optimization subroutine to perform. bound : int, None Bound the size of the probe variable. Returns ------- G : float The asymmetric private interaction information. """ network = _pair_network(dist, rvs, crvs, bound) if _no_path_of_length_three(network): return 0.0 h_x, h_y = network._marginal_entropies # I[X:Y|U] - I[X:Y] = v_XY - v_X - v_Y, and independence from X pins # v_X = 0, i.e. R_1 = H[X]. So maximize v_XY - v_Y = R_0 + R_2 - H[Y]. # Pinning v_X alone (rather than also pinning v_XY, which is only # without loss of generality via a lemma) keeps the trivial probe # feasible, which the optimizer needs as a starting point. point = network.rate_point( [-1.0, 0.0, -1.0], niter=niter, maxiter=maxiter, allow_signed=True, rate_equalities=[([0.0, 1.0, 0.0], h_x)], ) value = point.common + point.private[1] - h_y ceiling = min(network._joint_entropy - h for h in (h_x, h_y)) return min(max(value, 0.0), ceiling)
[docs] @unitful def symmetric_private_interaction_information(dist, rvs=None, crvs=None, niter=None, maxiter=1000, bound=None): """ Compute the symmetric private interaction information. .. math:: G_{PPI}(X : Y) = \\max_{U \\perp X,\\, U \\perp Y} I[X{:}Y|U] - I[X{:}Y] = \\max_{U \\perp X,\\, U \\perp Y} I[XY{:}U] the most that can be disclosed about the pair while telling an eavesdropper holding either source alone nothing :cite:`li2018extended`. It vanishes exactly when the support graph is acyclic. Parameters ---------- dist : Distribution The distribution of interest. rvs : list of lists, None The two sources. If None, all variables of `dist` are used. crvs : list, None Variables to condition on. niter : int, None Number of basin hops to perform during the optimization. maxiter : int The number of iterations of the optimization subroutine to perform. bound : int, None Bound the size of the probe variable. Returns ------- G : float The symmetric private interaction information. """ network = _pair_network(dist, rvs, crvs, bound) if _no_path_of_length_three(network): return 0.0 h_x, h_y = network._marginal_entropies # Independence from both sources pins v_X = v_Y = 0, i.e. R_1 = H[X] and # R_2 = H[Y]; then G = v_XY = R_0, to be maximized. point = network.rate_point( [-1.0, 0.0, 0.0], niter=niter, maxiter=maxiter, allow_signed=True, rate_equalities=[([0.0, 1.0, 0.0], h_x), ([0.0, 0.0, 1.0], h_y)], ) ceiling = min(network._joint_entropy - h for h in (h_x, h_y)) return min(max(point.common, 0.0), ceiling)