Enhancements to SimulationTester and Main Function with Config Flexibility

src/nepiada/baseline.py

@@ -1,7 +1,6 @@
 # Local imports
 import numpy as np
 import env.nepiada as nepiada
-from utils.config import BaselineConfig
 import pygame
 
 
@@ -193,11 +192,18 @@ def step(agent_name, agent_instance, observations, infos, env, config):
 
     return min_action
 
-def main(included_data=None):
+import json
+from utils.config import BaselineConfig
+
+def main(included_data=None, config_file=None):
     if included_data is None:
         included_data = ["observations", "rewards", "terminations", "truncations", "infos"]
 
-    env_config = BaselineConfig()
+    if config_file:
+        env_config = BaselineConfig(json_path=config_file)
+    else:
+        env_config = BaselineConfig()
+
     env = nepiada.parallel_env(config=env_config)
     observations, infos = env.reset()
 

src/nepiada/tester/test.py

@@ -1,36 +1,49 @@
 import numpy as np
 import sys
-sys.path.append("..") # Adds higher directory to python modules path.
+import csv
+import os
 
-# Import the main function from the simulation script
+sys.path.append("..")  # Adds higher directory to python modules path.
 from baseline import main as baseline_run
 import env.nepiada as nepiada
 
 class SimulationTester:
-    def __init__(self, included_data=None):
+    def __init__(self, included_data=None, num_runs=1, config_file=None):
         """
-        Initialize the SimulationTester with specified data components to include.
+        Initialize the SimulationTester with specified data components to include, number of simulation runs,
+        and an optional configuration file.
         :param included_data: List of strings indicating which data components to include in results.
+        :param num_runs: Number of times to run the simulation.
+        :param config_file: Optional configuration file for the simulation.
         """
+        self.num_runs = num_runs
+        self.config_file = config_file
         if included_data is None:
             self.included_data = ["observations", "rewards", "terminations", "truncations", "infos"]
         else:
             self.included_data = included_data
 
-    def run_simulation(self):
+    def run_simulation(self, config_file=None):
         """
         Run the simulation and store the results.
+        :param config_file: Optional configuration file for the simulation.
         """
-        self.results, self.agents, self.config, self.env = baseline_run(included_data=self.included_data)
+        if config_file is None:
+            config_file = self.config_file
+
+        if config_file:
+            # If a config file is provided, pass it to baseline_run
+            self.results, self.agents, self.config, self.env = baseline_run(included_data=self.included_data, config_file=config_file)
+        else:
+            # If no config file is provided, call baseline_run without the config_file parameter
+            self.results, self.agents, self.config, self.env = baseline_run(included_data=self.included_data)
+
 
     def calculate_convergence_score(self):
         """
-        This function calculates the convergence score based on the average reward acrooss all agent 
+        This function calculates the convergence score based on the average reward across all agents 
         in the last iteration of the algorithm.
 
-        The reward of the agents in turn are calculated using two metrics: global arrangement 
-        and local arrangement costs, which are described in Pavel and Dian's paper.
-
         IMPORTANT: An ideal / globally optimal NE will have a score of zero. Lower the score the closer 
         it is to globally optimal NE.
         """
@@ -52,21 +65,55 @@ def calculate_convergence_score(self):
 
         return convergence_score
 
+    def run_multiple_simulations(self):
+        """
+        Run the simulation multiple times and calculate the average convergence score.
+        """
+        scores = []
+        for _ in range(self.num_runs):
+            self.run_simulation(self.config_file)
+            score = self.calculate_convergence_score()
+            if score != -1:
+                scores.append(score)
+
+        return np.mean(scores) if scores else -1
+
     def print_results(self):
         """
-        Print the stored simulation results.
+        Print the stored simulation results and write them to a CSV file in a specified directory.
         """
         if not hasattr(self, 'results'):
             print("No results to display. Run the simulation first.")
             return
 
-        for step_result in self.results:
-            print(step_result)
+        # Ensure the simulation directory exists
+        if not os.path.exists(self.config.simulation_dir):
+            os.makedirs(self.config.simulation_dir)
+
+        csv_filename = os.path.join(self.config.simulation_dir, "simulation_results.csv")
+
+        with open(csv_filename, mode='w', newline='') as file:
+            writer = csv.writer(file)
+
+            # Writing header
+            headers = ['Step'] + list(self.results[0].keys())
+            writer.writerow(headers)
+
+            # Writing data and printing to console
+            for step, step_result in enumerate(self.results):
+                print(step_result)  # Printing to console
+                row = [step] + list(step_result.values())
+                writer.writerow(row)
+
+            convergence_score = self.calculate_convergence_score()
+            print("\nCalculating score, ideal NE is (0): ", convergence_score)
+            writer.writerow(["Convergence Score", convergence_score])
 
-        print("\nCalculating score, ideal NE is (0): ", self.calculate_convergence_score())
+        print(f"Results written to {csv_filename}")
 
 # Example usage
 if __name__ == "__main__":
-    tester = SimulationTester()
-    tester.run_simulation()
+    tester = SimulationTester(num_runs=1, config_file="test_cases/everyone_is_your_enemy.json")
+    average_score = tester.run_multiple_simulations()
+    print("Average Convergence Score over", tester.num_runs, "runs:", average_score)
     tester.print_results()

src/nepiada/tester/test_cases/default.json

@@ -0,0 +1,3 @@
+{
+    "iterations": 5
+}

src/nepiada/tester/test_cases/everyone_is_your_enemy.json

@@ -0,0 +1,5 @@
+{
+    "iterations": 5,
+    "num_good_agents": 3,
+    "num_adversarial_agents": 6
+}

src/nepiada/tester/test_cases/large_map.json

@@ -0,0 +1,5 @@
+{
+    "iterations": 5,
+    "screen_height": 250,
+    "screen_width": 250
+}

src/nepiada/utils/config.py

@@ -4,6 +4,8 @@
 from utils.noise import GaussianNoise
 from .anim_consts import WIDTH, HEIGHT
 
+import json
+
 
 class Config:
     """
@@ -32,7 +34,7 @@ class Config:
     # Initialization parameters
     dim: int = 2
     size: int = 50
-    iterations: int = 50
+    iterations: int = 1
     simulation_dir: str = "plots"
 
     # Agent related parameterss
@@ -61,10 +63,18 @@ class Config:
     empty_cell: int = -1
     global_reward_weight: int = 1
     local_reward_weight: int = 1
-    # screen_height: int = 400
-    # screen_width: int = 400
+    screen_height: int = 400
+    screen_width: int = 400
+
+    def __init__(self, json_path=None, **kwargs):
+
+        # If a JSON file path is provided, load and update attributes
+        if json_path:
+            with open(json_path, 'r') as file:
+                params = json.load(file)
+                for key, value in params.items():
+                    setattr(self, key, value)
 
-    def __init__(self):
         self._process_screen_size()
 
     def _process_screen_size(self):
@@ -89,14 +99,14 @@ def _process_screen_size(self):
 class BaselineConfig(Config):
     D: int = 1
 
-    def __init__(self):
-        super().__init__()
+    def __init__(self, json_path=None, **kwargs):
+        super().__init__(json_path=json_path, **kwargs)
 
 
 # Baseline specific configuration parameters
 class EpsilonBaselineConfig(Config):
     D: int = 1
-    epsilon: int = 0.2
+    epsilon: float = 0.2
 
-    def __init__(self):
-        super().__init__()
+    def __init__(self, json_path=None, **kwargs):
+        super().__init__(json_path=json_path, **kwargs)