RobbyGA can collector and evolution by circumstance
Mitchell's genetic algorithm that evolves strategies for Robby, a robot that collects cans in a room.
- Name:
robby_ga - Field:
computer_science/evolutionary_computation - Formalism: agent-based system
- References: Mitchell 2009; Gooding 2019
Select this system
from skcomplex.systems import create_simulation, create_system, describe
system = create_system('robby_ga')
entry = describe('robby_ga')
trajectory = create_simulation('robby_ga').prepare().run().trajectory_
Factory
@register_system(
field='computer_science/evolutionary_computation',
tags=('genetic algorithm', 'agents', 'evolution', 'robot'),
references=('Mitchell 2009', 'Gooding 2019'),
defaults={'until': 20},
)
def robby_ga(
n_strategies: int = 50,
*,
mutation: float = 0.005,
sessions: int = 10,
n_actions: int = 100,
regime: str = 'generational',
lifespan: int = 7,
starvation: float = 0.3,
switch: int | None = None,
random_state: int | np.random.Generator | None = 0,
) -> System:
"""Return Mitchell's genetic algorithm that evolves strategies for Robby, a robot that collects cans in a room.
Each step is one generation. Parents are selected with probabilities proportional to their rank, every pair of
parents exchanges the tails of their strategies after a random cut, and every action of a child mutates to another
action with probability `mutation`. Then every child is scored in random rooms. Random strategies score far below
zero because they bump into walls, and evolution finds strategies that avoid walls and collect cans.
Gooding's evolution by circumstance, the regime `'circumstance'`, replaces the selection with deaths: every
strategy starts at a random age, dies at the age `lifespan`, and the fraction `starvation` with the lowest scores
starves every generation. Random pairs of survivors, whatever their scores, breed children in the places of the
dead. The starvation is the only selection, so with it the scores rise as under the genetic algorithm, and without
it they drift. With a `switch` period the starvation switches on and off every `switch` generations, as in
Gooding's runs that switch it every 3000 iterations.
Args:
n_strategies:
The size of the population.
mutation:
The probability that an action of a child is replaced by another one.
sessions:
The number of rooms a strategy cleans to be scored.
n_actions:
The number of actions in a session.
regime:
The evolution, `'generational'` for Mitchell's genetic algorithm or `'circumstance'` for Gooding's
evolution by circumstance.
lifespan:
The age at which a strategy dies, in the evolution by circumstance.
starvation:
The fraction of the lowest scores that starves every generation, in the evolution by circumstance.
switch:
The number of generations after which the starvation switches on or off, or `None` to keep it on, in the
evolution by circumstance.
random_state:
The seed or generator of the random first generation and of its scores.
Returns:
system:
The system, with the group `strategies` of states `actions` and `score`, and `age` in the evolution by
circumstance.
Raises:
ParameterError:
The regime is unknown.
Examples:
>>> from skcomplex.systems.computer_science.evolutionary_computation import robby_ga
>>> robby_ga(5).elements[0].values['actions'].shape
(5, 243)
"""
if regime not in REGIMES:
error_msg = f'Unknown regime {regime!r}. Use one of {", ".join(REGIMES)}.'
raise ParameterError(error_msg)
rng = np.random.default_rng(random_state)
evolution = EvolutionInteraction(
'evolution',
target='strategies',
strategy='actions',
fitness='score',
selection='rank',
crossover=1.0,
mutation=mutation,
alphabet=N_ACTIONS,
)
scoring = RoomCleaningInteraction('scoring', target='strategies', sessions=sessions, n_actions=n_actions)
actions = rng.integers(0, N_ACTIONS, size=(n_strategies, N_SITUATIONS))
scores = scoring.score(actions, rng)
if regime == 'circumstance':
circumstance = CircumstanceInteraction(
'circumstance',
target='strategies',
lifespan=lifespan,
starvation=starvation,
mutation=mutation,
switch=switch,
)
ages = rng.integers(0, lifespan, size=n_strategies).astype(float)
strategies = GroupElement(
'strategies',
element=AgedStrategyElement,
size=n_strategies,
values={'actions': actions, 'score': scores, 'age': ages},
)
return System(elements=[strategies], interactions=[circumstance, scoring])
return System(
elements=[
GroupElement(
'strategies',
element=StrategyElement,
size=n_strategies,
values={'actions': actions, 'score': scores},
),
],
interactions=[evolution, scoring],
)
Components
Elements:
StrategyElement, fromElement
Interactions:
EvolutionInteractionRoomCleaningInteraction, fromInteraction
class StrategyElement(Element):
"""Strategy of Robby, the action for each of the 243 situations he can see, with its last measured score.
The actions are 0 to 3 for moving north, south, east, and west, 4 to stay, 5 to pick up a can, and 6 to move in a
random direction.
Args:
label:
The label of the agent or group.
visual:
How the element is drawn, or `None` for `default_visual_`.
actions:
The initial action per situation, an array of 243 integers from 0 to 6.
score:
The initial score.
Examples:
>>> from skcomplex.core.components.elements import GroupElement
>>> from skcomplex.systems.computer_science.evolutionary_computation import StrategyElement
>>> GroupElement('strategies', element=StrategyElement, size=3).get_params()['size']
3
"""
def __init__(
self,
label: str,
visual: VisualSpec | None = VisualSpec('agents', {'color_by': 'score'}),
*,
actions: Annotated[object, StateSpec()] = 0.0,
score: Annotated[float, StateSpec()] = 0.0,
) -> None:
super().__init__(label, visual=visual)
self.actions = actions
self.score = score
@classmethod
def _get_view(cls) -> type[StrategyView]:
"""Return the view of the strategies, which types their states."""
return StrategyView
class EvolutionInteraction(Interaction):
"""Replacement of the strategies of an agent group by offspring selected on a fitness, with crossover and mutation.
Every `period` steps, or rounds of its phase, the interaction selects parents by tournaments of `tournament`
entrants, by roulette with weights of the fitness minus its minimum, in proportion to the fitness itself
(`'proportional'`), or by rank with weights `1` for the least fit to `n` for the fittest of `n` members. The
`'generational'` replacement selects as many parents as there are members, each pair of which exchanges the tails
after a random cut with probability `crossover`, and passes the `elite` fittest strategies unchanged. The `'moran'`
replacement runs as many birth-death events as there are members, one after the other, each copying a selected
parent onto a member drawn uniformly, and the copy takes the fitness of its parent for the later events of the
step. The `'death-birth'` replacement removes every member with probability `death` and fills its place with the
offspring of a parent selected among the survivors, or among all members when none survives or `parents` is
`'all'`. The `'truncation'` replacement fills the places of the `fraction` least fit members with the offspring of
parents selected among the others. The `'best-mutant'` replacement lets every member draw a mutant of its strategy,
and every member takes the fittest mutant when it is fitter than the fittest strategy, or when every member
mutated, so a population of one strategy climbs a landscape as a whole.
Offspring mutate: each gene with probability `mutation` changes by the rule `change`, to another symbol of an
alphabet of `alphabet` symbols (`'other'`), to a symbol drawn anew from the alphabet (`'any'`), one step up or down
(`'step'`), or to a label one above the largest the members hold (`'new'`), or with probability `mutation` one
gene drawn at random of every offspring changes to another symbol (`'one'`). When `alphabet` is `None`, the
`'other'` and `'any'` rules add Gaussian noise of deviation `scale`. A function `mutation(strategy, rng)` returns
the mutated copy of a strategy instead. The fitness is a state of the members, or the function
`evaluate(strategies, system)` of the strategies, one row per member, which the replacement then writes into the
fitness state. The probabilities are parameters of the system when a subclass declares them with a
`ParameterSpec`, as the base does for `crossover`, `death`, and `fraction`. The draws come from the random stream
of the system.
Args:
label:
The label of the interaction, unique among the parts of its system.
target:
The path of the agent group whose strategies evolve.
strategy:
The name of the state of the members that holds the strategy, a sequence of genes, or the names of
several states, one gene each.
fitness:
The name of the state of the members that holds the fitness, or `None` for a neutral population of equal
fitness when no `evaluate` is given.
selection:
The selection of parents, `'tournament'`, `'roulette'`, `'proportional'`, or `'rank'`.
replacement:
The replacement of strategies, `'generational'`, `'moran'`, `'death-birth'`, `'truncation'`, or
`'best-mutant'`.
crossover:
The probability that a pair of parents exchanges the tails of their strategies, in the generational
replacement.
mutation:
The probability that a gene mutates, or a function `mutation(strategy, rng)` that returns the mutated copy
of a strategy.
elite:
The number of fittest strategies copied unchanged, in the generational replacement.
tournament:
The number of entrants of every tournament.
alphabet:
The number of symbols of a discrete strategy, or `None` for a real strategy.
scale:
The standard deviation of the mutation of a real strategy.
death:
The probability that a member is replaced in a step, in the death-birth replacement.
change:
How a gene mutates, `'other'`, `'any'`, `'one'`, `'step'`, or `'new'`.
parents:
The members the death-birth replacement selects parents among, `'survivors'` or `'all'`.
fraction:
The fraction of least fit members that the truncation replacement replaces.
period:
The number of steps, or rounds of the phase, between two replacements.
evaluate:
The function `evaluate(strategies, system)` that returns the fitness of every row of strategies, or `None`
to read the fitness state.
reset:
The values of states that every replaced member takes, or `None` for no reset.
phase:
The phase of the step the generation runs in, or `None` for the default phase.
Raises:
ValidationError:
A variant is unknown, a probability is outside `[0, 1]`, the mutation is neither a probability nor a
function, a count is invalid, or the evaluation is not callable.
Examples:
>>> from types import SimpleNamespace
>>> from skcomplex.core.components.interactions import EvolutionInteraction
>>> evolution = EvolutionInteraction('evolution', target='agents', selection='roulette', phase='breeding')
>>> evolution.prepare().run(SimpleNamespace()).laws_[0].phase
'breeding'
>>> EvolutionInteraction('evolution', target='agents', selection='rank', replacement='moran').selection
'rank'
>>> EvolutionInteraction(
... 'evolution', target='agents', replacement='truncation', fraction=0.2, change='any', period=50
... ).period
50
"""
law_classes: ClassVar[tuple[type[BaseLaw], ...]] = (RuleLaw,)
def __init__(
self,
label: str,
*,
target: Annotated[
str,
RoleSpec(elements=(ElementKind.GROUP, ElementKind.VARIABLE_GROUP), states=('strategy',)),
],
strategy: str | Sequence[str] = 'strategy',
fitness: str | None = 'fitness',
selection: str = 'tournament',
replacement: str = 'generational',
crossover: Annotated[float, ParameterSpec(domain=RealSet.closed(0, 1))] = 0.7,
mutation: float | Callable[[Array, np.random.Generator], Array] = 0.01,
elite: int = 0,
tournament: int = 2,
alphabet: int | None = 2,
scale: float = 0.1,
death: Annotated[float, ParameterSpec(domain=RealSet.closed(0, 1))] = 0.1,
change: str = 'other',
parents: str = 'survivors',
fraction: Annotated[float, ParameterSpec(domain=RealSet.closed(0, 1))] = 0.1,
period: int = 1,
evaluate: Callable[[Array, SystemViewProtocol], Array] | None = None,
reset: Mapping[str, float] | None = None,
phase: str | None = None,
) -> None:
super().__init__(label)
self.target = target
self.strategy = strategy
self.fitness = fitness
self.selection = selection
self.replacement = replacement
self.crossover = crossover
self.mutation = mutation
self.elite = elite
self.tournament = tournament
self.alphabet = alphabet
self.scale = scale
self.death = death
self.change = change
self.parents = parents
self.fraction = fraction
self.period = period
self.evaluate = evaluate
self.reset = reset
self.phase = phase
def _check_counts(self) -> None:
"""Raise an error for a negative elite, a tournament without entrants, an alphabet of one, or no period."""
if self.elite < 0 or self.tournament < 1:
error_msg = (
f'{self.label!r} has elite {self.elite} and tournament {self.tournament}. Give an elite of at least 0 '
'and a tournament of at least 1.'
)
raise ValidationError(error_msg)
if self.alphabet is not None and self.alphabet < MIN_ALPHABET:
error_msg = (
f'{self.label!r} has an alphabet of {self.alphabet} symbols. Give at least {MIN_ALPHABET}, or `None`.'
)
raise ValidationError(error_msg)
if self.period < 1:
error_msg = f'{self.label!r} has the period {self.period}. Give a period of at least 1.'
raise ValidationError(error_msg)
def _check_probabilities(self) -> None:
"""Raise an error for a probability outside `[0, 1]` or a mutation neither a probability nor callable."""
probabilities = [('crossover', self.crossover), ('death', self.death), ('fraction', self.fraction)]
if not callable(self.mutation):
probabilities.append(('mutation', self.mutation))
for name, value in probabilities:
if not isinstance(value, int | float) or not 0 <= value <= 1:
error_msg = f'The {name} probability of {self.label!r} is {value!r}. Give a value in [0, 1].'
raise ValidationError(error_msg)
def _check(self) -> None:
"""Raise an error for an unknown variant, a probability outside `[0, 1]`, or an invalid count.
Raises:
ValidationError:
The type declares states, a variant is unknown, a probability is outside `[0, 1]`, the mutation is
neither a probability nor a function, a count is invalid, or the evaluation is not callable.
"""
super()._check()
self._check_variant('selection', self.selection, SELECTIONS, (BaseSpec,))
self._check_variant('replacement', self.replacement, REPLACEMENTS, (BaseSpec,))
for name, value, variants in (
('change', self.change, CHANGES),
('parents', self.parents, PARENTS),
):
self._check_variant(name, value, variants)
if (isinstance(self.strategy, str) and not self.strategy) or not isinstance(self.strategy, str | Sequence):
error_msg = f'{self.label!r} has the strategy {self.strategy!r}. Give a state name or a sequence of them.'
raise ValidationError(error_msg)
if self.evaluate is not None and not callable(self.evaluate):
error_msg = f'The evaluation of {self.label!r} must be callable, but it is {self.evaluate!r}.'
raise ValidationError(error_msg)
self._check_probabilities()
self._check_counts()
def _value(self, name: str, system: SystemViewProtocol) -> float:
"""Return a probability, the parameter of the system when the class declares it as one."""
value = getattr(self, self._get_parameter(name))
if self._declares(self._get_parameter(name)) or isinstance(value, int | float):
return self._get_parameter_value(name, system)
return 0.0
def _evaluate(self, strategies: Array, system: SystemViewProtocol) -> Array | None:
"""Return the fitness of every row of strategies by the evaluation, or `None` when there is none.
Args:
strategies:
The strategies, one row per member, in the shape the states hold them.
system:
The system view of the interaction.
Returns:
fitness:
The fitness of every row, or `None` to read the fitness state.
"""
if self.evaluate is None:
return None
return np.asarray(self.evaluate(strategies, system), dtype=float).reshape(len(strategies))
def _evaluate_rows(self, flat: Array, genes: tuple[int, ...], system: SystemViewProtocol) -> Array | None:
"""Return the fitness of flat rows of genes, given back the shape of the states before evaluation."""
return self._evaluate(flat.reshape(len(flat), *genes), system)
def _select(self, fitness: Array, count: int, rng: np.random.Generator) -> npt.NDArray[np.intp]:
"""Return the indices of a number of parents selected on their fitness."""
size = len(fitness)
if self.selection == 'tournament':
entrants = rng.integers(size, size=(count, self.tournament))
return entrants[np.arange(count), np.argmax(fitness[entrants], axis=1)]
if self.selection == 'rank':
weights = np.empty(size)
weights[np.argsort(fitness, kind='stable')] = np.arange(1, size + 1)
elif self.selection == 'proportional':
weights = fitness.copy()
elif self.selection != 'roulette':
raise self._get_variant_error(self.selection, 'selection', '_select')
else:
weights = fitness - fitness.min()
weights = weights if weights.sum() > 0 else np.ones_like(weights)
return rng.choice(size, size=count, p=weights / weights.sum())
def _cross(self, parents: Array, probability: float, rng: np.random.Generator) -> Array:
"""Return the parents after one-point crossover of every consecutive pair."""
children = parents.copy()
length = children.shape[1]
for first in range(0, len(children) - 1, 2):
if length > 1 and rng.random() < probability:
cut = int(rng.integers(1, length))
tail = children[first, cut:].copy()
children[first, cut:] = children[first + 1, cut:]
children[first + 1, cut:] = tail
return children
def _shift(self, children: Array, mutated: npt.NDArray[np.bool_], rng: np.random.Generator) -> Array:
"""Return the children whose mutated genes change to another symbol, or by Gaussian noise."""
if self.alphabet is None:
return children + mutated * rng.normal(0.0, self.scale, children.shape)
shift = rng.integers(1, self.alphabet, children.shape)
return np.where(mutated, (children + shift) % self.alphabet, children)
def _redraw(self, children: Array, mutated: npt.NDArray[np.bool_], rng: np.random.Generator) -> Array:
"""Return the children whose mutated genes are drawn anew from the alphabet, or by Gaussian noise."""
if self.alphabet is None:
return children + mutated * rng.normal(0.0, self.scale, children.shape)
return np.where(mutated, rng.integers(0, self.alphabet, children.shape), children)
def _step(self, children: Array, mutated: npt.NDArray[np.bool_], rng: np.random.Generator) -> Array:
"""Return the children whose mutated genes move one step up or down, within the alphabet when it is finite."""
steps = np.where(rng.random(children.shape) < HALF, 1.0, -1.0)
stepped = children + mutated * steps
return stepped if self.alphabet is None else stepped % self.alphabet
def _mutate(self, children: Array, probability: float, rng: np.random.Generator, population: Array) -> Array:
"""Return the children mutated by the rule of the change, or by the mutation function.
Args:
children:
The strategies of the offspring, one row per offspring.
probability:
The probability of mutation.
rng:
The random number generator.
population:
The strategies the members hold, whose largest label the rule `'new'` exceeds.
Returns:
children:
The mutated strategies.
"""
if callable(self.mutation):
mutation = self.mutation
return np.array([np.ravel(mutation(child.copy(), rng)) for child in children], dtype=float)
if self.change == 'one':
mutated = np.zeros(children.shape, dtype=bool)
chosen = rng.random(len(children)) < probability
mutated[np.arange(len(children)), rng.integers(children.shape[1], size=len(children))] = chosen
return self._shift(children, mutated, rng)
mutated = rng.random(children.shape) < probability
if self.change == 'any':
return self._redraw(children, mutated, rng)
if self.change == 'step':
return self._step(children, mutated, rng)
if self.change == 'new':
labels = population.max(initial=0.0) + np.cumsum(mutated).reshape(children.shape)
return np.where(mutated, labels, children)
return self._shift(children, mutated, rng)
def _breed_generation(self, flat: Array, fitness: Array, system: SystemViewProtocol) -> Array:
"""Return the strategies of the next generation, one row per member."""
rng = system.rng
parents = flat[self._select(fitness, len(flat), rng)]
children = self._cross(parents, self._value('crossover', system), rng)
children = self._mutate(children, self._value('mutation', system), rng, flat)
if self.elite:
best = np.argsort(-fitness, kind='stable')[: self.elite]
children[: len(best)] = flat[best]
return children
def _breed_moran(
self,
flat: Array,
fitness: Array,
system: SystemViewProtocol,
genes: tuple[int, ...],
) -> tuple[Array, Array]:
"""Return the strategies and the fitness after one birth-death event per member, one after the other."""
rng, probability = system.rng, self._value('mutation', system)
strategies, fitness = flat.copy(), fitness.copy()
for _ in range(len(strategies)):
parent = int(self._select(fitness, 1, rng)[0])
dying = int(rng.integers(len(strategies)))
child = self._mutate(strategies[parent : parent + 1], probability, rng, strategies)
evaluated = self._evaluate_rows(child, genes, system)
strategies[dying] = child[0]
fitness[dying] = fitness[parent] if evaluated is None else evaluated[0]
return strategies, fitness
def _breed_death_birth(
self,
flat: Array,
fitness: Array,
system: SystemViewProtocol,
) -> tuple[Array, npt.NDArray[np.bool_]]:
"""Return the strategies after the dead members are filled by the offspring of the survivors, and the dead."""
rng = system.rng
dead = rng.random(len(flat)) < self._value('death', system)
if not dead.any():
return flat, dead
everyone = self.parents == 'all' or dead.all()
survivors = np.arange(len(flat)) if everyone else np.flatnonzero(~dead)
parents = survivors[self._select(fitness[survivors], int(dead.sum()), rng)]
strategies = flat.copy()
strategies[dead] = self._mutate(flat[parents], self._value('mutation', system), rng, flat)
return strategies, dead
def _breed_truncation(
self,
flat: Array,
fitness: Array,
system: SystemViewProtocol,
) -> tuple[Array, npt.NDArray[np.bool_]]:
"""Return the strategies after the least fit are filled by the offspring of the others, and the replaced."""
count = round(self._value('fraction', system) * len(flat))
replaced = np.zeros(len(flat), dtype=bool)
replaced[np.argsort(fitness, kind='stable')[:count]] = True
if not count:
return flat, replaced
others = np.arange(len(flat)) if replaced.all() else np.flatnonzero(~replaced)
parents = others[self._select(fitness[others], count, system.rng)]
strategies = flat.copy()
strategies[replaced] = self._mutate(flat[parents], self._value('mutation', system), system.rng, flat)
return strategies, replaced
def _breed_best_mutant(
self,
flat: Array,
fitness: Array,
system: SystemViewProtocol,
genes: tuple[int, ...],
) -> Array:
"""Return the strategies after every member takes the fittest mutant, when it is fitter or all mutated."""
mutants = self._mutate(flat, self._value('mutation', system), system.rng, flat)
mutated = np.any(mutants != flat, axis=1)
if not mutated.any():
return flat
evaluated = self._evaluate_rows(mutants[mutated], genes, system)
if evaluated is None:
error_msg = f'{self.label!r} replaces by the best mutant, which needs an evaluation of the strategies.'
raise SimulationError(error_msg)
best = int(np.argmax(evaluated))
if evaluated[best] <= fitness.max() and not mutated.all():
return flat
return np.repeat(mutants[mutated][best : best + 1], len(flat), axis=0)
def _breed(
self,
strategies: Array,
fitness: Array,
system: SystemViewProtocol,
) -> tuple[Array, Array | None, npt.NDArray[np.bool_]]:
"""Return the strategies after the replacement, their evaluated fitness or `None`, and the replaced."""
flat, genes = strategies.reshape(len(strategies), -1), strategies.shape[1:]
replaced = np.ones(len(flat), dtype=bool)
if self.replacement == 'moran':
children, fitness = self._breed_moran(flat, fitness, system, genes)
elif self.replacement == 'death-birth':
children, replaced = self._breed_death_birth(flat, fitness, system)
elif self.replacement == 'truncation':
children, replaced = self._breed_truncation(flat, fitness, system)
elif self.replacement == 'best-mutant':
children = self._breed_best_mutant(flat, fitness, system, genes)
elif self.replacement != 'generational':
raise self._get_variant_error(self.replacement, 'replacement', '_breed')
else:
children = self._breed_generation(flat, fitness, system)
children = children.reshape(strategies.shape)
return children, self._evaluate(children, system), replaced
def _read(self, agents: Sequence[AgentViewProtocol], system: SystemViewProtocol) -> tuple[Array, Array]:
"""Return the strategies, one row per member, and their fitness."""
if isinstance(self.strategy, str):
strategies = np.array([getattr(agent, self.strategy) for agent in agents], dtype=float)
else:
strategies = np.array([[getattr(agent, name) for name in self.strategy] for agent in agents], dtype=float)
evaluated = self._evaluate(strategies, system)
if evaluated is not None:
return strategies, evaluated
if self.fitness is None:
return strategies, np.ones(len(agents))
return strategies, np.array([getattr(agent, self.fitness) for agent in agents], dtype=float)
def _write(
self,
agents: Sequence[AgentViewProtocol],
children: Array,
fitness: Array | None,
replaced: npt.NDArray[np.bool_],
) -> None:
"""Write the strategies, the evaluated fitness, and the reset states of the replaced members."""
names = [self.strategy] if isinstance(self.strategy, str) else list(self.strategy)
for index, (agent, child, new) in enumerate(zip(agents, children, replaced, strict=True)):
genes = [child] if isinstance(self.strategy, str) else list(child)
for name, gene in zip(names, genes, strict=True):
setattr(agent, name, gene if np.ndim(gene) else float(gene))
if fitness is not None and self.fitness is not None:
setattr(agent, self.fitness, float(fitness[index]))
for name, value in (self.reset or {}).items() if new else ():
setattr(agent, name, value)
def _evolve(self, system: SystemViewProtocol) -> None:
"""Replace the strategies of the members by their offspring, once every period."""
calls = system.storage.get('calls', 0)
calls = calls + 1 if isinstance(calls, int) else 1
system.storage['calls'] = calls
agents = system.agents(self._get_role('target'))
if not agents or calls % self.period:
return
strategies, fitness = self._read(agents, system)
self._write(agents, *self._breed(strategies, fitness, system))
def _get_laws(self, namespace: SimpleNamespace) -> list[BaseLaw]:
"""Return the rule of one generation.
Args:
namespace:
The namespace of the symbols the laws can use, which the rule does not read.
Returns:
laws:
The rule that breeds the next generation.
"""
return [RuleLaw(self._evolve, phase=self.phase)]
class RoomCleaningInteraction(Interaction):
"""Score of every strategy of Robby, the mean of its scores in sessions of cleaning random rooms.
Every strategy cleans `sessions` random rooms of side `side`, with a can in each cell with probability `density`,
for `n_actions` actions, and its state `score` takes the mean score. A picked can scores 10, picking up nothing
scores -1, and bumping into a wall scores -5.
Args:
label:
The label of the interaction.
target:
The path of the group of strategies.
sessions:
The number of rooms a strategy cleans to be scored.
n_actions:
The number of actions in a session.
side:
The side of a room.
density:
The probability that a cell holds a can.
phase:
The phase of the step the scoring runs in, or `None` for the default phase.
Examples:
>>> from skcomplex.systems.computer_science.evolutionary_computation import RoomCleaningInteraction
>>> RoomCleaningInteraction('scoring', target='strategies').sessions
10
"""
def __init__(
self,
label: str,
*,
target: Annotated[str, RoleSpec()],
sessions: int = 10,
n_actions: int = 100,
side: int = 10,
density: float = 0.5,
phase: str | None = None,
) -> None:
super().__init__(label)
self.target = target
self.sessions = sessions
self.n_actions = n_actions
self.side = side
self.density = density
self.phase = phase
def _score_agents(self, system: SystemView) -> None:
"""Set the score of every strategy of the target to its mean score in random rooms."""
agents = system.agents('target', StrategyView)
strategies = np.array([agent.actions for agent in agents], dtype=np.int64)
for agent, score in zip(agents, self.score(strategies, system.rng), strict=True):
agent.score = float(score)
def _get_laws(self, s: SimpleNamespace) -> list[BaseLaw]:
"""Return the rule that scores the strategies.
Args:
s:
The namespace of the symbols the laws can use.
Returns:
laws:
The scoring rule.
"""
return [RuleLaw(self._score_agents, phase=self.phase)]
def score(self, strategies: IntArray, rng: np.random.Generator) -> Array:
"""Return the mean score of strategies in random rooms.
Args:
strategies:
The actions of every strategy, one row per strategy.
rng:
The random number generator of the rooms and the random moves.
Returns:
scores:
The mean score of every strategy.
"""
return RoomCleaningInteraction._evaluate(
strategies,
rng,
sessions=self.sessions,
actions=self.n_actions,
side=self.side,
density=self.density,
)
@staticmethod
def _new_rooms(rng: np.random.Generator, shape: tuple[int, ...], side: int, density: float) -> IntArray:
"""Return square rooms surrounded by walls, with a can in every inner cell with a probability."""
rooms = np.full((*shape, side + 2, side + 2), WALL, dtype=np.int64)
rooms[..., 1:-1, 1:-1] = (rng.random((*shape, side, side)) < density).astype(np.int64)
return rooms
@staticmethod
def _situations(rooms: IntArray, rows: IntArray, columns: IntArray) -> IntArray:
"""Return the index of what Robby sees north, south, east, west, and here, in base 3."""
index = np.indices(rows.shape)
seen = [rooms[(*index, rows + row, columns + column)] for row, column in [*MOVES.tolist(), [0, 0]]]
return sum(value * 3**power for power, value in enumerate(seen))
@staticmethod
def _evaluate(
strategies: IntArray,
rng: np.random.Generator,
*,
sessions: int,
actions: int,
side: int,
density: float,
) -> Array:
"""Return the mean score of every strategy over sessions of cleaning random rooms."""
shape = (strategies.shape[0], sessions)
rooms = RoomCleaningInteraction._new_rooms(rng, shape, side, density)
rows, columns = np.ones(shape, dtype=np.int64), np.ones(shape, dtype=np.int64)
scores = np.zeros(shape)
index = np.indices(shape)
for _ in range(actions):
action = strategies[index[0], RoomCleaningInteraction._situations(rooms, rows, columns)]
action = np.where(action == RANDOM_MOVE, rng.integers(0, 4, shape), action)
here = rooms[(*index, rows, columns)]
picking = action == PICK_UP
scores += np.where(picking, np.where(here == CAN, CAN_REWARD, EMPTY_PICK_UP_PENALTY), 0.0)
rooms[(*index, rows, columns)] = np.where(picking, EMPTY, here)
moving = action < STAY
step = MOVES[np.where(moving, action, 0)] * moving[..., None]
target_rows, target_columns = rows + step[..., 0], columns + step[..., 1]
blocked = rooms[(*index, target_rows, target_columns)] == WALL
scores += np.where(blocked, WALL_PENALTY, 0.0)
rows, columns = np.where(blocked, rows, target_rows), np.where(blocked, columns, target_columns)
return scores.mean(axis=1)