# substrate_generator/__init__.py
"""High-level interface for creating systems by specifying architecture."""
import string
from collections.abc import Callable
from collections.abc import Iterable
from collections.abc import Mapping
from collections.abc import Sequence
from functools import partial
from typing import Any
import numpy as np
from numpy.typing import NDArray
from pyphi.labels import NodeLabels
from pyphi.substrate import Substrate
from pyphi.utils import all_states
from . import ising
from . import unit_functions
from .mechanism_combinations import MECHANISM_COMBINATIONS
from .mechanism_combinations import composite
from .mechanisms import MECHANISMS
from .mechanisms import STATE_DEPENDENT
from .mechanisms import WEIGHTED
UNIT_FUNCTIONS = {
"ising": ising.probability,
"boolean": unit_functions.boolean_function,
"gaussian": unit_functions.gaussian,
"naka_rushton": unit_functions.naka_rushton,
"or": unit_functions.logical_or_function,
"and": unit_functions.logical_and_function,
"parity": unit_functions.logical_parity_function,
"nor": unit_functions.logical_nor_function,
"nand": unit_functions.logical_nand_function,
"nparity": unit_functions.logical_nparity_function,
}
# Register the ported substrate_modeler mechanisms under any name not already
# bound by the weighted-threshold logical gates above (so ``"and"``/``"or"``
# keep their existing meaning in :func:`build_substrate`).
UNIT_FUNCTIONS.update(
{name: func for name, func in MECHANISMS.items() if name not in UNIT_FUNCTIONS}
)
__all__ = [
"MECHANISMS",
"MECHANISM_COMBINATIONS",
"UNIT_FUNCTIONS",
"build_substrate",
"build_tpm",
"composite",
"create_substrate",
]
[docs]
def build_tpm(
unit_functions: str | Callable | Iterable[str | Callable],
weights: NDArray[Any],
**kwargs,
):
"""Build a binary state-by-node TPM from unit functions and a weight matrix.
Each unit function is evaluated at every one of the ``2**N`` substrate
states, returning the probability that its node is ON at the next step.
Parameters
----------
unit_functions : str or Callable or Iterable[str or Callable]
A single unit function (a name in :data:`UNIT_FUNCTIONS` or a callable
with signature ``f(element, weights, state, **kwargs) -> float``)
applied to every node, or an iterable of ``N`` such functions, one per
node.
weights : numpy.ndarray
Square ``N × N`` connection weight matrix; ``weights[i, j]`` couples
input ``i`` to node ``j``.
Other Parameters
----------------
**kwargs
Passed through to every unit function.
Returns
-------
numpy.ndarray
A state-by-node TPM of shape ``(2,) * N + (N,)``; entry
``tpm[state + (j,)]`` is the probability that node ``j`` is ON given
``state``.
Raises
------
ValueError
If ``weights`` is not square, or if an iterable of unit functions does
not have one entry per node.
"""
if weights.ndim != 2 or weights.shape[0] != weights.shape[1]:
raise ValueError("weights must be a square matrix")
N = weights.shape[0]
# Normalize unit_functions to a list
if isinstance(unit_functions, str):
# Single function name string - use for all nodes
unit_functions_list: list[str | Callable] = [unit_functions] * N
elif callable(unit_functions):
# Single function - use for all nodes
unit_functions_list = [unit_functions] * N
else:
# Iterable of functions
unit_functions_list = list(unit_functions)
if len(unit_functions_list) != weights.shape[0]:
raise ValueError(
"Number of unit functions must match number of nodes in weight matrix"
)
tpm = np.zeros([2] * N + [N])
for state in all_states(N):
for element, func in enumerate(unit_functions_list):
if isinstance(func, str):
unit_func = UNIT_FUNCTIONS[func]
else:
unit_func = func
tpm[(*state, element)] = unit_func(element, weights, state, **kwargs)
return tpm
[docs]
def build_substrate(
unit_functions: str | Callable | Iterable[str | Callable],
weights: NDArray[Any],
node_labels: NodeLabels | None = None,
**kwargs,
):
"""Build a PyPhi substrate from a weight matrix and unit functions.
The connectivity matrix is derived from the nonzero entries of ``weights``.
Parameters
----------
unit_functions : str or Callable or Iterable[str or Callable]
A single unit function applied to every node, or an iterable of one
function per node. Each function has signature
``f(element, weights, state, **kwargs) -> float`` returning a
probability; a string names an entry in :data:`UNIT_FUNCTIONS`.
weights : numpy.ndarray
Square weight matrix describing the substrate's connectivity.
node_labels : NodeLabels, optional
Labels for the nodes. Defaults to ``A, B, C, ...``.
Other Parameters
----------------
**kwargs
Passed through to every unit function.
Returns
-------
Substrate
A PyPhi substrate whose TPM is built by :func:`build_tpm` and whose
connectivity matrix marks the nonzero entries of ``weights``.
"""
if node_labels is None:
# Create default labels from uppercase letters
N = weights.shape[0]
node_labels = NodeLabels(string.ascii_uppercase[:N], range(N))
tpm = build_tpm(unit_functions, weights, **kwargs)
cm = (weights != 0).astype(int)
return Substrate(tpm, cm=cm, node_labels=node_labels)
def _sub_specs(spec: Mapping[str, Any]):
"""Yield the simple sub-specs of a node spec (the spec itself if not composite)."""
if "composite" in spec:
yield from spec["composite"]
else:
yield spec
def _connectivity_indices(spec: Mapping[str, Any]):
"""Yield every input index of a node spec (for connectivity markers)."""
for sub in _sub_specs(spec):
for i in sub["inputs"]:
yield int(i)
def _real_weight_pairs(spec: Mapping[str, Any]):
"""Yield ``(input_index, weight)`` for the *weighted* sub-mechanisms only.
Input weights come from ``params['input_weights']`` (or ``params['weights']``),
aligned to ``inputs`` in order; further inputs (e.g. modulators) are left as
connectivity markers.
"""
for sub in _sub_specs(spec):
if sub["mechanism"] not in WEIGHTED:
continue
params = sub.get("params", {})
input_weights = params.get("input_weights", params.get("weights"))
if input_weights is None:
continue
for i, w in zip(sub["inputs"], input_weights, strict=False):
yield int(i), float(w)
def _is_state_dependent(spec: Mapping[str, Any]) -> bool:
if "composite" in spec:
return any(sub["mechanism"] in STATE_DEPENDENT for sub in spec["composite"])
return spec.get("mechanism") in STATE_DEPENDENT
def _node_function(spec: Mapping[str, Any]) -> Callable:
"""Resolve a node spec to a bound unit function ``f(element, weights, state)``."""
if "composite" in spec:
return composite(
spec["composite"],
spec.get("mechanism_combination", "selective"),
**spec.get("combination_params", {}),
)
base = MECHANISMS[spec["mechanism"]]
return partial(base, inputs=tuple(spec["inputs"]), **spec.get("params", {}))
[docs]
def create_substrate(
node_params: Mapping[int, Mapping[str, Any]] | Sequence[Mapping[str, Any]],
labels: Sequence[str] | NodeLabels | None = None,
) -> Substrate:
"""Build a :class:`~pyphi.substrate.Substrate` from per-node specifications.
Each node names a ``mechanism`` (a key in :data:`MECHANISMS`), its
``inputs``, and a ``params`` dict. The resulting substrate TPM matches the
``dynamic_tpm`` of the ``substrate_modeler`` library, in which the present
state equals the past state.
Parameters
----------
node_params : Mapping[int, Mapping] or Sequence[Mapping]
Either an integer-keyed mapping (nodes taken in sorted key order) or an
ordered iterable of node specs. Each spec is a dict with keys:
- ``"mechanism"`` (str): mechanism name, **or**
- ``"composite"`` (list of sub-specs) plus optional
``"mechanism_combination"`` (str, default ``"selective"``) and
``"combination_params"`` (dict);
- ``"inputs"`` (tuple[int]): the unit's input indices;
- ``"params"`` (dict): mechanism parameters (e.g. ``input_weights``,
``determinism``, ``threshold``, ``weight_scale_mapping``).
labels : Sequence[str] or NodeLabels, optional
Node labels. Defaults to ``A, B, C, ...``.
Returns
-------
Substrate
A PyPhi substrate.
Raises
------
TypeError
If ``node_params`` is a mapping whose keys are not all integers.
Notes
-----
The connectivity matrix is assembled in two passes: every input edge is
first marked with weight 1, then the real edge weights of weighted
mechanisms (:data:`~pyphi.substrate_generator.mechanisms.WEIGHTED`) overwrite
those markers. State-dependent mechanisms
(:data:`~pyphi.substrate_generator.mechanisms.STATE_DEPENDENT`) read the
unit's own state, so a self-loop is inserted into the connectivity matrix
when the spec does not already include one.
"""
if isinstance(node_params, Mapping):
if not all(isinstance(k, int) for k in node_params):
raise TypeError(
"node_params mapping must be keyed by integer node index; "
"pass an ordered list otherwise"
)
specs = [node_params[k] for k in sorted(node_params)]
else:
specs = list(node_params)
n = len(specs)
weights = np.zeros((n, n))
functions: list[Callable] = []
for j, spec in enumerate(specs):
# Pass 1: connectivity markers for every input edge.
for i in _connectivity_indices(spec):
weights[i, j] = 1.0
# Pass 2: real edge weights override markers for weighted mechanisms.
for i, w in _real_weight_pairs(spec):
weights[i, j] = w
# Self-loop so a state-dependent unit's own-state dependence is in the CM.
if _is_state_dependent(spec) and weights[j, j] == 0:
weights[j, j] = 1.0
functions.append(_node_function(spec))
if labels is None:
labels = NodeLabels(string.ascii_uppercase[:n], range(n))
elif not isinstance(labels, NodeLabels):
labels = NodeLabels(tuple(labels), range(n))
return build_substrate(functions, weights, node_labels=labels)
[docs]
def random_substrate(
n: int,
*,
seed: int,
epsilon: float = 0.0,
cm: NDArray[np.int_] | None = None,
) -> Substrate:
"""A seeded random binary substrate of ``n`` units.
Each unit's ON probability, for every previous state, is drawn
independently and uniformly from ``(epsilon, 1 − epsilon)`` using an
isolated generator seeded with ``seed``, so the substrate is exactly
reproducible from ``(n, seed, epsilon)``.
Parameters
----------
n : int
Number of binary units.
seed : int
Seed for the isolated random generator.
epsilon : float, optional
Margin keeping probabilities away from 0 and 1 (deterministic
entries); ``0.0`` (the default) permits near-deterministic rows.
cm : numpy.ndarray, optional
Connectivity matrix. If ``None``, connectivity is inferred from
the sampled TPM (generically fully connected).
Returns
-------
Substrate
The sampled substrate.
"""
rng = np.random.default_rng(seed)
tpm = rng.uniform(epsilon, 1.0 - epsilon, size=(2**n, n))
return Substrate(tpm, cm=cm)
