# formalism/iit3/__init__.py
"""IIT 3.0 (Oizumi et al. 2014) algorithms.
Distribution-distance-based phi computation. Partition scheme:
``JOINT_BIPARTITION``. Compatible measures: ``EMD``, ``L1``, ``KLD``,
``ENTROPY_DIFFERENCE``, ``PSQ2``, ``MP2Q``, ``AID`` (absolute intrinsic
difference), ``ID`` (intrinsic difference).
The dispatch class :class:`IIT3Formalism` lives in :mod:`.formalism` and
calls back into this module's ``sia`` / ``ces`` / etc. for the actual
computation.
"""
from __future__ import annotations
import contextlib
import logging
from collections.abc import Iterable
from typing import TYPE_CHECKING
from typing import Any
from more_itertools import collapse
from pyphi import conf
from pyphi import connectivity
from pyphi import utils
from pyphi.conf import config
from pyphi.direction import Direction
from pyphi.measures.ces import ces_distance
from pyphi.models import DirectedBipartition
from pyphi.models import Distinction
from pyphi.models import Distinctions
from pyphi.models import IIT3SystemIrreducibilityAnalysis
from pyphi.models import NullResultReason
from pyphi.models import ResolvedDistinctions
from pyphi.models import UnresolvedDistinctions
from pyphi.models import _null_sia
from pyphi.models.ces import CauseEffectStructure
from pyphi.models.explanation import runner_up_from_candidates
from pyphi.parallel import map_reduce
from pyphi.partition import system_partition_types
from pyphi.relations import NullRelations
from pyphi.types import Mechanism
from pyphi.types import Purview
if TYPE_CHECKING:
from pyphi.labels import NodeLabels
from pyphi.system import System
# Create a logger for this module.
log = logging.getLogger(__name__)
def _check_active_version() -> None:
"""Raise unless the active formalism version is IIT 3.0.
Mechanism-level work dispatches through the *active* formalism, so
calling these functions under another version would silently mix
formalisms.
"""
from pyphi.conf._field_routing import ConfigurationError
version = config.formalism.iit.version
if version != "IIT_3_0":
raise ConfigurationError(
f"pyphi.formalism.iit3 functions dispatch mechanism-level work "
f"through the active formalism, which is {version!r}; pin "
f"IIT 3.0 first, e.g. with "
f"pyphi.config.override(**pyphi.conf.presets.iit3)"
)
[docs]
def concept(
system: System,
mechanism: Mechanism,
purviews: Iterable[Purview] | None = None,
cause_purviews: Iterable[Purview] | None = None,
effect_purviews: Iterable[Purview] | None = None,
**kwargs: Any,
) -> Distinction:
"""Return the IIT 3.0 concept specified by a mechanism.
A concept pairs the mechanism's maximally irreducible cause (its MIC) with
its maximally irreducible effect (its MIE). The empty mechanism specifies
the null concept.
Parameters
----------
system : System
The system the mechanism belongs to.
mechanism : tuple[int]
The mechanism for which to determine the concept.
purviews : tuple[tuple[int]], optional
A list of purviews to consider.
cause_purviews : tuple[tuple[int]], optional
A list of cause purviews to consider, overriding ``purviews``.
effect_purviews : tuple[tuple[int]], optional
A list of effect purviews to consider, overriding ``purviews``.
Returns
-------
Distinction
The concept of the given mechanism.
"""
from pyphi.formalism.queries import distinction
return distinction(
system,
mechanism,
purviews=purviews,
cause_purviews=cause_purviews,
effect_purviews=effect_purviews,
**kwargs,
)
def _compute_distinctions(
system: System,
mechanisms: Iterable[Mechanism] | None = None,
purviews: Iterable[Purview] | None = None,
cause_purviews: Iterable[Purview] | None = None,
effect_purviews: Iterable[Purview] | None = None,
directions: Iterable[Direction] | None = None,
only_positive_phi: bool = True,
**kwargs: Any,
) -> Distinctions:
"""Compute the bag of distinctions for a system, restricted by the
given mechanism / purview / direction filters.
Distinctions are computed over the mechanisms via
:func:`pyphi.parallel.map_reduce`, so the ``parallel`` keyword and the
other keys of ``config.infrastructure.parallel_distinction_evaluation``
govern the work.
Parameters
----------
system : System
The system for which to determine the
:class:`~pyphi.models.distinctions.Distinctions`.
mechanisms : tuple[tuple[int]], optional
Restrict possible mechanisms to those in this list. If ``None``, the
non-empty powerset of system nodes is used.
purviews : tuple[tuple[int]], optional
Same as in :func:`pyphi.formalism.iit3.concept`.
cause_purviews : tuple[tuple[int]], optional
Same as in :func:`pyphi.formalism.iit3.concept`.
effect_purviews : tuple[tuple[int]], optional
Same as in :func:`pyphi.formalism.iit3.concept`.
directions : Iterable[Direction], optional
Restrict possible directions to these.
only_positive_phi : bool
Whether to only return distinctions with positive φ (default
``True``).
parallel : bool, optional
Whether to compute distinctions in parallel. If given, overrides
:data:`config.infrastructure.parallel_distinction_evaluation`.
Returns
-------
Distinctions
An :class:`~pyphi.models.UnresolvedDistinctions` over the surviving
distinctions. It is unresolved because the active formalism's MICE search
may return tied specified states; IIT 3.0 callers route through
``ces_distance``, which accepts the base distinctions unchanged.
"""
total = None
if mechanisms is None:
mechanisms = utils.powerset(system.node_indices, nonempty=True)
total = 2 ** len(system.node_indices) - 1
else:
with contextlib.suppress(TypeError):
total = len(mechanisms) # type: ignore[arg-type] # mechanisms may be generator
def compute_concept(*args, **kwargs):
return concept(system, *args, **kwargs, progress=False)
reduce_func = _only_positive_phi if only_positive_phi else _any_phi
parallel_kwargs = conf.parallel_kwargs(
dict(config.infrastructure.parallel_distinction_evaluation), **kwargs
)
concepts = map_reduce(
compute_concept,
mechanisms,
map_kwargs={ # type: ignore[arg-type] # None values allowed in map_kwargs
"purviews": purviews,
"cause_purviews": cause_purviews,
"effect_purviews": effect_purviews,
"directions": directions,
},
reduce_func=reduce_func,
desc="Computing distinctions",
total=total,
# Larger mechanisms span larger cause/effect repertoires; a cheap
# directional proxy for per-mechanism cost.
size_func=lambda mechanism: 2 ** len(mechanism),
**parallel_kwargs, # type: ignore[arg-type] # parallel_kwargs contains map_reduce params
)
# ``find_mice`` may return tied specified states under the active
# formalism (IIT 4.0 in particular), so the conservative answer is
# ``UnresolvedDistinctions``. Callers in IIT 4.0 phi_structure
# immediately call ``resolve_congruence``; IIT 3.0 callers route
# through ``ces_distance`` which accepts the base ``Distinctions``.
return UnresolvedDistinctions(concepts)
[docs]
def ces(
system: System,
*,
sia: IIT3SystemIrreducibilityAnalysis | None = None,
distinctions: Distinctions | None = None,
sia_kwargs: dict | None = None,
distinctions_kwargs: dict | None = None,
) -> CauseEffectStructure:
"""Compute the cause-effect structure of a system under IIT 3.0.
Returns a :class:`CauseEffectStructure` wrapping the SIA, the resolved
distinctions, and an empty :class:`NullRelations` (IIT 3.0 does not
define relations between distinctions).
Pass ``sia=`` or ``distinctions=`` to reuse pre-computed values.
"""
import time
from pyphi.provenance import stamp_wall_time
_check_active_version()
start = time.perf_counter()
sia_kwargs = sia_kwargs or {}
distinctions_kwargs = distinctions_kwargs or {}
if sia is None:
sia = _sia(system, **sia_kwargs)
if distinctions is None:
cached = sia.distinctions
if cached is not None and not distinctions_kwargs:
distinctions = cached
else:
distinctions = _compute_distinctions(system, **distinctions_kwargs)
if not isinstance(distinctions, ResolvedDistinctions):
distinctions = ResolvedDistinctions(distinctions)
result = CauseEffectStructure(
sia=sia,
distinctions=distinctions,
relations=NullRelations(),
)
return stamp_wall_time(result, time.perf_counter() - start)
def _only_positive_phi(concepts: Iterable[Any]) -> list[Distinction]:
return list(filter(None, collapse(concepts)))
def _any_phi(concepts: Iterable[Any]) -> list[Distinction]:
return list(collapse(concepts))
[docs]
def conceptual_info(system: System, **kwargs: Any) -> float:
"""Return the conceptual information for a :class:`~pyphi.system.System`.
This is the distance from the system's
:class:`~pyphi.models.distinctions.Distinctions` to the null concept.
"""
_check_active_version()
ci = ces_distance(
_compute_distinctions(system, **kwargs), ResolvedDistinctions(()), system=system
)
return round(ci, config.numerics.precision) # type: ignore[arg-type] # config.Option descriptor
[docs]
def evaluate_partition(
partition: DirectedBipartition,
unpartitioned_system: System,
unpartitioned_ces: Distinctions,
**kwargs: Any,
) -> IIT3SystemIrreducibilityAnalysis:
"""Compute the system irreducibility for a given partition.
The partition is applied to the system, the partitioned system's
distinctions are computed, and φ is the CES distance
(:func:`pyphi.measures.ces.ces_distance`) between the unpartitioned and
partitioned distinctions.
Parameters
----------
partition : DirectedBipartition
The partition to evaluate.
unpartitioned_system : System
The system without a partition applied.
unpartitioned_ces : Distinctions
The cause-effect structure of the unpartitioned system.
Returns
-------
IIT3SystemIrreducibilityAnalysis
The Φ analysis for that partition.
"""
log.debug("Evaluating %s...", partition)
partitioned_system = unpartitioned_system.apply_partition(partition)
if config.formalism.iit.assume_partitions_cannot_create_new_concepts:
mechanisms = list(unpartitioned_ces.mechanisms)
else:
# Mechanisms can only produce concepts if they were concepts in the
# original system, or the partition splits the mechanism.
mechanisms = set(
list(unpartitioned_ces.mechanisms)
+ list(partitioned_system.partitioned_mechanisms)
)
kwargs = {"progress": False, **kwargs}
partitioned_ces = _compute_distinctions(partitioned_system, mechanisms, **kwargs)
log.debug("Finished evaluating %s.", partition)
phi_ = ces_distance(unpartitioned_ces, partitioned_ces, system=unpartitioned_system)
return IIT3SystemIrreducibilityAnalysis(
phi=phi_,
partitioned_distinctions=partitioned_ces,
partition=partitioned_system.partition,
node_indices=unpartitioned_system.node_indices,
node_labels=unpartitioned_system.substrate.node_labels,
current_state=unpartitioned_system.proper_state,
)
[docs]
def sia_partitions(
nodes: tuple[int, ...], node_labels: NodeLabels | None = None
) -> list[DirectedBipartition]:
"""Return all Φ cuts for the given nodes.
The scheme is chosen by the
:const:`config.formalism.iit.system_partition_scheme` option, which for IIT
3.0 must be one of the schemes in
:attr:`IIT3Formalism.compatible_system_partition_schemes`
(``DIRECTED_BIPARTITION`` or ``DIRECTED_BIPARTITION_CUT_ONE``); any other
value raises :class:`ValueError`.
Parameters
----------
nodes : tuple[int]
The node indices to partition.
node_labels : NodeLabels, optional
Enables printing the partition with labels.
Returns
-------
list[DirectedBipartition]
All unidirectional partitions.
Raises
------
ValueError
If the configured system partition scheme is not compatible with IIT
3.0.
"""
# TODO(4.0 consolidate 3.0 and 4.0 cuts)
scheme = config.formalism.iit.system_partition_scheme
from pyphi.formalism.iit3.formalism import IIT3Formalism
valid = IIT3Formalism.compatible_system_partition_schemes
if valid is not None and scheme not in valid:
raise ValueError(
"IIT 3.0 calculations must use one of the following system "
f"partition schemes: {sorted(valid)}; got {scheme}"
)
return system_partition_types[config.formalism.iit.system_partition_scheme]( # type: ignore[index] # config.Option descriptor
nodes, node_labels=node_labels
)
def _ces(system: System, **kwargs: Any) -> Distinctions:
"""Compute the unpartitioned :class:`~pyphi.models.distinctions.Distinctions`
using the partition-evaluation parallel settings.
"""
kwargs = {**dict(config.infrastructure.parallel_partition_evaluation), **kwargs}
return _compute_distinctions(system, **kwargs)
def _sia_map_reduce(
cuts: Iterable[DirectedBipartition],
system: System,
unpartitioned_ces: Distinctions,
**kwargs: Any,
) -> IIT3SystemIrreducibilityAnalysis:
"""Evaluate every partition and select the MIP via
``resolve_ties.sias``. The config knob
``config.formalism.iit.sia_tie_resolution`` controls the selection
strategy.
"""
from pyphi import resolve_ties
kwargs = {**dict(config.infrastructure.parallel_partition_evaluation), **kwargs}
null = _null_sia(system, reasons=[NullResultReason.NO_VALID_PARTITIONS])
cuts = list(cuts)
candidates = map_reduce(
evaluate_partition,
cuts,
map_kwargs={
"unpartitioned_system": system,
"unpartitioned_ces": unpartitioned_ces,
},
shortcircuit_func=utils.is_falsy,
desc="Evaluating cuts",
**kwargs,
)
if not candidates:
return null
ties = tuple(resolve_ties.sias(candidates, default=null))
if not ties:
return null
winner = ties[0]
# A short-circuited sweep evaluated only a prefix of the cuts, so its
# next-best candidate is not the runner-up over all of them.
if len(candidates) == len(cuts):
winner.runner_up = runner_up_from_candidates(candidates, winner.phi)
# Capture phi-tied peers (same minimum phi value) before any
# lex-based tiebreaker collapses the set to a single winner. These
# peers are equally valid MIPs for the system; preserving them on
# the winner lets callers (serialization, diagnostic display, and
# fixtures comparing across tied alternatives) see the full tie set.
phi_ties = tuple(resolve_ties.sias(candidates, strategy=["PHI"], default=null))
winner.set_ties(list(phi_ties) if len(phi_ties) > 1 else [winner])
return winner
def _sia(system: System, **kwargs: Any) -> IIT3SystemIrreducibilityAnalysis:
"""Return the minimal information partition of a system.
Φ is necessarily zero, and a null analysis is returned immediately, when the
system is empty, is not strongly connected, or is a single micro element
with no self-loop (or with a self-loop but
``config.formalism.iit.single_micro_nodes_with_selfloops_have_phi`` unset).
Otherwise the unpartitioned distinctions are computed once, every system
partition is evaluated, and the partition of minimum φ is selected.
Parameters
----------
system : System
The candidate set of nodes.
**kwargs
Options of the partition evaluation's
:func:`~pyphi.parallel.map_reduce` (``parallel``, ``progress``, ...).
Any other name raises ``TypeError``.
Returns
-------
IIT3SystemIrreducibilityAnalysis
A nested structure containing all the data from the intermediate
calculations. The top level contains the basic irreducibility
information for the given system.
"""
conf.reject_unknown_kwargs("sia", kwargs)
_check_active_version()
log.info("Calculating big-phi data for %s...", system)
# Check for degenerate cases
# =========================================================================
# Phi is necessarily zero if the system is:
# - not strongly connected;
# - empty;
# - an elementary micro mechanism (i.e. no nontrivial bipartitions).
# So in those cases we immediately return a null SIA.
if not system:
log.info("System %s is empty; returning null SIA immediately.", system)
return _null_sia(system, reasons=[NullResultReason.NO_SYSTEM])
if not connectivity.is_strong(system.cm, system.node_indices):
log.info(
"%s is not strongly connected; returning null SIA immediately.",
system,
)
return _null_sia(system, reasons=[NullResultReason.NO_STRONG_CONNECTIVITY])
# Handle elementary micro mechanism cases.
# Single macro element systems have nontrivial bipartitions because their
# bipartitions are over their micro elements.
if len(system.partition_indices) == 1:
# If the node lacks a self-loop, phi is trivially zero.
if not system.cm[system.node_indices][system.node_indices]:
log.info(
"Single micro nodes %s without selfloops cannot have "
"phi; returning null SIA immediately.",
system,
)
return _null_sia(system, reasons=[NullResultReason.MONAD_WITH_NO_SELFLOOP])
# Even if the node has a self-loop, we may still define phi to be zero.
if not config.formalism.iit.single_micro_nodes_with_selfloops_have_phi:
log.info(
"Single micro nodes %s with selfloops cannot have "
"phi; returning null SIA immediately.",
system,
)
return _null_sia(
system,
reasons=[NullResultReason.MONAD_WITH_SELFLOOP_DEFINED_TO_BE_ZERO_PHI],
)
# =========================================================================
log.debug("Finding unpartitioned Distinctions...")
unpartitioned_ces = _ces(system, progress=kwargs.get("progress"))
if not unpartitioned_ces:
log.info("Empty unpartitioned Distinctions; returning null SIA immediately.")
# Short-circuit if there are no concepts in the unpartitioned CES.
return _null_sia(system, reasons=[NullResultReason.EMPTY_CAUSE_EFFECT_STRUCTURE])
log.debug("Found unpartitioned Distinctions.")
# TODO: move this into sia_bipartitions?
# Only True if SINGLE_MICRO_NODES...=True, no?
if len(system.partition_indices) == 1:
cuts = [
DirectedBipartition(
Direction.EFFECT,
system.partition_indices,
system.partition_indices,
system.partition_node_labels,
)
]
else:
cuts = sia_partitions(system.partition_indices, system.partition_node_labels)
result = _sia_map_reduce(cuts, system, unpartitioned_ces, **kwargs)
# Attach the unpartitioned distinctions already computed above so that
# ``ces`` can reuse them rather than recomputing. They are a property of
# the system and state, not of the partition, so every phi-tied peer
# shares the same CES; attaching it across the tie set keeps the
# alternatives comparable under ``__eq__`` (which compares distinctions).
for peer in result.ties:
peer.distinctions = unpartitioned_ces
if config.infrastructure.clear_system_caches_after_computing_sia:
log.debug("Clearing system caches.")
system.clear_caches()
log.info("Finished calculating big-phi data for %s.", system)
return result
sia = _sia
[docs]
def phi(system: System) -> float:
"""Return the Φ value of a system."""
return sia(system).phi
