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
Factory
@register_model(
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 model, 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.models.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
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.models.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 `model`.
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.elements import Element
>>> from skcomplex.specs import StateSpec
>>> from skcomplex.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)
"""
@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. A new node joins `edges` other nodes, a node adding edges joins `edges`
other nodes, and a rewiring keeps the end `keep` of its edge. New ends are drawn with probability proportional to
`toward(node)`, a function of the namespace of a candidate node, or uniformly. A new node or edge takes the
`values`, each a value or a function of the random number generator that draws it. The groups on the nodes and the
edges of the network are variable. With `hyperedge`, a new node instead forms a hyperedge of that size with
`hyperedge - 1` existing nodes and joins each of them by an edge: the first two are the ends of an edge drawn
uniformly, so a size of three attaches a triangle along an edge, and further members are drawn by `toward`. The
new hyperedge is a new member of the variable group on the hyperedges of the network.
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, or `None` for an addition of one node per step.
action:
The change, `'add-node'`, `'remove-node'`, `'replace-node'`, `'add-edge'`, `'remove-edge'`, or
`'rewire'`.
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, 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 forms with existing nodes, or `None` for none.
Raises:
ValidationError:
The action or the kept end is unknown, both a rate and a probability are given, the number of edges is
not a whole number, an action other than `'add-node'` has no target, or a hyperedge is given to an action
other than `'add-node'` or has fewer than two members, raised by `check`.
Examples:
>>> import networkx as nx
>>> from skcomplex.core.elements import Element, GroupElement
>>> from skcomplex.core.interactions import TopologyInteraction
>>> from skcomplex.core.spaces import NetworkSpace
>>> from skcomplex.core.system import System
>>> from skcomplex.engine.runtime import Simulation
>>> nodes = GroupElement('nodes', element=Element, on='sites', variable=True, capacity=20)
>>> links = GroupElement('links', element=Element, on='edges', variable=True, capacity=20)
>>> expiry = TopologyInteraction('expiry', target='nodes', action='remove-edge')
>>> system = System(elements=[nodes, links], interactions=[expiry], space=NetworkSpace(nx.cycle_graph(5)))
>>> trajectory = Simulation(system, random_state=0).check().prepare().run(until=1).trajectory_
>>> len(trajectory.network(0).sources), len(trajectory.network(1).sources)
(5, 0)
A network that grows by triangles from one triangle:
>>> import xgi
>>> from skcomplex.core.spaces import DiscreteTime
>>> triangles = GroupElement('triangles', element=Element, on='hyperedges', variable=True, capacity=8)
>>> growth = TopologyInteraction('growth', action='add-node', hyperedge=3)
>>> space = NetworkSpace(nx.complete_graph(3), hyperedges=[[0, 1, 2]])
>>> system = System(
... elements=[nodes, links, triangles], interactions=[growth], space=space, time=DiscreteTime()
... )
>>> trajectory = Simulation(system, random_state=0).check().prepare().run(until=5).trajectory_
>>> trajectory.network(-1).hyperedges.shape
(6, 3)
"""
def __init__(
self,
label: str,
*,
target: Annotated[str | None, RoleSpec(optional=True)] = None,
action: str = 'add-edge',
rate: Chance | None = None,
probability: Chance | None = None,
condition: Callable[[SimpleNamespace], object] | None = None,
toward: Callable[[SimpleNamespace], object] | None = None,
edges: int = 1,
keep: str = 'source',
values: Mapping[str, object] | None = None,
layer: str | None = None,
hyperedge: int | None = None,
) -> None:
super().__init__(label, layer=layer)
self.target = target
self.action = action
self.rate = rate
self.probability = probability
self.condition = condition
self.toward = toward
self.edges = edges
self.keep = keep
self.values = values
self.hyperedge = hyperedge
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 or a
weight cannot be called, the number of edges is not a whole number of at least 0, an action other
than `'add-node'` has no target, or a hyperedge is given to an action other than `'add-node'`.
"""
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 != 'add-node':
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 != 'add-node':
error_msg = f'{self.label!r} runs {self.action!r} with a hyperedge. Only new nodes form hyperedges.'
raise ValidationError(error_msg)
for name in ('condition', 'toward'):
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.check()
def _get_requests(self) -> list[TopologyRequestSpec]:
"""Return the requests of the action, two for a replacement of a node."""
if self.action not in ACTIONS:
raise self._get_variant_error(self.action, 'action', '_get_requests')
values = tuple((self.values or {}).items())
kinds = ['add-node', 'remove-node'] if self.action == 'replace-node' else [self.action]
return [
TopologyRequestSpec(
kind,
self.edges,
self.toward,
self.keep,
self.layer,
values,
self.hyperedge if kind == 'add-node' else None,
)
for kind in kinds
]
@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 per request at a rate, or else an update law per 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 model view, `f(agent, model)`, and returns a number, a truth value, or the rows of a matrix, and a rate `r`
moves an agent with probability `1 - exp(-r)` in a step. With `over='cell'`, every member acts on the other members
of its cell: it moves each one that holds the source value with the probability, the rate, or the condition
`f(agent, other, model)` of that member `agent` and the acting member `other`, each by its own draw.
The `schedule` of the agent form orders the members. Under `'synchronous'` every member decides from the states at
the start of the step, under `'sequential'` the members act one after the other in their order, and under
`'random-sequential'` in a new random order every step, each reading the values the members before it set. Under
`'member'` every member acts in its own activation, together with the rules of its element and the other member
interactions, in the order the system lists them, so one rule of a member runs its movement, transfers,
transitions, and life cycle.
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.interactions import TransitionInteraction
>>> recovery = TransitionInteraction(
... 'recovery', target='hosts', state='state', source=1, destination=0, rate=lambda s: 0.5
... )
>>> recovery.check().over
'member'
>>> agents = TransitionInteraction(
... 'recovery', target='hosts', state='state', source=1, destination=2, probability=lambda a, m: 0.1,
... after=3, form='agent'
... )
>>> type(agents.check().prepare().run(SimpleNamespace()).laws_[0]).__name__
'RuleLaw'
>>> separation = TransitionInteraction(
... 'separation', target='workers', state='employer', source=lambda value: value >= 0, destination=-1,
... probability=lambda a, m: 0.05, form='agent'
... )
>>> separation.check().destination
-1
"""
def __init__(
self,
label: str,
*,
target: Annotated[str, RoleSpec()],
state: str = 'state',
source: SourceValues = 0.0,
destination: float = 1.0,
rate: Callable[..., SupportsFloat] | None = None,
probability: Callable[..., SupportsFloat] | None = None,
condition: Callable[..., object] | None = None,
matrix: Callable[..., Sequence[Sequence[object]]] | None = None,
emission: Callable[..., Sequence[Sequence[object]]] | None = None,
observed: str | None = None,
over: str = 'member',
neighborhood: BaseNeighborhood | None = None,
after: float | None = None,
schedule: str = 'synchronous',
form: str = 'symbolic',
phase: str | None = None,
) -> None:
super().__init__(label)
self.target = target
self.state = state
self.source = source
self.destination = destination
self.rate = rate
self.probability = probability
self.condition = condition
self.matrix = matrix
self.emission = emission
self.observed = observed
self.over = over
self.neighborhood = neighborhood
self.after = after
self.schedule = schedule
self.form = form
self.phase = phase
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, model: ModelViewProtocol) -> int:
"""Return the number of whole steps an agent has held its value, counted in the storage of the model."""
value = float(getattr(agent, self.state))
record = model.storage.get(f'{self.label}.{agent.id}')
held = record[1] + 1 if isinstance(record, tuple) and record[0] == value else 0
model.storage[f'{self.label}.{agent.id}'] = (value, held)
return held
def _draw_row(self, row: Sequence[object], model: ModelViewProtocol) -> float:
"""Return a value drawn by the probabilities of a row of a matrix."""
weights = np.asarray(row, dtype=float)
return float(model.rng.choice(weights.size, p=weights / weights.sum()))
def _evaluate[T](
self,
function: Callable[..., T],
agent: AgentViewProtocol,
other: AgentViewProtocol | None,
model: ModelViewProtocol,
) -> T:
"""Return a function of an agent, or of an agent and the member of its cell that acts on it."""
return function(agent, model) if other is None else function(agent, other, model)
def _moves_agent(
self,
agent: AgentViewProtocol,
model: ModelViewProtocol,
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, model))
if self.rate is not None:
probability = 1.0 - math.exp(-float(self._evaluate(self.rate, agent, other, model)))
elif self.probability is not None:
probability = float(self._evaluate(self.probability, agent, other, model))
else:
probability = 0.0
return bool(model.rng.random() < probability)
def _get_next(self, agent: AgentViewProtocol, model: ModelViewProtocol) -> 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, model)[int(value)], model)
held = self._get_held(agent, model) 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, model) else None
def _get_moves(self, agent: AgentViewProtocol, model: ModelViewProtocol) -> 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, model)
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, model, agent)
]
def _apply(self, moves: list[tuple[AgentViewProtocol, float]], model: ModelViewProtocol) -> 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, model)[int(value)], model))
def _move_member(self, agent: AgentViewProtocol, model: ModelViewProtocol) -> 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, model), model)
def move(self, model: ModelViewProtocol) -> None:
"""Move the state of every agent of the target that moves in this step, by the schedule.
Args:
model:
The model view of the interaction.
"""
agents = list(model.agents(self._get_role('target')))
if self.schedule == 'synchronous':
moves = [move for agent in agents for move in self._get_moves(agent, model)]
self._apply(moves, model)
return
order = model.rng.permutation(len(agents)) if self.schedule == 'random-sequential' else range(len(agents))
for index in order:
self._move_member(agents[int(index)], model)
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