Sequentially linear ecosystem
An ecosystem whose abundances follow linear dynamics between extinctions that rewrite its network.
- Name:
sequentially_linear_ecosystem - Field:
biology/ecology - Formalism: discrete map
- References: Thurner et al. 2018; Thurner, Klimek, and Hanel 2010
Select this system
from skcomplex.systems import create_simulation, create_system, describe
system = create_system('sequentially_linear_ecosystem')
entry = describe('sequentially_linear_ecosystem')
trajectory = create_simulation('sequentially_linear_ecosystem').prepare().run().trajectory_
Factory
@register_system(
field='biology/ecology',
tags=('adaptive network', 'extinction', 'diversity', 'linear dynamics', 'stochastic map'),
references=('Thurner et al. 2018', 'Thurner, Klimek, and Hanel 2010'),
defaults={'until': 300, 'dt': 1.0},
)
def sequentially_linear_ecosystem(
n: int = 60,
k: int = 4,
*,
rate: float = 0.2,
threshold: float = 1e-3,
seed: int | None = 1,
) -> System:
"""Return an ecosystem whose abundances follow linear dynamics between extinctions that rewrite its network.
Every species reads `k` random species through influences of random sign and strength, and grows linearly with
their abundances. Between extinctions the dynamics is linear, so the abundances approach the dominant eigenvector
of the current network. A species whose abundance falls below `threshold` goes extinct with its influences, and a
new species with `k` random influences takes its place, which changes the network and starts a new linear stretch.
Extinctions come in bursts separated by quiet periods.
Args:
n:
The number of species.
k:
The number of species every species reads.
rate:
The growth per step per unit of influence.
threshold:
The abundance below which a species goes extinct.
seed:
The seed of the network and of the initial strengths.
Returns:
system:
The system, with the abundances `species.x` and ages `species.age` of the variable group `species` and the
strengths `influences.w` of the variable group `influences` on the edges.
Examples:
>>> from skcomplex.systems.biology.ecology import sequentially_linear_ecosystem
>>> [component.label for component in sequentially_linear_ecosystem(n=10).elements]
['species', 'influences']
"""
rng = np.random.default_rng(seed)
graph = nx.DiGraph()
graph.add_nodes_from(range(n))
for node in range(n):
others = rng.choice(np.delete(np.arange(n), node), size=k, replace=False)
graph.add_edges_from((node, int(other)) for other in others)
space = NetworkSpace(graph)
edges = graph.number_of_edges()
species = GroupElement(
'species',
element=AbundanceElement,
space=space,
on='sites',
variable=True,
capacity=2 * n,
values={'x': np.full(n, 1 / n), 'threshold': threshold},
visual=VisualSpec('sites', {'states': ['x']}),
)
influences = GroupElement(
'influences',
element=WeightedEdgeElement,
space=space,
on='edges',
variable=True,
capacity=4 * edges,
values={'w': rng.uniform(-1, 1, edges), 'fresh': 0.0},
visual=HIDDEN,
)
return System(
elements=[species, influences],
interactions=[
LinearGrowthInteraction('growth', species='species', influences='influences', rate=rate),
EdgeDrawInteraction('draw', target='species', edges='influences'),
MapInteraction('ageing', target='species', state='age', function=lambda s: 1, mode='increment'),
TopologyInteraction(
'extinction',
target='species',
action='replace-node',
condition=lambda s: s.source.x < s.source.threshold,
edges=k,
),
],
time=DiscreteTime(),
)
Components
Elements:
AbundanceElement, fromElementWeightedEdgeElement, fromElement
Interactions:
LinearGrowthInteraction, fromInteractionEdgeDrawInteraction, fromNeighborInteractionMapInteractionTopologyInteraction
class AbundanceElement(Element):
"""Species of an ecosystem with its relative abundance and its age, which goes extinct below a threshold.
Args:
label:
The label of the species or of the group of species.
visual:
How the element is drawn, or `None` for `default_visual_`.
x:
The relative abundance.
age:
The number of steps since the species appeared.
threshold:
The abundance below which the species goes extinct.
Examples:
>>> from skcomplex.systems.biology.ecology import AbundanceElement
>>> AbundanceElement('species').threshold
0.001
"""
def __init__(
self,
label: str,
visual: VisualSpec | None = None,
*,
x: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 0.01,
age: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 0.0,
threshold: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.001,
) -> None:
super().__init__(label, visual=visual)
self.x = x
self.age = age
self.threshold = threshold
class WeightedEdgeElement(Element):
"""Edge of a network with a weight `w`, such as the strength of an influence or a coupling between two species.
A new edge is marked `fresh` until an `EdgeDrawInteraction` draws its weight.
Args:
label:
The label of the edge or of the group of edges.
visual:
How the element is drawn, or `None` for `default_visual_`.
w:
The weight.
fresh:
Whether the edge has not drawn its weight yet, 1 or 0.
Examples:
>>> from skcomplex.systems.biology.common.networks import WeightedEdgeElement
>>> WeightedEdgeElement('edge').fresh
1.0
"""
def __init__(
self,
label: str,
visual: VisualSpec | None = None,
*,
w: Annotated[float, StateSpec()] = 0.0,
fresh: Annotated[float, StateSpec(domain=FiniteEnumeratedSet([0, 1]))] = 1.0,
) -> None:
super().__init__(label, visual=visual)
self.w = w
self.fresh = fresh
class LinearGrowthInteraction(Interaction):
"""Linear growth of every species by the abundances of the species that influence it, normalised to one.
Every species grows by `rate` times the sum of the abundances of the species it reads weighted by the strengths `w`
of the influences, its abundance stays nonnegative, and the abundances are normalised to sum to one.
Args:
label:
The label of the interaction.
species:
The path of the group of species on the nodes.
influences:
The path of the variable group on the edges.
rate:
The growth per step per unit of influence.
Examples:
>>> from skcomplex.systems.biology.ecology import LinearGrowthInteraction
>>> LinearGrowthInteraction('growth', species='species', influences='influences').rate
0.2
"""
def __init__(
self,
label: str,
*,
species: Annotated[str, RoleSpec()],
influences: Annotated[str, RoleSpec()],
rate: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 0.2,
) -> None:
super().__init__(label)
self.species = species
self.influences = influences
self.rate = rate
def _get_laws(self, namespace: SimpleNamespace) -> list[BaseLaw]:
"""Return the linear growth and the normalisation of the abundances.
Args:
namespace:
The namespace with `namespace.species` and `namespace.influences`.
Returns:
laws:
The update of the abundance.
"""
species = namespace.species
grown = species.x + namespace.rate * neighbors(species.other.x, weight=namespace.influences.w)
kept = where(grown > 0, grown, 0)
return [UpdateLaw(species.x, kept / total(kept))]
class EdgeDrawInteraction(NeighborInteraction):
"""First draw of the weight of every new edge of a network, and a random drift of the weights of the other edges.
Every step of discrete time an edge marked `fresh` draws its weight uniformly from `[low, high]` and loses its mark,
and every other edge adds `noise` times a standard normal draw to its weight, or keeps it without noise.
Args:
label:
The label of the interaction.
target:
The path of the group on the nodes of the network.
edges:
The path of the group on its edges, whose members hold the weight and the mark.
noise:
The standard deviation of the drift of a weight in a step, 0 for none.
low:
The least weight of a new edge.
high:
The largest weight of a new edge.
state:
The name of the state of the edges that holds the weight.
Examples:
>>> from types import SimpleNamespace
>>> from sage.all__sagemath_symbolics import var
>>> from skcomplex.systems.biology.common.networks import EdgeDrawInteraction
>>> edge = SimpleNamespace(w=var('w'), fresh=var('fresh'))
>>> draw = EdgeDrawInteraction('draw', target='species', edges='links').prepare()
>>> [str(law.value) for law in draw.run(SimpleNamespace(edge=edge, noise=var('noise'))).laws_][1]
'0'
"""
def __init__(
self,
label: str,
*,
target: Annotated[str, RoleSpec()],
edges: Annotated[str, RoleSpec()],
noise: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.0,
low: float = -1.0,
high: float = 1.0,
state: str = 'w',
) -> None:
super().__init__(label)
self.target = target
self.edges = edges
self.noise = noise
self.low = low
self.high = high
self.state = state
def _get_name(self, draw: str) -> str:
"""Return the name of a random draw of the interaction."""
return re.sub(r'\W', '_', f'{draw}_{self.label}')
def _get_laws(self, namespace: SimpleNamespace) -> list[BaseLaw]:
"""Return the draw or the drift of the weight of an edge, and the removal of its mark.
Args:
namespace:
The namespace of an edge, with the member `namespace.edge` of the group on the edges.
Returns:
laws:
The updates of the weight and of the mark.
"""
edge = namespace.edge
weight = getattr(edge, self.state)
drawn = uniform(self._get_name('draw'), self.low, self.high)
kept = weight + namespace.noise * normal(self._get_name('drift')) if self.noise else weight
return [UpdateLaw(weight, where(edge.fresh > 1 / 2, drawn, kept)), UpdateLaw(edge.fresh, 0)]
class MapInteraction(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, system)` 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 system view.
Args:
label:
The label of the interaction, unique among the parts of its system.
target:
The path of the element or group whose state the map sets.
state:
The name of the state of the target.
function:
The function `f(namespace)` in the symbolic form, or `f(agent, system)` in the agent form.
sources:
The paths of the elements, or members of groups such as `cells[0]`, that the function reads.
mode:
`'replace'` or `'increment'`.
schedule:
`'synchronous'`, `'sequential'`, `'random-sequential'`, or `'member'` for the agent form.
form:
`'symbolic'` for update laws of the states of the target, or `'agent'` for a rule over an agent group.
phase:
The phase of the step the agent form runs in, or `None` for the default phase.
Raises:
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.components.elements import Element
>>> from skcomplex.core.components.interactions import MapInteraction
>>> from skcomplex.core.components.spaces import DiscreteTime
>>> from skcomplex.core.components.system import System
>>> from skcomplex.core.engine.runtime import Simulation
>>> from skcomplex.core.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).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.prepare().run(SimpleNamespace()).laws_[0]).__name__
'RuleLaw'
>>> from skcomplex.core.components.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])).prepare().run(until=2)
>>> trajectory.trajectory_.agents('weights', 2).values['w'].tolist()
[[4.0, 8.0]]
"""
law_classes: ClassVar[tuple[type[BaseLaw], ...]] = (RuleLaw, UpdateLaw)
def __init__(
self,
label: str,
*,
target: Annotated[str, RoleSpec(elements=(ElementKind.ANY,), states=('state',))],
state: str,
function: Callable[..., object],
sources: Annotated[Sequence[str], RoleSpec(variadic=True, elements=(ElementKind.ANY,))] = (),
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
def _check(self) -> None:
"""Raise an error for an unknown mode, schedule, or form, or a function that cannot be called.
Raises:
ValidationError:
The mode, the schedule, or the form is unknown, the symbolic form has the schedule `'member'`, or the
function is not callable.
"""
super()._check()
self._check_variant('mode', self.mode, MODES, (BaseSpec,))
for name, value, variants in (
('schedule', self.schedule, SCHEDULES),
('form', self.form, FORMS),
):
if value not in variants:
error_msg = f'The {name} of map {self.label!r} is {value!r}. Use one of {", ".join(variants)}.'
raise ValidationError(error_msg)
if self.schedule == 'member' and self.form != 'agent':
error_msg = (
f'The map {self.label!r} updates every member in its activation, which only the agent form does.'
)
raise ValidationError(error_msg)
if not callable(self.function):
error_msg = f'The function of map {self.label!r} must be callable, but it is {self.function!r}.'
raise ValidationError(error_msg)
def _apply_mode(self, current: Value, value: Value) -> Value:
"""Return the next value of a state from its current value and the value of the function, by the mode."""
if self.mode == 'increment':
return current + value
if self.mode == 'replace':
return value
raise self._get_variant_error(self.mode, 'mode', '_apply_mode')
def _get_vector(self, agent: AgentViewProtocol, system: SystemViewProtocol) -> Value:
"""Return the next value of a vector state of an agent."""
current = np.asarray(getattr(agent, self.state), dtype=np.float64)
value = np.asarray(self.function(agent, system), dtype=np.float64)
if value.shape != current.shape:
error_msg = (
f'The function of map {self.label!r} returns the shape {value.shape} for the state {self.state!r} of '
f'the shape {current.shape}. Return an array of the shape of the state.'
)
raise ValidationError(error_msg)
return self._apply_mode(current, value)
def _get_value(self, agent: AgentViewProtocol, system: SystemViewProtocol) -> Value:
"""Return the next value of the state of an agent, a number or the array of a vector state."""
if np.ndim(getattr(agent, self.state)) > 0:
return self._get_vector(agent, system)
found = self.function(agent, system)
if not isinstance(found, SupportsFloat):
error_msg = f'The function of map {self.label!r} returns {found!r} for an agent. Return a number.'
raise ValidationError(error_msg)
value = float(found)
return self._apply_mode(float(getattr(agent, self.state)), value)
def _set_value(self, agent: AgentViewProtocol, value: Value) -> None:
"""Set the state of an agent to its next value."""
setattr(agent, self.state, value)
def _update_member(self, agent: AgentViewProtocol, system: SystemViewProtocol) -> None:
"""Set the state of one member to its next value."""
if agent.alive:
self._set_value(agent, self._get_value(agent, system))
def update(self, system: SystemViewProtocol) -> None:
"""Set the state of every agent of the target to its next value, by the schedule.
Args:
system:
The system view of the interaction.
"""
self._update_agents(system, self.schedule, self._get_value, self._set_value)
def _get_order(self) -> object:
"""Return the order of the members of the symbolic form: none, their own order, or a uniform draw."""
if self.schedule == 'synchronous':
return None
if self.schedule == 'sequential':
return SR(0)
return uniform(re.sub(r'\W', '_', f'order_{self.label}'))
def _get_laws(self, namespace: SimpleNamespace) -> list[BasePart]:
"""Return the update of the state to its next value, or the rule of the agent form.
Args:
namespace:
The namespace of the symbols the laws can use, with `namespace.target` and `namespace.sources`.
Returns:
laws:
The update law of the state, or the rule that runs the map of the agents.
"""
if self.form == 'agent' and self.schedule == 'member':
return [RuleLaw(self._update_member, phase=self.phase, members=self._get_role('target'))]
if self.form == 'agent':
return [RuleLaw(self.update, phase=self.phase)]
value = getattr(self._get_member(namespace, 'target'), self.state)
mapped = cast(Value, self.function(namespace))
next_value = self._apply_mode(value, mapped)
return [UpdateLaw(value, next_value, increment=self.mode == 'increment', order=self._get_order())]
class TopologyInteraction(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. Without a target, the namespace of `'add-node'` holds the values drawn for
the node being added as `namespace.source`, which the condition and a probability read and a rate does not. 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 that also holds the node the new edge starts from as `node.source`, or uniformly. That node is the
new node of `'add-node'`, the adding node of `'add-edge'`, and the kept end of `'rewire'`. Where fewer candidates
have a positive weight than the edges to draw, the new ends are drawn 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. With `edges=0` or
`independent`, every other member is drawn by `toward`, and the new node draws its edges apart. The action
`'add-hyperedge'` runs for every node of the group on the nodes, which forms a hyperedge of size `hyperedge` with
nodes drawn by `toward` and is its last member, and `'remove-hyperedge'` runs for every hyperedge of the variable
group on the hyperedges that the target binds, whose namespace reads the hyperedge member as the source. With
`independent`, every candidate becomes a new end on its own with probability `toward`, so a node joins a random
number of nodes, and on a directed layer `direction` sends the new edges out of the node, into it, or both ways.
`'close-triangle'` joins two neighbours of the node that are not yet joined, or a node without such a pair to
`edges` nodes, `'contract-edge'` merges a neighbour drawn by `toward` into the node, which keeps the
union of their links, and `'duplicate-node'` adds a copy of the node that inherits each of its links with
probability `inherit` and joins it with probability `parent`. A replacement removes the node and adds the new one
in one request. With `candidates`, `'add-edge'` adds for every new edge the candidate of the smallest `score`, and
without a target it runs once per step between two nodes drawn uniformly. With `one`, the action runs at every step
for one member drawn uniformly among those present, firing with its probability there.
Args:
label:
The label of the interaction, unique among the parts of its system.
target:
The path of the group on the nodes of the network, of the variable group on its hyperedges for
`'remove-hyperedge'`, or `None` for an addition of one node or of edges between drawn nodes per step.
action:
The change, `'add-node'`, `'remove-node'`, `'replace-node'`, `'add-edge'`, `'remove-edge'`, `'rewire'`,
`'add-hyperedge'`, `'remove-hyperedge'`, `'close-triangle'`, `'contract-edge'`, or `'duplicate-node'`.
rate:
The rate of the change, a number or a function, or `None` for a probability per step.
probability:
The probability of the change per step, a number or a function, or `None` for probability one.
condition:
A function of the namespace that returns where the change may fire, or `None` everywhere.
toward:
A function of the namespace of a candidate node that returns its weight as a new end, reading the node
the new edge starts from as `namespace.source`, or `None` for equal weights.
edges:
The number of edges a new node or a node adding edges joins.
keep:
The end of an edge a rewiring keeps, `'source'` or `'target'`.
values:
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.
layer:
The layer of the network whose edges change, or `None` for the graph of the network.
hyperedge:
The size of the hyperedge every new node or every node of `'add-hyperedge'` forms with existing nodes, or
`None` for none.
independent:
Whether every candidate becomes a new end on its own with the probability `toward`, in place of `edges`
ends drawn by weight.
direction:
The direction of new edges on a directed layer, `'out'` of the node, `'in'` to it, or `'both'`.
candidates:
The number of candidate edges of `'add-edge'` for every edge it adds.
score:
A function of the namespace of a candidate end that returns the score of its edge, reading the other end
as `namespace.source`, whose smallest value is added, or `None` for the first candidate.
one:
Whether the action runs at every step for one member drawn uniformly among those present.
inherit:
The probability that the copy of `'duplicate-node'` inherits each link of the node.
parent:
The probability that the copy of `'duplicate-node'` joins the node.
Raises:
ValidationError:
The action, the kept end, or the direction 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'` and `'add-edge'` has no target, a
hyperedge is given to an action other than `'add-node'` and `'add-hyperedge'` or has fewer than two
members, `'add-hyperedge'` has no hyperedge, candidates or a score belong to another action than
`'add-edge'`, or a probability of duplication lies outside `[0, 1]`, raised by `prepare`.
Examples:
>>> import networkx as nx
>>> from skcomplex.core.components.elements import Element, GroupElement
>>> from skcomplex.core.components.interactions import TopologyInteraction
>>> from skcomplex.core.components.spaces import NetworkSpace
>>> from skcomplex.core.components.system import System
>>> from skcomplex.core.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).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.components.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).prepare().run(until=5).trajectory_
>>> trajectory.network(-1).hyperedges.shape
(6, 3)
"""
law_classes: ClassVar[tuple[type[BaseLaw], ...]] = (JumpLaw, UpdateLaw)
def __init__(
self,
label: str,
*,
target: Annotated[str | None, RoleSpec(optional=True, elements=(ElementKind.ANY,))] = 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,
independent: bool = False,
direction: str = 'out',
candidates: int = 1,
score: Callable[[SimpleNamespace], object] | None = None,
one: bool = False,
inherit: float = 1.0,
parent: float = 0.0,
) -> 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
self.independent = independent
self.direction = direction
self.candidates = candidates
self.score = score
self.one = one
self.inherit = inherit
self.parent = parent
def _check(self) -> None:
"""Raise an error for an unknown action or kept end, a rate next to a probability, or a bad number of edges.
Raises:
ValidationError:
The action or the kept end is unknown, both a rate and a probability are given, a condition, a
weight, or a score cannot be called, the number of edges is not a whole number of at least 0, an
action other than `'add-node'` and `'add-edge'` has no target, a hyperedge is given to an action other
than `'add-node'` and `'add-hyperedge'`, or `'add-hyperedge'` has none, or the request is invalid.
"""
super()._check()
self._check_variant('action', self.action, ACTIONS, (BaseSpec,))
if self.rate is not None and self.probability is not None:
error_msg = f'{self.label!r} fires at a rate and with a probability. Give one of them.'
raise ValidationError(error_msg)
if self.target is None and self.action not in {'add-node', 'add-edge'}:
error_msg = f'{self.label!r} runs {self.action!r} without a target. Give the group on the nodes.'
raise ValidationError(error_msg)
if self.hyperedge is not None and self.action not in {'add-node', 'add-hyperedge'}:
error_msg = (
f'{self.label!r} runs {self.action!r} with a hyperedge. Only new nodes and new hyperedges take a size.'
)
raise ValidationError(error_msg)
for name in ('condition', 'toward', 'score'):
value = getattr(self, name)
if value is not None and not callable(value):
error_msg = f'The {name} of {self.label!r} is {value!r}. Give a function of the namespace, or None.'
raise ValidationError(error_msg)
for request in self._get_requests():
request._validated()
def _get_requests(self) -> list[TopologyRequestSpec]:
"""Return the request of the action."""
if self.action not in ACTIONS:
raise self._get_variant_error(self.action, 'action', '_get_requests')
return [
TopologyRequestSpec(
self.action,
self.edges,
self.toward,
self.keep,
self.layer,
tuple((self.values or {}).items()),
self.hyperedge if self.action in {'add-node', 'add-hyperedge'} else None,
self.independent,
self.direction,
self.candidates,
self.score,
self.one,
self.inherit,
self.parent,
),
]
@staticmethod
def _evaluate(value: Chance, namespace: SimpleNamespace) -> object:
"""Return a number, or the value of a function of the namespace."""
return value(namespace) if callable(value) else value
def _get_laws(self, namespace: SimpleNamespace) -> list[BaseLaw]:
"""Return the requests of the action, fired at the rate or with the probability where the condition holds.
Args:
namespace:
The namespace of an edge, with `namespace.source` and `namespace.target`.
Returns:
laws:
A jump law of the request at a rate, or else an update law of the request with its probability.
"""
chance: object = 1
if self.rate is not None:
chance = self._evaluate(self.rate, namespace)
elif self.probability is not None:
chance = self._evaluate(self.probability, namespace)
if self.condition is not None:
chance = where(self.condition(namespace), chance, 0)
if self.rate is not None:
return [JumpLaw(chance, {request: 1}) for request in self._get_requests()]
return [UpdateLaw(request, chance) for request in self._get_requests()]
Equations
species.x[i]: x_species(t + 1) == -(active_species_topology - 1)*x_species(t) + active_species_topology*where(_neighbours_0*rate_growth + x_species(t) > 0, _neighbours_0*rate_growth + x_species(t), 0)/_reduced_0
species.age[i]: age_species(t + 1) == active_species_topology*(age_species(t) + 1) - (active_species_topology - 1)*age_species(t)