Interactions

Define interactions, the parts of a system whose laws couple elements to each other or to the environment.

AdvectionInteraction(label, *, target, state='u', drift=None, source=None, limiter='minmod', riemann='rusanov', sources=())

Bases: ConservationInteraction

Advection du/dt + divergence(a u) = source(ns) of a state u of a field by a drift field a(ns).

The flux of the conservation law is the drift times the state, written in conservative form, so the total of the state on a periodic field stays constant and, on a spectral field, the dealiasing of products applies to it. The drift is one expression for a field of one dimension, or one per dimension, of the states and parameters of the field, and may read other states than the one it carries, such as the flow of a fluid derived from a streamfunction, or the drift of a probability density in a Fokker-Planck equation.

Parameters:

Name	Type	Description	Default
label	str	The label of the interaction, unique among the parts of its system.	required
target	Annotated[str, RoleSpec()]	The path of the field whose state is carried.	required
state	str	The name of the carried state of the field.	'u'
drift	Callable[[SimpleNamespace], object] | None	A function of the namespace that returns the drift, or one drift per dimension.	None
source	Callable[[SimpleNamespace], object] | None	A function of the namespace that returns the source, or None for none.	None
limiter	str | None	The slope limiter of the reconstruction of a finite-volume field, 'minmod', 'van-leer', or None.	'minmod'
riemann	str | None	The numerical flux of a finite-volume field, 'rusanov' or 'godunov', or None for the divergence of the flux on a field of another discretization.	'rusanov'
sources	Annotated[Sequence[str], RoleSpec(variadic=True)]	The paths of other elements the drift and the source read, as the namespaces sources. A field over the position lattice of a target on a phase space is read broadcast along the class axes.	()

Raises:

Type	Description
ValidationError	The state is not a name, a function is not callable, or the limiter or the numerical flux is unknown.

Examples:

>>> from typing import Annotated
>>> import numpy as np
>>> from skcomplex.core.elements import Element, FieldElement
>>> from skcomplex.core.interactions import AdvectionInteraction
>>> from skcomplex.core.spaces import LatticeSpace
>>> from skcomplex.core.system import System
>>> from skcomplex.engine.runtime import Simulation
>>> from skcomplex.specs import StateSpec
>>> class CellElement(Element):
...     def __init__(self, label, u: Annotated[float, StateSpec()] = 0.0):
...         super().__init__(label)
...         self.u = u
>>> space = LatticeSpace((50,), spacing=0.1, boundary='periodic')
>>> values = {'u': np.exp(-((np.arange(50) - 25) / 5.0) ** 2)}
>>> volumes = 'finite-volume'
>>> field = FieldElement('tracer', element=CellElement, space=space, values=values, discretization=volumes)
>>> drift = AdvectionInteraction('drift', target='tracer', state='u', drift=lambda s: 1.0)
>>> system = System(elements=[field], interactions=[drift])
>>> trajectory = Simulation(system, dt=0.01).check().prepare().run(until=1.0).trajectory_
>>> total = trajectory.values('tracer.u')[-1].sum()
>>> round(float(total), 6) == round(float(values['u'].sum()), 6)
True

Source code in src/skcomplex/core/interactions/_advection.py

def __init__(
    self,
    label: str,
    *,
    target: Annotated[str, RoleSpec()],
    state: str = 'u',
    drift: Callable[[SimpleNamespace], object] | None = None,
    source: Callable[[SimpleNamespace], object] | None = None,
    limiter: str | None = 'minmod',
    riemann: str | None = 'rusanov',
    sources: Annotated[Sequence[str], RoleSpec(variadic=True)] = (),
) -> None:
    super().__init__(
        label,
        target=target,
        state=state,
        source=source,
        limiter=limiter,
        riemann=riemann,
        sources=sources,
    )
    self.drift = drift

BaseInteraction(label)

Bases: BasePart[BaseSpace, SimpleNamespace]

Base of every interaction, a part of a system whose laws couple elements to each other or to the environment.

A subclass takes its roles and parameters as arguments of its own __init__, annotated with Annotated and the spec RoleSpec or ParameterSpec, and stores each under its own name, as in scikit-learn. It returns its laws from _get_laws, where namespace.<role> is the namespace of the element bound to the role and namespace.<parameter> is a symbol. Interactions hold no state: state on couplings lives in groups on the edges or hyperedges of a network space. A general interaction reads the members of its roles, such as 'source' or 'target', by the name _get_role returns for them, the role itself by default, so a subclass that names its roles otherwise returns its own names, in the agent forms and, through _get_member, in the symbolic forms. It reads its parameters by the name _get_parameter returns for them, and a parameter the class does not declare reads the plain attribute, so a subclass renames a parameter or fixes its value by setting the attribute.

Parameters:

Name	Type	Description	Default
label	str	The label of the interaction, unique among the parts of its system.	required

Attributes:

Name	Type	Description
laws_	Sequence[BasePart]	The laws the interaction contributes for the namespace of its last run.

Examples:

>>> from typing import Annotated
>>> from skcomplex.core.interactions._base import BaseInteraction
>>> from skcomplex.specs import ParameterSpec, RoleSpec
>>> class GrowthInteraction(BaseInteraction):
...     def __init__(
...         self, label, target: Annotated[str, RoleSpec()], r: Annotated[float, ParameterSpec()] = 1.0
...     ):
...         super().__init__(label)
...         self.target = target
...         self.r = r
...
...     def _get_laws(self, namespace):
...         return []
>>> GrowthInteraction('growth', target='rabbits', r=0.5).get_params()
{'label': 'growth', 'r': 0.5, 'target': 'rabbits'}

Source code in src/skcomplex/core/base.py

def __init__(self, label: str) -> None:
    """Store the label of the part.

    Args:
        label:
            The label of the part, unique among the parts of its system.
    """
    self.label = label

CascadeInteraction(label, *, target, state='x', threshold=1.0, share=0.0, mode='redistribute', drive=None, neighborhood=None, layer=None)

Bases: NeighborInteraction

Cascade of the members whose state reaches a threshold, which relax and pass shares to their neighbours.

At every step the drive changes the state, and then every member whose state reaches threshold relaxes, all at once, until no member does. In the mode 'redistribute' a relaxing member loses threshold and each of its neighbours gains share. In the mode 'reset' a relaxing member falls to 0 and each of its neighbours gains share times the state it relaxed from. In the mode 'random' a relaxing member loses threshold in parcels of share, and every parcel goes to one of its neighbours drawn at random. Neighbours outside a fixed boundary take no share, so it leaves the system, and the mode 'random' treats every lattice as bounded. Every cascade is recorded as an event with its size, the number of relaxations, its duration, and its area.

Parameters:

Name	Type	Description	Default
label	str	The label of the interaction, unique among the parts of its system.	required
target	Annotated[str, RoleSpec()]	The path of the group on the nodes or sites.	required
state	str	The name of the state of the members.	'x'
threshold	float	The value of the state at which a member relaxes, fixed during a run.	1.0
share	float	What a neighbour of a relaxing member gains, an amount, a fraction of the state in the mode 'reset', or the amount of one parcel in the mode 'random', fixed during a run.	0.0
mode	str	'redistribute', 'reset', or 'random'.	'redistribute'
drive	Callable[..., Array] | None	A function f(values, *, random) of the array of the states and the random generator of the simulation that returns the states after the drive of a step, or None for no drive.	None
neighborhood	BaseNeighborhood | None	The neighbourhood of the member itself first and its neighbours, or None for the von Neumann neighbourhood of a square lattice with its centre. A network takes AdjacencyNeighborhood(center=True), and the mode 'random' takes a lattice neighbourhood.	None
layer	str | None	The layer of the network whose edges the interaction runs on, or None for the graph of the network.	None

Raises:

Type	Description
ValidationError	The mode is unknown, the drive is not callable, the neighbourhood leaves out the member itself, or the mode 'random' has no lattice neighbourhood or a threshold that is not a whole number of parcels.

Examples:

>>> from skcomplex.core.interactions import CascadeInteraction
>>> cascade = CascadeInteraction('cascade', target='sites', state='height', threshold=4, share=1)
>>> cascade.check().mode
'redistribute'

Source code in src/skcomplex/core/interactions/_cascade.py

def __init__(
    self,
    label: str,
    *,
    target: Annotated[str, RoleSpec()],
    state: str = 'x',
    threshold: float = 1.0,
    share: float = 0.0,
    mode: str = 'redistribute',
    drive: Callable[..., Array] | None = None,
    neighborhood: BaseNeighborhood | None = None,
    layer: str | None = None,
) -> None:
    super().__init__(label, layer=layer)
    self.target = target
    self.state = state
    self.threshold = threshold
    self.share = share
    self.mode = mode
    self.drive = drive
    self.neighborhood = neighborhood

ChoiceInteraction(label, *, target, state, options, utility, rule='logit', intensity=1.0, schedule='synchronous', activity=1.0, form='symbolic', phase=None, proposal='others', ties='random')

Bases: Interaction

Choice of every member of a group among options, drawn every step from the utilities of the options by a rule.

The rule 'logit', which is the heat-bath rule, draws option k of utility u_k with probability exp(beta u_k) / sum_j exp(beta u_j), where beta is the intensity of choice. The rules 'metropolis' and 'fermi' propose one of the other options uniformly, or with the proposal 'all' one of all the options including the current one, and accept it with probability min(1, exp(beta (u_new - u_old))) or 1 / (1 + exp(-beta (u_new - u_old))), so the rule 'fermi' with the proposal 'all' is the heat-bath rule of a uniform proposal. The rule 'best' draws uniformly among the options of largest utility under the ties 'random', and takes the first of them under 'first'. Every step, each member revises its choice with probability activity and keeps it otherwise.

In the 'symbolic' form, utility(namespace, option) returns the utility as a symbolic expression, where namespace.target holds the states and parameters of the group, and every member draws at once. In the 'agent' form, the group is an agent group and utility(agent, model, option) returns the utility as a number, so it can read the agent, its neighbours, and the history and storage of the model view. Its members draw at once, from the choices of the last step, under the 'synchronous' schedule, one after the other in their order or in a new random order, each seeing the choices made before it, under the 'sequential' or 'random-sequential' schedule, and each in its own activation under the 'member' schedule, together with the rules of its element and the other member interactions, in the order the system lists them. An option of utility -inf is never drawn in the agent form.

Parameters:

Name	Type	Description	Default
label	str	The label of the interaction, unique among the parts of its system.	required
target	Annotated[str, RoleSpec()]	The path of the element or group whose members choose.	required
state	str	The name of the discrete state of the members that holds the chosen option.	required
options	Sequence[float]	The values the state can take, at least two.	required
utility	Callable[..., float | Expression]	The function utility(namespace, option) in the symbolic form, or utility(agent, model, option) in the agent form, that returns the utility of an option.	required
rule	str	The rule of the draw, 'logit', 'metropolis', 'fermi', or 'best'.	'logit'
intensity	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The intensity of choice beta, zero for uniform choice.	1.0
schedule	str	The order in which the members revise, 'synchronous' or, in the agent form, 'sequential', 'random-sequential', or 'member'.	'synchronous'
activity	Annotated[float, ParameterSpec(domain=closed(0, 1))]	The probability that a member revises its choice in a step.	1.0
form	str	'symbolic' for update laws of the states of a group, or 'agent' for a rule over an agent group.	'symbolic'
phase	str | None	The phase of the step the agent form runs in, or None for the default phase.	None
proposal	str	The options the rules 'metropolis' and 'fermi' propose from, 'others' for the options other than the current one, or 'all' for every option.	'others'
ties	str	The option the rule 'best' takes among options of equal largest utility, 'random' for one drawn uniformly or 'first' for the first in the order of the options.	'random'

Raises:

Type	Description
ValidationError	The rule, the schedule, the form, the proposal, or the ties are unknown, the schedule is not synchronous in the symbolic form, there are fewer than two options, or the utility is not callable.

Examples:

>>> from skcomplex.core.interactions import ChoiceInteraction
>>> from skcomplex.core.laws.utils import mean
>>> choice = ChoiceInteraction(
...     'choice', target='spins', state='spin', options=(-1, 1), utility=lambda s, o: o * mean(s.target.spin)
... )
>>> choice.check().rule
'logit'
>>> agents = ChoiceInteraction(
...     'choice', target='traders', state='plan', options=(0, 1), utility=lambda a, m, o: o, form='agent'
... )
>>> type(agents.check().prepare().run(SimpleNamespace()).laws_[0]).__name__
'RuleLaw'
>>> ChoiceInteraction(
...     'heat-bath', target='spins', state='spin', options=(-1, 0, 1), utility=lambda s, o: o, rule='fermi',
...     proposal='all'
... ).check().proposal
'all'

Source code in src/skcomplex/core/interactions/_choice.py

def __init__(
    self,
    label: str,
    *,
    target: Annotated[str, RoleSpec()],
    state: str,
    options: Sequence[float],
    utility: Callable[..., float | Expression],
    rule: str = 'logit',
    intensity: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 1.0,
    schedule: str = 'synchronous',
    activity: Annotated[float, ParameterSpec(domain=RealSet.closed(0, 1))] = 1.0,
    form: str = 'symbolic',
    phase: str | None = None,
    proposal: str = 'others',
    ties: str = 'random',
) -> None:
    super().__init__(label)
    self.target = target
    self.state = state
    self.options = options
    self.utility = utility
    self.rule = rule
    self.intensity = intensity
    self.schedule = schedule
    self.activity = activity
    self.form = form
    self.phase = phase
    self.proposal = proposal
    self.ties = ties

choose(model)

Set the state of every agent that revises in this step to the option it draws, by the schedule.

Parameters:

Name	Type	Description	Default
model	ModelViewProtocol	The model view of the interaction.	required

Source code in src/skcomplex/core/interactions/_choice.py

def choose(self, model: ModelViewProtocol) -> None:
    """Set the state of every agent that revises in this step to the option it draws, by the schedule.

    Args:
        model:
            The model view of the interaction.
    """
    self._update_agents(model, self.schedule, self._get_choice, self._set_choice)

ConservationInteraction(label, *, target, state='u', flux=None, source=None, limiter='minmod', riemann='rusanov', sources=())

Bases: Interaction

Conservation law du/dt + divergence(flux(ns)) = source(ns) of a state u of a field.

The flux is one expression for a field of one dimension, or one expression per dimension, of the states of the field, and the interaction lowers it to a finite-volume flux law with a slope limiter and a numerical flux, or, without a numerical flux, to the rate -divergence(flux) that the field discretizes by its own operators, as a spectral or finite-difference field does. Without a source the total of the state on a periodic field stays constant. The conservation laws of several states of one field form one system, whose numerical flux reads the wave speeds of all of them, and a flux may read parameters that vary across the cells. A flux or a source function takes the namespace of the interaction, with the namespace target of the field, the time t, and the symbols of the parameters a subclass declares.

Parameters:

Name	Type	Description	Default
label	str	The label of the interaction, unique among the parts of its system.	required
target	Annotated[str, RoleSpec()]	The path of the field whose state is conserved.	required
state	str	The name of the conserved state of the field.	'u'
flux	Callable[[SimpleNamespace], object] | None	A function of the namespace that returns the flux, or one flux per dimension.	None
source	Callable[[SimpleNamespace], object] | None	A function of the namespace that returns the source, or None for none.	None
limiter	str | None	The slope limiter of the reconstruction, 'minmod', 'van-leer', or None for first order.	'minmod'
riemann	str | None	The numerical flux of a finite-volume field, 'rusanov' or 'godunov', the exact solver for one state, or None for the divergence of the flux on a field of another discretization.	'rusanov'
sources	Annotated[Sequence[str], RoleSpec(variadic=True)]	The paths of other elements the flux and the source read, as the namespaces sources. A field over the position lattice of a target on a phase space is read broadcast along the class axes.	()

Raises:

Type	Description
ValidationError	The state is not a name, a function is not callable, or the limiter or the numerical flux is unknown.

Examples:

>>> from typing import Annotated
>>> import numpy as np
>>> from skcomplex.core.elements import Element, FieldElement
>>> from skcomplex.core.interactions import ConservationInteraction
>>> from skcomplex.core.spaces import LatticeSpace
>>> from skcomplex.core.system import System
>>> from skcomplex.engine.runtime import Simulation
>>> from skcomplex.specs import StateSpec
>>> class CellElement(Element):
...     def __init__(self, label, u: Annotated[float, StateSpec()] = 0.0):
...         super().__init__(label)
...         self.u = u
>>> space = LatticeSpace((50,), spacing=0.1, boundary='periodic')
>>> values = {'u': 1.0 + np.sin(2 * np.pi * np.arange(50) / 50)}
>>> volumes = 'finite-volume'
>>> field = FieldElement('wave', element=CellElement, space=space, values=values, discretization=volumes)
>>> burgers = ConservationInteraction('burgers', target='wave', state='u', flux=lambda s: s.target.u**2 / 2)
>>> system = System(elements=[field], interactions=[burgers])
>>> trajectory = Simulation(system, dt=0.01).check().prepare().run(until=0.5).trajectory_
>>> round(float(trajectory.values('wave.u')[-1].sum()), 6)
50.0

Source code in src/skcomplex/core/interactions/_conservation.py

def __init__(
    self,
    label: str,
    *,
    target: Annotated[str, RoleSpec()],
    state: str = 'u',
    flux: Callable[[SimpleNamespace], object] | None = None,
    source: Callable[[SimpleNamespace], object] | None = None,
    limiter: str | None = 'minmod',
    riemann: str | None = 'rusanov',
    sources: Annotated[Sequence[str], RoleSpec(variadic=True)] = (),
) -> None:
    super().__init__(label)
    self.target = target
    self.state = state
    self.flux = flux
    self.source = source
    self.limiter = limiter
    self.riemann = riemann
    self.sources = sources

CouplingInteraction(label, *, target, source=None, read='x', write='x', function=None, reduce='sum', mode='edges', size=None, weights=None, include_self=False, lag=0.0, delay=0.0, form='rate', combine=None, neighborhood=None, layer=None, radii=(), clamp=None, phase=None)

Bases: NeighborInteraction

Coupling of every member to other members, through a reduction over them of a function of both members.

Every member i gets the reduction over other members j of function(namespace), where namespace.member reads the states of i, namespace.other those of j, namespace.edge the member of the edge group that joins them, or None, and the namespace also holds the parameters and the time t. The function is the difference x_j - x_i of the state read by default. The members j are, by mode:

• 'edges': the neighbours of i along the edges of its lattice or network, or within the neighborhood.
• 'group': every other member of the group, or every member of the group source when it is given.

In the mode 'group' the roles may also bind separate elements. A target that lists elements couples every element to the other elements of the list, a single target element is a hub that reduces over every member of the group source, and a source that is one element or a list of elements couples every member of the target to them. The reduction over separate elements is written out term by term. - 'next': the next member of the group in member order, the first after the last. - 'hyperedges': the other members of every hyperedge around i of size members, or of every size, on a network with hyperedges. The function then reads namespace.member, namespace.others, the other members of one hyperedge in its order, and namespace.members, all its members, which reductions over them read, and the reduction runs over the hyperedges around i. The default function is the sum over the other members of the difference of read.

With include_self, the reduction also runs over j = i. Every term f_ij has a weight w_ij: on edges, the state weights of the edge group, which a subclass binds with a second role, over the group, the entry W_ij of the matrix weights or of the shaped parameter of the interaction that weights names, which a subclass declares, and in every mode a callable weights(namespace) of the namespace of the term, such as an indicator of the states of both members. The reductions are 'sum', sum_j w_ij f_ij, 'mean', the sum divided by the number of members j, 'max' and 'min', the largest and smallest w_ij f_ij, 'normalized-mean', sum_j w_ij f_ij / sum_j w_ij, and 'circular-mean', the angle arctan2(sum_j w_ij sin f_ij, sum_j w_ij cos f_ij). The state read of j is read delay earlier, and the one of i shifted by the phase lag lag, when they are not 0.

In the form 'rate' the reduction adds to the rate of the state write, and in the form 'map' it sets the next value of write every step of discrete time to combine(namespace, reduction), where namespace.member reads the states of i and the namespace holds the parameters and the time, or to the reduction itself by default. The sum and the mean of the rate on the edges of a lattice or network run once per edge along both directions, and every other coupling runs once per member. On a FieldElement, a rate that is a constant times the difference of the ends becomes that constant times the field's Laplacian times its squared spacing, and on a lattice the other end is read across the field's boundary conditions, so beyond fixed edges it reads the fixed value.

In the form 'agent' the target is a group of agents, and every agent runs the coupling in its own activation in the phase phase. Every RadiusSpec of radii gives the vector function(agent, other, model) of every other agent of the source group, or of the target group, within its radius, reduced by its sum or mean. Only the radii of the highest priority that hold another agent count, and their reductions add up. The states write of the agent form a vector v, which becomes combine(agent, model, reduction, priority), or v plus the reduction by default, where reduction and priority are None when no radius holds another agent. clamp then bounds the angle the vector turns and its norm.

Parameters:

Name	Type	Description	Default
label	str	The label of the interaction, unique among the parts of its system.	required
target	Annotated[str | Sequence[str], RoleSpec()]	The path of the group of the members, or in the mode 'group' the path of one element or the paths of separate elements.	required
source	Annotated[str | Sequence[str] | None, RoleSpec(optional=True)]	The path of the group, the element, or the paths of the separate elements of the other members in the mode 'group', or None for the members of the target.	None
read	str	The name of the state that the default function, the delay, and the lag read.	'x'
write	str | Sequence[str]	The name of the state that the coupling changes, or the names of the components of the vector of the agent form.	'x'
function	NamespaceFunction | None	The term of one other member, a callable of the namespace, or None for the difference of read.	None
reduce	str	'sum', 'mean', 'max', 'min', 'normalized-mean', or 'circular-mean'.	'sum'
mode	str	'edges', 'group', 'next', or 'hyperedges'.	'edges'
size	int | None	The number of members of the hyperedges of the mode 'hyperedges', or None for every size.	None
weights	Weights	The name of the state of the edge group in the mode 'edges', a matrix with one row per member and one column per other member or the name of a parameter of the interaction declared ParameterSpec(shape=('target', 'source')) in the mode 'group', a callable of the namespace of a term in any mode, or None for no weights.	None
include_self	bool	Whether the reduction also runs over the member itself.	False
lag	Annotated[float, ParameterSpec()]	The phase lag added to the state read of the member.	0.0
delay	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The delay of the state read of the other member, 0 for none.	0.0
form	str	'rate', 'map', or 'agent'.	'rate'
combine	Combination | AgentCombination | None	The next value in the form 'map', a callable of the namespace of the member and the reduction, or in the form 'agent' a callable of the agent, the model view, the reduction, and its priority, or None for the default.	None
neighborhood	BaseNeighborhood | None	The neighbourhood of the mode 'edges', such as RadiusNeighborhood(1.0) in a continuous space, or None for the edges of the lattice or network.	None
layer	str | None	The layer of the network whose edges the interaction runs on, or None for the graph of the network.	None
radii	Sequence[RadiusSpec]	The terms of the agent form over the radii around an agent.	()
clamp	ClampSpec | None	The clamp of the turn and the norm of the vector of the agent form, or None.	None
phase	str | None	The phase of the step the agent and the map forms run in, or None for the default phase.	None

Raises:

Type	Description
ValidationError	A variant is unknown, the function or the combination is not callable, the weights, the source, the reduction, or the neighbourhood do not suit the mode and the form, or the radii, the clamp, the phase, or the written states do not suit the form, raised by check.

Examples:

>>> from types import SimpleNamespace
>>> from sage.all__sagemath_symbolics import sin, var
>>> from skcomplex.core.interactions import CouplingInteraction
>>> source, target = SimpleNamespace(theta=var('a')), SimpleNamespace(theta=var('b'))
>>> sine = lambda s: sin(s.other.theta - s.member.theta)
>>> coupling = CouplingInteraction('coupling', target='oscillators', read='theta', write='theta', function=sine)
>>> [law.value for law in coupling.check().prepare().run(SimpleNamespace(source=source, target=target)).laws_]
[sin(a - b), sin(-a + b)]

Source code in src/skcomplex/core/interactions/_coupling.py

def __init__(
    self,
    label: str,
    *,
    target: Annotated[str | Sequence[str], RoleSpec()],
    source: Annotated[str | Sequence[str] | None, RoleSpec(optional=True)] = None,
    read: str = 'x',
    write: str | Sequence[str] = 'x',
    function: NamespaceFunction | None = None,
    reduce: str = 'sum',
    mode: str = 'edges',
    size: int | None = None,
    weights: Weights = None,
    include_self: bool = False,
    lag: Annotated[float, ParameterSpec()] = 0.0,
    delay: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.0,
    form: str = 'rate',
    combine: Combination | AgentCombination | None = None,
    neighborhood: BaseNeighborhood | None = None,
    layer: str | None = None,
    radii: Sequence[RadiusSpec] = (),
    clamp: ClampSpec | None = None,
    phase: str | None = None,
) -> None:
    super().__init__(label, layer=layer)
    self.target = target
    self.source = source
    self.read = read
    self.write = write
    self.function = function
    self.reduce = reduce
    self.mode = mode
    self.size = size
    self.weights = weights
    self.include_self = include_self
    self.lag = lag
    self.delay = delay
    self.form = form
    self.combine = combine
    self.neighborhood = neighborhood
    self.radii = radii
    self.clamp = clamp
    self.phase = phase

CrossingInteraction(label, *, target, guard, reset=None, direction=0, terminal=False)

Bases: Interaction

Event at the crossing of zero by a guard, which sets states of the target to new values or ends the run.

The guard and the new values are functions of the namespace, which holds namespace.target, the states and parameters of the target, and the time namespace.t. A guard that reads the members of a group gives every member its own event, which sets the states of that member alone. Solvers locate the crossing within their tolerance, and the trajectory records every crossing among its events, under the label of the interaction and with the member.

Parameters:

Name	Type	Description	Default
label	str	The label of the interaction, unique among the parts of its system.	required
target	Annotated[str, RoleSpec()]	The path of the element or group whose states the guard reads and the event sets.	required
guard	Callable[[SimpleNamespace], object]	The function guard(namespace) of the expression whose crossing of zero triggers the event.	required
reset	Mapping[str, Callable[[SimpleNamespace], object]] | None	A mapping from the name of a state of the target to the function f(namespace) of its new value, applied all at once, or None to keep the states.	None
direction	int	1 for crossings from negative to positive, -1 for the opposite, and 0 for both.	0
terminal	bool	Whether the crossing ends the run.	False

Raises:

Type	Description
ValidationError	The direction is not -1, 0, or 1, or the guard or a new value is not a function.

Examples:

>>> from typing import Annotated
>>> from skcomplex.core.elements import Element
>>> from skcomplex.core.interactions import CrossingInteraction
>>> from skcomplex.core.laws import RateLaw
>>> from skcomplex.core.system import System
>>> from skcomplex.engine.runtime import Simulation
>>> from skcomplex.specs import StateSpec
>>> class ClockElement(Element):
...     def __init__(self, label, x: Annotated[float, StateSpec()] = 0.0):
...         super().__init__(label)
...         self.x = x
...
...     def _get_laws(self, s):
...         return [RateLaw(s.x, 1.0)]
>>> wrap = CrossingInteraction(
...     'wrap', target='clock', guard=lambda s: s.target.x - 1, reset={'x': lambda s: 0}, direction=1
... )
>>> system = System(elements=[ClockElement('clock')], interactions=[wrap])
>>> trajectory = Simulation(system, dt=0.5).check().prepare().run(until=2.2).trajectory_
>>> [round(event.time, 6) for event in trajectory.events]
[1.0, 2.0]

Source code in src/skcomplex/core/interactions/_crossing.py

def __init__(
    self,
    label: str,
    *,
    target: Annotated[str, RoleSpec()],
    guard: Callable[[SimpleNamespace], object],
    reset: Mapping[str, Callable[[SimpleNamespace], object]] | None = None,
    direction: int = 0,
    terminal: bool = False,
) -> None:
    super().__init__(label)
    self.target = target
    self.guard = guard
    self.reset = reset
    self.direction = direction
    self.terminal = terminal

DiffusionInteraction(label, *, target, state='u', coefficient=1.0)

Bases: ConservationInteraction

Diffusion du/dt = D laplacian(u) of a state u of a field, the conservation law of the flux -D gradient(u).

The coefficient D is uniform in space, a number or a function of the namespace that returns an expression of parameters and time. The interaction lowers to the Laplacian of the field, which its discretization evaluates, so the total of the state on a periodic field stays constant and a point mass spreads as the heat kernel.

Parameters:

Name	Type	Description	Default
label	str	The label of the interaction, unique among the parts of its system.	required
target	Annotated[str, RoleSpec()]	The path of the field whose state diffuses.	required
state	str	The name of the state of the field that diffuses.	'u'
coefficient	Callable[[SimpleNamespace], object] | float	A function of the namespace that returns the diffusion coefficient, or a number.	1.0

Raises:

Type	Description
ValidationError	The state is not a name, or the coefficient is neither callable nor a number.

Examples:

>>> from typing import Annotated
>>> import numpy as np
>>> from skcomplex.core.elements import Element, FieldElement
>>> from skcomplex.core.interactions import DiffusionInteraction
>>> from skcomplex.core.spaces import LatticeSpace
>>> from skcomplex.core.system import System
>>> from skcomplex.engine.runtime import Simulation
>>> from skcomplex.specs import StateSpec
>>> class CellElement(Element):
...     def __init__(self, label, u: Annotated[float, StateSpec()] = 0.0):
...         super().__init__(label)
...         self.u = u
>>> space = LatticeSpace((40,), spacing=0.1, boundary='periodic')
>>> field = FieldElement('rod', element=CellElement, space=space, values={'u': np.eye(40)[20]})
>>> heat = DiffusionInteraction('heat', target='rod', state='u', coefficient=0.1)
>>> trajectory = Simulation(System(elements=[field], interactions=[heat]), dt=0.01).check().prepare()
>>> round(float(trajectory.run(until=1.0).trajectory_.values('rod.u')[-1].sum()), 6)
1.0

Source code in src/skcomplex/core/interactions/_diffusion.py

def __init__(
    self,
    label: str,
    *,
    target: Annotated[str, RoleSpec()],
    state: str = 'u',
    coefficient: Callable[[SimpleNamespace], object] | float = 1.0,
) -> None:
    super().__init__(label, target=target, state=state)
    self.coefficient = coefficient

EncounterInteraction(label, *, candidate, field=None, rate=None, kernel=None, outcome='conservative', state='f', cross_section=None, factor=None)

Bases: Interaction

Encounters of a candidate density with a field density over activity classes, which move, remove, or create.

The candidate and the field are groups of DensityElement on the sites of lattices of activity classes, one member per class. A candidate at class h meets a field at class k at the rate eta[h, k]. In the outcome 'conservative' the candidate moves to class i with the probability kernel[h, k, i], in the outcome 'destructive' it is removed, and in the outcome 'proliferative' it creates a new candidate at class i with the probability kernel[h, k, i], or at its own class without a kernel. The rate and the kernel are functions of the namespace, which holds namespace.candidate, namespace.field, and the parameters of the interaction.

On a PhaseSpace the classes are the points of the class axes, and encounters pair classes at the same position. A rate that is one expression may read the coordinate of the class of the candidate. Without a field the encounter is a linear redistribution: a conservative candidate at class h moves to class i at the rate eta[h] with the probability kernel[h, i].

With a cross_section the candidate is a field on a PhaseSpace with two interval class axes, and its rate at every position is the binary collision integral over those axes, evaluated by the fast spectral method on a padded grid. The density is conserved exactly, and the momentum and the energy within the spectral accuracy of the method.

Parameters:

Name	Type	Description	Default
label	str	The label of the interaction, unique among the parts of its system.	required
candidate	Annotated[str, RoleSpec()]	The path of the group of the candidate density.	required
field	Annotated[str | None, RoleSpec(optional=True)]	The path of the group of the field density, or None for a linear redistribution.	None
rate	Callable[[SimpleNamespace], object] | None	The function rate(namespace) of the encounter rate, a number or an expression, or an array eta[h, k] of them, or eta[h] without a field, or None with a cross-section.	None
kernel	Callable[[SimpleNamespace], object] | None	The function kernel(namespace) of the array kernel[h, k, i] of the probabilities of the class of the moved or created candidate, or kernel[h, i] without a field, or None for a destructive encounter or the candidate's own class.	None
outcome	str	'conservative', 'destructive', or 'proliferative'.	'conservative'
state	str	The name of the state of the densities.	'f'
cross_section	Callable[..., object] | None	The function B(speed, cosine) of arrays of the relative speed and the cosine of the deflection angle, or None for encounters at a rate.	None
factor	Callable[[SimpleNamespace], object] | None	A function of the namespace that returns a factor of the collision frequency, such as a function of the density moment, or None for none.	None

Raises:

Type	Description
ValidationError	The outcome is unknown, the state is no name, a function is not callable, a conservative encounter has no kernel, an encounter without a field is not conservative, an encounter has neither a rate nor a cross-section, or a cross-section comes with a rate, a kernel, a field, or an outcome.

Examples:

>>> import numpy as np
>>> from skcomplex.core.interactions import EncounterInteraction
>>> kernel = np.full((3, 3, 3), 1 / 3)
>>> mixing = EncounterInteraction(
...     'mixing', candidate='population', field='population', rate=lambda s: 1.0, kernel=lambda s: kernel
... )
>>> mixing.check().outcome
'conservative'
>>> collide = EncounterInteraction('collide', candidate='gas', cross_section=lambda speed, cosine: 1.0)
>>> collide.check().rate is None
True

Source code in src/skcomplex/core/interactions/_encounter.py

def __init__(
    self,
    label: str,
    *,
    candidate: Annotated[str, RoleSpec()],
    field: Annotated[str | None, RoleSpec(optional=True)] = None,
    rate: Callable[[SimpleNamespace], object] | None = None,
    kernel: Callable[[SimpleNamespace], object] | None = None,
    outcome: str = 'conservative',
    state: str = 'f',
    cross_section: Callable[..., object] | None = None,
    factor: Callable[[SimpleNamespace], object] | None = None,
) -> None:
    super().__init__(label)
    self.candidate = candidate
    self.field = field
    self.rate = rate
    self.kernel = kernel
    self.outcome = outcome
    self.state = state
    self.cross_section = cross_section
    self.factor = factor

EulerLagrangeInteraction(label, *, target, states='x', lagrangian=None, dissipation=None, forcing=None)

Bases: Interaction

Equations of motion d/dt dL/dq' - dL/dq = Q - dR/dq' of the coordinates q of a Lagrangian L(ns).

The coordinates are states of order 2, and the Lagrangian, the Rayleigh dissipation function R, and the generalized forcing Q read them and their first derivatives derivative(q, 1). The interaction returns one equation law per coordinate, which the compiler solves jointly for the second derivatives, so a Lagrangian with a mass matrix, such as a double pendulum's, needs no inversion by hand. The equations of one coordinate from several interactions add, so their Lagrangians, dissipations, and forcings sum: a coupling, such as a spring between two pendulums, is an interaction of its own whose Lagrangian holds only the coupling energy. The coordinates are the states of every role that binds one element or group, target by default, and every component of a vector state is a coordinate.

A Lagrangian, dissipation, or forcing function takes the namespace of the interaction, with the namespaces of the roles, the time t, and the symbols of the parameters a subclass declares. The forcing returns one generalized force per coordinate, in the order of the roles and of the states, or one expression for a single coordinate.

Parameters:

Name	Type	Description	Default
label	str	The label of the interaction, unique among the parts of its system.	required
target	Annotated[str, RoleSpec()]	The path of the element or group whose states are the coordinates.	required
states	str | Sequence[str]	The name of the state of order 2 that is a coordinate, or the names of several states.	'x'
lagrangian	Callable[[SimpleNamespace], object] | None	A function of the namespace that returns the Lagrangian.	None
dissipation	Callable[[SimpleNamespace], object] | None	A function of the namespace that returns the Rayleigh dissipation function, or None for none.	None
forcing	Callable[[SimpleNamespace], object] | None	A function of the namespace that returns the generalized forces, or None for none.	None

Raises:

Type	Description
ValidationError	A state is not a name, or a function is not callable.

Examples:

>>> from typing import Annotated
>>> from sage.all__sagemath_symbolics import cos
>>> from skcomplex.core.elements import Element
>>> from skcomplex.core.interactions import EulerLagrangeInteraction
>>> from skcomplex.core.laws.utils import derivative
>>> from skcomplex.core.system import System
>>> from skcomplex.engine.runtime import Simulation
>>> from skcomplex.specs import StateSpec
>>> class AngleElement(Element):
...     def __init__(self, label, q: Annotated[float, StateSpec(order=2)] = 0.1, q_d1: float = 0.0):
...         super().__init__(label)
...         self.q = q
...         self.q_d1 = q_d1
>>> swing = EulerLagrangeInteraction(
...     'swing',
...     target='bob',
...     states='q',
...     lagrangian=lambda s: derivative(s.target.q, 1) ** 2 / 2 + cos(s.target.q),
... )
>>> system = System(elements=[AngleElement('bob')], interactions=[swing])
>>> trajectory = Simulation(system, dt=0.01).check().prepare().run(until=3.2).trajectory_
>>> round(float(trajectory.values('bob.q')[-1]), 2)
-0.1

Source code in src/skcomplex/core/interactions/_euler_lagrange.py

def __init__(
    self,
    label: str,
    *,
    target: Annotated[str, RoleSpec()],
    states: str | Sequence[str] = 'x',
    lagrangian: Callable[[SimpleNamespace], object] | None = None,
    dissipation: Callable[[SimpleNamespace], object] | None = None,
    forcing: Callable[[SimpleNamespace], object] | None = None,
) -> None:
    super().__init__(label)
    self.target = target
    self.states = states
    self.lagrangian = lagrangian
    self.dissipation = dissipation
    self.forcing = forcing

EvolutionInteraction(label, *, target, strategy='strategy', fitness='fitness', selection='tournament', replacement='generational', crossover=0.7, mutation=0.01, elite=0, tournament=2, alphabet=2, scale=0.1, death=0.1, change='other', parents='survivors', fraction=0.1, period=1, evaluate=None, reset=None, phase=None)

Bases: Interaction

Replacement of the strategies of an agent group by offspring selected on a fitness, with crossover and mutation.

Every period steps, or rounds of its phase, the interaction selects parents by tournaments of tournament entrants, by roulette with weights of the fitness minus its minimum, in proportion to the fitness itself ('proportional'), or by rank with weights 1 for the least fit to n for the fittest of n members. The 'generational' replacement selects as many parents as there are members, each pair of which exchanges the tails after a random cut with probability crossover, and passes the elite fittest strategies unchanged. The 'moran' replacement runs as many birth-death events as there are members, one after the other, each copying a selected parent onto a member drawn uniformly, and the copy takes the fitness of its parent for the later events of the step. The 'death-birth' replacement removes every member with probability death and fills its place with the offspring of a parent selected among the survivors, or among all members when none survives or parents is 'all'. The 'truncation' replacement fills the places of the fraction least fit members with the offspring of parents selected among the others. The 'best-mutant' replacement lets every member draw a mutant of its strategy, and every member takes the fittest mutant when it is fitter than the fittest strategy, or when every member mutated, so a population of one strategy climbs a landscape as a whole.

Offspring mutate: each gene with probability mutation changes by the rule change, to another symbol of an alphabet of alphabet symbols ('other'), to a symbol drawn anew from the alphabet ('any'), one step up or down ('step'), or to a label one above the largest the members hold ('new'), or with probability mutation one gene drawn at random of every offspring changes to another symbol ('one'). When alphabet is None, the 'other' and 'any' rules add Gaussian noise of deviation scale. A function mutation(strategy, rng) returns the mutated copy of a strategy instead. The fitness is a state of the members, or the function evaluate(strategies, model) of the strategies, one row per member, which the replacement then writes into the fitness state. The probabilities are parameters of the model when a subclass declares them with a ParameterSpec, as the base does for crossover, death, and fraction. The draws come from the random stream of the model.

Parameters:

Name	Type	Description	Default
label	str	The label of the interaction, unique among the parts of its system.	required
target	Annotated[str, RoleSpec()]	The path of the agent group whose strategies evolve.	required
strategy	str | Sequence[str]	The name of the state of the members that holds the strategy, a sequence of genes, or the names of several states, one gene each.	'strategy'
fitness	str | None	The name of the state of the members that holds the fitness, or None for a neutral population of equal fitness when no evaluate is given.	'fitness'
selection	str	The selection of parents, 'tournament', 'roulette', 'proportional', or 'rank'.	'tournament'
replacement	str	The replacement of strategies, 'generational', 'moran', 'death-birth', 'truncation', or 'best-mutant'.	'generational'
crossover	Annotated[float, ParameterSpec(domain=closed(0, 1))]	The probability that a pair of parents exchanges the tails of their strategies, in the generational replacement.	0.7
mutation	float | Callable[[Array, Generator], Array]	The probability that a gene mutates, or a function mutation(strategy, rng) that returns the mutated copy of a strategy.	0.01
elite	int	The number of fittest strategies copied unchanged, in the generational replacement.	0
tournament	int	The number of entrants of every tournament.	2
alphabet	int | None	The number of symbols of a discrete strategy, or None for a real strategy.	2
scale	float	The standard deviation of the mutation of a real strategy.	0.1
death	Annotated[float, ParameterSpec(domain=closed(0, 1))]	The probability that a member is replaced in a step, in the death-birth replacement.	0.1
change	str	How a gene mutates, 'other', 'any', 'one', 'step', or 'new'.	'other'
parents	str	The members the death-birth replacement selects parents among, 'survivors' or 'all'.	'survivors'
fraction	Annotated[float, ParameterSpec(domain=closed(0, 1))]	The fraction of least fit members that the truncation replacement replaces.	0.1
period	int	The number of steps, or rounds of the phase, between two replacements.	1
evaluate	Callable[[Array, ModelViewProtocol], Array] | None	The function evaluate(strategies, model) that returns the fitness of every row of strategies, or None to read the fitness state.	None
reset	Mapping[str, float] | None	The values of states that every replaced member takes, or None for no reset.	None
phase	str | None	The phase of the step the generation runs in, or None for the default phase.	None

Raises:

Type	Description
ValidationError	A variant is unknown, a probability is outside [0, 1], the mutation is neither a probability nor a function, a count is invalid, or the evaluation is not callable.

Examples:

>>> from types import SimpleNamespace
>>> from skcomplex.core.interactions import EvolutionInteraction
>>> evolution = EvolutionInteraction('evolution', target='agents', selection='roulette', phase='breeding')
>>> evolution.check().prepare().run(SimpleNamespace()).laws_[0].phase
'breeding'
>>> EvolutionInteraction('evolution', target='agents', selection='rank', replacement='moran').check().selection
'rank'
>>> EvolutionInteraction(
...     'evolution', target='agents', replacement='truncation', fraction=0.2, change='any', period=50
... ).check().period
50

Source code in src/skcomplex/core/interactions/_evolution.py

def __init__(
    self,
    label: str,
    *,
    target: Annotated[str, RoleSpec()],
    strategy: str | Sequence[str] = 'strategy',
    fitness: str | None = 'fitness',
    selection: str = 'tournament',
    replacement: str = 'generational',
    crossover: Annotated[float, ParameterSpec(domain=RealSet.closed(0, 1))] = 0.7,
    mutation: float | Callable[[Array, np.random.Generator], Array] = 0.01,
    elite: int = 0,
    tournament: int = 2,
    alphabet: int | None = 2,
    scale: float = 0.1,
    death: Annotated[float, ParameterSpec(domain=RealSet.closed(0, 1))] = 0.1,
    change: str = 'other',
    parents: str = 'survivors',
    fraction: Annotated[float, ParameterSpec(domain=RealSet.closed(0, 1))] = 0.1,
    period: int = 1,
    evaluate: Callable[[Array, ModelViewProtocol], Array] | None = None,
    reset: Mapping[str, float] | None = None,
    phase: str | None = None,
) -> None:
    super().__init__(label)
    self.target = target
    self.strategy = strategy
    self.fitness = fitness
    self.selection = selection
    self.replacement = replacement
    self.crossover = crossover
    self.mutation = mutation
    self.elite = elite
    self.tournament = tournament
    self.alphabet = alphabet
    self.scale = scale
    self.death = death
    self.change = change
    self.parents = parents
    self.fraction = fraction
    self.period = period
    self.evaluate = evaluate
    self.reset = reset
    self.phase = phase

FeedbackInteraction(label, *, rule=None, phase=None)

Bases: Interaction

Single actor that reads the model once per round of its phase and sets parameters from what it reads.

Its rule runs once per round of its phase, after the agents of the phase. It reads observables with model.observable(name), reporters with model.history(name), and agents with model.agents(name), and sets parameters of any group or interaction with model.set_parameter(name, value). A change is visible to every rule at once and recorded in the trajectory like any parameter. Subclasses declare their own parameters, such as a target, and override act; otherwise rule(model) acts.

Parameters:

Name	Type	Description	Default
label	str	The label of the feedback.	required
rule	Callable[[ModelT], None] | None	The function of the model view that acts, used when a subclass does not override act.	None
phase	str | None	The phase the feedback acts in, or None for the default phase.	None

Examples:

>>> from skcomplex.core.interactions import FeedbackInteraction
>>> FeedbackInteraction('controller', rule=lambda model: None).check().phase is None
True

Source code in src/skcomplex/core/interactions/_feedback.py

def __init__(
    self,
    label: str,
    *,
    rule: Callable[[ModelT], None] | None = None,
    phase: str | None = None,
) -> None:
    super().__init__(label)
    self.rule = rule
    self.phase = phase

act(model)

Read the model and set parameters.

Parameters:

Name	Type	Description	Default
model	ModelT	The model view scoped to the feedback.	required

Raises:

Type	Description
ValidationError	The feedback has neither a rule nor its own act.

Source code in src/skcomplex/core/interactions/_feedback.py

def act(self, model: ModelT) -> None:
    """Read the model and set parameters.

    Args:
        model:
            The model view scoped to the feedback.

    Raises:
        ValidationError:
            The feedback has neither a rule nor its own `act`.
    """
    if self.rule is None:
        error_msg = f'The feedback {self.label!r} has no rule. Pass `rule` or override `act`.'
        raise ValidationError(error_msg)
    self.rule(model)

GradientInteraction(label, *, target, states='x', potential=None, sources=(), divisor=None, variational=False, conserved=False)

Bases: Interaction

Motion of states down the gradient of a potential U(ns), as a gradient flow or as Newtonian motion.

Every state x the interaction moves gets the rate -dU/dx / d on its highest time derivative, with d the divisor of its member: a state or parameter of the member, an expression of them such as a moment of inertia m l^2, or one divisor per state, such as the masses of two coordinates of one element. On a state of order 1 this is the gradient flow x' = -dU/dx / d, with d a friction, and on a state of order 2 it is Newton's law x'' = -dU/dx / d, with d an inertia. A vector state gets the gradient of every component, and a tensor state the matrix gradient, the projection of the derivative onto the matrices of its symmetry.

On a field the potential is the density f of a functional F = integral f, which may read the field, its derivatives partial(x, k) or gradient(x), and its laplacian(x). A variational interaction moves the field by the functional derivative dF/dx = df/dx - sum_k partial(df/dx_k, k) + laplacian(df/d laplacian(x)), as in the Allen-Cahn equation, and a conserved interaction gives the rate laplacian(dF/dx) / d instead, as in the Cahn-Hilliard equation, so the total of the field stays constant.

The interaction moves the states of every role that binds one element or group, target by default. A subclass may bind several such roles instead, such as both members of a pair potential, and every one of them moves. The potential function takes the namespace of the interaction, with the namespaces of the roles and of the sources, the time t, and the symbols of the parameters a subclass declares.

Parameters:

Name	Type	Description	Default
label	str	The label of the interaction, unique among the parts of its system.	required
target	Annotated[str, RoleSpec()]	The path of the element or group whose states move.	required
states	str | Sequence[str]	The name of the state that moves, or the names of several states.	'x'
potential	Callable[[SimpleNamespace], object] | None	A function of the namespace that returns the potential, or the density of the functional of a field.	None
sources	Annotated[Sequence[str], RoleSpec(variadic=True)]	The paths of the elements or groups the potential reads but does not move.	()
divisor	str | Callable[[SimpleNamespace], object] | Mapping[str, str | Callable[[SimpleNamespace], object]] | None	The name of the state or parameter of every moving member that divides its rate, a function of the namespace of the member that returns the divisor, a mapping from the name of every moving state to its own divisor, or None.	None
variational	bool	Whether the potential is the density of a functional of a field, moved by its functional derivative.	False
conserved	bool	Whether the rate is the Laplacian of the derivative of the potential, which conserves the total of a field.	False

Raises:

Type	Description
ValidationError	A state or the divisor is not a name, or the potential is not callable.

Examples:

>>> from skcomplex.core.elements import CompartmentElement
>>> from skcomplex.core.interactions import GradientInteraction
>>> from skcomplex.core.system import System
>>> from skcomplex.engine.runtime import Simulation
>>> well = GradientInteraction('well', target='walker', states='n', potential=lambda s: s.target.n**2 / 2)
>>> system = System(elements=[CompartmentElement('walker', n=1.0)], interactions=[well])
>>> trajectory = Simulation(system, dt=0.01).check().prepare().run(until=1.0).trajectory_
>>> round(float(trajectory.values('walker.n')[-1]), 4)
0.3679

Source code in src/skcomplex/core/interactions/_gradient.py

def __init__(
    self,
    label: str,
    *,
    target: Annotated[str, RoleSpec()],
    states: str | Sequence[str] = 'x',
    potential: Callable[[SimpleNamespace], object] | None = None,
    sources: Annotated[Sequence[str], RoleSpec(variadic=True)] = (),
    divisor: (
        str | Callable[[SimpleNamespace], object] | Mapping[str, str | Callable[[SimpleNamespace], object]] | None
    ) = None,
    variational: bool = False,
    conserved: bool = False,
) -> None:
    super().__init__(label)
    self.target = target
    self.states = states
    self.potential = potential
    self.sources = sources
    self.divisor = divisor
    self.variational = variational
    self.conserved = conserved

GroupingInteraction(label, *, target, groups='pairs', act=None, size=2, by=None, capacity=None, phase=None, form='agent', write=None, function=None, reduce='sum')

Bases: Interaction

Grouping of the members of a group into groups drawn every step, with an action on every group.

The groups are one of four kinds:

• 'pairs' draws disjoint random pairs, and an odd member out forms no group.
• 'partition' draws a random partition into groups of size members, the last group holding the remainder.
• 'neighborhood' forms one group per member, the member first and then the members in the cells or nodes around it.
• 'state' forms one group per value of the state by, of at most capacity agents drawn at random among those that hold the value in the agent form.

In the agent form the groups act in random order, each through act(members, model), a function of the list of agent views and the model view, or the method act_on a subclass overrides. The grouping is an agent rule, so the system runs in discrete time.

In the symbolic form every member of a group takes the value of function(namespace) for the state write in the next step, reduced by reduce over the groups that hold it, and a member in no group keeps its value. The namespace of a group holds member, the member the value is for, others, the other members in the order of the group, members, which total, mean, count, maximum, and minimum reduce with over=namespace.members, the time t, and the parameters of the interaction. Pairs and partitions are drawn again at every step from the random stream of the simulation, and neighbourhoods are those of the space. The groups by state hold no others, and their reductions over the members are keyed by the state. Within a step, the grouping runs in its phase and reads the states the phases before it set.

Parameters:

Name	Type	Description	Default
label	str	The label of the interaction, unique among the parts of its system.	required
target	Annotated[str, RoleSpec()]	The path of the group whose members are grouped.	required
groups	str	The kind of groups, 'pairs', 'partition', 'neighborhood', or 'state', or in the agent form a PartnerSpec or a function pairing(members, members, rng, capacity) that returns pairs, which group every member with its partners. 'pairs' is a shorthand of PartnerSpec('all', 'uniform', capacity=1).	'pairs'
act	Act | None	The function of the members of a group and the model view that acts on the group in the agent form, used when a subclass does not override act_on.	None
size	int	The number of members of a group of a partition.	2
by	str | None	The name of the state whose values the groups 'state' share.	None
capacity	int | None	The largest number of agents of a group 'state' in the agent form, or None for no limit.	None
phase	str | None	The phase of the step the grouping runs in, or None for the default phase.	None
form	str	'agent' for an action on the agents of every group, or 'symbolic' for a function of every group.	'agent'
write	str | None	The name of the state that the symbolic form sets.	None
function	GroupFunction | None	The function of the namespace of one group that returns the value of write for its member, in the symbolic form.	None
reduce	str	'sum' or 'mean', how the symbolic form combines the values of the groups that hold a member.	'sum'

Raises:

Type	Description
ValidationError	The kind of groups, the form, or the reduction is unknown, the size is below two, the groups 'state' have no state, the capacity is not positive, the symbolic form has no state or function, or a setting of one form is given to the other, raised by check.

Examples:

>>> from typing import Annotated
>>> from skcomplex.core.elements import Element, GroupElement
>>> from skcomplex.core.interactions import GroupingInteraction
>>> from skcomplex.core.system import System
>>> from skcomplex.engine.runtime import Simulation
>>> from skcomplex.specs import StateSpec
>>> class VoterElement(Element):
...     def __init__(self, label, visual=None, *, s: Annotated[float, StateSpec()] = 0.0):
...         super().__init__(label, visual=visual)
...         self.s = s
>>> def adopt_majority(members, model):
...     majority = float(2 * sum(member.s for member in members) > len(members))
...     for member in members:
...         member.s = majority
>>> voters = GroupElement('voters', element=VoterElement, size=3, values={'s': [1.0, 1.0, 0.0]})
>>> galam = GroupingInteraction('galam', target='voters', groups='partition', size=3, act=adopt_majority)
>>> trajectory = Simulation(System(elements=[voters], interactions=[galam])).check().prepare().run(until=1)
>>> trajectory.trajectory_.agents('voters', 1).values['s'].tolist()
[1.0, 1.0, 1.0]

In the symbolic form every group of the partition takes the opinion of its majority:

>>> from skcomplex.core.laws.utils import mean, where
>>> galam = GroupingInteraction(
...     'galam',
...     target='voters',
...     groups='partition',
...     size=3,
...     form='symbolic',
...     write='s',
...     function=lambda g: where(mean(g.members.s, over=g.members) > 1 / 2, 1, 0),
... )
>>> simulation = Simulation(System(elements=[voters], interactions=[galam]), random_state=0)
>>> simulation.check().prepare().run(until=1).trajectory_.values('voters.s')[-1].tolist()
[1.0, 1.0, 1.0]

Source code in src/skcomplex/core/interactions/_grouping.py

def __init__(
    self,
    label: str,
    *,
    target: Annotated[str, RoleSpec()],
    groups: str = 'pairs',
    act: Act | None = None,
    size: int = 2,
    by: str | None = None,
    capacity: int | None = None,
    phase: str | None = None,
    form: str = 'agent',
    write: str | None = None,
    function: GroupFunction | None = None,
    reduce: str = 'sum',
) -> None:
    super().__init__(label)
    self.target = target
    self.groups = groups
    self.act = act
    self.size = size
    self.by = by
    self.capacity = capacity
    self.phase = phase
    self.form = form
    self.write = write
    self.function = function
    self.reduce = reduce

act_on(members, model)

Act on the members of one group.

Parameters:

Name	Type	Description	Default
members	list[AgentViewProtocol]	The agent views of the members of the group.	required
model	ModelViewProtocol	The model view of the rule.	required

Raises:

Type	Description
ValidationError	The grouping has neither an action nor its own act_on.

Source code in src/skcomplex/core/interactions/_grouping.py

def act_on(self, members: list[AgentViewProtocol], model: ModelViewProtocol) -> None:
    """Act on the members of one group.

    Args:
        members:
            The agent views of the members of the group.

        model:
            The model view of the rule.

    Raises:
        ValidationError:
            The grouping has neither an action nor its own `act_on`.
    """
    if self.act is None:
        error_msg = f'The grouping {self.label!r} has no action. Pass `act` or override `act_on`.'
        raise ValidationError(error_msg)
    self.act(members, model)

HamiltonianInteraction(label, *, target, coordinates='q', momenta='p', hamiltonian=None)

Bases: Interaction

Hamilton's equations q' = dH/dp and p' = -dH/dq of the canonical pairs (q, p) of a Hamiltonian H(ns).

The coordinates and the momenta are states of order 1, which may be angles, of every role that binds one element or group, target by default, and the coordinates pair with the momenta in order. Every component of a vector state is a coordinate or a momentum. The interaction returns one rate law per coordinate and per momentum, and the rates of one state add, so the Hamiltonians of separate interactions sum: a coupling between two oscillators is an interaction of its own whose Hamiltonian holds only the coupling energy. Dissipation or forcing is a rate interaction on the momenta.

A Hamiltonian function takes the namespace of the interaction, with the namespaces of the roles, the time t, and the symbols of the parameters a subclass declares.

Parameters:

Name	Type	Description	Default
label	str	The label of the interaction, unique among the parts of its system.	required
target	Annotated[str, RoleSpec()]	The path of the element or group whose states are the canonical pairs.	required
coordinates	str | Sequence[str]	The name of the state of order 1 that is a coordinate, or the names of several states.	'q'
momenta	str | Sequence[str]	The name of the state of order 1 that is the momentum of the coordinate, or the names of several states in the order of the coordinates.	'p'
hamiltonian	Callable[[SimpleNamespace], object] | None	A function of the namespace that returns the Hamiltonian.	None

Raises:

Type	Description
ValidationError	A state is not a name, the coordinates and the momenta differ in count, the Hamiltonian is not callable, or a coordinate and its momentum differ in size or are not of order 1.

Examples:

>>> from typing import Annotated
>>> from skcomplex.core.elements import Element
>>> from skcomplex.core.interactions import HamiltonianInteraction
>>> from skcomplex.core.system import System
>>> from skcomplex.engine.runtime import Simulation
>>> from skcomplex.specs import StateSpec
>>> class OscillatorElement(Element):
...     def __init__(self, label, q: Annotated[float, StateSpec()] = 1.0, p: Annotated[float, StateSpec()] = 0):
...         super().__init__(label)
...         self.q = q
...         self.p = p
>>> energy = lambda s: (s.target.p**2 + s.target.q**2) / 2
>>> spring = HamiltonianInteraction('spring', target='mass', hamiltonian=energy)
>>> system = System(elements=[OscillatorElement('mass')], interactions=[spring])
>>> trajectory = Simulation(system, dt=0.01).check().prepare().run(until=3.14159).trajectory_
>>> round(float(trajectory.values('mass.q')[-1]), 3)
-1.0

Source code in src/skcomplex/core/interactions/_hamiltonian.py

def __init__(
    self,
    label: str,
    *,
    target: Annotated[str, RoleSpec()],
    coordinates: str | Sequence[str] = 'q',
    momenta: str | Sequence[str] = 'p',
    hamiltonian: Callable[[SimpleNamespace], object] | None = None,
) -> None:
    super().__init__(label)
    self.target = target
    self.coordinates = coordinates
    self.momenta = momenta
    self.hamiltonian = hamiltonian

HyperedgeInteraction(label, size=None)

Bases: BaseInteraction

Base class of interactions whose laws run once for every hyperedge of the network of the nodes a role binds.

The first role binds a group with one member per node of a network space with hyperedges. A second role may bind a group on the hyperedges of the same network, or a group on the edges of one of its layers. The laws are written for one hyperedge of size members and read:

• namespace.member, the focal member, whose states rate, noise, and jump laws change;
• namespace.others, the other members in the order of the hyperedge, a tuple of namespaces;
• namespace.members, the members of the hyperedge, which total, mean, count, maximum, and minimum reduce with over=namespace.members;
• namespace.hyperedge, the member of the group on hyperedges;
• namespace.edge, the member of the group on edges that joins the first two members of the hyperedge;
• the parameters of the interaction.

A law of the focal member runs once per member of every hyperedge and adds into that member. A law that targets the member of the group on hyperedges or on edges runs once per hyperedge, with the first member of the hyperedge as the focal member. Without a size, the laws run on the hyperedges of every size, one size at a time.

Parameters:

Name	Type	Description	Default
label	str	The label of the interaction, unique among the parts of its system.	required
size	int | None	The number of members of the hyperedges the laws run on, or None for every size.	None

Raises:

Type	Description
ValidationError	The size is not a whole number of at least two.

Examples:

>>> from typing import Annotated
>>> from skcomplex.core.interactions import HyperedgeInteraction
>>> from skcomplex.core.laws import RateLaw
>>> from skcomplex.core.laws.utils import mean
>>> from skcomplex.specs import RoleSpec
>>> class ConsensusInteraction(HyperedgeInteraction):
...     def __init__(self, label, voters: Annotated[str, RoleSpec()], size=None):
...         super().__init__(label, size=size)
...         self.voters = voters
...
...     def _get_laws(self, s):
...         return [RateLaw(s.member.x, mean(s.members.x, over=s.members) - s.member.x)]
>>> ConsensusInteraction('consensus', voters='voters', size=3).check().size
3

Source code in src/skcomplex/core/interactions/_hyperedge.py

def __init__(self, label: str, size: int | None = None) -> None:
    super().__init__(label)
    self.size = size

ImitationInteraction(label, *, target, state='state', payoff='payoff', options=(0.0, 1.0), rate=1.0, rule='random', intensity=1.0, form='rate', probability=1.0, layer=None, phase=None, schedule='synchronous', ties='first')

Bases: NeighborInteraction

Imitation of a discrete state of the neighbours of every member, at a rate or with a probability per step.

By the rule 'random' a member copies a neighbour chosen at random, by 'majority' it takes the option that more of its neighbours hold than any other, and by 'best' it copies the neighbour whose payoff is largest, when that payoff is larger than its own. By 'better' it picks a neighbour at random and copies it when the neighbour's payoff is larger than its own. By 'fermi' it picks a neighbour at random and copies it with probability 1 / (1 + exp(-beta (P_j - P_i))), where P_j and P_i are the payoffs of the neighbour and of the member and beta is the intensity. A member without such a neighbour keeps its state. The rules read the neighbourhood of the space, the graph of a network or the neighbourhood of a lattice.

In the 'rate' form every member imitates at the rate rate, by jumps between every pair of options in continuous time. In the 'probability' form every member imitates with probability probability in every step of a discrete-time system, all members at once in its phase, reading the states the phases before it set. In the 'agent' form the target is an agent group, and every agent imitates its neighbours in the space with probability probability in its phase. The agents imitate at once, reading the states of the step before, under the schedule 'synchronous', one after the other in their order or in a new random order every step under 'sequential' or 'random-sequential', each reading the states the agents before it set, or each in its own activation under 'member'. The rule 'better' runs in the agent form only. Among neighbours of equal largest payoff, the rule 'best' of the agent form copies the first under the ties 'first' and one drawn uniformly under 'random'.

Parameters:

Name	Type	Description	Default
label	str	The label of the interaction, unique among the parts of its system.	required
target	Annotated[str, RoleSpec()]	The path of the group on the nodes or sites.	required
state	str	The name of the discrete state of the members.	'state'
payoff	str	The name of the state that holds the payoff of the members, which the rules 'best' and 'fermi' read.	'payoff'
options	Sequence[float]	The values the state takes.	(0.0, 1.0)
rate	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The rate at which every member imitates, in the rate form.	1.0
rule	str	'random', 'majority', 'best', or 'fermi'.	'random'
intensity	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The intensity beta of the Fermi rule.	1.0
form	str	'rate' for imitation in continuous time, 'probability' for imitation in discrete time, or 'agent' for imitation of the neighbours of the agents of an agent group.	'rate'
probability	Annotated[float, ParameterSpec(domain=closed(0, 1))]	The probability that a member imitates in a step, in the probability form.	1.0
layer	str | None	The layer of the network whose edges the interaction runs on, or None for the graph of the network.	None
phase	str | None	The phase of the step the agent and the probability forms run in, or None for the default phase.	None
schedule	str	The order of the agents in the agent form, 'synchronous', 'sequential', 'random-sequential', or 'member'.	'synchronous'
ties	str	The neighbour the rule 'best' of the agent form copies among ties, 'first' or 'random'.	'first'

Raises:

Type	Description
ValidationError	The rule, the form, the schedule, or the ties are unknown, a schedule or ties other than the defaults are given outside the agent form, or the options are not distinct.

Examples:

>>> from skcomplex.core.interactions import ImitationInteraction
>>> ImitationInteraction('imitation', target='voters', rule='majority').check().options
(0.0, 1.0)
>>> ImitationInteraction('imitation', target='players', rule='fermi', form='probability').check().form
'probability'
>>> ImitationInteraction('imitation', target='traders', rule='better', form='agent', phase='learn').check().rule
'better'

Source code in src/skcomplex/core/interactions/_imitation.py

def __init__(
    self,
    label: str,
    *,
    target: Annotated[str, RoleSpec()],
    state: str = 'state',
    payoff: str = 'payoff',
    options: Sequence[float] = (0.0, 1.0),
    rate: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 1.0,
    rule: str = 'random',
    intensity: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 1.0,
    form: str = 'rate',
    probability: Annotated[float, ParameterSpec(domain=RealSet.closed(0, 1))] = 1.0,
    layer: str | None = None,
    phase: str | None = None,
    schedule: str = 'synchronous',
    ties: str = 'first',
) -> None:
    super().__init__(label, layer=layer)
    self.target = target
    self.state = state
    self.payoff = payoff
    self.options = options
    self.rate = rate
    self.rule = rule
    self.intensity = intensity
    self.form = form
    self.probability = probability
    self.phase = phase
    self.schedule = schedule
    self.ties = ties

ImpulseInteraction(label, *, target, reset, times=(), period=None, start=0.0, count=None)

Bases: Interaction

Impulses that set states of the target to new values at given times, or at periodic times.

The impulses fall at every time of times and, with a period, at start + j * period for j from 0, count times or without end. An impulse at the initial time of a run does not fire. The new values are functions of the namespace, which holds namespace.target, the states and parameters of the target, and the time namespace.t. An impulse sets the states of every member of a group at once, and the trajectory records every impulse among its events, under the label of the interaction.

Parameters:

Name	Type	Description	Default
label	str	The label of the interaction, unique among the parts of its system.	required
target	Annotated[str, RoleSpec()]	The path of the element or group whose states the impulses set.	required
reset	Mapping[str, Callable[[SimpleNamespace], object]]	A mapping from the name of a state of the target to the function f(namespace) of its new value.	required
times	Sequence[float]	The times of the impulses.	()
period	float | None	The period of the periodic impulses, or None for none.	None
start	float	The time of the first periodic impulse.	0.0
count	int | None	The number of periodic impulses, or None for impulses without end.	None

Raises:

Type	Description
ValidationError	The impulses set no state, a new value is not a function, the impulses have no time, the period is not positive, or the count is not a positive integer.

Examples:

>>> from skcomplex.core.interactions import ImpulseInteraction
>>> reset = {'c': lambda s: s.target.c + 1.0}
>>> dose = ImpulseInteraction('dose', target='body', reset=reset, period=8.0, count=3)
>>> dose.check().start
0.0

Source code in src/skcomplex/core/interactions/_impulse.py

def __init__(
    self,
    label: str,
    *,
    target: Annotated[str, RoleSpec()],
    reset: Mapping[str, Callable[[SimpleNamespace], object]],
    times: Sequence[float] = (),
    period: float | None = None,
    start: float = 0.0,
    count: int | None = None,
) -> None:
    super().__init__(label)
    self.target = target
    self.reset = reset
    self.times = times
    self.period = period
    self.start = start
    self.count = count

Interaction(label)

Bases: BaseInteraction

Base class of interactions whose laws run once for the elements, or for every member of the groups, they bind.

A subclass takes its roles and parameters as arguments of its own __init__, annotating each with a RoleSpec or a ParameterSpec inside Annotated, and returns its laws from _get_laws, which returns no laws by default.

Parameters:

Name	Type	Description	Default
label	str	The label of the interaction, unique among the parts of its system.	required

Examples:

>>> from typing import Annotated
>>> from skcomplex.core.interactions import Interaction
>>> from skcomplex.specs import ParameterSpec, RoleSpec
>>> from sage.all__sagemath_symbolics import oo
>>> from sage.sets.real_set import RealSet
>>> class PredationInteraction(Interaction):
...     def __init__(
...         self,
...         label,
...         predator: Annotated[str, RoleSpec()],
...         prey: Annotated[str, RoleSpec()],
...         a: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 0.1,
...     ):
...         super().__init__(label)
...         self.predator = predator
...         self.prey = prey
...         self.a = a
>>> PredationInteraction('predation', predator='foxes', prey='rabbits')
PredationInteraction(label='predation', predator='foxes', prey='rabbits')

Source code in src/skcomplex/core/base.py

def __init__(self, label: str) -> None:
    """Store the label of the part.

    Args:
        label:
            The label of the part, unique among the parts of its system.
    """
    self.label = label

JumpInteraction(label, *, target, changes=None, rate=None, probability=None, waiting=None, reset=None)

Bases: Interaction

Random event that changes states of the target by signed amounts, at a rate, by a probability, or after a wait.

With a rate, the event fires at the times of a Poisson process of that propensity in continuous time. With a probability, it fires in a step of discrete time with that probability. With a waiting time, usually a draw from a distribution with nonnegative support, it fires after a time drawn anew at every firing, as in renewal processes, and every member of a bound group fires on its own clock. When the event fires it adds the changes to the states and, with a waiting time, sets the states of reset to their new values. Every function reads namespace.target, the states and parameters of the target, and the time namespace.t.

Parameters:

Name	Type	Description	Default
label	str	The label of the interaction, unique among the parts of its system.	required
target	Annotated[str, RoleSpec()]	The path of the element or group whose states the event changes.	required
changes	Mapping[str, float] | None	A mapping from the name of a state of the target to the signed amount the event adds to it.	None
rate	Callable[[SimpleNamespace], object] | None	The function rate(namespace) of the propensity of the event, or None.	None
probability	Callable[[SimpleNamespace], object] | None	The function probability(namespace) of the probability that the event fires in a step, or None.	None
waiting	Callable[[SimpleNamespace], object] | None	The function waiting(namespace) of the time until the next firing, or None.	None
reset	Mapping[str, Callable[[SimpleNamespace], object]] | None	With a waiting time, a mapping from the name of a state of the target to the function f(namespace) of its new value, one value per component of a vector state, or None.	None

Raises:

Type	Description
ValidationError	The event has not exactly one clock, changes nothing, or resets states without a waiting time, or a function is not callable.

Examples:

>>> from skcomplex.core.elements import CompartmentElement
>>> from skcomplex.core.interactions import JumpInteraction
>>> from skcomplex.core.system import System
>>> from skcomplex.engine.runtime import Simulation
>>> arrivals = JumpInteraction('arrivals', target='queue', changes={'n': 1}, rate=lambda s: 2.0)
>>> system = System(elements=[CompartmentElement('queue')], interactions=[arrivals])
>>> simulation = Simulation(system, dt=1.0, random_state=0).check().prepare()
>>> counts = simulation.run(until=5).trajectory_.values('queue.n')
>>> bool(all(count == round(count) for count in counts))
True

Source code in src/skcomplex/core/interactions/_jump.py

def __init__(
    self,
    label: str,
    *,
    target: Annotated[str, RoleSpec()],
    changes: Mapping[str, float] | None = None,
    rate: Callable[[SimpleNamespace], object] | None = None,
    probability: Callable[[SimpleNamespace], object] | None = None,
    waiting: Callable[[SimpleNamespace], object] | None = None,
    reset: Mapping[str, Callable[[SimpleNamespace], object]] | None = None,
) -> None:
    super().__init__(label)
    self.target: str | None = target
    self.changes = changes
    self.rate = rate
    self.probability = probability
    self.waiting = waiting
    self.reset = reset

LifeCycleInteraction(label, *, target, birth=None, death=None, rate=None, offspring=1, placement='here', inherit=(), split=(), mutate=None, schedule='random-sequential', phase=None, group=None, values=None, keep=None, reset=None, replace=False, renew=False, cull=None, kin=None, arrival=None, pick='first')

Bases: Interaction

Life cycle of the agents of a group, which give birth to offspring and die, every step of discrete time.

Every step the agents act once each, in their order under the schedule 'sequential', in a new random order under 'random-sequential', or each in its own activation under 'member', together with the rules of its element and the other member interactions, in the order the system lists them, without the reports of the step and without arrivals. An agent dies with the probability death, and an agent that survives gives birth with the probability birth, or as many times as a Poisson number of the mean rate. A probability is a number or a function of the agent and the model view, and a function that returns a condition gives probability one where it holds. Every birth creates offspring new agents, a number or a function of the parent and the model view, such as a Poisson draw, of the group group, or of the parent's group without one. A new agent stands in the cell of its parent, with the placement 'here', in a neighbouring cell that no agent of the group holds, with the placement 'neighbor', or at the location a function of the parent and the model view returns, where None means no place, and a birth without a place creates fewer agents.

A new agent copies the states inherit of its parent and takes the values that values(parent, model) returns, such as a generation one higher or drawn states. The parent splits each state of split with its offspring, keeping the share keep, a number or a function of the parent and the model view such as a random cut, or else an equal share, and the offspring share the rest equally. Then mutate(values, model) returns the values of a new agent from those it would get, and the other states take their defaults. A parent that gave birth takes the values reset, and with replace it is removed, so its offspring replace it, also when the birth created none. With renew an agent that dies is renewed in place instead of removed: it keeps its identity, its place, and the states the new values leave out, and takes the values of an offspring of its own. With the probability cull an agent that does not die removes, after its births, one member of group, or of the target without a group, for which kin(agent, member) holds, the first in the order of the group under the pick 'first', such as one it owns, or a random one under 'random'. Every step a Poisson number of the mean arrival, a number or a function of the model view, of new agents arrive without a parent, at the location of the placement function or else at a random place, with the values values(None, model). The life cycle reports the births, deaths, and arrivals of every step. It is an agent rule, so the system runs in discrete time.

Parameters:

Name	Type	Description	Default
label	str	The label of the interaction, unique among the parts of its system.	required
target	Annotated[str, RoleSpec()]	The path of the variable group whose agents give birth and die.	required
birth	Chance | None	The probability that an agent gives birth in a step, or None for no births by probability.	None
death	Chance | None	The probability that an agent dies in a step, or None for no deaths.	None
rate	Chance | None	The mean number of births of an agent in a step, a number or a function, or None for no births by rate.	None
offspring	Count	The number of new agents of a birth.	1
placement	Placement	Where a new agent stands, 'here', 'neighbor', or a function of the parent, None for an arrival, and the model view that returns its location.	'here'
inherit	Sequence[str]	The names of the states a new agent copies from its parent.	()
split	Sequence[str]	The names of the states the parent shares with its offspring.	()
mutate	Mutation | None	A function of the values of a new agent and the model view that returns its values, or None.	None
schedule	str	The order of the agents, 'sequential', 'random-sequential', or 'member'.	'random-sequential'
phase	str | None	The phase of the step the life cycle runs in, or None for the default phase.	None
group	Annotated[str | None, RoleSpec(optional=True)]	The path of the variable group the new agents join, or None for the group of the target.	None
values	Values | None	A function of the parent, None for an arrival, and the model view that returns values of a new agent by state name, or None.	None
keep	Chance | None	The share of every split state the parent keeps, a number or a function, or None for an equal share.	None
reset	Mapping[str, object] | None	The values a parent takes after it gives birth, by state name, or None.	None
replace	bool	Whether a parent that gives birth is removed.	False
renew	bool	Whether an agent that dies is renewed in place instead of removed.	False
cull	Chance | None	The probability that an agent removes one member of group, or of the target without a group, in a step, or None for no removals.	None
kin	Kinship | None	A function of an agent and a member of group that returns whether the agent may remove the member, or None for every member.	None
arrival	Arrival | None	The mean number of new agents that arrive in a step without a parent, or None for no arrivals.	None
pick	str	Which member of group a removal takes among those it may remove, 'first' or 'random'.	'first'

Raises:

Type	Description
ValidationError	The placement, the schedule, or the pick is unknown, both a birth probability and a rate are given, a function is not callable, or the schedule 'member' has arrivals, raised by check.

Examples:

>>> from typing import Annotated
>>> from skcomplex.core.elements import Element, GroupElement
>>> from skcomplex.core.interactions import LifeCycleInteraction
>>> from skcomplex.core.spaces import LatticeSpace
>>> from skcomplex.core.system import System
>>> from skcomplex.engine.runtime import Simulation
>>> from skcomplex.specs import StateSpec
>>> class CellElement(Element):
...     def __init__(self, label, visual=None, *, size: Annotated[float, StateSpec()] = 0.0):
...         super().__init__(label, visual=visual)
...         self.size = size
>>> cells = GroupElement('cells', element=CellElement, size=1, values={'size': 8.0}, variable=True)
>>> division = LifeCycleInteraction('division', target='cells', birth=1.0, split=('size',))
>>> system = System(elements=[cells], interactions=[division], space=LatticeSpace((3, 3)))
>>> trajectory = Simulation(system, random_state=0).check().prepare().run(until=2).trajectory_
>>> trajectory.agents('cells', 2).values['size'].tolist()
[2.0, 2.0, 2.0, 2.0]

Source code in src/skcomplex/core/interactions/_life_cycle.py

def __init__(
    self,
    label: str,
    *,
    target: Annotated[str, RoleSpec()],
    birth: Chance | None = None,
    death: Chance | None = None,
    rate: Chance | None = None,
    offspring: Count = 1,
    placement: Placement = 'here',
    inherit: Sequence[str] = (),
    split: Sequence[str] = (),
    mutate: Mutation | None = None,
    schedule: str = 'random-sequential',
    phase: str | None = None,
    group: Annotated[str | None, RoleSpec(optional=True)] = None,
    values: Values | None = None,
    keep: Chance | None = None,
    reset: Mapping[str, object] | None = None,
    replace: bool = False,
    renew: bool = False,
    cull: Chance | None = None,
    kin: Kinship | None = None,
    arrival: Arrival | None = None,
    pick: str = 'first',
) -> None:
    super().__init__(label)
    self.target = target
    self.birth = birth
    self.death = death
    self.rate = rate
    self.offspring = offspring
    self.placement = placement
    self.inherit = inherit
    self.split = split
    self.mutate = mutate
    self.schedule = schedule
    self.phase = phase
    self.group = group
    self.values = values
    self.keep = keep
    self.reset = reset
    self.replace = replace
    self.renew = renew
    self.cull = cull
    self.kin = kin
    self.arrival = arrival
    self.pick = pick

LocalRuleInteraction(label, *, target, state, rule, neighborhood, reads=None, schedule='synchronous', stochastic=False)

Bases: Interaction

Local rule of a cellular automaton, which sets the next value of every cell from its ordered neighbourhood.

The rule is a table of the next value of every configuration of the neighbourhood, TableSpec, a rule of the sum of the neighbourhood, TotalisticSpec, a rule of the cell and the sum of its neighbours, OuterTotalisticSpec, an object whose table() returns a TableSpec, or a function. A function receives one array per neighbour of every read state, in neighbourhood order, with one value per cell, the uniform array random when the rule is stochastic, and the parameters of the bound group it names, and returns the next values of the states, one array per state.

Parameters:

Name	Type	Description	Default
label	str	The label of the interaction, unique among the parts of its system.	required
target	Annotated[str, RoleSpec()]	The path of the group on the cells of a lattice, the nodes of a network, or a Cayley graph.	required
state	str | tuple[str, ...]	The name of the state the rule sets, or a tuple of the names of the states it sets together.	required
rule	object	The table, the totalistic rule, or the function.	required
neighborhood	BaseNeighborhood	The ordered neighbourhood of every cell, such as VonNeumannNeighborhood(center=True).	required
reads	Sequence[str] | None	The names of the states whose neighbourhoods a function reads, or None for the states it sets.	None
schedule	str	'synchronous', 'random-sequential', 'random-site', 'checkerboard', or 'colored'.	'synchronous'
stochastic	bool	Whether a function receives the uniform random array random.	False

Raises:

Type	Description
ValidationError	The state or the read states are not names, or the neighbourhood is not a neighbourhood.

Examples:

>>> from skcomplex.core.interactions import LocalRuleInteraction
>>> from skcomplex.core.spaces import VonNeumannNeighborhood
>>> from skcomplex.specs import TotalisticSpec
>>> parity = TotalisticSpec({1: 1, 3: 1})
>>> around = VonNeumannNeighborhood(n_dim=1, center=True)
>>> rule = LocalRuleInteraction('parity', target='cells', state='cell', rule=parity, neighborhood=around)
>>> rule.check().schedule
'synchronous'

Source code in src/skcomplex/core/interactions/_local_rule.py

def __init__(
    self,
    label: str,
    *,
    target: Annotated[str, RoleSpec()],
    state: str | tuple[str, ...],
    rule: object,
    neighborhood: BaseNeighborhood,
    reads: Sequence[str] | None = None,
    schedule: str = 'synchronous',
    stochastic: bool = False,
) -> None:
    super().__init__(label)
    self.target = target
    self.state = state
    self.rule = rule
    self.neighborhood = neighborhood
    self.reads = reads
    self.schedule = schedule
    self.stochastic = stochastic

MapInteraction(label, *, target, state, function, sources=(), mode='replace', schedule='synchronous', form='symbolic', phase=None)

Bases: Interaction

Map of a state of discrete time, whose next value is a function, or the state plus it.

In the mode 'replace' the next value of the state is f, and in the mode 'increment' it is the state plus f. Under the 'synchronous' schedule every member of a bound group updates at once, from the states at the start of the step. Under the 'sequential' schedule the members update one after the other, in their order, and under the 'random-sequential' schedule in a new random order every step, each member reading the values the members before it set. Under the 'member' schedule of the agent form, every member updates in its own activation, together with the rules of its element and the other member interactions, in the order the system lists them.

In the 'symbolic' form, function(namespace) returns a symbolic expression, or a vector of them for a vector state, where namespace.target holds the states and parameters of the target, namespace.sources the namespaces of the sources, and namespace.t the time, and may hold draws such as normal('noise'). In the 'agent' form, the target is an agent group and function(agent, model) returns the value as a number, or as an array of the shape of a vector state, so it can read the agent, its neighbours, and the history and storage of the model view.

Parameters:

Name	Type	Description	Default
label	str	The label of the interaction, unique among the parts of its system.	required
target	Annotated[str, RoleSpec()]	The path of the element or group whose state the map sets.	required
state	str	The name of the state of the target.	required
function	Callable[..., object]	The function f(namespace) in the symbolic form, or f(agent, model) in the agent form.	required
sources	Annotated[Sequence[str], RoleSpec(variadic=True)]	The paths of the elements, or members of groups such as cells[0], that the function reads.	()
mode	str	'replace' or 'increment'.	'replace'
schedule	str	'synchronous', 'sequential', 'random-sequential', or 'member' for the agent form.	'synchronous'
form	str	'symbolic' for update laws of the states of the target, or 'agent' for a rule over an agent group.	'symbolic'
phase	str | None	The phase of the step the agent form runs in, or None for the default phase.	None

Raises:

Type	Description
ValidationError	The mode, the schedule, or the form is unknown, the symbolic form has the schedule 'member', the function is not callable, or the agent form returns a value that does not match the state, raised by run of the simulation.

Examples:

>>> from skcomplex.core.elements import Element
>>> from skcomplex.core.interactions import MapInteraction
>>> from skcomplex.core.spaces import DiscreteTime
>>> from skcomplex.core.system import System
>>> from skcomplex.engine.runtime import Simulation
>>> from skcomplex.specs import StateSpec
>>> from typing import Annotated, SupportsFloat
>>> class ValueElement(Element):
...     def __init__(self, label, x: Annotated[float, StateSpec()] = 0.5):
...         super().__init__(label)
...         self.x = x
>>> logistic = MapInteraction(
...     'map', target='value', state='x', function=lambda s: 3 * s.target.x * (1 - s.target.x)
... )
>>> system = System(elements=[ValueElement('value')], interactions=[logistic], time=DiscreteTime())
>>> Simulation(system).check().prepare().run(until=2).trajectory_.values('value.x').tolist()
[0.5, 0.75, 0.5625]
>>> agents = MapInteraction('map', target='walkers', state='x', function=lambda a, m: a.x + 1, form='agent')
>>> type(agents.check().prepare().run(SimpleNamespace()).laws_[0]).__name__
'RuleLaw'
>>> from skcomplex.core.elements import GroupElement
>>> class WeightsElement(Element):
...     def __init__(self, label, visual=None, *, w: Annotated[object, StateSpec(vector=True)] = 0.0):
...         super().__init__(label, visual=visual)
...         self.w = w
>>> group = GroupElement('weights', element=WeightsElement, size=1, values={'w': [1.0, 2.0]})
>>> doubling = MapInteraction(
...     'double', target='weights', state='w', function=lambda a, m: [2 * x for x in a.w], form='agent'
... )
>>> trajectory = Simulation(System(elements=[group], interactions=[doubling])).check().prepare().run(until=2)
>>> trajectory.trajectory_.agents('weights', 2).values['w'].tolist()
[[4.0, 8.0]]

Source code in src/skcomplex/core/interactions/_map.py

def __init__(
    self,
    label: str,
    *,
    target: Annotated[str, RoleSpec()],
    state: str,
    function: Callable[..., object],
    sources: Annotated[Sequence[str], RoleSpec(variadic=True)] = (),
    mode: str = 'replace',
    schedule: str = 'synchronous',
    form: str = 'symbolic',
    phase: str | None = None,
) -> None:
    super().__init__(label)
    self.target = target
    self.state = state
    self.function = function
    self.sources = sources
    self.mode = mode
    self.schedule = schedule
    self.form = form
    self.phase = phase

update(model)

Set the state of every agent of the target to its next value, by the schedule.

Parameters:

Name	Type	Description	Default
model	ModelViewProtocol	The model view of the interaction.	required

Source code in src/skcomplex/core/interactions/_map.py

def update(self, model: ModelViewProtocol) -> None:
    """Set the state of every agent of the target to its next value, by the schedule.

    Args:
        model:
            The model view of the interaction.
    """
    self._update_agents(model, self.schedule, self._get_value, self._set_value)

MatchingInteraction(label, *, requesters, providers, rule=RandomSearchSpec(), limit='limit', quantity='quantity', allowance='allowance', transfers=(), phase=None, on_match=None)

Bases: Interaction

Matching of the members of a group of requesters with those of a group of providers by a rule.

Requesters state the highest clearing value they accept in their limit state, the quantity they want in their quantity state, and the most value they can give in their allowance state, and providers the lowest clearing value they accept and the quantity they offer. A name given as a pair names the state of the requesters first and of the providers second, and a missing state counts as no limit, one unit, and no allowance. The rule is a spec that names the algorithm that pairs the offers and holds its settings: random search, partners, partner switching, a queue, iterative, continuous, batch, or stable matching. The value of a match is its clearing value per unit. Every match lowers the quantities of both sides and moves every state of transfers: a negative sign moves value * quantity from the requester to the provider, and a positive sign moves quantity from the provider to the requester, so the totals stay the same. on_match(match, model) adds any other effect. The matching runs once per round of its phase and reports the quantity-weighted clearing value value, the matched volume, and the unmatched quantity left, as the reporters <label>.value, <label>.volume, and <label>.unmatched.

Parameters:

Name	Type	Description	Default
label	str	The label of the matching.	required
requesters	Annotated[str, RoleSpec()]	The path of the group of requesters.	required
providers	Annotated[str, RoleSpec()]	The path of the group of providers.	required
rule	MatchingRule | Pairing	The spec of the algorithm that matches the members, such as ContinuousMatchingSpec(), or a function rule(requesters, providers, rng, capacity) of the agents with a positive quantity that returns their pairs, which match at the midpoint of both limits. A PartnerSpec picks among the providers that still offer a quantity at a limit the requester accepts, in random order of the requesters.	RandomSearchSpec()
limit	str | tuple[str, str]	The state with the limit of both sides, or a pair of the requesters' and the providers' states. An agent without the default state 'limit' has no limit as a requester and the limit 0 as a provider.	'limit'
quantity	str | tuple[str, str]	The state with the quantity of both sides, or a pair of the requesters' and the providers' states. An agent without the default state 'quantity' offers the quantity 1.	'quantity'
allowance	str	The state of a requester with the most value it can give. A requester without the default state 'allowance' has no limit on what it gives.	'allowance'
transfers	Sequence[tuple[str, int]]	The states every match moves, each with the sign -1 for a move from the requester to the provider or 1 for a move from the provider to the requester.	()
phase	str | None	The phase the matching runs in, or None for the default phase.	None
on_match	Callable[[MatchSpec, ModelT], None] | None	A function of every match and the model view, run after the transfers, or None.	None

Raises:

Type	Description
ValidationError	A transfer has a sign other than -1 or 1, or the rule is neither a matching spec nor a function, or it is invalid, raised by check, or an agent lacks a state other than a default one that names its limit, quantity, or allowance, raised by run.

Examples:

>>> from skcomplex.core.interactions import MatchingInteraction
>>> from skcomplex.specs import ContinuousMatchingSpec
>>> matching = MatchingInteraction(
...     'exchange',
...     requesters='traders',
...     providers='traders',
...     rule=ContinuousMatchingSpec(),
...     limit=('bid', 'ask'),
... )
>>> matching.check().reports
('value', 'volume', 'unmatched')

Source code in src/skcomplex/core/interactions/_matching.py

def __init__(
    self,
    label: str,
    *,
    requesters: Annotated[str, RoleSpec()],
    providers: Annotated[str, RoleSpec()],
    rule: MatchingRule | Pairing = RandomSearchSpec(),
    limit: str | tuple[str, str] = 'limit',
    quantity: str | tuple[str, str] = 'quantity',
    allowance: str = 'allowance',
    transfers: Sequence[tuple[str, int]] = (),
    phase: str | None = None,
    on_match: Callable[[MatchSpec, ModelT], None] | None = None,
) -> None:
    super().__init__(label)
    self.requesters = requesters
    self.providers = providers
    self.rule = rule
    self.limit = limit
    self.quantity = quantity
    self.allowance = allowance
    self.transfers = transfers
    self.phase = phase
    self.on_match = on_match

match(model)

Match the members, settle the matches, and report the value, the volume, and the unmatched quantity.

Parameters:

Name	Type	Description	Default
model	ModelT	The model view of the matching.	required

Source code in src/skcomplex/core/interactions/_matching.py

def match(self, model: ModelT) -> None:
    """Match the members, settle the matches, and report the value, the volume, and the unmatched quantity.

    Args:
        model:
            The model view of the matching.
    """
    requesters, providers = self._get_offers(model)
    matches = self._get_matches(requesters, providers, model)
    for match in matches:
        self._settle(match, model)
    unmatched = sum(offer.quantity for offer in requesters + providers)
    self._report_matches(matches, unmatched, model)

MovementInteraction(label, *, target, step='neighbor', probability=1.0, rate=None, condition=None, accept=None, exclusive=False, score=None, scale=1.0, exponent=1.5, cost=None, vision=None, schedule='random-sequential', phase=None, ties=None)

Bases: Interaction

Movement of the agents of a group by steps, every step of discrete time.

A step is one of five kinds:

• 'neighbor' moves an agent to a uniformly drawn cell or node next to its own, on a lattice or a network.
• 'gaussian' adds to the position of an agent a normal vector whose components have the standard deviation scale, in a continuous space.
• 'levy' moves an agent in a uniformly drawn direction by a Pareto length, whose least value is scale and whose tail exponent is exponent, in a continuous space.
• 'best' moves an agent to the cell or node next to its own with the highest score(agent, cell, model). With a vision, a number or a function of the agent and the model view, it looks instead along every direction of the neighbourhood of a lattice up to that many cells, and at the cell of the agent, in the order of the axes, the positive direction first. Among cells of equal highest score, the ties 'random' draw one uniformly, 'first' take the first in that order, and 'nearest' take the nearest, the first of them among equal distances. By default the ties are 'random' without a vision and 'nearest' with one.
• A function step(agent, model) returns the destination, or None to stay.

An agent for which condition(agent, model) holds attempts one step every step of time, or a Poisson number of steps with the mean rate, and takes each attempt with the probability probability and then with the probability accept(agent, destination, model). An exclusive movement never moves an agent to a cell another agent of the group holds. The agents move one after the other, in their order under the schedule 'sequential', in a new random order every step under 'random-sequential', or each in its own activation under 'member', together with the rules of its element and the other member interactions, in the order the system lists them. Every step taken charges its cost, a TransferInteraction from the agent to itself or a function of the agent and the model view. The movement is an agent rule, so the system runs in discrete time.

Parameters:

Name	Type	Description	Default
label	str	The label of the interaction, unique among the parts of its system.	required
target	Annotated[str, RoleSpec()]	The path of the group whose agents move.	required
step	Step	The kind of step, 'neighbor', 'gaussian', 'levy', or 'best', or a function of the agent and the model view that returns the destination.	'neighbor'
probability	Annotated[float, ParameterSpec(domain=closed(0, 1))]	The probability that an agent takes an attempted step.	1.0
rate	float | None	The mean number of attempted steps of an agent in one unit of time, or None for one attempt every step.	None
condition	Callable[[AgentViewProtocol, ModelViewProtocol], bool] | None	A function of the agent and the model view that returns whether the agent moves, or None for every agent.	None
accept	Callable[[AgentViewProtocol, object, ModelViewProtocol], float] | None	A function of the agent, its destination, and the model view that returns the probability of taking the step, or None to take every step.	None
exclusive	bool	Whether an agent moves only to cells or nodes no other agent of the group holds.	False
score	Callable[[AgentViewProtocol, object, ModelViewProtocol], float] | None	The function of the agent, a neighbouring cell, and the model view that the step 'best' maximizes.	None
scale	Annotated[float, ParameterSpec(domain=open(0, oo))]	The standard deviation of every component of a Gaussian step, or the least length of a Lévy step.	1.0
exponent	Annotated[float, ParameterSpec(domain=open(0, oo))]	The tail exponent of the length of a Lévy step.	1.5
cost	Cost | None	A TransferInteraction that the agent pays to itself or to the environment for every step taken, a function of the agent and the model view run for every step taken, or None.	None
vision	Vision | None	The number of cells the step 'best' looks along every direction of a lattice, a number or a function of the agent and the model view, or None for the cells next to the agent.	None
schedule	str	The order of the agents, 'sequential', 'random-sequential', or 'member'.	'random-sequential'
phase	str | None	The phase of the step the movement runs in, or None for the default phase.	None
ties	str | None	The cell the step 'best' takes among cells of equal highest score, 'random', 'first', or 'nearest', or None for 'random' without a vision and 'nearest' with one.	None

Raises:

Type	Description
ValidationError	The step, the schedule, or the ties are unknown, the rate is not positive, a function is not callable, or the step 'best' has no score, raised by check.

Examples:

>>> from skcomplex.core.elements import Element, GroupElement
>>> from skcomplex.core.interactions import MovementInteraction
>>> from skcomplex.core.spaces import LatticeSpace
>>> from skcomplex.core.system import System
>>> from skcomplex.engine.runtime import Simulation
>>> walkers = GroupElement('walkers', element=Element, size=2, location=[(2, 2), (2, 2)])
>>> walk = MovementInteraction('walk', target='walkers', step='neighbor')
>>> space = LatticeSpace((5, 5), neighborhood='von_neumann')
>>> system = System(elements=[walkers], interactions=[walk], space=space)
>>> trajectory = Simulation(system, random_state=0).check().prepare().run(until=1).trajectory_
>>> [bool(abs(cell - 2).sum() == 1) for cell in trajectory.agents('walkers', 1).locations]
[True, True]

Source code in src/skcomplex/core/interactions/_movement.py

def __init__(
    self,
    label: str,
    *,
    target: Annotated[str, RoleSpec()],
    step: Step = 'neighbor',
    probability: Annotated[float, ParameterSpec(domain=RealSet.closed(0, 1))] = 1.0,
    rate: float | None = None,
    condition: Callable[[AgentViewProtocol, ModelViewProtocol], bool] | None = None,
    accept: Callable[[AgentViewProtocol, object, ModelViewProtocol], float] | None = None,
    exclusive: bool = False,
    score: Callable[[AgentViewProtocol, object, ModelViewProtocol], float] | None = None,
    scale: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 1.0,
    exponent: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 1.5,
    cost: Cost | None = None,
    vision: Vision | None = None,
    schedule: str = 'random-sequential',
    phase: str | None = None,
    ties: str | None = None,
) -> None:
    super().__init__(label)
    self.target = target
    self.step = step
    self.probability = probability
    self.rate = rate
    self.condition = condition
    self.accept = accept
    self.exclusive = exclusive
    self.score = score
    self.scale = scale
    self.exponent = exponent
    self.cost = cost
    self.vision = vision
    self.schedule = schedule
    self.phase = phase
    self.ties = ties

NeighborInteraction(label, layer=None)

Bases: BaseInteraction

Base class of interactions whose laws run once for every edge of the network of the nodes a role binds.

A role binds a group with one member per node or site, and a second role may bind a group on the edges of the same network and layer. The laws are written for one edge and read namespace.source and namespace.target, the members at both ends, namespace.edge, the member of the edge group, and the parameters. On a directed graph the ends follow the direction of the edge, and on an undirected graph the source comes first in node order. Rate, noise, and jump laws may target any of them, and update laws the edge member. delayed reads the past state of either end.

On fixed edges, namespace.source.degree and namespace.target.degree are the degrees of the ends, the number of edges of the layer at every node, both ways on a directed graph. A law that reads no edge member and, outside its reductions, only one end is a law of that end's member, and runs once per member rather than once per edge. Inside its reductions the other end stands for the other members of the group, so mean(f(namespace.source.x, namespace.target.x), over=namespace.source) is the mean of f over every member for the target. Cascade, drive, and local rule laws are laws of the members, and a cascade records its cascades as events.

Instead of a group on the edges, a second role may bind a second group of nodes. The laws then run once per member of the group of the first role, which namespace.target reads, and their reductions run over the members of the second group, which namespace.source reads, without edges. A role declared RoleSpec(optional=True) and bound to None binds nothing.

The roles may also bind separate elements. When the first role binds one element or a list of elements, the laws run once per element, which namespace.target reads. When the second role binds a group, namespace.source reads its members inside reductions, and otherwise namespace.source is the tuple of the namespaces of the elements the second role binds, or of the other elements of the list the first role binds.

The laws read every end under the name _get_role returns for 'source', 'target', and 'edge', the end itself by default, so a subclass that names its roles otherwise reads the ends under its own names.

Parameters:

Name	Type	Description	Default
label	str	The label of the interaction, unique among the parts of its system.	required
layer	str | None	The layer of the network whose edges the interaction runs on, or None for the layer of the edge group it binds, or else the graph of the network.	None

Examples:

>>> from typing import Annotated
>>> from skcomplex.core.interactions import NeighborInteraction
>>> from skcomplex.specs import ParameterSpec, RoleSpec
>>> class DiffusionInteraction(NeighborInteraction):
...     def __init__(
...         self,
...         label,
...         nodes: Annotated[str, RoleSpec()],
...         D: Annotated[float, ParameterSpec()] = 1.0,
...         layer=None,
...     ):
...         super().__init__(label, layer=layer)
...         self.nodes = nodes
...         self.D = D
>>> DiffusionInteraction('diffusion', nodes='cells', layer='roads').layer
'roads'

Source code in src/skcomplex/core/interactions/_neighbor.py

def __init__(self, label: str, layer: str | None = None) -> None:
    super().__init__(label)
    self.layer = layer

OptimizationInteraction(label, *, target, state, objective, bounds, algorithm=DifferentialEvolutionSpec(), phase=None)

Bases: Interaction

Move a group of candidates on an objective, one generation of a pymoo algorithm per step of discrete time.

Every candidate holds a vector state in a box, from which the first generation starts. Every step asks the algorithm for candidates, evaluates the objective on their array, tells the algorithm the values, and writes the candidates of the generation into the members. The interaction reports best, the lowest value so far, and mean, the mean value of the generation, every step.

Parameters:

Name	Type	Description	Default
label	str	The label of the interaction.	required
target	Annotated[str, RoleSpec()]	The path of the group of candidates.	required
state	str	The name of the vector state of the candidates.	required
objective	Callable[[Array], ArrayLike]	The function of the array of candidates, one row per candidate, that returns their values.	required
bounds	tuple[ArrayLike, ArrayLike]	The lower and the upper bounds of the box, each a number or one number per component.	required
algorithm	AlgorithmSpec	The settings of the algorithm, which select it.	DifferentialEvolutionSpec()
phase	str | None	The name of the phase of the step the rule runs in, or None for the default phase.	None

Raises:

Type	Description
ValidationError	The algorithm is not an algorithm spec, or the objective is not callable, raised by check.

Examples:

>>> from skcomplex.core.interactions import OptimizationInteraction
>>> objective = lambda values: (values**2).sum(axis=1)
>>> search = OptimizationInteraction('search', target='swarm', state='x', objective=objective, bounds=(-1, 1))
>>> search.check().label
'search'

Source code in src/skcomplex/core/interactions/_optimization.py

def __init__(
    self,
    label: str,
    *,
    target: Annotated[str, RoleSpec()],
    state: str,
    objective: Callable[[Array], npt.ArrayLike],
    bounds: tuple[npt.ArrayLike, npt.ArrayLike],
    algorithm: AlgorithmSpec = DifferentialEvolutionSpec(),
    phase: str | None = None,
) -> None:
    super().__init__(label)
    self.target = target
    self.state = state
    self.objective = objective
    self.bounds = bounds
    self.algorithm = algorithm
    self.phase = phase

PairCouplingInteraction(label, first, second, cutoff=1.0, position='x', *, write='x', function=None, potential=None, divisor=None, period=None)

Bases: PairInteraction

Coupling of every pair of members closer than a cutoff, through a function of the pair or a pair potential.

Every member of a pair adds a value to the rate of its state write, which gives the highest derivative of a state of any order. With function, the value is function(namespace), where namespace.first is the member that adds it and namespace.second the other member. With potential, a function potential(namespace, r) of the distance r between the values of the vector state position of the pair, the value is minus the gradient of the potential with respect to the member's own position. A divisor names a state or parameter of the members that divides the value. With several states write, the function returns one value per state, in their order, so one pair couples several states jointly.

Parameters:

Name	Type	Description	Default
label	str	The label of the interaction, unique among the parts of its system.	required
first	Annotated[str, RoleSpec()]	The path of the group whose members are the first of every pair.	required
second	Annotated[str, RoleSpec()]	The path of the group whose members are the second of every pair, which may be the first group.	required
cutoff	Annotated[float, ParameterSpec(domain=open(0, oo))]	The largest distance between the members of a pair.	1.0
position	str	The name of the vector state whose distances select the pairs, and which the potential is a function of.	'x'
write	str | Sequence[str]	The name of the state whose rate the coupling adds to, or a sequence of names, one per value of the function.	'x'
function	NamespaceFunction | None	The value a member adds, or a sequence of values, one per state write, a callable of the namespace of the pair, or None when the potential is given.	None
potential	PairPotential | None	The pair potential, a callable of the namespace of the pair and the distance, or None when the function is given.	None
divisor	str | None	The name of the state or parameter of the members that divides the value, or None for none.	None
period	float | None	The side of a periodic box, for minimum image distances, or None for the period of the space.	None

Raises:

Type	Description
ValidationError	Not exactly one of the function and the potential is a callable, or the potential has several states to write, raised by check.

Examples:

>>> from types import SimpleNamespace
>>> from sage.all__sagemath_symbolics import var, vector
>>> from skcomplex.core.interactions import PairCouplingInteraction
>>> first = SimpleNamespace(x=vector([var('x1')]), m=var('m1'))
>>> second = SimpleNamespace(x=vector([var('x2')]), m=var('m2'))
>>> spring = PairCouplingInteraction('spring', 'a', 'b', potential=lambda s, r: r**2 / 2, divisor='m')
>>> [law.value for law in spring.check().prepare().run(SimpleNamespace(first=first, second=second)).laws_]
[(-(x1 - x2)/m1), ((x1 - x2)/m2)]

Source code in src/skcomplex/core/interactions/_pair_coupling.py

def __init__(
    self,
    label: str,
    first: Annotated[str, RoleSpec()],
    second: Annotated[str, RoleSpec()],
    cutoff: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 1.0,
    position: str = 'x',
    *,
    write: str | Sequence[str] = 'x',
    function: NamespaceFunction | None = None,
    potential: PairPotential | None = None,
    divisor: str | None = None,
    period: float | None = None,
) -> None:
    super().__init__(label, first, second, cutoff=cutoff, position=position)
    self.write = write
    self.function = function
    self.potential = potential
    self.divisor = divisor
    self.period = period

PairInteraction(label, first, second, cutoff=1.0, position='x')

Bases: BaseInteraction

Base class of interactions whose laws run once for every pair of members closer than a cutoff.

The laws are written for one pair and read namespace.first and namespace.second, the members of the pair, and the parameters. A subclass adds its own roles and parameters to the ones of every pair interaction, and a subclass that names the members otherwise returns their names from _get_role, which the laws read them under.

Parameters:

Name	Type	Description	Default
label	str	The label of the interaction, unique among the parts of its system.	required
first	Annotated[str, RoleSpec()]	The path of the group whose members are the first of every pair.	required
second	Annotated[str, RoleSpec()]	The path of the group whose members are the second of every pair, which may be the first group.	required
cutoff	Annotated[float, ParameterSpec(domain=open(0, oo))]	The largest distance between the members of a pair.	1.0
position	str	The name of the vector state that holds the positions of the members.	'x'

Examples:

>>> from typing import Annotated
>>> from skcomplex.core.interactions import PairInteraction
>>> from skcomplex.specs import ParameterSpec
>>> class RepulsionInteraction(PairInteraction):
...     def __init__(
...         self, label, first, second, cutoff=1.0, position='x', k: Annotated[float, ParameterSpec()] = 1.0
...     ):
...         super().__init__(label, first, second, cutoff=cutoff, position=position)
...         self.k = k
>>> RepulsionInteraction('repulsion', 'particles', 'particles').cutoff
1.0

Source code in src/skcomplex/core/interactions/_pair.py

def __init__(
    self,
    label: str,
    first: Annotated[str, RoleSpec()],
    second: Annotated[str, RoleSpec()],
    cutoff: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 1.0,
    position: str = 'x',
) -> None:
    super().__init__(label)
    self.first = first
    self.second = second
    self.cutoff = cutoff
    self.position = position

RateInteraction(label, *, target, state='x', rate=0.0, sources=(), noise=None, divisor=None, process=None, index=None, convention='ito', hurst=None, conserved=False)

Bases: Interaction

Rate rate(ns) and noise noise(ns) dW added to the highest time derivative of a state of a target.

On a state of order 1 the rate is its first derivative, and on a state of order k it adds to the k-th derivative, so rates on one state superpose. With a divisor d the rate and the noise are divided by d, the name of a state or parameter of the target, such as an inertia, a friction, or a capacitance, which backends can read.

A rate, a noise, or a divisor function takes the namespace of the interaction, with the namespaces target and sources of the bound elements, the time t, and the symbols of the parameters a subclass declares. A number stands for a constant, and a number on a vector state applies to every component. The noise is a Wiener process per component and member by default, a process of a WienerProcessSpec to share or correlate it across states and members, or a fractional Brownian motion of a Hurst index.

Parameters:

Name	Type	Description	Default
label	str	The label of the interaction, unique among the parts of its system.	required
target	Annotated[str, RoleSpec()]	The path of the element or group whose state the rate changes.	required
state	str	The name of the state of the target that the rate changes.	'x'
rate	Callable[[SimpleNamespace], object] | float	A function of the namespace that returns the rate, or a number.	0.0
sources	Annotated[Sequence[str], RoleSpec(variadic=True)]	The paths of the elements or groups the rate and the noise read but do not change.	()
noise	Callable[[SimpleNamespace], object] | float | None	A function of the namespace that returns the amplitude of the noise, a number, or None for no noise.	None
divisor	str | Callable[[SimpleNamespace], object] | None	The name of the state or parameter of the target that divides the rate and the noise, a function of the namespace of the target that returns the divisor, such as a moment of inertia m l^2, or None.	None
process	Annotated[WienerProcessSpec | None, RoleSpec(noise=True)]	The Wiener processes that drive the noise, or None for one independent process per component and member.	None
index	int | None	The process of process that drives the noise, or None for every process, one per component of a vector state.	None
convention	str	The stochastic calculus of the noise, 'ito', 'stratonovich', or 'anti-ito'.	'ito'
hurst	Callable[[SimpleNamespace], object] | float | None	The Hurst index in (0, 1) of a fractional Brownian motion that drives the noise, a number or a function of the namespace that returns a parameter, or None for a Wiener process.	None
conserved	bool	Whether the noise is the divergence of a random flux on the links of the lattice of a field, so it keeps the total of the state.	False

Raises:

Type	Description
ValidationError	The state is not a name, a function is neither callable nor a number, the convention is unknown, the Hurst index is outside (0, 1), or the noise options cannot combine.

Examples:

>>> from skcomplex.core.elements import CompartmentElement
>>> from skcomplex.core.interactions import RateInteraction
>>> from skcomplex.core.system import System
>>> from skcomplex.engine.runtime import Simulation
>>> decay = RateInteraction('decay', target='stock', state='n', rate=lambda s: -0.5 * s.target.n)
>>> system = System(elements=[CompartmentElement('stock', n=8.0)], interactions=[decay])
>>> trajectory = Simulation(system, dt=0.01).check().prepare().run(until=2.0).trajectory_
>>> round(float(trajectory.values('stock.n')[-1]), 3)
2.943

Source code in src/skcomplex/core/interactions/_rate.py

def __init__(
    self,
    label: str,
    *,
    target: Annotated[str, RoleSpec()],
    state: str = 'x',
    rate: Callable[[SimpleNamespace], object] | float = 0.0,
    sources: Annotated[Sequence[str], RoleSpec(variadic=True)] = (),
    noise: Callable[[SimpleNamespace], object] | float | None = None,
    divisor: str | Callable[[SimpleNamespace], object] | None = None,
    process: Annotated[WienerProcessSpec | None, RoleSpec(noise=True)] = None,
    index: int | None = None,
    convention: str = 'ito',
    hurst: Callable[[SimpleNamespace], object] | float | None = None,
    conserved: bool = False,
) -> None:
    super().__init__(label)
    self.target = target
    self.state = state
    self.rate = rate
    self.sources = sources
    self.noise = noise
    self.divisor = divisor
    self.process = process
    self.index = index
    self.convention = convention
    self.hurst = hurst
    self.conserved = conserved

ReactionInteraction(label, *, target=None, reactants=None, products=None, k=1.0, propensity=None, modifiers=None, mode='rate', form='symbolic', phase=None, schedule='synchronous')

Bases: JumpInteraction

Reaction that moves units from its reactants to its products at a mass-action or a given propensity.

A reactant, a product, or a modifier is a pair of the path of an element, or of a member of a group such as cells[0], and the name of its state that holds the units, mapped to its stoichiometric coefficient. A coefficient is a positive number, or a function of the namespace of the reaction that returns an expression of its parameters, such as lambda s: 1 / s.target.epsilon. So S + I -> 2 I has the reactants {('S', 'n'): 1, ('I', 'n'): 1} and the products {('I', 'n'): 2}, and two states of one element are ('cell', 'u') and ('cell', 'v'). A species may also be the name of a state of the target alone, such as {'u': 1, 'v': 2}. A target that is a group or a field reacts in every member, with the species of every member its own states, as the reactions at every site of a reaction-diffusion field.

A subclass may bind its species under roles of its own names, such as susceptible and infected: it returns its entries from _get_reactants, _get_products, and _get_modifiers, pairs of the path a role binds and a state name, and the reaction reads every pair from the namespace of the role that binds its path.

The propensity is k times the product of every reactant and modifier state to the power of its coefficient, or the expression the propensity function returns. The reaction changes every state by its coefficient among the products minus its coefficient among the reactants, so it conserves the total of the states whose changes sum to zero, and modifiers enter the propensity but do not change.

In the mode 'rate' the reaction is a rate equation. In the mode 'jump' it is the random event of a JumpInteraction at the propensity that moves the units of the coefficients. In the mode 'langevin' it is the chemical Langevin equation, the rate equation with one Wiener process of amplitude the square root of the propensity driving every state the reaction changes.

In the 'agent' form the target is an agent group and every member reacts on its own states, the species named by state names alone, in every step from the states at its start. The propensity of a member is k times the product of its reactant and modifier states to the power of their coefficients, or the number propensity(agent, model) returns, and a coefficient that is a function is the number coefficient(agent, model) returns. In the mode 'rate' the member adds the changes times its propensity, one unit step of the rate equation, and in the mode 'jump' it adds the changes with the probability 1 - exp(-propensity) that an event fires in the step. The members react at once from the states at the start of the rule under the schedule 'synchronous', one after the other in their order or in a new random order every step under 'sequential' or 'random-sequential', or each in its own activation under 'member'.

Parameters:

Name	Type	Description	Default
label	str	The label of the reaction, unique among the parts of its system.	required
target	Annotated[str | None, RoleSpec()]	The path of the element, group, or field whose states the species named by a state alone are, or None.	None
reactants	Annotated[Stoichiometry | None, RoleSpec(variadic=True)]	The coefficients of the reactants, or None for a source.	None
products	Annotated[Stoichiometry | None, RoleSpec(variadic=True)]	The coefficients of the products, or None for a sink.	None
k	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The rate constant.	1.0
propensity	Callable[..., object] | None	A function of the namespace of the reaction that returns the propensity, or None for mass action. The namespace holds the tuples reactants, products, and modifiers of the namespaces of the elements their pairs bind, in the order of their entries, target, the namespace of the target or of one of its members, and the symbol k. In the agent form, a function propensity(agent, model) that returns a number.	None
modifiers	Annotated[Stoichiometry | None, RoleSpec(variadic=True)]	The coefficients of the modifiers, whose states the propensity reads but the reaction does not change, or None.	None
mode	str	The form of the laws, 'rate', 'jump', or 'langevin'.	'rate'
form	str	'symbolic' for the laws of the states, or 'agent' for a rule in which every member of an agent group reacts on its own states.	'symbolic'
phase	str | None	The phase of the step the agent form runs in, or None for the default phase.	None
schedule	str	The order of the members in the agent form, 'synchronous', 'sequential', 'random-sequential', or 'member'.	'synchronous'

Raises:

Type	Description
ValidationError	The mode, the form, or the schedule is unknown, the symbolic form is not synchronous, the reaction has neither reactants nor products, an entry is neither a pair of a path and a state name nor a state name of a given target, a coefficient is neither a positive number nor a function, or the agent form has no target, has an entry that is a pair, or has the mode 'langevin'.

Examples:

>>> from skcomplex.core.elements import CompartmentElement
>>> from skcomplex.core.interactions import ReactionInteraction
>>> from skcomplex.core.system import System
>>> from skcomplex.engine.runtime import Simulation
>>> infection = ReactionInteraction(
...     'infection', reactants={('S', 'n'): 1, ('I', 'n'): 1}, products={('I', 'n'): 2}, k=0.0005
... )
>>> recovery = ReactionInteraction('recovery', reactants={('I', 'n'): 1}, products={('R', 'n'): 1}, k=0.1)
>>> system = System(
...     elements=[CompartmentElement('S', n=990), CompartmentElement('I', n=10), CompartmentElement('R')],
...     interactions=[infection, recovery],
... )
>>> states = Simulation(system, dt=1.0).check().prepare().run(until=50).trajectory_.states
>>> round(float(states[-1].sum()), 6)
1000.0

The decay n -> 0 at every site of a field, at the rate k n of the site:

>>> import numpy as np
>>> from skcomplex.core.elements import FieldElement
>>> from skcomplex.core.spaces import LatticeSpace
>>> decay = ReactionInteraction('decay', target='cells', reactants={'n': 1}, k=1.0)
>>> values = {'n': np.arange(3.0)}
>>> cells = FieldElement('cells', element=CompartmentElement, space=LatticeSpace((3,)), values=values)
>>> simulation = Simulation(System(elements=[cells], interactions=[decay]), dt=1.0).check().prepare()
>>> np.round(simulation.run(until=1).trajectory_.lattice('cells.n')[-1], 3).tolist()
[0.0, 0.368, 0.736]

A coefficient that is a parameter, the decay at the rate x that removes 1 / epsilon units per event:

>>> from typing import Annotated
>>> from skcomplex.core.elements import Element
>>> from skcomplex.specs import ParameterSpec, StateSpec
>>> class ScaledElement(Element):
...     def __init__(
...         self,
...         label,
...         x: Annotated[float, StateSpec()] = 1.0,
...         epsilon: Annotated[float, ParameterSpec()] = 0.5,
...     ):
...         super().__init__(label)
...         self.x = x
...         self.epsilon = epsilon
>>> scaled = ReactionInteraction(
...     'decay', target='box', reactants={'x': lambda s: 1 / s.target.epsilon}, propensity=lambda s: s.target.x
... )
>>> simulation = Simulation(System(elements=[ScaledElement('box')], interactions=[scaled]), dt=0.01)
>>> round(float(simulation.check().prepare().run(until=0.5).trajectory_.values('box.x')[-1]), 3)
0.368

Every agent of a group decays on its own states, by k n in a step of the rate equation:

>>> from skcomplex.core.elements import GroupElement
>>> from skcomplex.core.spaces import DiscreteTime
>>> class CellElement(Element):
...     def __init__(self, label, n: Annotated[float, StateSpec()] = 3.0):
...         super().__init__(label)
...         self.n = n
>>> cells = GroupElement('cells', element=CellElement, size=2)
>>> decay = ReactionInteraction('decay', target='cells', reactants={'n': 1}, k=0.1, form='agent')
>>> system = System(elements=[cells], interactions=[decay], time=DiscreteTime())
>>> trajectory = Simulation(system).check().prepare().run(until=2).trajectory_
>>> np.round(trajectory.agents('cells', 2).values['n'], 6).tolist()
[2.43, 2.43]

Source code in src/skcomplex/core/interactions/_reaction.py

def __init__(
    self,
    label: str,
    *,
    target: Annotated[str | None, RoleSpec()] = None,
    reactants: Annotated[Stoichiometry | None, RoleSpec(variadic=True)] = None,
    products: Annotated[Stoichiometry | None, RoleSpec(variadic=True)] = None,
    k: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 1.0,
    propensity: Callable[..., object] | None = None,
    modifiers: Annotated[Stoichiometry | None, RoleSpec(variadic=True)] = None,
    mode: str = 'rate',
    form: str = 'symbolic',
    phase: str | None = None,
    schedule: str = 'synchronous',
) -> None:
    super().__init__(label, target='')
    self.target = target
    self.reactants = reactants
    self.products = products
    self.k = k
    self.propensity = propensity
    self.modifiers = modifiers
    self.mode = mode
    self.form = form
    self.phase = phase
    self.schedule = schedule

react(model)

React in every agent of the target on its own states, by the schedule.

Parameters:

Name	Type	Description	Default
model	ModelViewProtocol	The model view of the interaction.	required

Source code in src/skcomplex/core/interactions/_reaction.py

def react(self, model: ModelViewProtocol) -> None:
    """React in every agent of the target on its own states, by the schedule.

    Args:
        model:
            The model view of the interaction.
    """
    self._update_agents(model, self.schedule, self._get_agent_step, self._apply_agent_step)

ReinforcementInteraction(label, *, target, scores, payoff, rate=0.1, rule='average', form='symbolic', phase=None, chosen=None, schedule='synchronous')

Bases: Interaction

Reinforcement of the scores of options, each moving towards the payoff its option realized in the step.

Every step, the score s_k of option k of every member moves towards the realized payoff P_k at the rate a, s_k <- s_k + a (P_k - s_k), so a score is an exponentially weighted mean of past payoffs, and the last payoff at the rate 1, by the rule 'average'. By the rule 'cumulative' the score adds the payoff, s_k <- s_k + P_k, so it is the sum of past payoffs, such as a virtual score. In the agent form the scores may be the components of one vector state, one per option, and a group of one member holds scores that several groups share. In the 'symbolic' form, payoff(namespace, k) returns the payoff of option k as a symbolic expression, where namespace.target holds the states and parameters of the group. In the 'agent' form, the group is an agent group and payoff(agent, model, k) returns the payoff as a number, so it can read the agent and the history and storage of the model view. With chosen, the name of a state that holds the index of the option a member chose, only the score of that option moves, and the payoff of the other options is not read. In the agent form every agent reinforces at once under the schedule 'synchronous', one after the other in their order or in a new random order every step under 'sequential' or 'random-sequential', and each in its own activation under 'member', together with the rules of its element and the other member interactions, in the order the system lists them, such as after a choice and a transfer in one visit.

Parameters:

Name	Type	Description	Default
label	str	The label of the interaction, unique among the parts of its system.	required
target	Annotated[str, RoleSpec()]	The path of the element or group whose members hold the scores.	required
scores	str | Sequence[str]	The names of the states that hold the score of every option, in the order of the options, or in the agent form the name of one vector state whose components are the scores.	required
payoff	Callable[..., float | Expression]	The function payoff(namespace, k) in the symbolic form, or payoff(agent, model, k) in the agent form, that returns the realized payoff of option k.	required
rate	Annotated[float, ParameterSpec(domain=closed(0, 1))]	The rate a at which the scores move towards the payoffs, between 0 and 1, by the rule 'average'.	0.1
rule	str	'average' to move the scores towards the payoffs, or 'cumulative' to add the payoffs to the scores.	'average'
form	str	'symbolic' for update laws of the states of a group, or 'agent' for a rule over an agent group.	'symbolic'
phase	str | None	The phase of the step the agent form runs in, or None for the default phase.	None
chosen	str | None	The name of the state that holds the index of the option a member chose, so only its score moves, or None to move every score.	None
schedule	str	The order of the agent form, 'synchronous' for every agent at once, 'sequential' or 'random-sequential' for one agent after the other, or 'member' for every agent in its own activation.	'synchronous'

Raises:

Type	Description
ValidationError	The form, the rule, or the schedule is unknown, the symbolic form runs member by member, there are no scores, or the payoff is not callable.

Examples:

>>> from skcomplex.core.interactions import ReinforcementInteraction
>>> reinforcement = ReinforcementInteraction(
...     'reinforcement', target='learners', scores=('first', 'second'), payoff=lambda s, k: k, rate=0.5
... )
>>> reinforcement.check().rate
0.5
>>> ReinforcementInteraction(
...     'virtual', target='players', scores='scores', payoff=lambda a, m, k: k, rule='cumulative', form='agent'
... ).check().rule
'cumulative'
>>> ReinforcementInteraction(
...     'satisfaction', target='customers', scores=('s1', 's2'), payoff=lambda a, m, k: 1.0, chosen='shop',
...     schedule='member', form='agent',
... ).check().chosen
'shop'

Source code in src/skcomplex/core/interactions/_reinforcement.py

def __init__(
    self,
    label: str,
    *,
    target: Annotated[str, RoleSpec()],
    scores: str | Sequence[str],
    payoff: Callable[..., float | Expression],
    rate: Annotated[float, ParameterSpec(domain=RealSet.closed(0, 1))] = 0.1,
    rule: str = 'average',
    form: str = 'symbolic',
    phase: str | None = None,
    chosen: str | None = None,
    schedule: str = 'synchronous',
) -> None:
    super().__init__(label)
    self.target = target
    self.scores = scores
    self.payoff = payoff
    self.rate = rate
    self.rule = rule
    self.form = form
    self.phase = phase
    self.chosen = chosen
    self.schedule = schedule

reinforce(model)

Move the scores of every agent towards the payoffs of its options, by the schedule.

Parameters:

Name	Type	Description	Default
model	ModelViewProtocol	The model view of the interaction.	required

Source code in src/skcomplex/core/interactions/_reinforcement.py

def reinforce(self, model: ModelViewProtocol) -> None:
    """Move the scores of every agent towards the payoffs of its options, by the schedule.

    Args:
        model:
            The model view of the interaction.
    """
    self._update_agents(model, self.schedule, self._get_scores, self._write)

RelaxationInteraction(label, *, target, state='x', goal=0.0, time=1.0, noise=None)

Bases: RateInteraction

Relaxation (goal(ns) - x) / time(ns) of a state x of a target towards a goal, with optional noise.

With a constant goal and time and an additive noise of amplitude sigma, the state follows the Ornstein-Uhlenbeck process dx = (goal - x) / time dt + sigma dW, whose stationary variance is sigma**2 time / 2. Without noise it is first-order smoothing, or adaptive expectation, towards the goal. A goal or a time function takes the namespace of the interaction, as for RateInteraction, and a number stands for a constant. A goal that is a number applies to every component of a vector state.

Parameters:

Name	Type	Description	Default
label	str	The label of the interaction, unique among the parts of its system.	required
target	Annotated[str, RoleSpec()]	The path of the element or group whose state relaxes.	required
state	str	The name of the state of the target that relaxes.	'x'
goal	Callable[[SimpleNamespace], object] | float	A function of the namespace that returns the goal, or a number.	0.0
time	Callable[[SimpleNamespace], object] | float	A function of the namespace that returns the relaxation time, or a positive number.	1.0
noise	Callable[[SimpleNamespace], object] | float | None	A function of the namespace that returns the amplitude of the noise, a number, or None for no noise.	None

Raises:

Type	Description
ValidationError	The state is not a name, or a function is neither callable nor a number.

Examples:

>>> from skcomplex.core.elements import CompartmentElement
>>> from skcomplex.core.interactions import RelaxationInteraction
>>> from skcomplex.core.system import System
>>> from skcomplex.engine.runtime import Simulation
>>> smoothing = RelaxationInteraction('smoothing', target='belief', state='n', goal=10.0, time=0.5)
>>> system = System(elements=[CompartmentElement('belief', n=0.0)], interactions=[smoothing])
>>> trajectory = Simulation(system, dt=0.01).check().prepare().run(until=5.0).trajectory_
>>> round(float(trajectory.values('belief.n')[-1]), 3)
10.0

Source code in src/skcomplex/core/interactions/_relaxation.py

def __init__(
    self,
    label: str,
    *,
    target: Annotated[str, RoleSpec()],
    state: str = 'x',
    goal: Callable[[SimpleNamespace], object] | float = 0.0,
    time: Callable[[SimpleNamespace], object] | float = 1.0,
    noise: Callable[[SimpleNamespace], object] | float | None = None,
) -> None:
    super().__init__(label, target=target, state=state, noise=noise)
    self.goal = goal
    self.time = time

SamplingInteraction(label, *, target, state, objective, move='flip', intensity=1.0, rule='metropolis', schedule='checkerboard', neighborhood=None, values=(), width=1.0, cooling=None)

Bases: Interaction

Monte Carlo sweeps of a state of the members of a lattice or network group, one sweep per unit of time.

Every sweep proposes a move of the state of every member and accepts it from the change of the local objective of the member, the terms of the objective that involve its value, at the intensity. The local objective is a function of the namespace, which holds namespace.target, the states and parameters of the member, and namespace.target.other, a neighbour, whose values neighbors(...) sums over neighborhood. A move is 'flip', the negative value, 'random-value', a value drawn from values, 'rotate', the value plus a uniform number in [-width, width], 'gaussian', the value plus a normal number of standard deviation width, or a function move(values, random) of the array of the current values of some members and the random generator that returns their proposed values. The move 'cluster' is the Wolff algorithm instead: a sweep flips clusters until the flipped members number as many as the members, each cluster grown from a random member through the bonds to neighbours with probability 1 - exp(min(0, -intensity * (f(-s, n) - f(s, n)))), where f(s, n) is the term of the local objective of a member of value s with a neighbour of value n. It takes an objective of such pair terms that the flip of both values keeps, and ignores the rule and the schedule.

Parameters:

Name	Type	Description	Default
label	str	The label of the interaction, unique among the parts of its system.	required
target	Annotated[str, RoleSpec()]	The path of the group on the sites of a lattice or the nodes of a network.	required
state	str	The name of the state the moves change.	required
objective	Callable[[SimpleNamespace], object]	The function objective(namespace) of the local objective of a member.	required
move	str | Callable[..., object]	'flip', 'random-value', 'rotate', 'gaussian', 'cluster', or a function that proposes values.	'flip'
intensity	Annotated[float, ParameterSpec(domain=open(0, oo))]	The factor of the change of the objective in the acceptance exp(-intensity * change).	1.0
rule	str	The acceptance, 'metropolis', 'glauber', or 'heat-bath', which draws among all values of a discrete move by the weights exp(-intensity * objective).	'metropolis'
schedule	str	'checkerboard', 'colored', or 'random-sequential'.	'checkerboard'
neighborhood	BaseNeighborhood | None	The neighbourhood the objective reduces over, or None for the von Neumann neighbourhood of a square lattice. A network takes AdjacencyNeighborhood().	None
values	Sequence[float]	The values a random-value move draws from.	()
width	float	The largest change of a rotation, or the standard deviation of a Gaussian move.	1.0
cooling	Callable[[Expression], object] | None	The function of the time symbol that multiplies one over the intensity, or None for a constant intensity.	None

Raises:

Type	Description
ValidationError	The move, the rule, or the schedule is unknown, or the objective is not callable.

Examples:

>>> from skcomplex.core.elements import Element, LatticeElement
>>> from skcomplex.core.interactions import SamplingInteraction
>>> from skcomplex.core.laws.utils import neighbors
>>> from skcomplex.core.spaces import DiscreteTime, LatticeSpace
>>> from skcomplex.core.system import System
>>> from skcomplex.engine.runtime import Simulation
>>> from skcomplex.specs import StateSpec
>>> from typing import Annotated
>>> class SpinElement(Element):
...     def __init__(self, label, s: Annotated[float, StateSpec()] = 1.0):
...         super().__init__(label)
...         self.s = s
>>> objective = lambda n: -n.target.s * neighbors(n.target.other.s)
>>> ising = SamplingInteraction('ising', target='spins', state='s', objective=objective, intensity=1.0)
>>> spins = LatticeElement('spins', element=SpinElement, space=LatticeSpace((8, 8)))
>>> system = System(elements=[spins], interactions=[ising], time=DiscreteTime())
>>> states = Simulation(system, random_state=0).check().prepare().run(until=20).trajectory_.states
>>> bool(abs(states[-1].mean()) > 0.9)
True

Source code in src/skcomplex/core/interactions/_sampling.py

def __init__(
    self,
    label: str,
    *,
    target: Annotated[str, RoleSpec()],
    state: str,
    objective: Callable[[SimpleNamespace], object],
    move: str | Callable[..., object] = 'flip',
    intensity: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 1.0,
    rule: str = 'metropolis',
    schedule: str = 'checkerboard',
    neighborhood: BaseNeighborhood | None = None,
    values: Sequence[float] = (),
    width: float = 1.0,
    cooling: Callable[[Expression], object] | None = None,
) -> None:
    super().__init__(label)
    self.target = target
    self.state = state
    self.objective = objective
    self.move = move
    self.intensity = intensity
    self.rule = rule
    self.schedule = schedule
    self.neighborhood = neighborhood
    self.values = values
    self.width = width
    self.cooling = cooling

SatisfiabilityInteraction(label, *, target, state, constraints, reveal=1, solver='cadical153', phase=None)

Bases: Interaction

Reveal constraints on the Boolean states of a group step by step, and decide them with one incremental solver.

Every step adds the next reveal constraints to one PySAT solver, which keeps what it learned, and sets the states of the members to a satisfying assignment when one exists. The interaction reports satisfiable, 1 or 0, and revealed, the number of active constraints, every step. A state that is not Boolean is reported by the compiler with the code law-domain.

Parameters:

Name	Type	Description	Default
label	str	The label of the interaction.	required
target	Annotated[str, RoleSpec()]	The path of the group.	required
state	str	The name of the Boolean state of the members.	required
constraints	Sequence[ClauseSpec | CardinalitySpec]	The constraints, in the order they are revealed.	required
reveal	int	The number of constraints revealed per step.	1
solver	str	The name of the PySAT solver.	'cadical153'
phase	str | None	The name of the phase of the step the rule runs in, or None for the default phase.	None

Raises:

Type	Description
ValidationError	A constraint is not a spec or the number revealed per step is not positive, raised by check.

Examples:

>>> from skcomplex.core.interactions import SatisfiabilityInteraction
>>> from skcomplex.specs import ClauseSpec
>>> interaction = SatisfiabilityInteraction('sat', target='bits', state='on', constraints=[ClauseSpec((1, -2))])
>>> interaction.check().reveal
1

Source code in src/skcomplex/core/interactions/_satisfiability.py

def __init__(
    self,
    label: str,
    *,
    target: Annotated[str, RoleSpec()],
    state: str,
    constraints: Sequence[ClauseSpec | CardinalitySpec],
    reveal: int = 1,
    solver: str = 'cadical153',
    phase: str | None = None,
) -> None:
    super().__init__(label)
    self.target = target
    self.state = state
    self.constraints = constraints
    self.reveal = reveal
    self.solver = solver
    self.phase = phase

SignAdjustmentInteraction(label, *, target, state, signal, step=0.1, mode='multiplicative', lower=None, upper=None, form='symbolic', phase=None, schedule='synchronous')

Bases: Interaction

Step of a state of discrete time by the sign of a signal, kept within bounds.

In the mode 'multiplicative' the next value of the state x is x (1 + step sign(signal)), and in the mode 'additive' it is x + step sign(signal), so the state rises while the signal is positive, falls while it is negative, and stays while it is zero, by trial and error rather than by the size of the signal. The next value is then limited to [lower, upper].

In the 'symbolic' form, the signal, the step, and the bounds are numbers or functions of the namespace, where namespace.target holds the states and parameters of the target. In the 'agent' form, the target is an agent group and they are numbers or functions f(agent, model) that return numbers. The agents step at once from the states at the start of the rule under the schedule 'synchronous', one after the other in their order or in a new random order every step under 'sequential' or 'random-sequential', each reading the states the agents before it set, or each in its own activation under 'member', together with the rules of its element and the other member interactions, in the order the system lists them.

Parameters:

Name	Type	Description	Default
label	str	The label of the interaction, unique among the parts of its system.	required
target	Annotated[str, RoleSpec()]	The path of the element or group whose state steps.	required
state	str	The name of the state of the target.	required
signal	Callable[..., object]	The function of the signal whose sign sets the direction of the step.	required
step	Value	The size of the step, relative in the mode 'multiplicative', a number or a function.	0.1
mode	str	'multiplicative' or 'additive'.	'multiplicative'
lower	Value | None	The lowest next value, a number or a function, or None for none.	None
upper	Value | None	The highest next value, a number or a function, or None for none.	None
form	str	'symbolic' for an update law of the state of the target, or 'agent' for a rule over an agent group.	'symbolic'
phase	str | None	The phase of the step the agent form runs in, or None for the default phase.	None
schedule	str	The order of the agents in the agent form, 'synchronous', 'sequential', 'random-sequential', or 'member'.	'synchronous'

Raises:

Type	Description
ValidationError	The mode, the form, or the schedule is unknown, the symbolic form is not synchronous, or the signal is not callable, raised by check.

Examples:

>>> from typing import Annotated
>>> from skcomplex.core.elements import Element
>>> from skcomplex.core.interactions import SignAdjustmentInteraction
>>> from skcomplex.core.spaces import DiscreteTime
>>> from skcomplex.core.system import System
>>> from skcomplex.engine.runtime import Simulation
>>> from skcomplex.specs import StateSpec
>>> class QuoteElement(Element):
...     def __init__(self, label, x: Annotated[float, StateSpec()] = 1.0):
...         super().__init__(label)
...         self.x = x
>>> search = SignAdjustmentInteraction(
...     'search', target='quote', state='x', signal=lambda s: 2 - s.target.x, step=0.5, upper=2.0
... )
>>> system = System(elements=[QuoteElement('quote')], interactions=[search], time=DiscreteTime())
>>> Simulation(system).check().prepare().run(until=3).trajectory_.values('quote.x').tolist()
[1.0, 1.5, 2.0, 2.0]

Source code in src/skcomplex/core/interactions/_sign_adjustment.py

def __init__(
    self,
    label: str,
    *,
    target: Annotated[str, RoleSpec()],
    state: str,
    signal: Callable[..., object],
    step: Value = 0.1,
    mode: str = 'multiplicative',
    lower: Value | None = None,
    upper: Value | None = None,
    form: str = 'symbolic',
    phase: str | None = None,
    schedule: str = 'synchronous',
) -> None:
    super().__init__(label)
    self.target = target
    self.state = state
    self.signal = signal
    self.step = step
    self.mode = mode
    self.lower = lower
    self.upper = upper
    self.form = form
    self.phase = phase
    self.schedule = schedule

adjust(model)

Step the state of every agent of the target, by the schedule.

Parameters:

Name	Type	Description	Default
model	ModelViewProtocol	The model view of the interaction.	required

Source code in src/skcomplex/core/interactions/_sign_adjustment.py

def adjust(self, model: ModelViewProtocol) -> None:
    """Step the state of every agent of the target, by the schedule.

    Args:
        model:
            The model view of the interaction.
    """
    self._update_agents(model, self.schedule, self._get_value, self._set_value)

TopologyInteraction(label, *, target=None, action='add-edge', rate=None, probability=None, condition=None, toward=None, edges=1, keep='source', values=None, layer=None, hyperedge=None)

Bases: NeighborInteraction

Change of the topology of a network: nodes and edges that are added or removed, and edges that are rewired.

The action runs for every node of the group a role binds, 'add-node', 'remove-node', 'replace-node', which removes the node and adds a new one, and 'add-edge', or for every edge of its network, 'remove-edge' and 'rewire'. Without a target, 'add-node' runs once per step, adding one node from outside the network to the one network whose groups are variable, as in growing networks. It fires at a rate, with a probability per step, or with probability one, and only where the condition holds. The rate, the probability, and the condition are numbers or functions of the namespace of an edge, whose ends are namespace.source and namespace.target, and an action of the nodes reads the node as the source. A new node joins edges other nodes, a node adding edges joins edges other nodes, and a rewiring keeps the end keep of its edge. New ends are drawn with probability proportional to toward(node), a function of the namespace of a candidate node, or uniformly. A new node or edge takes the values, each a value or a function of the random number generator that draws it. The groups on the nodes and the edges of the network are variable. With hyperedge, a new node instead forms a hyperedge of that size with hyperedge - 1 existing nodes and joins each of them by an edge: the first two are the ends of an edge drawn uniformly, so a size of three attaches a triangle along an edge, and further members are drawn by toward. The new hyperedge is a new member of the variable group on the hyperedges of the network.

Parameters:

Name	Type	Description	Default
label	str	The label of the interaction, unique among the parts of its system.	required
target	Annotated[str | None, RoleSpec(optional=True)]	The path of the group on the nodes of the network, or None for an addition of one node per step.	None
action	str	The change, 'add-node', 'remove-node', 'replace-node', 'add-edge', 'remove-edge', or 'rewire'.	'add-edge'
rate	Chance | None	The rate of the change, a number or a function, or None for a probability per step.	None
probability	Chance | None	The probability of the change per step, a number or a function, or None for probability one.	None
condition	Callable[[SimpleNamespace], object] | None	A function of the namespace that returns where the change may fire, or None everywhere.	None
toward	Callable[[SimpleNamespace], object] | None	A function of the namespace of a candidate node that returns its weight as a new end, or None for equal weights.	None
edges	int	The number of edges a new node or a node adding edges joins.	1
keep	str	The end of an edge a rewiring keeps, 'source' or 'target'.	'source'
values	Mapping[str, object] | None	The initial states of a new node or edge by name, each a value or a function of the random number generator that draws it, which default to the defaults of the element type.	None
layer	str | None	The layer of the network whose edges change, or None for the graph of the network.	None
hyperedge	int | None	The size of the hyperedge every new node forms with existing nodes, or None for none.	None

Raises:

Type	Description
ValidationError	The action or the kept end is unknown, both a rate and a probability are given, the number of edges is not a whole number, an action other than 'add-node' has no target, or a hyperedge is given to an action other than 'add-node' or has fewer than two members, raised by check.

Examples:

>>> import networkx as nx
>>> from skcomplex.core.elements import Element, GroupElement
>>> from skcomplex.core.interactions import TopologyInteraction
>>> from skcomplex.core.spaces import NetworkSpace
>>> from skcomplex.core.system import System
>>> from skcomplex.engine.runtime import Simulation
>>> nodes = GroupElement('nodes', element=Element, on='sites', variable=True, capacity=20)
>>> links = GroupElement('links', element=Element, on='edges', variable=True, capacity=20)
>>> expiry = TopologyInteraction('expiry', target='nodes', action='remove-edge')
>>> system = System(elements=[nodes, links], interactions=[expiry], space=NetworkSpace(nx.cycle_graph(5)))
>>> trajectory = Simulation(system, random_state=0).check().prepare().run(until=1).trajectory_
>>> len(trajectory.network(0).sources), len(trajectory.network(1).sources)
(5, 0)

A network that grows by triangles from one triangle:

>>> import xgi
>>> from skcomplex.core.spaces import DiscreteTime
>>> triangles = GroupElement('triangles', element=Element, on='hyperedges', variable=True, capacity=8)
>>> growth = TopologyInteraction('growth', action='add-node', hyperedge=3)
>>> space = NetworkSpace(nx.complete_graph(3), hyperedges=[[0, 1, 2]])
>>> system = System(
...     elements=[nodes, links, triangles], interactions=[growth], space=space, time=DiscreteTime()
... )
>>> trajectory = Simulation(system, random_state=0).check().prepare().run(until=5).trajectory_
>>> trajectory.network(-1).hyperedges.shape
(6, 3)

Source code in src/skcomplex/core/interactions/_topology.py

def __init__(
    self,
    label: str,
    *,
    target: Annotated[str | None, RoleSpec(optional=True)] = None,
    action: str = 'add-edge',
    rate: Chance | None = None,
    probability: Chance | None = None,
    condition: Callable[[SimpleNamespace], object] | None = None,
    toward: Callable[[SimpleNamespace], object] | None = None,
    edges: int = 1,
    keep: str = 'source',
    values: Mapping[str, object] | None = None,
    layer: str | None = None,
    hyperedge: int | None = None,
) -> None:
    super().__init__(label, layer=layer)
    self.target = target
    self.action = action
    self.rate = rate
    self.probability = probability
    self.condition = condition
    self.toward = toward
    self.edges = edges
    self.keep = keep
    self.values = values
    self.hyperedge = hyperedge

TrainingInteraction(label, *, model, trained, loss, data, optimizer='adam', learning_rate=0.001, momentum=0.9, beta1=0.9, beta2=0.999, batch_size=32, truncation=None, metrics=())

Bases: Interaction

Training of the adaptive parameters of a nested system by gradient steps on a loss over unrolled runs.

The role model binds the path of a nested system, which the training unrolls over the sequences of a dataset from its initial states, inside one step of the discrete time of the system that holds the interaction. That step is one epoch of shuffled batches, each followed by one update of the optimizer, and the interaction returns the GradientStepLaw that takes it. The hyperparameters of the optimizer are parameters of the interaction, so a schedule or Simulation.update_parameters changes them between epochs. 'sgd' reads the learning rate, 'momentum' also the momentum, 'rmsprop' the learning rate and beta2, the decay of its mean squared gradient, and 'adam' the learning rate, beta1, and beta2. It differs from GradientInteraction, which moves states down the gradient of a potential in the system's own time.

Parameters:

Name	Type	Description	Default
label	str	The label of the interaction, unique among the parts of its system.	required
model	Annotated[str, RoleSpec()]	The path of the nested system whose parameters the interaction trains.	required
trained	Sequence[str]	The dotted paths of the adaptive parameters of the nested system to train, such as readout.W.	required
loss	ObservableSpec	The observable of the nested system whose weighted mean over the steps of a batch is minimized.	required
data	DatasetSpec	The sequences of the parameters of the nested system that set its inputs and targets at every step.	required
optimizer	str	The name of the optimizer, one of OPTIMIZERS.	'adam'
learning_rate	Annotated[float, ParameterSpec(domain=closed_open(0, oo))]	The learning rate of the optimizer.	0.001
momentum	Annotated[float, ParameterSpec(domain=closed_open(0, 1))]	The momentum of 'momentum'.	0.9
beta1	Annotated[float, ParameterSpec(domain=closed_open(0, 1))]	The decay of the mean gradient of 'adam'.	0.9
beta2	Annotated[float, ParameterSpec(domain=closed_open(0, 1))]	The decay of the mean squared gradient of 'adam' and 'rmsprop'.	0.999
batch_size	int	The number of sequences of a batch.	32
truncation	int | None	The number of inner steps the gradient flows back through the inner states, or None for the whole run. A truncation of 1 holds the previous inner states as constant inputs, as Elman's original training does.	None
metrics	Sequence[ObservableSpec]	The observables of the nested system recorded at the end of every epoch.	()

Raises:

Type	Description
ValidationError	The optimizer is unknown, the batch size or the truncation is not a positive integer, no parameter is trained, or the loss, the data, or a metric has the wrong type or label.

Examples:

>>> import numpy as np
>>> from skcomplex.core.interactions import TrainingInteraction
>>> from skcomplex.specs import DatasetSpec, ObservableSpec
>>> data = DatasetSpec({'inputs.x': np.zeros((4, 3, 1))})
>>> training = TrainingInteraction(
...     'training', model='net', trained=['readout.W'], loss=ObservableSpec('error', 0.0), data=data
... )
>>> training.check().optimizer
'adam'

Source code in src/skcomplex/core/interactions/_training.py

def __init__(
    self,
    label: str,
    *,
    model: Annotated[str, RoleSpec()],
    trained: Sequence[str],
    loss: ObservableSpec,
    data: DatasetSpec,
    optimizer: str = 'adam',
    learning_rate: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 1e-3,
    momentum: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, 1))] = 0.9,
    beta1: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, 1))] = 0.9,
    beta2: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, 1))] = 0.999,
    batch_size: int = 32,
    truncation: int | None = None,
    metrics: Sequence[ObservableSpec] = (),
) -> None:
    super().__init__(label)
    self.model = model
    self.trained = trained
    self.loss = loss
    self.data = data
    self.optimizer = optimizer
    self.learning_rate = learning_rate
    self.momentum = momentum
    self.beta1 = beta1
    self.beta2 = beta2
    self.batch_size = batch_size
    self.truncation = truncation
    self.metrics = metrics

TransferInteraction(label, *, source, target, state, into=None, amount, flow='', pairing='members', legs=(), fallback=None, guard=None, rate=None, reverse_rate=None, offset=0, boundary='periodic', form='symbolic', schedule='random-sequential', actor='source', partners=None, remove=None, phase=None, settlement=None)

Bases: Interaction

Transfer of amounts of states from the members of a source role to the members of a target role.

A leg moves an amount out of a state of the source member, or out of Source(), into a state of the target member, or into Sink(). The transfer has a first leg, state into into by amount, and any further legs, each (state, into, amount) or (state, into, amount, flow). The amounts of all legs are read before any of them moves, so a leg sees the states as they were before the transfer. A guard skips the transfer where it does not hold.

The symbolic form reports the amount of every leg as a flow. A leg named by its flow is reported under that name: under the element or group, as <path>.<flow>, when the source and the target are one element or group, since the flow then moves a quantity within it like a flow of its own, and under the interaction, as <label>.<flow>, otherwise, as it is for a subclass that names its roles otherwise. A leg without a name is reported under the interaction as <label>.flow_<index>.

The form chooses how the transfer lowers:

• 'symbolic' lowers to TransferLaws between the source and the target of one element, or member i of both groups. The amounts, the guard, and the rates are numbers or functions of the namespace, and the roles are read as namespace.source and namespace.target. With a rate, the amount is the size of a jump at the rate, and with a reverse_rate jumps also move the amounts back from the target to the source at that rate, so a leg from Source() returns its amount to the environment. With an offset, the source is a group and every member transfers along an edge of the group order, to the member offset places along it, which namespace.target reads. Past the ends of a 'fixed' boundary no member is there, and the edge moves nothing. A draw in an amount is one draw per edge, which both of its ends read, so the amounts of a step move in parallel. A leg end '<role>.<state>' names the state of the element or group another role binds, so one transfer moves amounts between several holders, such as the payments of a circuit of sectors, and a subclass that declares parameters reads them in its amounts. A subclass that names its roles otherwise returns their names from _get_role, or the namespaces of the source and the target of every pair from _get_holders.
• 'agent' lowers to an agent rule, whose amounts and guard are numbers or functions of the source agent, the target agent, and the model view. The pairing 'members' pairs member i of both groups, 'random' pairs every source member with a random target member, 'pairs' draws disjoint random pairs, 'neighbors' pairs every source member with the target members in the cells or nodes around it, 'neighbor-pairs' draws disjoint random pairs of neighbours, every source member in random order taking a random unpaired target member around it, and 'cell' pairs the source and target members that stand in one cell or node. With a fallback group, the members of that group are the targets of a step in which no target member lives. With the pairing 'site' the target role binds a field, and every source member moves its amounts into the site of the field at its cell or node, so a negative amount takes from the site. The functions of this pairing take the cell of the source agent in place of a target agent. With remove, the source or the target member of every pair that transfers is removed, as an agent eaten whole. The named pairings other than 'members' and 'site' are shorthands of a PartnerSpec, which pairs every source member with partners within a radius, by score, or with a fallback. A settlement receives both partners and the amounts of the legs, read once, and changes the states in place of the legs, so one draw credits both ends. A subclass that names its roles otherwise returns their names from _get_role.

Under the schedule 'random-sequential' the pairs act in random order. Under 'member' every member of the actor role acts in its own activation, together with the rules of its element and the other member interactions, in the order the system lists them: it transfers with its partners, the members of the other role that the pairing 'random', 'neighbors', or 'cell' gives it, or with partners of them drawn at random, and with itself under the pairing 'members' within one group. A leg end of the agent form may name '<reference>.<state>', the state of the member that a reference state of the agent names, such as 'bank.deposits' for the deposits of a customer's bank, declared StateSpec(reference='banks').

Parameters:

Name	Type	Description	Default
label	str	The label of the interaction, unique among the parts of its system.	required
source	Annotated[str, RoleSpec()]	The path of the element or group the amounts leave.	required
target	Annotated[str, RoleSpec()]	The path of the element or group the amounts enter, the source again for a transfer within members.	required
state	str | Source	The name of the state of the source the first leg takes from, or Source() for the environment.	required
into	str | Sink | None	The name of the state of the target the first leg adds to, Sink() for the environment, or None for state.	None
amount	Amount	The amount of the first leg, a number or a function.	required
flow	str	The name the symbolic form reports the first leg under, or the empty string for flow_0.	''
pairing	str | Pairing	How members are paired, one of 'members', 'random', 'pairs', 'neighbors', 'neighbor-pairs', 'cell', or 'site', the agent form alone taking the others than 'members', or, in the agent form, a PartnerSpec, or a function pairing(sources, targets, rng, capacity) of the source and target agents that returns their pairs, whose capacity is partners, or 1 without it. Under the schedule 'member' the spec or the function receives the actor alone as its first sequence and the actor's partners are the second items of its pairs.	'members'
legs	Sequence[Leg]	Further legs, each a tuple of a state or Source(), a state, Sink(), or None, an amount, and optionally the name of its flow.	()
fallback	Annotated[str | None, RoleSpec()]	The path of the group whose members the agent form pairs with instead of the target members in a step in which no target member lives, or None for no fallback.	None
guard	Callable[..., object] | None	A function that returns whether a pair transfers, a condition of the namespace or a boolean of the agents, or None for every pair.	None
rate	Amount | None	The rate of jumps of the symbolic form, a number or a function of the namespace, or None for a continuous flow.	None
reverse_rate	Amount | None	The rate of the jumps of the symbolic form that move the amounts back from the target to the source, a number or a function of the namespace, or None for no reverse jumps.	None
offset	int	The number of places along the group order from every source member to its target member, or 0 to pair the members of the roles.	0
boundary	str	Past the ends of the group order, 'periodic' closes a ring and 'fixed' holds no member.	'periodic'
form	str	The lowering, 'symbolic' to transfer laws or 'agent' to an agent rule.	'symbolic'
schedule	str	The order of the agent form, 'random-sequential' over the pairs, or 'member' over the actors.	'random-sequential'
actor	str	The role whose members act under the schedule 'member', 'source' or 'target'.	'source'
partners	int | None	The number of partners an actor draws at random under the schedule 'member', or None for all of them.	None
remove	str | None	The member of a pair that transfers that is removed, 'source', 'target', or None.	None
phase	str | None	The phase the rule of the agent form runs in, or None for the default phase.	None
settlement	Settlement | None	A function settlement(source, target, amounts, model) that settles a pair of agents from the amounts of the legs, read once, in place of moving them, or None to move them.	None

Raises:

Type

Description

ValidationError

The form, the pairing, the schedule, the actor, the removal, or the boundary is unknown, a leg is malformed, two legs share a flow name, the symbolic form has an agent pairing, a fallback, or a settlement, the agent form a rate, an offset, or a flow name, a reverse rate has no rate, an offset is not a whole number or pairs two groups, a leg end of the symbolic form names no role, the settlement cannot be called or pairs with a site, or the schedule 'member' has a pairing it cannot draw, raised by check.

Examples:

>>> from skcomplex.core.elements import Element
>>> from skcomplex.core.interactions import TransferInteraction
>>> from skcomplex.core.system import System
>>> from skcomplex.engine.runtime import Simulation
>>> from skcomplex.specs import StateSpec
>>> from typing import Annotated
>>> class TankElement(Element):
...     def __init__(self, label, visual=None, *, level: Annotated[float, StateSpec()] = 0.0):
...         super().__init__(label, visual=visual)
...         self.level = level
>>> flow = TransferInteraction(
...     'flow', source='upper', target='lower', state='level', amount=lambda s: s.source.level / 2
... )
>>> system = System(elements=[TankElement('upper', level=4.0), TankElement('lower')], interactions=[flow])
>>> trajectory = Simulation(system).check().prepare().run(until=1.0).trajectory_
>>> round(float(trajectory.values('upper.level')[-1] + trajectory.values('lower.level')[-1]), 6)
4.0

Source code in src/skcomplex/core/interactions/_transfer.py

def __init__(
    self,
    label: str,
    *,
    source: Annotated[str, RoleSpec()],
    target: Annotated[str, RoleSpec()],
    state: str | Source,
    into: str | Sink | None = None,
    amount: Amount,
    flow: str = '',
    pairing: str | Pairing = 'members',
    legs: Sequence[Leg] = (),
    fallback: Annotated[str | None, RoleSpec()] = None,
    guard: Callable[..., object] | None = None,
    rate: Amount | None = None,
    reverse_rate: Amount | None = None,
    offset: int = 0,
    boundary: str = 'periodic',
    form: str = 'symbolic',
    schedule: str = 'random-sequential',
    actor: str = 'source',
    partners: int | None = None,
    remove: str | None = None,
    phase: str | None = None,
    settlement: Settlement | None = None,
) -> None:
    super().__init__(label)
    self.source = source
    self.target = target
    self.state = state
    self.into = into
    self.amount = amount
    self.flow = flow
    self.pairing = pairing
    self.legs = legs
    self.fallback = fallback
    self.guard = guard
    self.rate = rate
    self.reverse_rate = reverse_rate
    self.offset = offset
    self.boundary = boundary
    self.form = form
    self.schedule = schedule
    self.actor = actor
    self.partners = partners
    self.remove = remove
    self.phase = phase
    self.settlement = settlement

transfer(source, target, model)

Move the amounts of every leg from one source agent to one target agent, when the guard holds.

Parameters:

Name	Type	Description	Default
source	AgentViewProtocol	The agent the amounts leave.	required
target	AgentViewProtocol	The agent the amounts enter.	required
model	ModelViewProtocol	The model view of the rule.	required

Returns:

Name	Type	Description
transferred	bool	Whether the guard held, so the amounts moved.

Raises:

Type	Description
ValidationError	An amount function returns something other than a number.
SimulationError	A leg end names the state of a referenced member, and the reference state names no member.

Source code in src/skcomplex/core/interactions/_transfer.py

def transfer(self, source: AgentViewProtocol, target: AgentViewProtocol, model: ModelViewProtocol) -> bool:
    """Move the amounts of every leg from one source agent to one target agent, when the guard holds.

    Args:
        source:
            The agent the amounts leave.

        target:
            The agent the amounts enter.

        model:
            The model view of the rule.

    Returns:
        transferred:
            Whether the guard held, so the amounts moved.

    Raises:
        ValidationError:
            An amount function returns something other than a number.

        SimulationError:
            A leg end names the state of a referenced member, and the reference state names no member.
    """
    if self.guard is not None and not self.guard(source, target, model):
        return False
    legs = self._get_legs()
    amounts = [self._evaluate(amount, source, target, model) for _, _, amount, _ in legs]
    if self.settlement is not None:
        self.settlement(source, target, amounts, model)
        return True
    ends = [
        (
            self._get_end(source, state) if isinstance(state, str) else None,
            self._get_end(target, into) if isinstance(into, str) else None,
        )
        for state, into, _, _ in legs
    ]
    for (taken, given), amount in zip(ends, amounts, strict=True):
        if taken is not None:
            holder, name = taken
            setattr(holder, name, getattr(holder, name) - amount)
        if given is not None:
            holder, name = given
            setattr(holder, name, getattr(holder, name) + amount)
    return True

TransitionInteraction(label, *, target, state='state', source=0.0, destination=1.0, rate=None, probability=None, condition=None, matrix=None, emission=None, observed=None, over='member', neighborhood=None, after=None, schedule='synchronous', form='symbolic', phase=None)

Bases: Interaction

Move of a discrete state of the target from one value to another, or between all its values by a matrix.

A member whose state holds source moves to destination at a rate, as a jump in continuous time, with a probability in every step of discrete time, or in the step in which a condition holds. With over='neighbors' the rate and the probability are those of one neighbour, written with namespace.target.other for the neighbour: the rates of the neighbours add up, and their probabilities combine as 1 - prod(1 - p), the probability that at least one of them moves the member. With a matrix instead, the state takes the values 0 to n - 1 and steps by the rows of a transition matrix in discrete time, the row of its current value. An emission matrix then draws the observed state of the target from the row that the next value of the state selects, as a hidden Markov source. With a residence time after, a member moves by a probability or a condition only once it has held the source value for after steps, at the earliest at the end of the step that completes them.

In the 'symbolic' form, every function reads namespace.target, the states and parameters of the target, and the time namespace.t. In the 'agent' form, the target is an agent group and every function takes the agent and the model view, f(agent, model), and returns a number, a truth value, or the rows of a matrix, and a rate r moves an agent with probability 1 - exp(-r) in a step. With over='cell', every member acts on the other members of its cell: it moves each one that holds the source value with the probability, the rate, or the condition f(agent, other, model) of that member agent and the acting member other, each by its own draw.

The schedule of the agent form orders the members. Under 'synchronous' every member decides from the states at the start of the step, under 'sequential' the members act one after the other in their order, and under 'random-sequential' in a new random order every step, each reading the values the members before it set. Under 'member' every member acts in its own activation, together with the rules of its element and the other member interactions, in the order the system lists them, so one rule of a member runs its movement, transfers, transitions, and life cycle.

Parameters:

Name	Type	Description	Default
label	str	The label of the interaction, unique among the parts of its system.	required
target	Annotated[str, RoleSpec()]	The path of the element or group whose state moves.	required
state	str	The name of the discrete state of the target.	'state'
source	SourceValues	The value the state moves from, a collection of such values, or a function of the value of the state that returns whether the state moves from it, such as lambda value: value >= 0, a condition in the symbolic form. A state that moves from several values moves to the destination from each.	0.0
destination	float	The value the state moves to.	1.0
rate	Callable[..., SupportsFloat] | None	The function of the rate of the move, or None.	None
probability	Callable[..., SupportsFloat] | None	The function of the probability of the move in a step, or None.	None
condition	Callable[..., object] | None	The function of the condition under which the state moves in a step, or None.	None
matrix	Callable[..., Sequence[Sequence[object]]] | None	The function of the transition matrix, one row and one column per value of the state, or None.	None
emission	Callable[..., Sequence[Sequence[object]]] | None	With a matrix, the function of the emission matrix, one row per value of the state and one column per value of the observed state, or None.	None
observed	str | None	The name of the state of the target that the emission matrix draws, or None.	None
over	str	'member' for a rate or a probability of the member, 'neighbors' for that of one neighbour, or 'cell' for the agent form whose members act on the members of their cell.	'member'
neighborhood	BaseNeighborhood | None	The neighbourhood of over='neighbors', or None for the neighbours of the network or the lattice.	None
after	float | None	The number of steps a member holds the source value before it can move, or None.	None
schedule	str	The order of the members of the agent form, 'synchronous', 'sequential', 'random-sequential', or 'member'.	'synchronous'
form	str	'symbolic' for laws of the states of the target, or 'agent' for a rule over an agent group.	'symbolic'
phase	str | None	The phase of the step the agent form runs in, or None for the default phase.	None

Raises:

Type	Description
ValidationError	The transition has not exactly one kind, the source and the destination are equal, an emission has no matrix or no observed state, over is unknown or goes with a form or a kind it cannot take, the residence time is negative or goes with a rate, a matrix, or members acting on their cell, the form or the schedule is unknown, the symbolic form has a schedule, or a function is not callable.

Examples:

>>> from skcomplex.core.interactions import TransitionInteraction
>>> recovery = TransitionInteraction(
...     'recovery', target='hosts', state='state', source=1, destination=0, rate=lambda s: 0.5
... )
>>> recovery.check().over
'member'
>>> agents = TransitionInteraction(
...     'recovery', target='hosts', state='state', source=1, destination=2, probability=lambda a, m: 0.1,
...     after=3, form='agent'
... )
>>> type(agents.check().prepare().run(SimpleNamespace()).laws_[0]).__name__
'RuleLaw'
>>> separation = TransitionInteraction(
...     'separation', target='workers', state='employer', source=lambda value: value >= 0, destination=-1,
...     probability=lambda a, m: 0.05, form='agent'
... )
>>> separation.check().destination
-1

Source code in src/skcomplex/core/interactions/_transition.py

def __init__(
    self,
    label: str,
    *,
    target: Annotated[str, RoleSpec()],
    state: str = 'state',
    source: SourceValues = 0.0,
    destination: float = 1.0,
    rate: Callable[..., SupportsFloat] | None = None,
    probability: Callable[..., SupportsFloat] | None = None,
    condition: Callable[..., object] | None = None,
    matrix: Callable[..., Sequence[Sequence[object]]] | None = None,
    emission: Callable[..., Sequence[Sequence[object]]] | None = None,
    observed: str | None = None,
    over: str = 'member',
    neighborhood: BaseNeighborhood | None = None,
    after: float | None = None,
    schedule: str = 'synchronous',
    form: str = 'symbolic',
    phase: str | None = None,
) -> None:
    super().__init__(label)
    self.target = target
    self.state = state
    self.source = source
    self.destination = destination
    self.rate = rate
    self.probability = probability
    self.condition = condition
    self.matrix = matrix
    self.emission = emission
    self.observed = observed
    self.over = over
    self.neighborhood = neighborhood
    self.after = after
    self.schedule = schedule
    self.form = form
    self.phase = phase

move(model)

Move the state of every agent of the target that moves in this step, by the schedule.

Parameters:

Name	Type	Description	Default
model	ModelViewProtocol	The model view of the interaction.	required

Source code in src/skcomplex/core/interactions/_transition.py

def move(self, model: ModelViewProtocol) -> None:
    """Move the state of every agent of the target that moves in this step, by the schedule.

    Args:
        model:
            The model view of the interaction.
    """
    agents = list(model.agents(self._get_role('target')))
    if self.schedule == 'synchronous':
        moves = [move for agent in agents for move in self._get_moves(agent, model)]
        self._apply(moves, model)
        return
    order = model.rng.permutation(len(agents)) if self.schedule == 'random-sequential' else range(len(agents))
    for index in order:
        self._move_member(agents[int(index)], model)