Fix bug and remove simulator (#7)

Author: hasnainroopawalla

.gitignore

@@ -130,3 +130,6 @@ dmypy.json
 
 # Pyre type checker
 .pyre/
+
+# Simulator
+aco_routing/simulator.py

aco_routing/__init__.py

@@ -0,0 +1,3 @@
+from aco_routing.graph import *
+from aco_routing.dijkstra import *
+from aco_routing.aco import *

aco_routing/aco.py

@@ -2,8 +2,8 @@
 import random
 from typing import List, Tuple
 
-from aco_routing.utils.graph import Graph
-from aco_routing.utils.ant import Ant
+from aco_routing.graph import Graph
+from aco_routing.ant import Ant
 
 
 @dataclass
@@ -15,11 +15,11 @@ def _forward_ants(self, ants: List[Ant], max_iterations: int) -> None:
 
         Args:
             ants (List[Ant]): A List of Ants.
-            max_iterations (int, optional): The maximum number of steps an ant is allowed is to take in order to reach the destination.
-                If it fails to find a path, it is tagged as unfit. Defaults to 50.
+            max_iterations (int): The maximum number of steps an ant is allowed is to take in order to reach the destination.
+                If it fails to find a path, it is tagged as unfit.
         """
-        for idx, ant in enumerate(ants):
-            for i in range(max_iterations):
+        for _, ant in enumerate(ants):
+            for _ in range(max_iterations):
                 if ant.reached_destination():
                     ant.is_fit = True
                     break
@@ -31,7 +31,7 @@ def _backward_ants(self, ants: List[Ant]) -> None:
         Args:
             ants (List[Ant]): A List of Ants.
         """
-        for idx, ant in enumerate(ants):
+        for _, ant in enumerate(ants):
             if ant.is_fit:
                 self.graph.deposit_pheromones_along_path(ant.path)
 
@@ -40,8 +40,8 @@ def _deploy_search_ants(
         source: str,
         destination: str,
         num_ants: int,
-        random_spawns: bool = False,
-        cycles: int = 100,
+        cycles: int,
+        random_spawns: bool,
         max_iterations: int = 50,
     ) -> None:
         """Deploys search ants which traverse the graph to find the shortest path.
@@ -50,12 +50,12 @@ def _deploy_search_ants(
             source (str): The source node in the graph.
             destination (str): The destination node in the graph.
             num_ants (int): The number of ants to be spawned.
-            random_spawns (bool): A flag to determine if the ants should be spawned at random nodes or always at the source node.
-            cycles (int, optional): The number of cycles of generating and deploying ants (forward and backward). Defaults to 100.
-            max_iterations (int, optional): The maximum number of steps an ant is allowed is to take in order to reach the destination.
-                If it fails to find a path, it is tagged as unfit. Defaults to 50.
+            cycles (int): The number of cycles of generating and deploying ants (forward and backward).
+            random_spawns (bool): Indicates if the search ants should spawn at random nodes in the graph.
+            max_iterations (int, optional): The maximum number of steps an ant is allowed is to take in order to reach the destination,
+                after which it is tagged as unfit. Defaults to 50.
         """
-        for cycle in range(cycles):
+        for _ in range(cycles):
             ants: List[Ant] = []
             for _ in range(num_ants):
                 spawn_point = (
@@ -79,24 +79,42 @@ def _deploy_solution_ant(self, source: str, destination: str) -> List[str]:
             List[str]: The shortest path found by the ants (A list of node IDs).
         """
         # Spawn an ant which favors pheromone values over edge costs.
-        ant = Ant(self.graph, source, destination, alpha=0.99, beta=0.01)
+        ant = Ant(self.graph, source, destination, is_solution_ant=True)
         while not ant.reached_destination():
             ant.take_step()
         return ant.path
 
     def find_shortest_path(
-        self, source: str, destination: str
+        self,
+        source: str,
+        destination: str,
+        num_ants: int,
+        max_iterations: int,
+        cycles: int,
+        random_spawn: bool = True,
     ) -> Tuple[List[str], float]:
         """Finds the shortest path from the source to the destination in the graph using the traditional Ant Colony Optimization technique.
 
         Args:
             source (str): The source node in the graph.
             destination (str): The destination node in the graph.
+            num_ants (int): The number of search ants to be deployed.
+            max_iterations (int): The maximum number of steps an ant is allowed is to take in order to reach the destination,
+                after which it is tagged as unfit. Defaults to 50.
+            cycles (int): The number of cycles/waves of search ants to be deployed.
+            random_spawn (bool, optional): Indicates if the search ants should spawn at random nodes in the graph. Defaults to True.
 
         Returns:
-            List[str]: The shortest path found by the ants (A list of node IDs).
+            List[str]: The shortest path found by the ants (a list of node IDs).
             float: The total travel time of the shortest path.
         """
-        self._deploy_search_ants(source, destination, num_ants=20, random_spawns=True)
+        self._deploy_search_ants(
+            source,
+            destination,
+            num_ants=num_ants,
+            max_iterations=max_iterations,
+            cycles=cycles,
+            random_spawns=random_spawn,
+        )
         shortest_path = self._deploy_solution_ant(source, destination)
         return shortest_path, self.graph.compute_path_travel_time(shortest_path)

aco_routing/utils/ant.py renamed to aco_routing/ant.py

@@ -2,7 +2,7 @@
 import random
 from typing import Dict, List, Set
 
-from aco_routing.utils.graph import Edge, Graph
+from aco_routing.graph import Edge, Graph
 
 
 @dataclass
@@ -13,11 +13,12 @@ class Ant:
         graph (Graph): The Graph object.
         source (str): The source node of the ant.
         destination (str): The destination node of the ant.
-        alpha (float): The amount of importance given to the pheromone by the ant. Defaults to 0.9.
-        beta (float): The amount of importance given to the travel time value by the ant. Defaults to 0.1.
-        visited_nodes (Set): A set of nodes that have been visited by the ant.
-        path (List[str]): A List of node IDs of the path taken by the ant so far.
-        is_fit (bool): A flag which indicates if the ant has reached the destination (fit) or not (unfit). Defaults to False.
+        alpha (float, optional): The amount of importance given to the pheromone by the ant. Defaults to 0.9.
+        beta (float, optional): The amount of importance given to the travel time value by the ant. Defaults to 0.1.
+        visited_nodes (Set, optional): A set of nodes that have been visited by the ant.
+        path (List[str], optional): A List of node IDs of the path taken by the ant so far.
+        is_fit (bool, optional): Indicates if the ant has reached the destination (fit) or not (unfit). Defaults to False.
+        is_solution_ant (bool, optional): Indicates if the ant is the final/solution ant. Defaults to False.
     """
 
     graph: Graph
@@ -28,6 +29,7 @@ class Ant:
     visited_nodes: Set = field(default_factory=set)
     path: List[str] = field(default_factory=list)
     is_fit: bool = False
+    is_solution_ant: bool = False
 
     def __post_init__(self) -> None:
         self.current_node = self.source
@@ -149,11 +151,19 @@ def _pick_next_node(
         Returns:
             str: The ID of the next node to be visited by the ant.
         """
+        if self.is_solution_ant:
+            # The final/solution ant greedily chooses the next node with the highest pheromone value.
+            return max(
+                unvisited_neighbors, key=lambda k: unvisited_neighbors[k].pheromones
+            )
         edges_total = self._calculate_edges_total(unvisited_neighbors, alpha, beta)
+
         probabilities = self._calculate_edge_probabilites(
             unvisited_neighbors, edges_total, alpha, beta
         )
         sorted_probabilities = self._sort_edge_probabilites(probabilities)
+
+        # Pick the next node based on the Roulette Wheel selection technique.
         return self._choose_neighbor_using_roulette_wheel(sorted_probabilities)
 
     def take_step(self) -> None:
@@ -171,8 +181,11 @@ def take_step(self) -> None:
         # Find unvisited neighbors of the current node.
         unvisited_neighbors = self._get_unvisited_neighbors(all_neighbors)
 
-        # Pick the next node based on the Roulette Wheel selection technique.
+        # Pick the next node of the ant.
         next_node = self._pick_next_node(unvisited_neighbors, self.alpha, self.beta)
 
+        if not next_node:
+            return
+
         self.path.append(next_node)
         self.current_node = next_node

aco_routing/dijkstra.py

@@ -1,12 +1,12 @@
 from dataclasses import dataclass
 from typing import Dict, List, Tuple
 
-from aco_routing.utils.graph import Graph
+from aco_routing.graph import Graph
 
 
 @dataclass
 class Dijkstra:
-    """The basline Dijkstra's Algorithm to find the shortest path between 2 nodes in the graph.
+    """The basline Dijkstra's Algorithm to find the shortest path between two nodes in the graph.
     Reference: https://stackoverflow.com/a/61078380
     """
 

aco_routing/utils/graph.py renamed to aco_routing/graph.py

@@ -1,7 +1,4 @@
-from aco_routing.utils.graph import Graph
-from aco_routing.dijkstra import Dijkstra
-from aco_routing.utils.simulator import Simulator
-from aco_routing.aco import ACO
+from aco_routing import Graph, Dijkstra, ACO
 
 graph = Graph()
 
@@ -23,10 +20,10 @@
 dijkstra = Dijkstra(graph)
 aco = ACO(graph)
 
+aco_path, aco_cost = aco.find_shortest_path(
+    source, destination, num_ants=100, max_iterations=50, cycles=100
+)
 dijkstra_path, dijkstra_cost = dijkstra.find_shortest_path(source, destination)
-aco_path, aco_cost = aco.find_shortest_path(source, destination)
 
 print(f"ACO - path: {aco_path}, cost: {aco_cost}")
 print(f"Dijkstra - path: {dijkstra_path}, cost: {dijkstra_cost}")
-
-Simulator(graph).simulate(source, destination, num_episodes=100, plot=True)

aco_routing/utils/__init__.py

@@ -1,2 +0,0 @@
-matplotlib
-tqdm