first fixed

Author: colinfrisch

examples/virus_antibody/agents.py

@@ -9,113 +9,86 @@
 from collections import deque
 
 import numpy as np
-
-sys.path.insert(0, os.path.abspath("../../mesa"))
 from mesa.experimental.continuous_space import ContinuousSpaceAgent
 
 
 class AntibodyAgent(ContinuousSpaceAgent):
     """An Antibody agent. They move randomly until they see a virus, go fight it.
     If they lose, stay KO for a bit, lose health and back to random moving.
     """
+    speed = 1.5
+    sight_range = 10
+    ko_timeout = 15
+    memory_capacity = 3
+    health = 2
 
     def __init__(
         self,
         model,
         space,
-        sight_range,
         duplication_rate,
-        ko_timeout,
-        memory_capacity,
         initial_position=(0, 0),
         direction=(1, 1),
     ):
         super().__init__(model=model, space=space)
 
-        # Movement & state
+        # Movement & characteristics
         self.position = initial_position
-        self.speed = 1.5
         self.direction = np.array(direction, dtype=float)
-
-        # Characteristics
-        self.sight_range = sight_range
-        self.health = 2
         self.duplication_rate = duplication_rate
 
         # Memory
-        self.st_memory: deque = deque()
+        self.st_memory: deque = deque(maxlen=self.memory_capacity)
         self.lt_memory: list = []
-        self.memory_capacity = memory_capacity
 
         # Target & KO state
         self.target = None  # will hold a weakref.ref or None
-        self.ko_timeout = ko_timeout
         self.ko_steps_left = 0
 
     def step(self):
+        nearby_agents,_ = self.space.get_agents_in_radius(self.position, self.sight_range)
+        nearby_viruses = [a for a in nearby_agents if isinstance(a, VirusAgent)]
+        nearby_antibodies = [a for a in nearby_agents if isinstance(a, AntibodyAgent) and a.unique_id != self.unique_id]
+
         # Acquire a virus target if we don't already have one
-        if self.target is None:
-            closest = self.find_closest_virus()
-            if closest:
-                self.target = weakref.ref(closest)
+        if self.target is None and nearby_viruses:
+            closest = nearby_viruses[0]
+            self.target = weakref.ref(closest)
 
         # Communicate and maybe duplicate
-        self.communicate()
+        self.communicate(nearby_antibodies)
         if self.random.random() < self.duplication_rate:
             self.duplicate()
 
         # Then move
         self.move()
 
-    def find_closest_virus(self):
-        agents, _ = self.space.get_agents_in_radius(self.position, self.sight_range)
-        viruses = [a for a in agents if isinstance(a, VirusAgent)]
-        return viruses[0] if viruses else None
-
-    def communicate(self) -> bool:
-        agents, _ = self.space.get_agents_in_radius(self.position, self.sight_range)
-        peers = [
-            a
-            for a in agents
-            if isinstance(a, AntibodyAgent) and a.unique_id != self.unique_id
-        ]
-        if not peers:
-            return False
-
-        for other in peers:
+    def communicate(self, nearby_antibodies) -> bool:
+        for other in nearby_antibodies:
             to_share = [
                 dna for dna in self.st_memory if dna and dna not in other.lt_memory
             ]
             if to_share:
                 other.st_memory.extend(to_share)
                 other.lt_memory.extend(to_share)
-                while len(other.st_memory) > self.memory_capacity:
-                    other.st_memory.popleft()
         return True
 
     def duplicate(self):
         clone = AntibodyAgent(
             self.model,
             self.space,
-            sight_range=self.sight_range,
             duplication_rate=self.duplication_rate,
-            ko_timeout=self.ko_timeout,
-            memory_capacity=self.memory_capacity,
             initial_position=self.position,
             direction=self.direction,
         )
         # Copy over memory
-        clone.st_memory = deque(item for item in self.st_memory if item)
+        clone.st_memory = deque(maxlen=self.memory_capacity)
+        clone.st_memory.extend([item for item in self.st_memory if item])
         clone.lt_memory = [item for item in self.lt_memory if item]
         clone.target = None
         clone.ko_steps_left = 0
 
-        self.model.antibodies_set.add(clone)
-
     def move(self):
-        # If we've been removed from the space, bail out
-        if getattr(self, "space", None) is None:
-            return
 
         # Dereference weakref if needed
         target = (
@@ -163,22 +136,17 @@ def move(self):
             self.position = new_pos
 
     def engage_virus(self, virus) -> str:
-        # If it's already gone
-        if virus not in self.model.agents:
-            self.target = None
-            return "no_target"
 
         dna = copy.deepcopy(virus.dna)
         if dna in self.st_memory or dna in self.lt_memory:
             virus.remove()
             self.target = None
-            return "win"
+
         else:
             # KO (or death)
             self.health -= 1
             if self.health <= 0:
                 self.remove()
-                return "dead"
 
             self.st_memory.append(dna)
             self.lt_memory.append(dna)
@@ -190,6 +158,7 @@ def engage_virus(self, virus) -> str:
 
 class VirusAgent(ContinuousSpaceAgent):
     """A virus agent: random movement, mutation, duplication, passive to antibodies."""
+    speed = 1
 
     def __init__(
         self,
@@ -205,28 +174,23 @@ def __init__(
         self.position = position
         self.mutation_rate = mutation_rate
         self.duplication_rate = duplication_rate
-        self.speed = 1
         self.direction = np.array((1, 1), dtype=float)
         self.dna = dna if dna is not None else self.generate_dna()
 
     def step(self):
-        # If already removed from the space, don't do anything
-        if getattr(self, "space", None) is None:
-            return
         if self.random.random() < self.duplication_rate:
             self.duplicate()
         self.move()
 
     def duplicate(self):
-        clone = VirusAgent(
+        VirusAgent(
             self.model,
             self.space,
             mutation_rate=self.mutation_rate,
             duplication_rate=self.duplication_rate,
             position=self.position,
             dna=self.generate_dna(self.dna),
         )
-        self.model.viruses_set.add(clone)
 
     def generate_dna(self, dna=None):
         if dna is None:
@@ -240,9 +204,6 @@ def generate_dna(self, dna=None):
         return dna
 
     def move(self):
-        if getattr(self, "space", None) is None:
-            return
-
         # Random walk
         perturb = np.array(
             [
@@ -255,5 +216,4 @@ def move(self):
         if norm > 0:
             self.direction /= norm
 
-        # Step
-        self.position = self.position + self.direction * self.speed
+        self.position += self.direction * self.speed

examples/virus_antibody/app.py

@@ -6,7 +6,7 @@
 from matplotlib.markers import MarkerStyle
 from model import VirusAntibodyModel
 
-sys.path.insert(0, os.path.abspath("../../mesa"))
+sys.path.insert(0, os.path.abspath("../../../mesa"))
 
 from mesa.experimental.devs import ABMSimulator
 from mesa.visualization import (

examples/virus_antibody/model.py

@@ -7,7 +7,7 @@
 import os
 import sys
 
-sys.path.insert(0, os.path.abspath("../../mesa"))
+sys.path.insert(0, os.path.abspath("../../../mesa"))
 
 import numpy as np
 from agents import AntibodyAgent, VirusAgent
@@ -31,11 +31,10 @@ def __init__(
         initial_viruses=20,
         width=100,
         height=100,
+
         # Antibody parameters
         antibody_duplication_rate=0.01,
-        antibody_sight_range=10,
-        antibody_ko_timeout=15,
-        antibody_memory_capacity=3,
+
         # Virus parameters
         virus_duplication_rate=0.01,
         virus_mutation_rate=0.01,
@@ -54,7 +53,6 @@ def __init__(
 
         Indirect Args (not chosen in the graphic interface for clarity reasons):
             antibody_memory_capacity: Number of virus DNA an antibody can remember
-            antibody_sight_range: Radius within which antibodies can detect viruses
             antibody_ko_timeout : Number of step after which an antibody can move after a KO
 
         """
@@ -69,9 +67,6 @@ def __init__(
 
         # antibody parameters
         self.antibody_duplication_rate = antibody_duplication_rate
-        self.antibody_sight_range = antibody_sight_range
-        self.antibody_ko_timeout = antibody_ko_timeout
-        self.antibody_memory_capacity = antibody_memory_capacity
 
         # virus parameters
         self.virus_duplication_rate = virus_duplication_rate
@@ -103,16 +98,13 @@ def __init__(
             size=(self.initial_antibody, 2)
         ) * np.array(self.space.size)
         directions = self.rng.uniform(-1, 1, size=(self.initial_antibody, 2))
-        self.antibodies_set = AntibodyAgent.create_agents(
+        AntibodyAgent.create_agents(
             self,
             self.initial_antibody,
             self.space,
             initial_position=antibodies_positions,
             direction=directions,
-            sight_range=self.antibody_sight_range,
             duplication_rate=self.antibody_duplication_rate,
-            ko_timeout=self.antibody_ko_timeout,
-            memory_capacity=self.antibody_memory_capacity,
         )
 
         # Create and place the Virus agents
@@ -122,7 +114,7 @@ def __init__(
         )
         directions = self.rng.uniform(-1, 1, size=(self.initial_viruses, 2))
 
-        self.viruses_set = VirusAgent.create_agents(
+        VirusAgent.create_agents(
             self,
             self.initial_viruses,
             self.space,