Laws

Define laws, the rules by which states change.

BaseLaw(label)

Bases: BasePart[str, object]

Base of every law, a rule by which states change, which an element, an interaction, or a system returns.

A law takes the states it changes and the expressions it reads as arguments of its own __init__, which stores each under its own name, as in scikit-learn. It returns the states it changes from _get_targets, what it requires of a system, such as continuous or discrete time, from _get_requirements, and its value at given values of the symbols it reads from _run, which sets it as value_.

Parameters:

Name	Type	Description	Default
label	str	The name of the law, or the empty string for a law without one.	required

Attributes:

Name	Type	Description
targets_		The states the law changes.
reads_		The symbols the law reads.
owner_		The path of the element or interaction that returns the law, or None.
value_	object	The value of the law at the values of the symbols of its last run.

Examples:

>>> from sage.all__sagemath_symbolics import var
>>> from skcomplex.core.laws import RateLaw
>>> x, k = var('x'), var('k')
>>> law = RateLaw(x, -k * x).check().prepare('decay')
>>> law.targets_, sorted(map(str, law.reads_)), law.owner_
((x,), ['k', 'x'], 'decay')
>>> law.run({'x': 2.0, 'k': 0.5}).value_
-1.0

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

BilinearLaw(target, first, second, rate, outcomes=None, *, label='')

Bases: BaseLaw

Gain or loss of a target density at a signed rate of the product of two densities.

The amount of the first density f in class h that meets the amount of the second density g in class k adds to class i of the target at the rate r[h, k] with probability B[h, k, i], so the law adds sum_hk r[h, k] B[h, k, i] f_h g_k to the rate of the target at i. A positive rate gives a gain that leaves the first density in place, and a negative rate with the first density as the target gives a loss.

Parameters:

Name	Type	Description	Default
target	Expression	The symbol of the density the law changes.	required
first	Expression	The symbol of the first density.	required
second	Expression	The symbol of the second density.	required
rate	object	The signed rate, a number or a parameter expression, or an array r[h, k] of them.	required
outcomes	object	The array B[h, k, i] of the probabilities of the class in the target, or None for the class of the first density, which needs a target of its size.	None
label	str	The name of the law, or the empty string for a law without one.	''

Examples:

>>> from sage.all__sagemath_symbolics import var
>>> from skcomplex.core.laws import BilinearLaw
>>> f, g = var('f'), var('g')
>>> BilinearLaw(f, f, g, -0.5).check().prepare().run({}).value_
(-0.5, None)

Source code in src/skcomplex/core/laws/_bilinear.py

def __init__(
    self,
    target: Expression,
    first: Expression,
    second: Expression,
    rate: object,
    outcomes: object = None,
    *,
    label: str = '',
) -> None:
    super().__init__(label)
    self.target = target
    self.first = first
    self.second = second
    self.rate = rate
    self.outcomes = outcomes

BoundaryLaw(target, value, *, label='')

Bases: BaseLaw

Value a state takes after every solver step, such as a reflection at a wall or a wrap around a periodic box.

Boundaries leave the form of a system as is, so they combine with every other law in any time.

Parameters:

Name	Type	Description	Default
target	Target	The symbol of a scalar state, or the column of symbols of a vector state.	required
value	object	A symbolic expression, a Call, or a callable f(t, values) that gives the new value.	required
label	str	The name of the law, or the empty string for a law without one.	''

Examples:

>>> from sage.all__sagemath_symbolics import floor, var
>>> from skcomplex.core.laws import BoundaryLaw
>>> x = var('x')
>>> BoundaryLaw(x, x - floor(x)).check().prepare().run({'x': 2.25}).value_
0.25

Source code in src/skcomplex/core/laws/_boundary.py

def __init__(self, target: Target, value: object, *, label: str = '') -> None:
    super().__init__(label)
    self.target = target
    self.value = value

CascadeLaw(drive, relax, stable, max_iterations=100000, activity=None, *, label='')

Bases: BaseLaw

Cascade of a lattice: drive laws once per step, then the laws that relax it until it is stable.

Every cascade that changes the lattice is recorded as an event with its size, the number of site updates, its duration, the number of iterations, and its area, the number of distinct sites updated. A site update is a site that changes in an iteration, or a site that activity marks as active before it, such as a toppling site.

Parameters:

Name	Type	Description	Default
drive	Sequence[BaseLaw]	The laws applied once per step, local rules or drives.	required
relax	Sequence[BaseLaw]	The laws applied in turn until the lattice is stable.	required
stable	Callable[..., bool]	A function of the arrays of the targets of the laws that relax the lattice, in lattice shape, that returns whether the lattice is stable.	required
max_iterations	int	The largest number of iterations in one step.	100000
activity	Callable[..., object] | None	A function of the arrays of the targets that returns a boolean array of the active sites of an iteration, or None to count the sites that change.	None
label	str	The name of the cascade events, or the empty string for the path of the element.	''

Examples:

>>> from sage.all__sagemath_symbolics import var
>>> from skcomplex.core.laws import CascadeLaw, LocalRuleLaw
>>> from skcomplex.core.spaces import VonNeumannNeighborhood
>>> height = var('height')
>>> topple = LocalRuleLaw(height, VonNeumannNeighborhood(center=True), lambda z, *n: z - 4 * (z >= 4))
>>> CascadeLaw([], [topple], lambda z: bool((z < 4).all())).check().prepare().targets_
(height,)

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

def __init__(
    self,
    drive: Sequence[BaseLaw],
    relax: Sequence[BaseLaw],
    stable: Callable[..., bool],
    max_iterations: int = 100_000,
    activity: Callable[..., object] | None = None,
    *,
    label: str = '',
) -> None:
    super().__init__(label)
    self.drive = drive
    self.relax = relax
    self.stable = stable
    self.max_iterations = max_iterations
    self.activity = activity

CrossingLaw(guard, reset=None, direction=0, terminal=False, *, label='')

Bases: BaseLaw

Event at the crossing of zero by a guard, when states take new values or the run ends.

Solvers locate the crossing within their tolerance. Crossings leave the form of a system as is, so they combine with rate laws, and the trajectory records every crossing with its time and the states before and after it.

Parameters:

Name	Type	Description	Default
guard	object	A symbolic expression of states and time whose crossing of zero triggers the event, or a function f(t, values) of time and the mapping from the dotted name of every state and parameter to its value, or CallSpec(f, relative=True) to read the names of the component. Over a group it returns one value per member. It runs in Python, and a function does not lower the order of derivatives or read delays.	required
reset	Mapping[Expression, object] | None	A mapping from state symbol to its new value, applied all at once, or None to keep the states. A value may be a function like the guard's.	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
label	str	The name of the crossing in the trajectory, or the empty string for the path of the element.	''

Examples:

>>> from sage.all__sagemath_symbolics import var
>>> from skcomplex.core.laws import CrossingLaw
>>> y, z = var('y'), var('z')
>>> law = CrossingLaw(y, {z: -0.8 * z}, direction=-1).check().prepare()
>>> law.run({'z': -2.0}).value_
{z: 1.6}

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

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

DissipatorLaw(operator, rate=1.0, *, label='')

Bases: BaseLaw

Lindblad term of a Markovian master equation, with a collapse operator and a rate.

The master equation gains rate * (L rho L^† - (L^† L rho + rho L^† L) / 2).

Parameters:

Name	Type	Description	Default
operator	object	The collapse operator L, an expression of local operators of the namespaces the law receives.	required
rate	object	The non-negative rate, a number or a Sage expression of parameters and time.	1.0
label	str	The name of the law, or the empty string for a law without one.	''

Examples:

>>> from sage.all__sagemath_symbolics import SR, var
>>> from skcomplex.core.laws import DissipatorLaw
>>> from skcomplex.core.laws.utils import local_operator
>>> minus = local_operator(SR.symbol('q'), SR.symbol('minus'))
>>> DissipatorLaw(minus, 2 * var('gamma')).check().prepare().run({'gamma': 0.05}).value_
(local_operator(q, minus), 0.1)

Source code in src/skcomplex/core/laws/_dissipator.py

def __init__(self, operator: object, rate: object = 1.0, *, label: str = '') -> None:
    super().__init__(label)
    self.operator = operator
    self.rate = rate

DriveLaw(target, function, *, label='')

Bases: BaseLaw

Change of a whole lattice state at once by a function of its array, such as adding 1 at a random site.

The function receives the state as an array of the lattice shape, keyword arguments for the parameters of the element that it names, and random, the random generator of the simulation, and returns the new array.

Parameters:

Name	Type	Description	Default
target	Expression	The symbol of a lattice state.	required
function	Callable[..., object]	The function f(values, *, random, **parameters).	required
label	str	The name of the law, or the empty string for a law without one.	''

Examples:

>>> from sage.all__sagemath_symbolics import var
>>> from skcomplex.core.laws import DriveLaw
>>> DriveLaw(var('height'), lambda values, *, random: values + 1).check().prepare().targets_
(height,)

Source code in src/skcomplex/core/laws/_drive.py

def __init__(self, target: Expression, function: Callable[..., object], *, label: str = '') -> None:
    super().__init__(label)
    self.target = target
    self.function = function

EmissionLaw(target, source, matrix, *, label='')

Bases: BaseLaw

Observation of a hidden Markov source, a state drawn from the row of an emission matrix a hidden state selects.

The emission reads the hidden state's value after the same step, so the hidden and observed sequences align.

Parameters:

Name	Type	Description	Default
target	Expression	The symbol of the observed scalar state.	required
source	Expression	The symbol of the hidden state, which an update or transition law of the same element sets.	required
matrix	Sequence[Sequence[object]]	The emission matrix, one row per value of the hidden state and one column per value of the observed state, as numbers or expressions of parameters.	required
label	str	The name of the law, or the empty string for a law without one.	''

Examples:

>>> from sage.all__sagemath_symbolics import var
>>> from skcomplex.core.laws import EmissionLaw
>>> observed, hidden = var('observed'), var('hidden')
>>> EmissionLaw(observed, hidden, [[0.7, 0.3], [0.2, 0.8]]).check().prepare().targets_
(observed,)

Source code in src/skcomplex/core/laws/_emission.py

def __init__(
    self,
    target: Expression,
    source: Expression,
    matrix: Sequence[Sequence[object]],
    *,
    label: str = '',
) -> None:
    super().__init__(label)
    self.target = target
    self.source = source
    self.matrix = matrix

EncounterLaw(candidate, field, rate, transition, test=None, *, label='')

Bases: BaseLaw

Bilinear transfer of a density between its classes at the rate of its product with a second density.

The amount of the candidate density f in class h meets the amount of the field density g in class k at the rate eta[h, k], and moves to class i with the probability A[h, k, i]. The law adds the gain and the loss

df_i/dt = sum_hk eta[h, k] A[h, k, i] f_h g_k - f_i sum_k eta[i, k] g_k,

where the loss acts on the candidate and the gain on the test density, the candidate itself by default, so the total over the classes is conserved. The candidate, the field, and the test are states of groups on the sites of lattices of classes, one member per class. A field density that no law changes is a fixed external input.

Parameters:

Name	Type	Description	Default
candidate	Expression	The symbol of the candidate density, a state of a group read member by member.	required
field	Expression | None	The symbol of the field density, or None for a linear redistribution, with one field class of value one.	required
rate	object	The encounter rate, a number or a parameter expression, or an array eta[h, k] of them.	required
transition	object	The array A[h, k, i] of transition probabilities, numbers or parameter expressions, which sum to one over i.	required
test	Expression | None	The symbol of the density the candidate moves into, or None for the candidate's own density.	None
label	str	The name of the law, or the empty string for a law without one.	''

Examples:

>>> import numpy as np
>>> from sage.all__sagemath_symbolics import var
>>> from skcomplex.core.laws import EncounterLaw
>>> f, g = var('f'), var('g')
>>> rate, transition = EncounterLaw(f, g, 1.0, np.full((2, 2, 2), 0.5)).check().prepare().run({}).value_
>>> rate, np.shape(transition)
(1.0, (2, 2, 2))

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

def __init__(
    self,
    candidate: Expression,
    field: Expression | None,
    rate: object,
    transition: object,
    test: Expression | None = None,
    *,
    label: str = '',
) -> None:
    super().__init__(label)
    self.candidate = candidate
    self.field = field
    self.rate = rate
    self.transition = transition
    self.test = test

EquationLaw(lhs, rhs=None, *, target=None, label='')

Bases: BaseLaw

Implicit relation in which the highest time derivatives of states appear, such as Newton's second law.

The states are Sage functions of time, such as function('theta')(t), and their derivatives are Sage derivatives, theta.diff(t, 2). The compiler solves all equations of a system jointly for the highest derivatives, which must appear linearly, and adds the rate laws on the same states to the solution. An equation may name the state whose equation it is, its target, and the equations of one target, from any element or interaction, add into one before the solve, so a term of it may hold no highest derivative, such as the part a coupling Lagrangian adds to the Euler-Lagrange equation of a coordinate.

Parameters:

Name	Type	Description	Default
lhs	object	The left side, or a Sage relation lhs == rhs.	required
rhs	object	The right side, or None when lhs is a relation.	None
target	object	The state whose equation this is, so that the equations of one target add, or None for an equation of its own.	None
label	str	The name of the law, or the empty string for a law without one.	''

Attributes:

Name	Type	Description
lhs_		The left side.
rhs_		The right side.

Examples:

>>> from sage.all__sagemath_symbolics import function, sin, var
>>> from skcomplex.core.laws import EquationLaw
>>> t, g, l = var('t g l')
>>> theta = function('theta')(t)
>>> EquationLaw(l * theta.diff(t, 2) == -g * sin(theta)).check().prepare().rhs_
-g*sin(theta(t))

Source code in src/skcomplex/core/laws/_equation.py

def __init__(self, lhs: object, rhs: object = None, *, target: object = None, label: str = '') -> None:
    super().__init__(label)
    self.lhs = lhs
    self.rhs = rhs
    self.target = target

ExtremalLaw(targets, over, rule, *, key=None, extremum='min', repeat=1, label='')

Bases: BaseLaw

Rule applied to the site with the smallest or largest value of a key state and to its neighbours.

Every application finds the extremal site, with ties broken by the random stream of the simulation, gathers the site and its neighbours without duplicates, the site first, and calls f(*values, random, **parameters) with one array per target and a uniform random array of the same length. Every application is recorded as an event.

Parameters:

Name	Type	Description	Default
targets	Expression | tuple[Expression, ...]	The symbol of a lattice state, or a tuple of them that the rule updates together.	required
over	BaseNeighborhood	The neighbourhood of the extremal site that the rule also updates.	required
rule	Callable[..., object]	The function f(*values, random, **parameters).	required
key	Expression | None	The lattice state whose extremum is searched, or None for the first target.	None
extremum	str	'min' for the smallest value or 'max' for the largest.	'min'
repeat	int	The number of applications per step.	1
label	str	The name of the events, or the empty string for the path of the element.	''

Examples:

>>> from sage.all__sagemath_symbolics import var
>>> from skcomplex.core.laws import ExtremalLaw
>>> from skcomplex.core.spaces import BoxNeighborhood
>>> law = ExtremalLaw(var('fitness'), BoxNeighborhood(), lambda values, *, random: random).check()
>>> law.extremum, law.repeat
('min', 1)

Source code in src/skcomplex/core/laws/_extremal.py

def __init__(
    self,
    targets: Expression | tuple[Expression, ...],
    over: BaseNeighborhood,
    rule: Callable[..., object],
    *,
    key: Expression | None = None,
    extremum: str = 'min',
    repeat: int = 1,
    label: str = '',
) -> None:
    super().__init__(label)
    self.targets = targets
    self.over = over
    self.rule = rule
    self.key = key
    self.extremum = extremum
    self.repeat = repeat

FluxLaw(target, flux, *, limiter='minmod', riemann='rusanov', speed=None, label='')

Bases: BaseLaw

Flux of a conserved state of a finite-volume field, du/dt + divergence(flux) = 0.

The flux is a symbolic expression of the field's states for one dimension, or one expression per dimension. The states of a field with fluxes form a system of conservation laws, whose wave speeds come from the eigenvalues of the flux Jacobians. Rate laws on the same states add sources.

Parameters:

Name	Type	Description	Default
target	Target	The symbol of the conserved state.	required
flux	object	The flux, an expression or one expression per dimension.	required
limiter	str | None	The slope limiter of the MUSCL reconstruction, 'minmod', 'van-leer', or None for first order.	'minmod'
riemann	str	The numerical flux, 'rusanov' or 'godunov', the exact solver for one state.	'rusanov'
speed	object	A function f(t, values) of the mapping from the name of every conserved state and parameter of the flux to its array over the cells, which returns the wave speed of the Rusanov flux, or one function per dimension. None takes the speeds from the eigenvalues of the flux Jacobians.	None
label	str	The name of the law, or the empty string for a law without one.	''

Examples:

>>> from sage.all__sagemath_symbolics import var
>>> from skcomplex.core.laws import FluxLaw
>>> u = var('u')
>>> FluxLaw(u, u**2 / 2, riemann='godunov').check().prepare().run({'u': 2.0}).value_
2.0

Source code in src/skcomplex/core/laws/_flux.py

def __init__(
    self,
    target: Target,
    flux: object,
    *,
    limiter: str | None = 'minmod',
    riemann: str = 'rusanov',
    speed: object = None,
    label: str = '',
) -> None:
    super().__init__(label)
    self.target = target
    self.flux = flux
    self.limiter = limiter
    self.riemann = riemann
    self.speed = speed

FractionalNoiseLaw(target, value, order=None, *, hurst, convention='ito', label='')

Bases: NoiseLaw

Amplitude of a fractional Brownian motion of a Hurst index that drives the highest time derivative of a state.

The increments are fractional Gaussian noise, correlated over the whole run. A Hurst index of 1/2 gives the process of a NoiseLaw, below 1/2 the increments anticorrelate, and above 1/2 they correlate. For other indices than 1/2 the amplitude must not depend on any state.

Parameters:

Name	Type	Description	Default
target	Target	The symbol of a scalar state, or the column of symbols of a vector state.	required
value	object	A symbolic expression of parameters and time. A vector target gets one fractional Brownian motion per component.	required
order	int | None	The order of the derivative the noise drives, checked against the state's order, or None to use it.	None
hurst	object	The Hurst index 0 < H < 1, a number or a parameter.	required
convention	str	The stochastic calculus of the law, 'ito', 'stratonovich', or 'anti-ito', as for NoiseLaw.	'ito'
label	str	The name of the law, or the empty string for a law without one.	''

Examples:

>>> from sage.all__sagemath_symbolics import var
>>> from skcomplex.core.laws import FractionalNoiseLaw
>>> x, D = var('x'), var('D')
>>> FractionalNoiseLaw(x, 2 * D, hurst=0.7).check().hurst
0.7

Source code in src/skcomplex/core/laws/_fractional_noise.py

def __init__(
    self,
    target: Target,
    value: object,
    order: int | None = None,
    *,
    hurst: object,
    convention: str = 'ito',
    label: str = '',
) -> None:
    super().__init__(target, value, order, convention=convention, label=label)
    self.hurst = hurst

GeneratorLaw(operator, *, label='')

Bases: BaseLaw

Hermitian operator that adds to the generator H of the linear evolution dx/dt = -i H x on a tensor product.

Parameters:

Name	Type	Description	Default
operator	object	The Hermitian operator, an expression of local operators of the namespaces the law receives.	required
label	str	The name of the law, or the empty string for a law without one.	''

Examples:

>>> from sage.all__sagemath_symbolics import SR, var
>>> from skcomplex.core.laws import GeneratorLaw
>>> from skcomplex.core.laws.utils import local_operator
>>> z = local_operator(SR.symbol('q'), SR.symbol('z'))
>>> GeneratorLaw(var('h') * z).check().prepare().run({'h': 0.5}).value_
0.5*local_operator(q, z)

Source code in src/skcomplex/core/laws/_generator.py

def __init__(self, operator: object, *, label: str = '') -> None:
    super().__init__(label)
    self.operator = operator

GradientStepLaw(trained, loss, data, optimizer, batch_size, *, truncation=None, metrics=(), label='')

Bases: BaseLaw

Optimizer steps on adaptive parameters of a nested system, on the gradient of a loss over unrolled runs.

One step of the discrete time that holds the law is one epoch. In an epoch the sequences of the dataset are shuffled and split into batches, and for every batch the nested system runs from its initial states over every sequence, the loss is averaged over the steps and sequences with the step weights, and the optimizer changes the trained parameters by the gradient of that mean. The gradient is exact, by automatic differentiation of the unrolled runs. The law records the mean batch loss of the epoch and every metric over the whole dataset at the end of the epoch.

Parameters:

Name	Type	Description	Default
trained	Sequence[str]	The dotted paths of the adaptive parameters of the nested system that the law trains, 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, 'sgd', 'momentum', 'rmsprop', or 'adam'.	required
batch_size	int	The number of sequences of a batch. The last batch of an epoch holds the rest.	required
truncation	int | None	The number of inner steps the gradient flows back through the inner states, or None for the whole run.	None
metrics	Sequence[ObservableSpec]	The observables of the nested system recorded at the end of every epoch, as weighted means over every sequence.	()
label	str	The name of the law, or the empty string for a law without one.	''

Examples:

>>> import numpy as np
>>> from skcomplex.core.laws import GradientStepLaw
>>> from skcomplex.specs import DatasetSpec, ObservableSpec
>>> data = DatasetSpec({'inputs.x': np.zeros((4, 3, 1))})
>>> law = GradientStepLaw(['readout.W'], ObservableSpec('error', 0.0), data, 'adam', 2).check()
>>> law.prepare('training').targets_
()

Source code in src/skcomplex/core/laws/_gradient_step.py

def __init__(
    self,
    trained: Sequence[str],
    loss: ObservableSpec,
    data: DatasetSpec,
    optimizer: str,
    batch_size: int,
    *,
    truncation: int | None = None,
    metrics: Sequence[ObservableSpec] = (),
    label: str = '',
) -> None:
    super().__init__(label)
    self.trained = trained
    self.loss = loss
    self.data = data
    self.optimizer = optimizer
    self.batch_size = batch_size
    self.truncation = truncation
    self.metrics = metrics

JumpLaw(rate=None, changes=None, waiting=None, reset=None, *, label='')

Bases: BaseLaw

Random event that changes states, on an exponential clock of a given rate or on a clock of drawn waiting times.

With a rate, the event fires at the times of a Poisson process of that propensity and changes states by fixed increments, as in continuous-time Markov chains. With a waiting time instead, it fires after a waiting time drawn anew at every firing, and applies its reset, as in renewal processes such as the jumps of a random walk. Every member of a group then fires on its own clock, and the trajectory records every firing and counts them in <label>.renewals. Jumps combine with rate and noise laws, which evolve the states between firings.

Parameters:

Name	Type	Description	Default
rate	object	The propensity of the event, a symbolic expression, a Call, or a callable f(t, values), or None with a waiting time.	None
changes	Mapping[Expression, object] | None	A mapping from the symbol of a scalar state to its increment when the event fires. In a law of a group, a target may also be a member offset read shift_member(namespace.n, k).	None
waiting	object	The waiting time until the next firing, usually a draw from a distribution with nonnegative support, such as draw('wait', 'pareto', b=alpha), or None for the exponential clock of rate.	None
reset	Mapping[Expression, object] | None	With a waiting time, a mapping from state symbol to its new value, which may hold draws, or None.	None
label	str	With a waiting time, the name of the firings in the trajectory, or the empty string for the path of the element.	''

Examples:

>>> from sage.all__sagemath_symbolics import var
>>> from skcomplex.core.laws import JumpLaw
>>> from skcomplex.core.laws.utils import draw
>>> s, i = var('s'), var('i')
>>> JumpLaw(0.1 * s * i, {s: -1, i: 1}).check().prepare().targets_
(s, i)
>>> n = var('n')
>>> JumpLaw(waiting=draw('wait', 'expon'), reset={n: 0}).check().prepare().targets_
(n,)

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

def __init__(
    self,
    rate: object = None,
    changes: Mapping[Expression, object] | None = None,
    waiting: object = None,
    reset: Mapping[Expression, object] | None = None,
    *,
    label: str = '',
) -> None:
    super().__init__(label)
    self.rate = rate
    self.changes = changes
    self.waiting = waiting
    self.reset = reset

LocalRuleLaw(targets, over, rule, *, reads=None, schedule='synchronous', stochastic=False, label='')

Bases: BaseLaw

Next value of lattice cells computed from the ordered values of their neighbourhoods.

A function rule receives one array per neighbour of every read state, in neighbourhood order, with one value per site, and keyword arguments for the parameters of the element that it names, plus random, an array of uniform numbers per site, when the rule is stochastic, and degree, the number of neighbours of every site besides the site itself, when it names it. It returns the next values of the target, or a tuple with one array per target.

Parameters:

Name	Type	Description	Default
targets	Expression | tuple[Expression, ...]	The symbol of a lattice state, or a tuple of them that the rule updates together.	required
over	BaseNeighborhood	The neighbourhood whose values the rule reads: a lattice neighbourhood of offsets, an adjacency on a network, or a memory of group elements on a group space.	required
rule	object	A function, a Table, a Totalistic rule, an OuterTotalistic rule, or an object whose table() returns a Table, such as a rule code.	required
reads	tuple[Expression, ...] | None	The lattice states whose neighbourhoods a function rule reads, or None for the targets.	None
schedule	str	'synchronous' to update every site from the previous configuration, 'random-sequential' to update the sites one at a time in a random order, 'random-site' to update as many randomly chosen sites, 'checkerboard' to update the two sublattices of a square lattice in turn, or 'colored' to update the classes of a proper coloring of the neighbour graph in turn.	'synchronous'
stochastic	bool	Whether a function rule receives the uniform random array random.	False
label	str	The name of the law, or the empty string for a law without one.	''

Examples:

>>> from sage.all__sagemath_symbolics import var
>>> from skcomplex.core.laws import LocalRuleLaw
>>> from skcomplex.specs import TotalisticSpec
>>> from skcomplex.core.spaces import BoxNeighborhood
>>> LocalRuleLaw(var('cell'), BoxNeighborhood(), TotalisticSpec({2: 1})).check().prepare().targets_
(cell,)

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

def __init__(
    self,
    targets: Expression | tuple[Expression, ...],
    over: BaseNeighborhood,
    rule: object,
    *,
    reads: tuple[Expression, ...] | None = None,
    schedule: str = 'synchronous',
    stochastic: bool = False,
    label: str = '',
) -> None:
    super().__init__(label)
    self.targets = targets
    self.over = over
    self.rule = rule
    self.reads = reads
    self.schedule = schedule
    self.stochastic = stochastic

MatchLaw(left, right=None, pairing=None, capacity=1, phase=None, *, label='match')

Bases: BaseLaw

Pairing of the members of one agent group, or of two, every step, which the rules of the step read.

The members are paired uniformly at random by default. Any other pairing, such as stable matching, an order book, or an auction, is a function given as pairing, and a selection of partners within reach or among neighbours is a PartnerSpec. Rules read the pairs of the step with matches(model, label). Transfers, matchings, and groupings read their pairs from get_pairs.

Parameters:

Name	Type	Description	Default
left	str	The path of the first agent group.	required
right	str | None	The path of the second agent group, or None for pairs within the first.	None
pairing	Pairing | PartnerSpec | None	A function pairing(left, right, rng, capacity) of the agents of both groups that returns their pairs, a PartnerSpec, or None for uniform random pairing.	None
capacity	int	The largest number of pairs of every member.	1
phase	str | None	The phase of the step the pairing runs in, before the rules of the phase that read it.	None
label	str	The name of the pairs.	'match'

Examples:

>>> from skcomplex.core.laws import MatchLaw
>>> MatchLaw('buyers', 'sellers', label='trades').check().capacity
1

Source code in src/skcomplex/core/laws/_match.py

def __init__(
    self,
    left: str,
    right: str | None = None,
    pairing: Pairing | PartnerSpec | None = None,
    capacity: int = 1,
    phase: str | None = None,
    *,
    label: str = 'match',
) -> None:
    super().__init__(label)
    self.left = left
    self.right = right
    self.pairing = pairing
    self.capacity = capacity
    self.phase = phase

get_pairs(left, right, rng)

Return the pairs of the agents of both sides by the pairing.

Parameters:

Name	Type	Description	Default
left	Sequence[AgentViewProtocol]	The agents of the first side.	required
right	Sequence[AgentViewProtocol]	The agents of the second side, the same sequence for pairs within one group.	required
rng	Generator	The random generator.	required

Returns:

Name	Type	Description
pairs	list[tuple[object, object]]	The pairs of agents, the first of every pair from the first side.

Raises:

Type	Description
ValidationError	A selection of partners names a state the agents do not have.

Source code in src/skcomplex/core/laws/_match.py

def get_pairs(
    self,
    left: Sequence[AgentViewProtocol],
    right: Sequence[AgentViewProtocol],
    rng: np.random.Generator,
) -> list[tuple[object, object]]:
    """Return the pairs of the agents of both sides by the pairing.

    Args:
        left:
            The agents of the first side.

        right:
            The agents of the second side, the same sequence for pairs within one group.

        rng:
            The random generator.

    Returns:
        pairs:
            The pairs of agents, the first of every pair from the first side.

    Raises:
        ValidationError:
            A selection of partners names a state the agents do not have.
    """
    spec = self.pairing
    if not isinstance(spec, PartnerSpec):
        return (spec or random_pairing)(left, right, rng, self.capacity)
    if (spec.candidates, spec.choice, spec.capacity, spec.fallback) == ('all', 'uniform', 1, None):
        return self._get_disjoint_pairs(left, right, rng)
    return list(self._get_partners(spec, left, right, rng))

iterate_pairs(left, right, rng, feasible)

Yield the pairs of a selection of partners one at a time, so a caller settles each before the next one.

The agents act in random order, since the first to act take what the feasibility reads.

Parameters:

Name	Type	Description	Default
left	Sequence[AgentViewProtocol]	The agents of the first side.	required
right	Sequence[AgentViewProtocol]	The agents of the second side.	required
rng	Generator	The random generator.	required
feasible	Feasibility	The function feasible(agent, candidate) that says whether an agent can pair with a candidate now.	required

Returns:

Name	Type	Description
pairs	Iterator[tuple[AgentViewProtocol, AgentViewProtocol]]	The pairs of agents, the first of every pair from the first side.

Raises:

Type	Description
ValidationError	The pairing is not a selection of partners, or it names a state the agents do not have.

Source code in src/skcomplex/core/laws/_match.py

def iterate_pairs(
    self,
    left: Sequence[AgentViewProtocol],
    right: Sequence[AgentViewProtocol],
    rng: np.random.Generator,
    feasible: Feasibility,
) -> Iterator[tuple[AgentViewProtocol, AgentViewProtocol]]:
    """Yield the pairs of a selection of partners one at a time, so a caller settles each before the next one.

    The agents act in random order, since the first to act take what the feasibility reads.

    Args:
        left:
            The agents of the first side.

        right:
            The agents of the second side.

        rng:
            The random generator.

        feasible:
            The function `feasible(agent, candidate)` that says whether an agent can pair with a candidate now.

    Returns:
        pairs:
            The pairs of agents, the first of every pair from the first side.

    Raises:
        ValidationError:
            The pairing is not a selection of partners, or it names a state the agents do not have.
    """
    if not isinstance(self.pairing, PartnerSpec):
        error_msg = f'The match {self.label!r} iterates the pairs of a `PartnerSpec`, not {self.pairing!r}.'
        raise ValidationError(error_msg)
    return self._get_partners(self.pairing, left, right, rng, feasible)

NoiseLaw(target, value, order=None, *, convention='ito', process=None, conserved=False, label='')

Bases: BaseLaw

Amplitude of a Wiener process that drives the highest time derivative of a state.

Without a process, every component of the target and every member of a group gets its own independent Wiener process. Laws of one component, or of one member, that name the same local process share its increment. Laws that name a process of a WienerProcessSpec through a role share it across components and members.

A conserved noise on a state of a field on a lattice adds div(value eta) instead, the finite-difference divergence of a random flux with one Wiener process per link between neighbouring sites, so the total of the field changes only by its rate laws.

Parameters:

Name	Type	Description	Default
target	Target	The symbol of a scalar state, or the column of symbols of a vector state.	required
value	object	A symbolic expression, a Call, or a callable f(t, values). A vector target gets one Wiener process per component.	required
order	int | None	The order of the derivative the noise drives, checked against the state's order, or None to use it.	None
convention	str	The stochastic calculus of the law, 'ito', 'stratonovich', or 'anti-ito'. The compiler adds the drift correction of the convention, so every solver integrates the Itô form.	'ito'
process	Process	None for independent processes, a label local to the element, a tuple of labels with one per component of a vector target, or the handle namespace.<role>.dW or namespace.<role>.dW[k] of the processes of a WienerProcessSpec.	None
conserved	bool	Whether the noise is the divergence of a random flux on the links of the lattice of a field.	False
label	str	The name of the law, or the empty string for a law without one.	''

Examples:

>>> from sage.all__sagemath_symbolics import var
>>> from skcomplex.core.laws import NoiseLaw
>>> x = var('x')
>>> NoiseLaw(x, 0.2 * x, process='shared').check().prepare().run({'x': 10.0}).value_
2.0

Source code in src/skcomplex/core/laws/_noise.py

def __init__(
    self,
    target: Target,
    value: object,
    order: int | None = None,
    *,
    convention: str = 'ito',
    process: Process = None,
    conserved: bool = False,
    label: str = '',
) -> None:
    super().__init__(label)
    self.target = target
    self.value = value
    self.order = order
    self.convention = convention
    self.process = process
    self.conserved = conserved

RateLaw(target, value, order=None, *, cross_section=None, label='')

Bases: BaseLaw

Contribution to the highest time derivative of a state, the first for states of order 1.

Rates on the same state add up, so a system with only rate laws is a system of ordinary differential equations. On a state of order k, a rate contributes to the k-th derivative, so independent mechanisms superpose.

Parameters:

Name	Type	Description	Default
target	Target | Matrix	The symbol of a scalar state, the column of symbols of a vector state, or the matrix of a tensor state.	required
value	object	A symbolic expression, a matrix of the shape of a tensor state, a Call, or a callable f(t, values) where values maps dotted state and parameter names to their current values.	required
order	int | None	The order of the derivative the rate gives, checked against the state's order, or None to use it.	None
cross_section	Callable[..., object] | None	The function B(speed, cosine) of arrays of the relative speed and the cosine of the deflection angle, which a collision of the value reads, or None.	None
label	str	The name of the law, or the empty string for a law without one.	''

Examples:

>>> from sage.all__sagemath_symbolics import var
>>> from skcomplex.core.laws import RateLaw
>>> n = var('n')
>>> RateLaw(n, 2 * n).check().prepare().run({'n': 3.0}).value_
6.0

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

def __init__(
    self,
    target: Target | Matrix,
    value: object,
    order: int | None = None,
    *,
    cross_section: Callable[..., object] | None = None,
    label: str = '',
) -> None:
    super().__init__(label)
    self.target = target
    self.value = value
    self.order = order
    self.cross_section = cross_section

RuleLaw(function, phase=None, *, members=None, label='')

Bases: BaseLaw

Behaviour written as code, run once per step for every member of an agent group, or once per step for a model.

An element whose laws include rules is an agent, alone or in a group. A rule of an element receives the agent and the model, function(agent, model). A rule of an interaction receives the model alone, function(model), unless it names the role whose members run it: every member of the groups of that role then runs it in its own activation, after the rules of its element and the member rules of the interactions before it, as function(agent, model) with the model view of the interaction. A system with rules is agent-based and needs discrete time.

Parameters:

Name	Type	Description	Default
function	Callable[..., object]	The function that runs the rule.	required
phase	str | None	The name of the phase of the step the rule runs in, or None for the default phase.	None
members	str | None	The role of the interaction whose members run the rule, or None for a rule of the model or of an element.	None
label	str	The name of the law, or the empty string for a law without one.	''

Examples:

>>> from skcomplex.core.laws import RuleLaw
>>> moved = []
>>> RuleLaw(lambda agent, model: moved.append(agent)).check().prepare().run(('walker', None)).value_
>>> moved
['walker']

Source code in src/skcomplex/core/laws/_rule.py

def __init__(
    self,
    function: Callable[..., object],
    phase: str | None = None,
    *,
    members: str | None = None,
    label: str = '',
) -> None:
    super().__init__(label)
    self.function = function
    self.phase = phase
    self.members = members

SampleLaw(objective, move, sampler=None, *, label='')

Bases: BaseLaw

Monte Carlo sweep that samples states by moves accepted from the change of an objective, one per unit of time.

Parameters:

Name	Type	Description	Default
objective	ObjectiveSpec	The local objective of a site and its neighbourhood.	required
move	MoveSpec	The move proposed at a site.	required
sampler	SamplerSpec | None	The acceptance of moves, with its intensity and schedule, or None for Metropolis acceptance at the parameter intensity of the element.	None
label	str	The name of the law, or the empty string for a law without one.	''

Examples:

>>> from sage.all__sagemath_symbolics import var
>>> from skcomplex.core.laws import SampleLaw
>>> from skcomplex.core.spaces import VonNeumannNeighborhood
>>> from skcomplex.specs import ObjectiveSpec, MoveSpec, SamplerSpec
>>> spin = var('spin')
>>> law = SampleLaw(ObjectiveSpec(-spin, VonNeumannNeighborhood()), MoveSpec(spin), SamplerSpec('glauber', 2.0))
>>> law.check().prepare().targets_
(spin,)

Source code in src/skcomplex/core/laws/_sample.py

def __init__(
    self,
    objective: ObjectiveSpec,
    move: MoveSpec,
    sampler: SamplerSpec | None = None,
    *,
    label: str = '',
) -> None:
    super().__init__(label)
    self.objective = objective
    self.move = move
    self.sampler = sampler

Sink(label='environment')

Environment a transfer releases a quantity to, outside the system, such as a drain or the atmosphere.

Parameters:

Name	Type	Description	Default
label	str	The name of the environment in transaction-flow matrices.	'environment'

Examples:

>>> from skcomplex.core.laws import Sink
>>> Sink('drain').label
'drain'

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

def __init__(self, label: str = 'environment') -> None:
    self.label = label

Source(label='environment')

Environment a transfer draws a quantity from, outside the system, such as a bank that creates money.

Parameters:

Name	Type	Description	Default
label	str	The name of the environment in transaction-flow matrices.	'environment'

Examples:

>>> from skcomplex.core.laws import Source
>>> Source().label
'environment'

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

def __init__(self, label: str = 'environment') -> None:
    self.label = label

TransferLaw(source, sink, amount, rate=None, *, label='', owner='')

Bases: BaseLaw

Movement of an amount of a quantity from a source holder to a sink holder, in continuous or discrete time.

A transfer is one outflow of its source and one inflow of its sink. Its amount is a rate in a system of rate laws, the amount moved per step in a system of update laws, and the size of every jump when a rate is given. The trajectory records the amount at every frame under <path>.<label>, the path of the part that declares the transfer, or owner when given. One holder may be a state of a group read by shift_member at a member offset, so every member transfers with the member that many places along the group order. Past the ends of a fixed boundary no member is there, and the transfer moves nothing.

Parameters:

Name	Type	Description	Default
source	Expression | Source	The symbol of the state the quantity leaves, the symbol read at a member offset, or Source() for the environment.	required
sink	Expression | Sink	The symbol of the state the quantity enters, the symbol read at a member offset, or Sink() for the environment.	required
amount	object	The symbolic amount, a rate, a per-step amount, or a jump size.	required
rate	object	The propensity of a jump that moves amount at random times, or None for a deterministic flow.	None
label	str	The name of the flow, unique among the flows reported under one part, or the empty string for flow_<index>.	''
owner	str	The path of the part the flow is reported under, relative to the system of the part that declares the transfer, such as the element an interaction moves a quantity within, or the empty string for the part that declares it.	''

Examples:

>>> from sage.all__sagemath_symbolics import var
>>> from skcomplex.core.laws import Source, TransferLaw
>>> level = var('level')
>>> TransferLaw(Source(), level, 5, label='faucet').check().prepare().targets_
(level,)

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

def __init__(
    self,
    source: Expression | Source,
    sink: Expression | Sink,
    amount: object,
    rate: object = None,
    *,
    label: str = '',
    owner: str = '',
) -> None:
    super().__init__(label)
    self.source = source
    self.sink = sink
    self.amount = amount
    self.rate = rate
    self.owner = owner

TransitionLaw(target, matrix, *, label='')

Bases: BaseLaw

Step of a discrete-time Markov chain, a state over a finite alphabet drawn from a row of a transition matrix.

The state's current value selects the row, and the next value is drawn from it. The state takes a categorical domain or the integers from 0 to n - 1. A transition is a kind of update, so it combines with update laws.

Parameters:

Name	Type	Description	Default
target	Expression	The symbol of a scalar state.	required
matrix	Sequence[Sequence[object]] | Callable[..., object] | CallSpec	The square transition matrix, whose entry (i, j) is the probability of the step from i to j, as numbers or expressions of parameters, or a function f(t, values) that returns it, one matrix for all members or one per member, or CallSpec(f, relative=True) to read the names of the component. A function runs in Python and may read states.	required
label	str	The name of the law, or the empty string for a law without one.	''

Examples:

>>> from sage.all__sagemath_symbolics import var
>>> from skcomplex.core.laws import TransitionLaw
>>> x, q = var('x'), var('q')
>>> TransitionLaw(x, [[q, 1 - q], [0.4, 0.6]]).check().prepare().run({'q': 0.1}).value_[0]
[0.1, 0.9]

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

def __init__(
    self,
    target: Expression,
    matrix: Sequence[Sequence[object]] | Callable[..., object] | CallSpec,
    *,
    label: str = '',
) -> None:
    super().__init__(label)
    self.target = target
    self.matrix = matrix

UpdateLaw(target, value, *, label='', increment=False, source=None, order=None, phase=None)

Bases: BaseLaw

Next value of a state after one discrete time step.

A state takes one update law, unless each of its update laws declares how it combines with the others. An increment law gives the state plus a change, and a law with a source changes only a state that holds the source value and keeps every other value. Several such laws of one state combine in one step by adding their changes, x + sum_k (f_k - x), all computed from the states at the start of the step, so increments add and moves from distinct source values compose. Two laws with the same source, or a law that declares neither, are an error.

With an order, the members of a group update one after the other, in increasing order of the value of order at the start of the step, ties in the order of the members, and each member reads the values the members before it set. Without one, every member updates at once from the states at the start of the step.

With a phase, the law runs in that phase of the step, which a PhasesScheduleSpec of the discrete time orders. The phases run one after the other, each reading the states the phases before it set, and the laws of one phase update at once. A law without a phase runs in the default phase, first unless the schedule places it.

Parameters:

Name	Type	Description	Default
target	Target | TopologyRequestSpec	The symbol of a scalar state, the column of symbols of a vector state, or a TopologyRequestSpec, whose value is the probability that the request fires.	required
value	object	A symbolic expression, a Call, or a callable f(t, values) that gives the next value.	required
label	str	The name of the law, or the empty string for a law without one.	''
increment	bool	Whether the value is the state plus a change that adds to the changes of the other update laws of the state.	False
source	float | None	The value the law moves the state from, keeping every other value, or None.	None
order	object	A symbolic expression of the states, parameters, and draws of a member, whose values order the members that update one after the other, or None for members that update at once.	None
phase	str | None	The name of the phase of the step the law runs in, or None for the default phase.	None

Raises:

Type	Description
ValidationError	The value is a NumPy array, or an increment law has a source.

Examples:

>>> from sage.all__sagemath_symbolics import var
>>> from skcomplex.core.laws import UpdateLaw
>>> x = var('x')
>>> UpdateLaw(x, 4 * x * (1 - x)).check().prepare().run({'x': 0.5}).value_
1.0
>>> UpdateLaw(x, x + 1, increment=True).check().increment
True
>>> UpdateLaw(x, x + 1, phase='imitate').check().phase
'imitate'

Source code in src/skcomplex/core/laws/_update.py

def __init__(
    self,
    target: Target | TopologyRequestSpec,
    value: object,
    *,
    label: str = '',
    increment: bool = False,
    source: float | None = None,
    order: object = None,
    phase: str | None = None,
) -> None:
    super().__init__(label)
    self.target = target
    self.value = value
    self.increment = increment
    self.source = source
    self.order = order
    self.phase = phase