Update Moran process for atomic mutation and add tests

Author: marcharper

axelrod/moran.py

@@ -6,7 +6,7 @@
 
 import matplotlib.pyplot as plt
 import numpy as np
-from axelrod import DEFAULT_TURNS, Game, Player
+from axelrod import EvolvablePlayer, DEFAULT_TURNS, Game, Player
 
 from .deterministic_cache import DeterministicCache
 from .graph import Graph, complete_graph
@@ -56,7 +56,8 @@ def __init__(
         interaction_graph: Graph = None,
         reproduction_graph: Graph = None,
         fitness_transformation: Callable = None,
-        mutation_method="transition"
+        mutation_method="transition",
+        stop_on_fixation=True
     ) -> None:
         """
         An agent based Moran process class. In each round, each player plays a
@@ -109,6 +110,8 @@ def __init__(
             given
         fitness_transformation:
             A function mapping a score to a (non-negative) float
+        stop_on_fixation:
+            A bool indicating if the process should stop on fixation.
         """
         self.turns = turns
         self.prob_end = prob_end
@@ -121,6 +124,7 @@ def __init__(
         self.score_history = []  # type: List
         self.winning_strategy_name = None  # type: Optional[str]
         self.mutation_rate = mutation_rate
+        self.stop_on_fixation = stop_on_fixation
         m = mutation_method.lower()
         if m in ["atomic", "transition"]:
             self.mutation_method = m
@@ -165,6 +169,7 @@ def __init__(
         # Map players to graph vertices
         self.locations = sorted(interaction_graph.vertices)
         self.index = dict(zip(sorted(interaction_graph.vertices), range(len(players))))
+        self.fixated = self.fixation_check()
 
     def set_players(self) -> None:
         """Copy the initial players into the first population."""
@@ -184,21 +189,21 @@ def mutate(self, index: int) -> Player:
         """
 
         if self.mutation_method == "atomic":
-            mutant = self.players[index].clone()
-            mutant.mutate()
-            return mutant
-
-        # Choose another strategy at random from the initial population
-        r = random.random()
-        if r < self.mutation_rate:
-            s = str(self.players[index])
-            j = randrange(0, len(self.mutation_targets[s]))
-            p = self.mutation_targets[s][j]
-            new_player = p.clone()
-        else:
-            # Just clone the player
-            new_player = self.players[index].clone()
-        return new_player
+            if not issubclass(self.players[index].__class__, EvolvablePlayer):
+                raise TypeError("Player is not evolvable. Use a subclass of EvolvablePlayer.")
+            return self.players[index].mutate()
+
+        # Assuming mutation_method == "transition"
+        if self.mutation_rate > 0:
+            # Choose another strategy at random from the initial population
+            r = random.random()
+            if r < self.mutation_rate:
+                s = str(self.players[index])
+                j = randrange(0, len(self.mutation_targets[s]))
+                p = self.mutation_targets[s][j]
+                return p.clone()
+        # Just clone the player
+        return self.players[index].clone()
 
     def death(self, index: int = None) -> int:
         """
@@ -262,14 +267,13 @@ def fixation_check(self) -> bool:
         Boolean:
             True if fixation has occurred (population all of a single type)
         """
-        if self.mutation_rate > 0:
-            return False
         classes = set(str(p) for p in self.players)
+        self.fixated = False
         if len(classes) == 1:
             # Set the winning strategy name variable
             self.winning_strategy_name = str(self.players[0])
-            return True
-        return False
+            self.fixated = True
+        return self.fixated
 
     def __next__(self) -> object:
         """
@@ -287,7 +291,7 @@ def __next__(self) -> object:
             Returns itself with a new population
         """
         # Check the exit condition, that all players are of the same type.
-        if self.fixation_check():
+        if self.stop_on_fixation and self.fixation_check():
             raise StopIteration
         if self.mode == "bd":
             # Birth then death
@@ -298,16 +302,10 @@ def __next__(self) -> object:
             i = self.death()
             self.players[i] = None
             j = self.birth(i)
-        # Mutate
-        if self.mutation_rate:
-            new_player = self.mutate(j)
-        else:
-            new_player = self.players[j].clone()
-        # Replace player i with clone of player j
-        self.players[i] = new_player
+        # Mutate and/or replace player i with clone of player j
+        self.players[i] = self.mutate(j)
+        # Record population.
         self.populations.append(self.population_distribution())
-        # Check again for fixation
-        self.fixation_check()
         return self
 
     def _matchup_indices(self) -> Set[Tuple[int, int]]:
@@ -407,10 +405,10 @@ def play(self) -> List[Counter]:
          populations:
             Returns a list of all the populations
         """
-        if self.mutation_rate != 0:
+        if not self.stop_on_fixation or self.mutation_rate != 0:
             raise ValueError(
                 "MoranProcess.play() will never exit if mutation_rate is"
-                "nonzero. Use iteration instead."
+                "nonzero or stop_on_fixation is False. Use iteration instead."
             )
         while True:
             try:

axelrod/tests/unit/test_moran.py

@@ -182,8 +182,10 @@ def test_death_birth(self):
         for seed in seeds:
             axelrod.seed(seed)
             mp = MoranProcess((p1, p2), mode="db")
-            next(mp)
-            self.assertIsNotNone(mp.winning_strategy_name)
+            mp.play()
+        self.assertIsNotNone(mp.winning_strategy_name)
+        # Number of populations is 2: the original and the one after the first round.
+        self.assertEqual(len(mp.populations), 2)
 
     def test_death_birth_outcomes(self):
         """Show that birth-death and death-birth can produce different
@@ -218,7 +220,7 @@ def test_two_random_players(self):
     def test_two_players_with_mutation(self):
         p1, p2 = axelrod.Cooperator(), axelrod.Defector()
         axelrod.seed(5)
-        mp = MoranProcess((p1, p2), mutation_rate=0.2)
+        mp = MoranProcess((p1, p2), mutation_rate=0.2, stop_on_fixation=False)
         self.assertDictEqual(mp.mutation_targets, {str(p1): [p2], str(p2): [p1]})
         # Test that mutation causes the population to alternate between
         # fixations
@@ -256,7 +258,7 @@ def test_three_players_with_mutation(self):
         p2 = axelrod.Random()
         p3 = axelrod.Defector()
         players = [p1, p2, p3]
-        mp = MoranProcess(players, mutation_rate=0.2)
+        mp = MoranProcess(players, mutation_rate=0.2, stop_on_fixation=False)
         self.assertDictEqual(
             mp.mutation_targets,
             {str(p1): [p3, p2], str(p2): [p1, p3], str(p3): [p1, p2]},
@@ -377,6 +379,28 @@ def test_cooperator_can_win_with_fitness_transformation(self):
         populations = mp.play()
         self.assertEqual(mp.winning_strategy_name, "Cooperator")
 
+    def test_atomic_mutation_fsm(self):
+        axelrod.seed(12)
+        players = [axelrod.EvolvableFSMPlayer(num_states=2, initial_state=1, initial_action=C)
+                   for _ in range(5)]
+        mp = MoranProcess(players, turns=10, mutation_method="atomic")
+        population = mp.play()
+        self.assertEqual(
+            mp.winning_strategy_name,
+            'Evolvable FSM Player: ((0, C, 1, D), (0, D, 1, C), (1, C, 0, D), (1, D, 1, C)), 1, C, 2, 0.1')
+        self.assertEqual(len(mp.populations), 31)
+        self.assertTrue(mp.fixated)
+
+    def test_atomic_mutation_cycler(self):
+        axelrod.seed(10)
+        cycle_length = 5
+        players = [axelrod.EvolvableCycler(cycle_length=cycle_length)
+                   for _ in range(5)]
+        mp = MoranProcess(players, turns=10, mutation_method="atomic")
+        population = mp.play()
+        self.assertEqual(mp.winning_strategy_name, 'EvolvableCycler: CDCDD, 5, 0.2, 1')
+        self.assertEqual(len(mp.populations), 19)
+        self.assertTrue(mp.fixated)
 
 class GraphMoranProcess(unittest.TestCase):
     def test_complete(self):
@@ -496,13 +520,7 @@ def test_init(self):
         """Test the initialisation process"""
         self.assertEqual(
             set(self.amp.cached_outcomes.keys()),
-            set(
-                [
-                    ("Cooperator", "Defector"),
-                    ("Cooperator", "Cooperator"),
-                    ("Defector", "Defector"),
-                ]
-            ),
+            {("Cooperator", "Defector"), ("Cooperator", "Cooperator"), ("Defector", "Defector")},
         )
         self.assertEqual(self.amp.players, self.players)
         self.assertEqual(self.amp.turns, 0)