Activity-driven temporal networks
SIS spreading on an activity-driven temporal network, whose edges last one step.
- Name:
activity_driven_epidemics - Field:
biology/epidemiology - Formalism: discrete map
- References: Perra et al. 2012; Pastor-Satorras et al. 2015
Select this system
from skcomplex.systems import create_simulation, create_system, describe
system = create_system('activity_driven_epidemics')
entry = describe('activity_driven_epidemics')
trajectory = create_simulation('activity_driven_epidemics').prepare().run().trajectory_
Factory
@register_system(
field='biology/epidemiology',
tags=('temporal network', 'activity-driven', 'stochastic map', 'epidemic threshold'),
references=('Perra et al. 2012', 'Pastor-Satorras et al. 2015'),
defaults={'until': 150, 'dt': 1.0},
)
def activity_driven_epidemics(
n: int = 500,
*,
m: int = 2,
exponent: float = 2.2,
least: float = 0.01,
lam: float = 0.9,
mu: float = 0.1,
infected: float = 0.05,
seed: int | None = 1,
) -> System:
"""Return SIS spreading on an activity-driven temporal network, whose edges last one step.
At every step every node fires with probability its activity, drawn from `a^-exponent` above `least`, and joins
`m` random nodes. The infection travels along the edges of a step, which disappear at the next step, so it spreads
only when `lam / mu` exceeds a threshold set by the moments of the activity, `2 <a> / (m (<a> + sqrt(<a^2>)))` in
the limit of small probabilities, and not by the degrees of the network aggregated over time.
Args:
n:
The number of nodes.
m:
The number of edges an active node creates.
exponent:
The exponent of the power-law distribution of activities.
least:
The smallest activity.
lam:
The probability of infection along one edge.
mu:
The probability of recovery at a step.
infected:
The initial share of infected nodes.
seed:
The seed of the activities and the initial infections.
Returns:
system:
The system, with the states `hosts.i` and activities `hosts.a` of the group `hosts` on the nodes, and the
variable group `contacts` on the edges of every step.
Examples:
>>> from skcomplex.systems.biology.epidemiology import activity_driven_epidemics
>>> float(activity_driven_epidemics(n=10, seed=0).elements[0].values['a'].min()) >= 0.01
True
"""
rng = np.random.default_rng(seed)
activities = np.minimum(least * (1 - rng.random(n)) ** (-1 / (exponent - 1)), 1.0)
initial = (rng.random(n) < infected).astype(float)
space = NetworkSpace(nx.empty_graph(n))
hosts = GroupElement(
'hosts',
element=ActiveHostElement,
space=space,
on='sites',
variable=True,
capacity=n,
values={'i': initial, 'a': activities, 'lam': lam, 'mu': mu},
visual=VisualSpec('sites', {'states': ['i'], 'low': 0, 'high': 1}),
)
contacts = GroupElement(
'contacts',
element=Element,
space=space,
on='edges',
variable=True,
capacity=2 * m * n,
visual=HIDDEN,
)
return System(
elements=[hosts, contacts],
interactions=[
TopologyInteraction('firing', target='hosts', action='add-edge', probability=_get_activity, edges=m),
TopologyInteraction('expiry', target='hosts', action='remove-edge'),
TransitionInteraction(
'infection',
target='hosts',
state='i',
source=0,
destination=1,
probability=_get_infection,
),
TransitionInteraction(
'recovery',
target='hosts',
state='i',
source=1,
destination=0,
probability=_get_recovery,
),
],
time=DiscreteTime(),
)
Components
Elements:
ActiveHostElement, fromElementElement
Interactions:
TopologyInteractionTransitionInteraction
class ActiveHostElement(Element):
"""Node of an activity-driven network, susceptible at `i = 0` or infected at `i = 1`, in discrete time.
A `TransitionInteraction` recovers an infected node with probability `mu` in a step, and another infects a
susceptible node with probability `lam` by every infected neighbour along the edges of the previous step. A
`TopologyInteraction` joins the node to other nodes with the probability `a`, its activity.
Args:
label:
The label of the node or of the group of nodes.
visual:
How the element is drawn, or `None` for `default_visual_`.
i:
Whether the node is infected.
a:
The activity, the probability of firing at a step.
lam:
The probability of infection along one edge.
mu:
The probability of recovery at a step.
Examples:
>>> from skcomplex.systems.biology.epidemiology import ActiveHostElement
>>> ActiveHostElement('node').mu
0.1
"""
def __init__(
self,
label: str,
visual: VisualSpec | None = None,
*,
i: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 0.0,
a: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.1,
lam: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.3,
mu: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.1,
) -> None:
super().__init__(label, visual=visual)
self.i = i
self.a = a
self.lam = lam
self.mu = mu
class Element(BaseElement):
"""Base class of entities that carry state, such as particles, populations, agents, or firms.
A subclass takes its states, its parameters, and `visual` as arguments of its own `__init__`, annotating each state
with a `StateSpec` and each parameter with a `ParameterSpec` inside `Annotated`, and returns its intrinsic dynamics
from `_get_laws`, which returns no laws by default. A subclass drawn in its own way gives `visual` a default in its
signature.
An element whose laws have the agent trait, such as rules, alone or in a group, is an agent: under discrete time it
runs on the agent stepper, whose rules receive every member as an agent with its states as attributes. Its states
then hold any value, not only numbers. A member is removed only when its states tagged with a quantity are settled
to zero, and its states and parameters cannot take the names of the agent view, such as `location` or `alive`, or
of the attributes Mesa reads, such as `position` or `system`.
A subclass that connects to other elements declares its ports in the class variable `ports`, a mapping from the name
of a port to its `ConnectorSpec`, and takes one algebraic state per variable of a port, named `<port>_<variable>`.
A flow is positive into the element.
Args:
label:
The label of the element, unique among the parts of its system.
visual:
How the element is drawn, or `None` for `default_visual_`.
Attributes:
default_visual_:
The visual of the element when it sets none.
Examples:
>>> from typing import Annotated
>>> from skcomplex.core.components.elements import Element
>>> from skcomplex.core.specs import StateSpec
>>> from skcomplex.core.specs import VisualSpec
>>> from sage.all__sagemath_symbolics import oo
>>> from sage.sets.real_set import RealSet
>>> class PopulationElement(Element):
... def __init__(
... self,
... label,
... n: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 1.0,
... visual=VisualSpec('line'),
... ):
... super().__init__(label, visual=visual)
... self.n = n
>>> PopulationElement('rabbits', n=40)
PopulationElement(label='rabbits', n=40)
"""
ports: ClassVar[Mapping[str, ConnectorSpec]] = {}
@classmethod
def _get_view(cls) -> type | None:
"""Return the class of the views of the element's agents, or `None` for the engine's plain view.
An agent element names a subclass of the engine's `AgentView` that annotates its states and parameters, such as
`class WordView(AgentView): count: float`, so its rules read them with checked types.
"""
return None
def _get_laws(self, namespace: SimpleNamespace) -> Sequence[BasePart]:
"""Return the laws of the element for a namespace of symbols, none by default.
Args:
namespace:
The namespace of the symbols the laws can use.
Returns:
laws:
The laws of the element.
"""
return []
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()]
class TransitionInteraction(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 system view, `f(agent, system)`, 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, system)` 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.
Args:
label:
The label of the interaction, unique among the parts of its system.
target:
The path of the element or group whose state moves.
state:
The name of the discrete state of the target.
source:
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.
destination:
The value the state moves to.
rate:
The function of the rate of the move, or `None`.
probability:
The function of the probability of the move in a step, or `None`.
condition:
The function of the condition under which the state moves in a step, or `None`.
matrix:
The function of the transition matrix, one row and one column per value of the state, or `None`.
emission:
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`.
observed:
The name of the state of the target that the emission matrix draws, or `None`.
over:
`'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.
neighborhood:
The neighbourhood of `over='neighbors'`, or `None` for the neighbours of the network or the lattice.
after:
The number of steps a member holds the source value before it can move, or `None`.
schedule:
The order of the members of the agent form, `'synchronous'`, `'sequential'`, `'random-sequential'`, or
`'member'`.
form:
`'symbolic'` for 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 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.components.interactions import TransitionInteraction
>>> recovery = TransitionInteraction(
... 'recovery', target='hosts', state='state', source=1, destination=0, rate=lambda s: 0.5
... )
>>> recovery.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.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.destination
-1
"""
law_classes: ClassVar[tuple[type[BaseLaw], ...]] = (JumpLaw, UpdateLaw, RuleLaw, TransitionLaw, EmissionLaw)
def __init__(
self,
label: str,
*,
target: Annotated[str, RoleSpec(elements=(ElementKind.ANY,), states=('state', 'observed'))],
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
def _check_kind(self) -> str:
"""Raise an error for a transition without exactly one callable kind, or with an emission it cannot take."""
kinds = [kind for kind in KINDS if getattr(self, kind) is not None]
if len(kinds) != 1 or not callable(getattr(self, kinds[0])):
error_msg = (
f'The transition {self.label!r} has the forms {kinds}. Give exactly one of {", ".join(KINDS)}, as a '
'function of the namespace.'
)
raise ValidationError(error_msg)
unpaired = (self.emission is None) != (self.observed is None)
if unpaired or (self.emission is not None and self.matrix is None):
error_msg = (
f'The transition {self.label!r} has the emission {self.emission!r} of the state {self.observed!r}. '
'Give both with a matrix, or neither.'
)
raise ValidationError(error_msg)
return kinds[0]
def _check_form(self) -> None:
"""Raise an error for a transition without exactly one callable kind, or with arguments its kind ignores."""
kind = self._check_kind()
if self.over not in OVERS:
error_msg = f'The transition {self.label!r} runs over {self.over!r}. Use one of {", ".join(OVERS)}.'
raise ValidationError(error_msg)
if self.form not in FORMS:
error_msg = f'The form of transition {self.label!r} is {self.form!r}. Use one of {", ".join(FORMS)}.'
raise ValidationError(error_msg)
if self.schedule not in SCHEDULES or (self.form == 'symbolic' and self.schedule != 'synchronous'):
error_msg = (
f'The schedule of transition {self.label!r} is {self.schedule!r}. Use one of {", ".join(SCHEDULES)}, '
'and the agent form for another than `synchronous`.'
)
raise ValidationError(error_msg)
if self.over == 'cell' and (self.form != 'agent' or kind == 'matrix' or self.after is not None):
error_msg = (
f'The members of transition {self.label!r} act on their cell, which only the agent form does, by a '
'rate, a probability, or a condition and without a residence time.'
)
raise ValidationError(error_msg)
if self.over == 'neighbors' and (kind not in {'rate', 'probability'} or self.form == 'agent'):
error_msg = (
f'The transition {self.label!r} runs over neighbours, which only a rate or a probability of the '
'symbolic form does. In the agent form, read the neighbours of the agent in the function.'
)
raise ValidationError(error_msg)
single = not (callable(self.source) or isinstance(self.source, Collection))
if self.matrix is None and single and self.source == self.destination:
error_msg = f'The transition {self.label!r} moves from {self.source} to itself. Give distinct values.'
raise ValidationError(error_msg)
if self.after is not None and (kind not in {'probability', 'condition'} or not self.after >= 0):
error_msg = (
f'The transition {self.label!r} has the residence time {self.after!r}. Give a number of steps at least '
'0 with a probability or a condition.'
)
raise ValidationError(error_msg)
def _check_transition(self) -> None:
"""Raise an error for arguments the transition cannot take."""
self._check_form()
def _check(self) -> None:
"""Raise an error for a transition without exactly one kind, or with arguments its kind ignores.
Raises:
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 condition, a matrix, or the agent
form, the residence time is negative or goes with a rate or a matrix, the form is unknown, or a
function is not callable.
"""
super()._check()
self._check_transition()
def _get_moving(self, value: Expression, source: SourceValues) -> Expression:
"""Return 1 for a member whose state holds a source value, and 0 otherwise."""
if callable(source):
return where(SR(source(value)), 1, 0)
if isinstance(source, Collection):
return sum((where(value == item, 1, 0) for item in set(source)), SR(0))
return where(value == source, 1, 0)
def _get_jump(self, value: Expression, source: SourceValues, destination: float, rate: Expression) -> JumpLaw:
"""Return the jump of the state from a source value to the destination at a rate."""
if callable(source) or isinstance(source, Collection):
return JumpLaw(rate * self._get_moving(value, source), {value: destination - value})
return JumpLaw(rate * self._get_moving(value, source), {value: destination - source})
def _get_step(self, value: Expression, source: SourceValues, destination: float, moves: Expression) -> UpdateLaw:
"""Return the update of the state that moves a member from a source value where a condition holds."""
if callable(source) or isinstance(source, Collection):
return UpdateLaw(value, where(self._get_moving(value, source) > 0, where(moves, destination, value), value))
return UpdateLaw(value, where(value == source, where(moves, destination, source), value), source=source)
def _holds_source(self, value: float) -> bool:
"""Return whether the value of the state of an agent is a source value."""
if callable(self.source):
return bool(self.source(value))
if isinstance(self.source, Collection):
return value in set(self.source)
return value == self.source
def _combine_rates(self, rate: Expression) -> Expression:
"""Return the rate of the member, or the sum of the rates of its neighbours."""
return neighbors(rate, over=self.neighborhood) if self.over == 'neighbors' else rate
def _combine_probabilities(self, probability: Expression) -> Expression:
"""Return the probability of the member, or the probability that at least one neighbour moves it."""
if self.over == 'member':
return probability
return 1 - exp(neighbors(log(1 - probability), over=self.neighborhood))
def _get_draw(self) -> Expression:
"""Return the uniform draw of the transition in a step."""
return uniform(re.sub(r'\W', '_', f'transition_{self.label}'))
def _get_chain(self, namespace: SimpleNamespace, value: Expression) -> list[BasePart]:
"""Return the transition law of the state by its matrix, with the emission law of the observed state."""
laws: list[BasePart] = []
if self.matrix is not None:
laws.append(TransitionLaw(value, self.matrix(namespace), label=self.label))
if self.emission is not None and self.observed is not None:
observed = getattr(self._get_member(namespace, 'target'), self.observed)
laws.append(EmissionLaw(observed, value, self.emission(namespace), label=self.label))
return laws
def _get_held(self, agent: AgentViewProtocol, system: SystemViewProtocol) -> int:
"""Return the number of whole steps an agent has held its value, counted in the storage of the system."""
value = float(getattr(agent, self.state))
record = system.storage.get(f'{self.label}.{agent.id}')
held = record[1] + 1 if isinstance(record, tuple) and record[0] == value else 0
system.storage[f'{self.label}.{agent.id}'] = (value, held)
return held
def _draw_row(self, row: Sequence[object], system: SystemViewProtocol) -> float:
"""Return a value drawn by the probabilities of a row of a matrix."""
weights = np.asarray(row, dtype=float)
return float(system.rng.choice(weights.size, p=weights / weights.sum()))
def _evaluate[T](
self,
function: Callable[..., T],
agent: AgentViewProtocol,
other: AgentViewProtocol | None,
system: SystemViewProtocol,
) -> T:
"""Return a function of an agent, or of an agent and the member of its cell that acts on it."""
return function(agent, system) if other is None else function(agent, other, system)
def _moves_agent(
self,
agent: AgentViewProtocol,
system: SystemViewProtocol,
other: AgentViewProtocol | None = None,
) -> bool:
"""Return whether an agent that holds the source value moves in this step, moved by another one or alone."""
if self.condition is not None:
return bool(self._evaluate(self.condition, agent, other, system))
if self.rate is not None:
probability = 1.0 - math.exp(-float(self._evaluate(self.rate, agent, other, system)))
elif self.probability is not None:
probability = float(self._evaluate(self.probability, agent, other, system))
else:
probability = 0.0
return bool(system.rng.random() < probability)
def _get_next(self, agent: AgentViewProtocol, system: SystemViewProtocol) -> float | None:
"""Return the next value of the state of an agent, or `None` when it keeps its value."""
value = float(getattr(agent, self.state))
if self.matrix is not None:
return self._draw_row(self.matrix(agent, system)[int(value)], system)
held = self._get_held(agent, system) if self.after is not None else 0
if not self._holds_source(value) or (self.after is not None and held + 1 < self.after):
return None
return float(self.destination) if self._moves_agent(agent, system) else None
def _get_moves(self, agent: AgentViewProtocol, system: SystemViewProtocol) -> list[tuple[AgentViewProtocol, float]]:
"""Return the members an agent moves in this step, itself or the members of its cell, with their values."""
if self.over != 'cell':
value = self._get_next(agent, system)
return [] if value is None else [(agent, value)]
return [
(other, float(self.destination))
for other in agent.agents_here(agent.group)
if other is not agent
and self._holds_source(float(getattr(other, self.state)))
and self._moves_agent(other, system, agent)
]
def _apply(self, moves: list[tuple[AgentViewProtocol, float]], system: SystemViewProtocol) -> None:
"""Set the moved states, and draw the observed state of an emission."""
for agent, value in moves:
setattr(agent, self.state, value)
if self.emission is not None and self.observed is not None:
setattr(agent, self.observed, self._draw_row(self.emission(agent, system)[int(value)], system))
def _move_member(self, agent: AgentViewProtocol, system: SystemViewProtocol) -> None:
"""Move the state of one member, or of the members of its cell, from the states the members before it set."""
if agent.alive:
self._apply(self._get_moves(agent, system), system)
def move(self, system: SystemViewProtocol) -> None:
"""Move the state of every agent of the target that moves in this step, by the schedule.
Args:
system:
The system view of the interaction.
"""
agents = list(system.agents(self._get_role('target')))
if self.schedule == 'synchronous':
moves = [move for agent in agents for move in self._get_moves(agent, system)]
self._apply(moves, system)
return
order = system.rng.permutation(len(agents)) if self.schedule == 'random-sequential' else range(len(agents))
for index in order:
self._move_member(agents[int(index)], system)
def _get_laws(self, namespace: SimpleNamespace) -> list[BasePart]:
"""Return the jump, the update, the transition or emission law of the state, or the rule of the agent form.
Args:
namespace:
The namespace of the symbols the laws can use, with `namespace.target`.
Returns:
laws:
The law of the move, or the rule that moves the agents.
"""
if self.form == 'agent' and self.schedule == 'member':
return [RuleLaw(self._move_member, phase=self.phase, members=self._get_role('target'))]
if self.form == 'agent':
return [RuleLaw(self.move, phase=self.phase)]
value = getattr(self._get_member(namespace, 'target'), self.state)
if self.matrix is not None:
return self._get_chain(namespace, value)
if self.rate is not None:
rate = self._combine_rates(SR(self.rate(namespace)))
return [self._get_jump(value, self.source, self.destination, rate)]
if self.probability is not None:
probability = self._combine_probabilities(SR(self.probability(namespace)))
moves = self._get_draw() < probability
else:
moves = SR(self.condition(namespace)) if self.condition is not None else SR(0)
if self.after is not None:
moves = both(residence(value) + 1 >= self.after, moves)
return [self._get_step(value, self.source, self.destination, moves)]
Equations