Ready!

Author: mclevey

README.md

@@ -1 +1,7 @@
-# agent-based-modelling
+# Agent-based Modelling
+
+This content is ready to go! As usual, create a CodeSpace or work locally use the DevContainers plugin and Docker Desktop. The virutal environment should be selected by default this time, so try running the code right away. 😎 If not, select the Poetry environment. It make need a couple of minutes to fully load.
+
+Note that some cells in the notebook will take time to run, especially when doing large batch runs and sensitivity analyses. We should be able to run code in the live session together without issue.
+
+See you in class!

icsspy/abms/.DS_Store

@@ -0,0 +1,6 @@
+from .bounded_confidence import BoundedConfidenceModel
+from .sir import SIRModel
+
+# from .threshold.model import ThresholdModel
+
+__all__ = ["SIRModel", "BoundedConfidenceModel"]  # ThresholdModel

icsspy/abms/__init__.py

@@ -0,0 +1,113 @@
+import numpy as np
+from mesa import Agent, Model
+from mesa.datacollection import DataCollector
+from mesa.space import MultiGrid
+from mesa.time import RandomActivation
+
+
+class BoundedConfidenceAgent(Agent):
+    def __init__(self, unique_id, model, epsilon, max_agent_step_size=1):
+        super().__init__(unique_id, model)
+        self.opinion = self.random.uniform(
+            -1, 1
+        )  # Agent's opinion initialized between -1 and 1
+        self.epsilon = epsilon  # Confidence bound ($\epsilon$)
+        self.max_agent_step_size = max_agent_step_size  # Step size for movement
+        self.pos_history = []  # Track agent's position over time
+        self.interactions = {}  # Track interactions with a count
+
+    def step(self):
+        # Move to a random neighboring cell with the given step size
+        possible_moves = self.model.grid.get_neighborhood(
+            self.pos, moore=True, include_center=False, radius=self.max_agent_step_size
+        )
+        new_position = self.random.choice(possible_moves)
+        self.model.grid.move_agent(self, new_position)
+        self.pos_history.append(new_position)  # Record the new position
+
+        # Interact with neighbors within confidence bound
+        neighbors = self.model.grid.get_neighbors(
+            self.pos, moore=True, include_center=False
+        )
+        for neighbor in neighbors:
+            if abs(self.opinion - neighbor.opinion) <= self.epsilon:
+                # Update opinion if within confidence bound
+                self.opinion = (self.opinion + neighbor.opinion) / 2
+
+            # Record interactions
+            if neighbor.unique_id in self.interactions:
+                self.interactions[neighbor.unique_id] += 1
+            else:
+                self.interactions[neighbor.unique_id] = 1
+
+
+class BoundedConfidenceModel(Model):
+    def __init__(
+        self,
+        grid_width,
+        grid_height,
+        N,
+        epsilon,
+        max_agent_step_size=1,
+        max_steps=1000,
+        convergence_threshold=0.001,
+    ):
+        super().__init__()
+        self.num_agents = N
+        self.grid = MultiGrid(grid_width, grid_height, True)
+        self.schedule = RandomActivation(self)
+        self.epsilon = epsilon
+        self.max_agent_step_size = max_agent_step_size
+        self.max_steps = max_steps
+        self.convergence_threshold = convergence_threshold
+
+        # Initialize step_count
+        self.step_count = 0  # This will track the number of steps taken
+
+        # Create agents
+        for i in range(self.num_agents):
+            step_size = (
+                max_agent_step_size if i % 2 == 0 else 1
+            )  # Half agents have large step size, half small
+            agent = BoundedConfidenceAgent(
+                i, self, epsilon, max_agent_step_size=step_size
+            )
+            self.grid.place_agent(
+                agent,
+                (
+                    self.random.randrange(self.grid.width),
+                    self.random.randrange(self.grid.height),
+                ),
+            )
+            agent.pos_history.append(
+                agent.pos
+            )  # Initialize the position history with the starting position
+            self.schedule.add(agent)
+
+        self.datacollector = DataCollector(
+            agent_reporters={"Opinion": "opinion"},
+            model_reporters={"Avg_Similarity": self.calculate_similarity},
+        )
+
+    def calculate_similarity(self):
+        """Calculate average pairwise similarity in agent opinions."""
+        opinions = np.array([agent.opinion for agent in self.schedule.agents])
+        pairwise_diffs = np.abs(opinions[:, None] - opinions[None, :])
+        similarity = 1 - pairwise_diffs  # Similarity is inverse of difference
+        avg_similarity = np.mean(similarity)
+        return avg_similarity
+
+    def step(self):
+        self.datacollector.collect(self)
+        self.schedule.step()
+        self.step_count += 1
+
+        # Check stopping conditions
+        if self.check_convergence() or self.step_count >= self.max_steps:
+            self.running = False
+
+    def check_convergence(self):
+        """Check if the opinions have converged."""
+        opinions = np.array([agent.opinion for agent in self.schedule.agents])
+        max_diff = np.max(np.abs(opinions[:, None] - opinions[None, :]))
+        return max_diff < self.convergence_threshold

icsspy/abms/bounded_confidence.py

@@ -0,0 +1,149 @@
+from mesa import Agent, Model
+from mesa.datacollection import DataCollector
+from mesa.space import MultiGrid
+from mesa.time import RandomActivation
+
+
+class SIRAgent(Agent):
+    def __init__(
+        self, unique_id, model, recovery_time_range=(8, 12), max_agent_step_size=1
+    ):
+
+        super().__init__(unique_id, model)
+        self.state = "S"
+        self.infected_time = 0
+        self.recovery_time_range = recovery_time_range
+        self.max_agent_step_size = max_agent_step_size
+        self.pos_history = []
+        self.interactions = {}
+
+    def step(self):
+        if self.state == "I":
+            self.infected_time += 1
+            # check if agent is within the recovery window
+            if self.infected_time >= self.recovery_time_range[0]:
+                # if so, recover probabilistically
+                if (
+                    self.random.random() < 0.5
+                    or self.infected_time >= self.recovery_time_range[1]
+                ):
+                    self.state = "R"
+
+            # if still infected, try to infect others
+            if self.state == "I":
+                neighbors = self.model.grid.get_neighbors(
+                    self.pos, moore=True, include_center=False
+                )
+                for neighbor in neighbors:
+                    if (
+                        neighbor.state == "S"
+                        and self.random.random() < self.model.infection_rate
+                    ):
+                        neighbor.state = "I"
+
+        possible_moves = self.model.grid.get_neighborhood(
+            self.pos, moore=True, include_center=False, radius=self.max_agent_step_size
+        )
+
+        new_position = self.random.choice(possible_moves)
+        self.model.grid.move_agent(self, new_position)
+        self.pos_history.append(new_position)
+
+        # record interactions with other agents in this step
+        neighbors = self.model.grid.get_neighbors(
+            self.pos, moore=True, include_center=False
+        )
+
+        for neighbor in neighbors:
+            if neighbor.unique_id in self.interactions:
+                self.interactions[neighbor.unique_id] += 1
+            else:
+                self.interactions[neighbor.unique_id] = 1
+
+
+class SIRModel(Model):
+    def __init__(
+        self,
+        grid_width,
+        grid_height,
+        N,
+        infection_rate,
+        recovery_time_range,
+        max_agent_step_size=1,
+        n_initial_infections=1,
+        max_iterations=1000,
+        change_threshold=0.001,
+    ):
+
+        super().__init__()
+        self.num_agents = N
+        self.grid = MultiGrid(grid_width, grid_height, True)
+        self.schedule = RandomActivation(self)
+        self.infection_rate = infection_rate
+        self.recovery_time_range = recovery_time_range
+        self.max_agent_step_size = max_agent_step_size
+        self.max_iterations = max_iterations
+        self.current_iteration = 0
+        self.change_threshold = change_threshold
+        # track the ratio of infected agents in the previous step
+        self.previous_infected_ratio = None
+
+        # create agents
+        for i in range(self.num_agents):
+            # half the agents will have a large step size, half small
+            step_size = max_agent_step_size if i % 2 == 0 else 1
+            a = SIRAgent(i, self, recovery_time_range, max_agent_step_size=step_size)
+            self.grid.place_agent(
+                a,
+                (
+                    self.random.randrange(self.grid.width),
+                    self.random.randrange(self.grid.height),
+                ),
+            )
+            # initialize the position history with the starting position
+            a.pos_history.append(a.pos)
+            self.schedule.add(a)
+
+        # randomly infect a specified number of agents
+        initial_infected_agents = self.random.sample(
+            self.schedule.agents, n_initial_infections
+        )
+        for agent in initial_infected_agents:
+            agent.state = "I"
+            agent.infected_time = 0  # initialize infection duration
+
+        self.datacollector = DataCollector(
+            model_reporters={
+                "Susceptible": lambda m: self.count_state("S"),
+                "Infected": lambda m: self.count_state("I"),
+                "Recovered": lambda m: self.count_state("R"),
+            },
+            agent_reporters={
+                "State": "state",
+                "Infection_Duration": "infection_duration",
+            },
+        )
+        # this line is necessary for doing multiple model runs simultaneously
+        # we will do this later in the lecture
+        self.running = True
+
+    def step(self):
+        self.datacollector.collect(self)
+        self.schedule.step()
+        self.current_iteration += 1
+
+        # check for stopping condition based on maximum iterations
+        if self.current_iteration >= self.max_iterations:
+            self.running = False
+
+        # check for stopping condition based on change in infected ratio
+        current_infected_ratio = self.count_state("I")
+        if self.previous_infected_ratio is not None:
+            change = abs(current_infected_ratio - self.previous_infected_ratio)
+            if change < self.change_threshold:
+                self.running = False
+        self.previous_infected_ratio = current_infected_ratio
+
+    def count_state(self, state_name):
+        count = sum([1 for a in self.schedule.agents if a.state == state_name])
+        return count / self.num_agents