Tragedy of the meadow commons
Herders whose cows graze a shared meadow, each adding cows while a cow pays, until the grass collapses.
- Name:
tragedy_of_the_commons - Field:
economics/resources - Formalism: agent-based model
- References: Hamill and Gilbert 2016 sec. 10.2; Hardin 1968
Factory
@register_model(
field='economics/resources',
tags=('commons', 'agents', 'overgrazing', 'regulation', 'grid'),
references=('Hamill and Gilbert 2016 sec. 10.2', 'Hardin 1968'),
defaults={'until': 100},
observables=_commons_observables,
)
def tragedy_of_the_commons(
n_herders: int = 20,
*,
width: int = 20,
height: int = 20,
cows: int = 1,
quota: float | None = None,
regrowth: float = 0.2,
capacity: float = 2.0,
upkeep: float = 0.3,
random_state: int | np.random.Generator | None = 0,
) -> System:
"""Return herders whose cows graze a shared meadow, each adding cows while a cow pays, until the grass collapses.
Every step every cow moves to the richest pasture it can reach, eats, and earns its herder milk, and the grass
grows back logistically. Every herder adds a cow whenever a cow earned more than its upkeep, since the gain of a
cow goes to its herder while the cost of the grass it eats falls on all. With more herders than the meadow
carries, the herd grows past what the grass sustains, the grass collapses below a tenth of its capacity, and the
cows starve, so the herd shrinks after it. A regulator with an adaptive quota of cows per herder, lowered when the
grass runs short and raised when it recovers, keeps the meadow from collapse (Hamill and Gilbert 2016 sec. 10.2).
Args:
n_herders:
The number of herders.
width:
The number of columns of the meadow.
height:
The number of rows of the meadow.
cows:
The initial number of cows of every herder.
quota:
The initial quota of cows per herder of a regulator that adapts it, or `None` for no regulation.
regrowth:
The logistic growth rate of the grass.
capacity:
The most grass a cell holds.
upkeep:
The cost of keeping a cow for a step.
random_state:
The seed or generator of the cells of the cows.
Returns:
system:
The model, with the groups `meadow`, `cows`, and `herders`, whose adaptive parameter `quota` the regulator
`regulator` changes, and the grazing `grazing`.
Examples:
>>> from skcomplex.models.economics.resources import tragedy_of_the_commons
>>> [group.size for group in tragedy_of_the_commons(4, width=5, height=5).elements]
[25, 4, 4]
"""
rng = np.random.default_rng(random_state)
cells = [(x, y) for x in range(width) for y in range(height)]
meadow = GroupElement(
'meadow',
element=PastureElement,
size=len(cells),
values={'grass': capacity, 'capacity': capacity, 'regrowth': regrowth},
location=cells,
)
herders = GroupElement(
'herders',
element=HerderElement,
size=n_herders,
values={
'upkeep': upkeep,
'quota': UNLIMITED if quota is None else quota,
},
location=[cells[0]] * n_herders,
)
n_cows = n_herders * cows
herd = GroupElement(
'cows',
element=CowElement,
size=n_cows,
values={'owner': np.repeat(np.arange(n_herders, dtype=float), cows)},
variable=True,
location=[cells[index] for index in rng.integers(len(cells), size=n_cows)],
)
regulator = QuotaRegulatorInteraction('regulator', phase='herding', regulated=float(quota is not None))
grazing = GrazingInteraction(
'grazing',
requesters='cows',
providers='meadow',
rule=PartnerSpec('local', 'score', score=GrazingInteraction.score, clearing='provider'),
herders='herders',
)
return System(
elements=[meadow, herd, herders],
interactions=[grazing, *_get_investment(), regulator],
space=LatticeSpace((width, height), boundary='periodic', neighborhood='moore'),
time=DiscreteTime(
schedule=PhasesScheduleSpec([PhaseSpec('grazing'), PhaseSpec('growth'), PhaseSpec('herding')]),
),
)
Components
class PastureElement(Element):
"""Cell of the meadow whose grass grows back logistically, from a seed bank when it is grazed bare.
Args:
label:
The label of the agent or group.
visual:
How the element is drawn, or `None` for `default_visual_`.
grass:
The grass on the cell.
capacity:
The most grass the cell holds.
regrowth:
The logistic growth rate of the grass.
seed:
The share of the capacity that grows back as if it stood on a bare cell.
Examples:
>>> from skcomplex.core.elements import GroupElement
>>> from skcomplex.models.economics.resources import PastureElement
>>> GroupElement('meadow', element=PastureElement, size=4, values={'capacity': 2.0}).size
4
"""
def __init__(
self,
label: str,
visual: VisualSpec | None = VisualSpec('agents', {'color_by': 'grass', 'color_range': (0.0, 2.0)}),
*,
grass: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 2.0,
capacity: Annotated[float, StateSpec(domain=RealSet.open(0, oo))] = 2.0,
regrowth: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.2,
seed: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.05,
) -> None:
super().__init__(label, visual=visual)
self.grass = grass
self.capacity = capacity
self.regrowth = regrowth
self.seed = seed
@classmethod
def _get_view(cls) -> type[PastureView]:
"""Return the view of the agents of the element."""
return PastureView
def _get_laws(self, s: SimpleNamespace) -> list[BaseLaw]:
"""Return the regrowth rule of the phase of growth.
Args:
s:
The namespace of the symbols the laws can use.
Returns:
laws:
The laws of the component.
"""
return [RuleLaw(self.grow, phase='growth')]
def grow(self, agent: PastureView, model: ModelView) -> None:
"""Grow the grass logistically up to the capacity.
Args:
agent:
The pasture.
model:
The model.
"""
stand = max(agent.grass, model.parameter('seed') * agent.capacity)
growth = model.parameter('regrowth') * stand * (1 - stand / agent.capacity)
agent.grass = min(agent.capacity, agent.grass + growth)
class CowElement(Element):
"""Cow of a herder that gains the energy of the grass it eats, burns its metabolism, and starves without food.
Args:
label:
The label of the agent or group.
visual:
How the element is drawn, or `None` for `default_visual_`.
owner:
The id of the herder that owns the cow, given initially as its position in the group `herders`.
energy:
The energy of the cow.
Examples:
>>> from skcomplex.core.elements import GroupElement
>>> from skcomplex.models.economics.resources import CowElement
>>> GroupElement('cows', element=CowElement, size=3, values={'energy': 4.0}).size
3
"""
def __init__(
self,
label: str,
visual: VisualSpec | None = VisualSpec('agents', {'color_by': 'energy'}),
*,
owner: Annotated[float, StateSpec(reference='herders')] = -1.0,
energy: Annotated[float, StateSpec()] = 4.0,
) -> None:
super().__init__(label, visual=visual)
self.owner = owner
self.energy = energy
@classmethod
def _get_view(cls) -> type[CowView]:
"""Return the view of the agents of the element."""
return CowView
def _get_laws(self, s: SimpleNamespace) -> list[BaseLaw]:
"""Return no laws, since the grazing moves the cows and their energy.
Args:
s:
The namespace of the symbols the laws can use.
Returns:
laws:
The laws of the component.
"""
return []
class HerderElement(Element):
"""Herder who adds a cow whenever a cow earned more than its upkeep, and sells one when the herd exceeds its quota.
The life cycle `investment` of the model buys a cow, at a random cell, or sells the first cow the herder owns,
and two maps clear the income and the herd the grazing tallies, all member interactions in the phase of herding.
Args:
label:
The label of the agent or group.
visual:
How the element is drawn, or `None` for `default_visual_`.
income:
The milk income of the herder in the last step.
herd:
The number of cows of the herder in the last step.
upkeep:
The cost of keeping a cow for a step.
newborn:
The energy of a new cow.
quota:
The largest herd of a herder, an adaptive parameter that a regulator may change.
Examples:
>>> from skcomplex.core.elements import GroupElement
>>> from skcomplex.models.economics.resources import HerderElement
>>> GroupElement('herders', element=HerderElement, size=5).size
5
"""
def __init__(
self,
label: str,
visual: VisualSpec | None = None,
*,
income: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 0.0,
herd: Annotated[float, StateSpec(domain=RealSet.closed_open(0, oo))] = 0.0,
upkeep: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.3,
newborn: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo))] = 4.0,
quota: Annotated[float, ParameterSpec(domain=RealSet.open(0, oo), adaptive=True)] = UNLIMITED,
) -> None:
super().__init__(label, visual=visual)
self.income = income
self.herd = herd
self.upkeep = upkeep
self.newborn = newborn
self.quota = quota
@classmethod
def _get_view(cls) -> type[HerderView]:
"""Return the view of the agents of the element."""
return HerderView
class GrazingInteraction(MatchingInteraction[ModelView]):
"""Grazing of the meadow, a matching of the cows with the pastures, after which every cow eats and burns energy.
The rule of the matching pairs the cows with the pastures, such as
`PartnerSpec('local', 'score', score=GrazingInteraction.score, clearing='provider')`, which pairs every cow, in
random order, with the free pasture of most grass in its cell or the cells around it, ties broken at random. A
pasture takes one cow, its default quantity of one unit. Every matched cow moves to its pasture and eats up to its
appetite as a transfer of grass to its energy, and its herder earns milk for what it ate. Then every cow burns
its metabolism and counts in the herd of its herder, and a cow whose energy runs out dies.
Args:
label:
The label of the grazing.
requesters:
The path of the group of cows.
providers:
The path of the group of pastures.
rule:
The spec of the algorithm that pairs the cows with the pastures.
phase:
The phase the grazing runs in.
herders:
The path of the group of herders.
appetite:
The most grass a cow eats in a step.
metabolism:
The energy a cow burns in a step.
milk:
The income of a herder per unit of grass its cows eat.
Examples:
>>> from skcomplex.models.economics.resources import GrazingInteraction
>>> GrazingInteraction('grazing', requesters='cows', providers='meadow', herders='herders').appetite
1.0
"""
def __init__(
self,
label: str,
*,
requesters: Annotated[str, RoleSpec()],
providers: Annotated[str, RoleSpec()],
rule: MatchingRule = RandomSearchSpec(),
phase: str | None = 'grazing',
herders: Annotated[str, RoleSpec()],
appetite: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 1.0,
metabolism: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.5,
milk: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 1.0,
) -> None:
super().__init__(label, requesters=requesters, providers=providers, rule=rule, phase=phase)
self.herders = herders
self.appetite = appetite
self.metabolism = metabolism
self.milk = milk
@staticmethod
def score(cow: object, pasture: object) -> float:
"""Return the score of a pasture for a cow, the grass on the pasture.
Args:
cow:
The agent view of the cow.
pasture:
The agent view of the pasture.
Returns:
score:
The grass on the pasture, or minus infinity for an agent that is not a pasture.
"""
del cow
return float(pasture.grass) if isinstance(pasture, PastureView) else -np.inf
@staticmethod
def _get_herder(cow: CowView) -> HerderView:
"""Return the herder that owns a cow, which its reference state `owner` names."""
herder = cow.referenced('owner')
if not isinstance(herder, HerderView):
error_msg = f'Cow {cow.id} names no herder as its owner.'
raise SimulationError(error_msg)
return herder
def _settle(self, match: MatchSpec, model: ModelView) -> None:
"""Move the cow of a match to its pasture, feed it, and pay its herder for what it ate.
Args:
match:
The match of a cow with a pasture.
model:
The model view of the grazing.
"""
cow, pasture = match.requester, match.provider
if not isinstance(cow, CowView) or not isinstance(pasture, PastureView):
return
cow.move_to(pasture.location)
eaten = min(pasture.grass, model.parameter('appetite'))
transfer(pasture, cow, eaten, 'grass', 'energy')
self._get_herder(cow).income += model.parameter('milk') * eaten
def match(self, model: ModelView) -> None:
"""Pair and feed the cows, then let every cow burn its metabolism, count it in its herd, and starve.
Args:
model:
The model view of the grazing.
"""
super().match(model)
for cow in model.agents('requesters', CowView):
cow.energy -= model.parameter('metabolism')
self._get_herder(cow).herd += 1
if cow.energy < 0:
cow.remove()
class LifeCycleInteraction(Interaction):
"""Life cycle of the agents of a group, which give birth to offspring and die, every step of discrete time.
Every step the agents act once each, in their order under the schedule `'sequential'`, in a new random order under
`'random-sequential'`, or each in its own activation under `'member'`, together with the rules of its element and
the other member interactions, in the order the system lists them, without the reports of the step and without
arrivals. An agent dies with the probability `death`, and an agent that survives gives birth with the probability
`birth`, or as many times as a Poisson number of the mean `rate`. A probability is a number or a function of the
agent and the model view, and a function that returns a condition gives probability one where it holds. Every birth
creates `offspring` new agents, a number or a function of the parent and the model view, such as a Poisson draw, of
the group `group`, or of the parent's group without one. A new agent stands in the cell of its parent, with the
placement `'here'`, in a neighbouring cell that no agent of the group holds, with the placement `'neighbor'`, or at
the location a function of the parent and the model view returns, where `None` means no place, and a birth without a
place creates fewer agents.
A new agent copies the states `inherit` of its parent and takes the values that `values(parent, model)` returns,
such as a generation one higher or drawn states. The parent splits each state of `split` with its offspring,
keeping the share `keep`, a number or a function of the parent and the model view such as a random cut, or else an
equal share, and the offspring share the rest equally. Then `mutate(values, model)` returns the values of a new
agent from those it would get, and the other states take their defaults. A parent that gave birth takes the values
`reset`, and with `replace` it is removed, so its offspring replace it, also when the birth created none. With
`renew` an agent that dies is renewed in place instead of removed: it keeps its identity, its place, and the states
the new values leave out, and takes the values of an offspring of its own. With the probability `cull` an agent
that does not die removes, after its births, one member of `group`, or of the target without a group, for which
`kin(agent, member)` holds, the first in the order of the group under the pick `'first'`, such as one it owns, or
a random one under `'random'`. Every
step a Poisson number of the mean `arrival`, a number or a function of the model view, of new agents arrive without
a parent, at the location of the placement function or else at a random place, with the values
`values(None, model)`. The life cycle reports the `births`, `deaths`, and `arrivals` of every step. It is an agent
rule, so the system runs in discrete time.
Args:
label:
The label of the interaction, unique among the parts of its system.
target:
The path of the variable group whose agents give birth and die.
birth:
The probability that an agent gives birth in a step, or `None` for no births by probability.
death:
The probability that an agent dies in a step, or `None` for no deaths.
rate:
The mean number of births of an agent in a step, a number or a function, or `None` for no births by rate.
offspring:
The number of new agents of a birth.
placement:
Where a new agent stands, `'here'`, `'neighbor'`, or a function of the parent, `None` for an arrival, and
the model view that returns its location.
inherit:
The names of the states a new agent copies from its parent.
split:
The names of the states the parent shares with its offspring.
mutate:
A function of the values of a new agent and the model view that returns its values, or `None`.
schedule:
The order of the agents, `'sequential'`, `'random-sequential'`, or `'member'`.
phase:
The phase of the step the life cycle runs in, or `None` for the default phase.
group:
The path of the variable group the new agents join, or `None` for the group of the target.
values:
A function of the parent, `None` for an arrival, and the model view that returns values of a new agent by
state name, or `None`.
keep:
The share of every split state the parent keeps, a number or a function, or `None` for an equal share.
reset:
The values a parent takes after it gives birth, by state name, or `None`.
replace:
Whether a parent that gives birth is removed.
renew:
Whether an agent that dies is renewed in place instead of removed.
cull:
The probability that an agent removes one member of `group`, or of the target without a group, in a
step, or `None` for no removals.
kin:
A function of an agent and a member of `group` that returns whether the agent may remove the member, or
`None` for every member.
arrival:
The mean number of new agents that arrive in a step without a parent, or `None` for no arrivals.
pick:
Which member of `group` a removal takes among those it may remove, `'first'` or `'random'`.
Raises:
ValidationError:
The placement, the schedule, or the pick is unknown, both a birth probability and a rate are given, a
function is not callable, or the schedule `'member'` has arrivals, raised by `check`.
Examples:
>>> from typing import Annotated
>>> from skcomplex.core.elements import Element, GroupElement
>>> from skcomplex.core.interactions import LifeCycleInteraction
>>> from skcomplex.core.spaces import LatticeSpace
>>> from skcomplex.core.system import System
>>> from skcomplex.engine.runtime import Simulation
>>> from skcomplex.specs import StateSpec
>>> class CellElement(Element):
... def __init__(self, label, visual=None, *, size: Annotated[float, StateSpec()] = 0.0):
... super().__init__(label, visual=visual)
... self.size = size
>>> cells = GroupElement('cells', element=CellElement, size=1, values={'size': 8.0}, variable=True)
>>> division = LifeCycleInteraction('division', target='cells', birth=1.0, split=('size',))
>>> system = System(elements=[cells], interactions=[division], space=LatticeSpace((3, 3)))
>>> trajectory = Simulation(system, random_state=0).check().prepare().run(until=2).trajectory_
>>> trajectory.agents('cells', 2).values['size'].tolist()
[2.0, 2.0, 2.0, 2.0]
"""
reports: ClassVar[tuple[str, ...]] = ('births', 'deaths', 'arrivals')
def __init__(
self,
label: str,
*,
target: Annotated[str, RoleSpec()],
birth: Chance | None = None,
death: Chance | None = None,
rate: Chance | None = None,
offspring: Count = 1,
placement: Placement = 'here',
inherit: Sequence[str] = (),
split: Sequence[str] = (),
mutate: Mutation | None = None,
schedule: str = 'random-sequential',
phase: str | None = None,
group: Annotated[str | None, RoleSpec(optional=True)] = None,
values: Values | None = None,
keep: Chance | None = None,
reset: Mapping[str, object] | None = None,
replace: bool = False,
renew: bool = False,
cull: Chance | None = None,
kin: Kinship | None = None,
arrival: Arrival | None = None,
pick: str = 'first',
) -> None:
super().__init__(label)
self.target = target
self.birth = birth
self.death = death
self.rate = rate
self.offspring = offspring
self.placement = placement
self.inherit = inherit
self.split = split
self.mutate = mutate
self.schedule = schedule
self.phase = phase
self.group = group
self.values = values
self.keep = keep
self.reset = reset
self.replace = replace
self.renew = renew
self.cull = cull
self.kin = kin
self.arrival = arrival
self.pick = pick
def _check_variants(self) -> None:
"""Raise an error for an unknown placement, schedule, or pick, or arrivals of a member schedule."""
self._check_variant('placement', self.placement, PLACEMENTS, functions=True)
self._check_variant('schedule', self.schedule, SCHEDULES)
self._check_variant('pick', self.pick, PICKS)
if self.schedule == 'member' and self.arrival is not None:
error_msg = f'{self.label!r} runs member by member, so no agent brings arrivals. Give `arrival=None`.'
raise ValidationError(error_msg)
def _check_space(self, space: BaseSpace | None) -> None:
"""Raise an error for a placement in a neighbouring cell or node in a continuous space."""
if self.placement == 'neighbor':
self._check_grid('placement', self.placement, space, "placement='here', or a placement function")
def _check(self) -> None:
"""Raise an error for an unknown placement, a birth by both probability and rate, or a bad function.
Raises:
ValidationError:
The placement, the schedule, or the pick is unknown, both a birth probability and a rate are given,
a value is neither a number nor a function, or the schedule `'member'` has arrivals.
"""
super()._check()
self._check_variants()
if self.birth is not None and self.rate is not None:
error_msg = f'{self.label!r} gives births by a probability and by a rate. Give one of them.'
raise ValidationError(error_msg)
for name in ('birth', 'death', 'rate', 'offspring', 'keep', 'cull', 'arrival'):
value = getattr(self, name)
if value is not None and not callable(value) and not isinstance(value, int | float):
error_msg = f'The {name} of {self.label!r} is {value!r}. Give a number or a function.'
raise ValidationError(error_msg)
for name in ('mutate', 'values', 'kin'):
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, or None.'
raise ValidationError(error_msg)
@staticmethod
def _evaluate(value: Chance | Count, agent: AgentViewProtocol, model: ModelViewProtocol) -> float:
"""Return a number, or the value of a function of an agent and the model view, as a float."""
return float(value(agent, model) if callable(value) else value)
def _happens(self, chance: Chance | None, agent: AgentViewProtocol, model: ModelViewProtocol) -> bool:
"""Return whether an event of a probability happens to an agent in this step."""
if chance is None:
return False
probability = self._evaluate(chance, agent, model)
return probability >= 1.0 or (probability > 0.0 and model.rng.random() < probability)
def _count_births(self, agent: AgentViewProtocol, model: ModelViewProtocol) -> int:
"""Return the number of births of a parent in this step."""
if self.rate is not None:
return int(model.rng.poisson(self._evaluate(self.rate, agent, model)))
return int(self._happens(self.birth, agent, model))
def _get_places(
self,
agent: AgentViewProtocol,
count: int,
occupied: set[object],
model: ModelViewProtocol,
) -> list[object]:
"""Return the locations of the new agents of a parent, fewer when places run out."""
if callable(self.placement):
places = [self.placement(agent, model) for _ in range(count)]
return [place for place in places if place is not None]
if self.placement == 'here':
return [agent.location] * count
free = [cell for cell in agent.neighbor_cells() if cell not in occupied]
chosen = [free[int(index)] for index in model.rng.permutation(len(free))[:count]]
occupied.update(chosen)
return chosen
def _get_values(self, agent: AgentViewProtocol | None, model: ModelViewProtocol) -> dict[str, object]:
"""Return the inherited and given values of a new agent of a parent, or of an arrival."""
values: dict[str, object] = {} if agent is None else {name: getattr(agent, name) for name in self.inherit}
if self.values is not None:
values.update(self.values(agent, model))
return values
def _create(
self,
agent: AgentViewProtocol | None,
place: object,
values: dict[str, object],
model: ModelViewProtocol,
) -> None:
"""Create a new agent with mutated values at a place, `None` for a random one, in the group of the cycle."""
if self.mutate is not None:
values = self.mutate(values, model)
if self.group is not None:
model.create(self._get_role('group'), place, **values)
elif agent is not None:
agent.create(None, place, **values)
else:
model.create(self._get_role('target'), place, **values)
def _bear(self, agent: AgentViewProtocol, places: list[object], model: ModelViewProtocol) -> None:
"""Create the new agents of a parent at their places, with inherited, given, split, and mutated values."""
kept = 1 / (len(places) + 1) if self.keep is None else self._evaluate(self.keep, agent, model)
totals = {name: float(getattr(agent, name)) for name in self.split}
for name, amount in totals.items():
setattr(agent, name, amount * kept)
shares = {name: (amount - amount * kept) / max(len(places), 1) for name, amount in totals.items()}
for place in places:
values = self._get_values(agent, model)
values.update(shares)
self._create(agent, place, values, model)
def _renew(self, agent: AgentViewProtocol, model: ModelViewProtocol) -> None:
"""Give an agent that dies the values of an offspring of its own, in place."""
values = self._get_values(agent, model)
if self.mutate is not None:
values = self.mutate(values, model)
for name, value in values.items():
setattr(agent, name, value)
def _remove_kin(self, agent: AgentViewProtocol, model: ModelViewProtocol) -> None:
"""Remove the first or a random member of the group of the cycle that is kin of an agent, if there is one."""
kin = self.kin
group = self._get_role('target' if self.group is None else 'group')
members = (item for item in model.agents(group) if item.alive and (kin is None or kin(agent, item)))
if self.pick == 'random':
candidates = list(members)
if candidates:
candidates[int(model.rng.integers(len(candidates)))].remove()
return
member = next(members, None)
if member is not None:
member.remove()
def _die(self, agent: AgentViewProtocol, occupied: set[object], model: ModelViewProtocol) -> None:
"""Renew an agent in place, or remove it and free its cell."""
if self.renew:
self._renew(agent, model)
return
occupied.discard(agent.location_key)
agent.remove()
def _act(self, agent: AgentViewProtocol, occupied: set[object], model: ModelViewProtocol) -> tuple[int, int]:
"""Let an agent die, or give birth and remove kin, and return its numbers of new agents and deaths."""
if self._happens(self.death, agent, model):
self._die(agent, occupied, model)
return 0, 1
born, died = self._give_birth(agent, occupied, model)
if self._happens(self.cull, agent, model):
self._remove_kin(agent, model)
return born, died
def _give_birth(self, agent: AgentViewProtocol, occupied: set[object], model: ModelViewProtocol) -> tuple[int, int]:
"""Let an agent give birth, and return its number of new agents and whether it was replaced."""
born = 0
events = self._count_births(agent, model)
if events > 0:
births = events * int(self._evaluate(self.offspring, agent, model))
places = self._get_places(agent, births, occupied, model) if births else []
if places or self.split:
self._bear(agent, places, model)
born = len(places)
for name, value in (self.reset or {}).items():
setattr(agent, name, value)
if self.replace:
occupied.discard(agent.location_key)
agent.remove()
return born, 1
return born, 0
def _arrive(self, model: ModelViewProtocol) -> int:
"""Create the agents that arrive without a parent in this step, and return their number."""
if self.arrival is None:
return 0
mean = self.arrival(model) if callable(self.arrival) else self.arrival
count = int(model.rng.poisson(float(mean)))
for _ in range(count):
place = self.placement(None, model) if callable(self.placement) else None
self._create(None, place, self._get_values(None, model), model)
return count
def _live_member(self, agent: AgentViewProtocol, model: ModelViewProtocol) -> None:
"""Run the life cycle of one agent in its own activation."""
if not agent.alive:
return
neighbor = self.placement == 'neighbor'
occupied = {other.location_key for other in model.agents(self._get_role('target'))} if neighbor else set()
self._act(agent, occupied, model)
def _live(self, model: ModelViewProtocol) -> None:
"""Remove or renew the agents that die and create the offspring of those that give birth, by the schedule."""
agents = model.agents(self._get_role('target'))
occupied = {agent.location_key for agent in agents} if self.placement == 'neighbor' else set()
births = deaths = 0
order = model.rng.permutation(len(agents)) if self.schedule == 'random-sequential' else range(len(agents))
for index in order:
agent = agents[int(index)]
if not agent.alive:
continue
born, died = self._act(agent, occupied, model)
births, deaths = births + born, deaths + died
model.report('births', births)
model.report('deaths', deaths)
model.report('arrivals', self._arrive(model))
def _get_laws(self, namespace: SimpleNamespace) -> list[BaseLaw]:
"""Return the rule of the life cycle.
Args:
namespace:
The namespace of the symbols the laws can use.
Returns:
laws:
The rule that removes and creates agents every step.
"""
if self.schedule == 'member':
return [RuleLaw(self._live_member, phase=self.phase, members=self._get_role('target'))]
return [RuleLaw(self._live, phase=self.phase)]
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, model)` returns the value as a number, or as an array of the
shape of a vector state, so it can read the agent, its neighbours, and the history and storage of the model view.
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, model)` 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.elements import Element
>>> from skcomplex.core.interactions import MapInteraction
>>> from skcomplex.core.spaces import DiscreteTime
>>> from skcomplex.core.system import System
>>> from skcomplex.engine.runtime import Simulation
>>> from skcomplex.specs import StateSpec
>>> from typing import Annotated, SupportsFloat
>>> class ValueElement(Element):
... def __init__(self, label, x: Annotated[float, StateSpec()] = 0.5):
... super().__init__(label)
... self.x = x
>>> logistic = MapInteraction(
... 'map', target='value', state='x', function=lambda s: 3 * s.target.x * (1 - s.target.x)
... )
>>> system = System(elements=[ValueElement('value')], interactions=[logistic], time=DiscreteTime())
>>> Simulation(system).check().prepare().run(until=2).trajectory_.values('value.x').tolist()
[0.5, 0.75, 0.5625]
>>> agents = MapInteraction('map', target='walkers', state='x', function=lambda a, m: a.x + 1, form='agent')
>>> type(agents.check().prepare().run(SimpleNamespace()).laws_[0]).__name__
'RuleLaw'
>>> from skcomplex.core.elements import GroupElement
>>> class WeightsElement(Element):
... def __init__(self, label, visual=None, *, w: Annotated[object, StateSpec(vector=True)] = 0.0):
... super().__init__(label, visual=visual)
... self.w = w
>>> group = GroupElement('weights', element=WeightsElement, size=1, values={'w': [1.0, 2.0]})
>>> doubling = MapInteraction(
... 'double', target='weights', state='w', function=lambda a, m: [2 * x for x in a.w], form='agent'
... )
>>> trajectory = Simulation(System(elements=[group], interactions=[doubling])).check().prepare().run(until=2)
>>> trajectory.trajectory_.agents('weights', 2).values['w'].tolist()
[[4.0, 8.0]]
"""
def __init__(
self,
label: str,
*,
target: Annotated[str, RoleSpec()],
state: str,
function: Callable[..., object],
sources: Annotated[Sequence[str], RoleSpec(variadic=True)] = (),
mode: str = 'replace',
schedule: str = 'synchronous',
form: str = 'symbolic',
phase: str | None = None,
) -> None:
super().__init__(label)
self.target = target
self.state = state
self.function = function
self.sources = sources
self.mode = mode
self.schedule = schedule
self.form = form
self.phase = phase
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, model: ModelViewProtocol) -> 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, model), 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, model: ModelViewProtocol) -> 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, model)
found = self.function(agent, model)
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, model: ModelViewProtocol) -> None:
"""Set the state of one member to its next value."""
if agent.alive:
self._set_value(agent, self._get_value(agent, model))
def update(self, model: ModelViewProtocol) -> None:
"""Set the state of every agent of the target to its next value, by the schedule.
Args:
model:
The model view of the interaction.
"""
self._update_agents(model, self.schedule, self._get_value, self._set_value)
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 QuotaRegulatorInteraction(FeedbackInteraction[ModelView]):
"""Regulator that halves the quota of cows per herder when the grass runs short and raises it when it recovers.
The quota is the adaptive parameter `quota` of every herder. It halves, down to one cow, when the grass falls below
the share `low` of the capacity of the meadow, and grows by one cow when the grass is above the share `high`.
Without regulation it stays as it is.
Args:
label:
The label of the regulator.
rule:
Unused, since the regulator overrides `act`.
phase:
The phase the regulator acts in.
regulated:
One when the regulator changes the quota, zero otherwise.
low:
The share of the capacity of the meadow below which the quota halves.
high:
The share of the capacity of the meadow above which the quota rises by one cow.
Examples:
>>> from skcomplex.models.economics.resources import QuotaRegulatorInteraction
>>> QuotaRegulatorInteraction('regulator', regulated=1.0).regulated
1.0
"""
def __init__(
self,
label: str,
*,
rule: Callable[[ModelView], None] | None = None,
phase: str | None = None,
regulated: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.0,
low: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.5,
high: Annotated[float, ParameterSpec(domain=RealSet.closed_open(0, oo))] = 0.8,
) -> None:
super().__init__(label, rule=rule, phase=phase)
self.regulated = regulated
self.low = low
self.high = high
def act(self, model: ModelView) -> None:
"""Change the quota with the share of grass left on the meadow.
Args:
model:
The model view of the regulator.
"""
if not model.parameter('regulated'):
return
pastures = model.agents('meadow', PastureView)
share = sum(pasture.grass for pasture in pastures) / sum(pasture.capacity for pasture in pastures)
for herder in model.agents('herders', HerderView):
if share < model.parameter('low'):
herder.quota = max(1.0, float(np.floor(herder.quota / 2)))
elif share > model.parameter('high'):
herder.quota += 1