add astar and astar tests, add color to logging

Author: guilyx

main.py

@@ -1,6 +1,5 @@
-import src.automated_planner as parser
+from src.automated_planner import AutomatedPlanner
 import argparse
-import logging
 import os
 
 
@@ -13,23 +12,15 @@ def main():
     args_parser.add_argument("problem", type=str, help="PDDL problem file")
     args_parser.add_argument("-v", "--verbose", help="Increases the output's verbosity")
     args = args_parser.parse_args()
-    logging.basicConfig(
-        filename="logs/main.log",
-        format="%(levelname)s:%(message)s",
-        filemode="w",
-        level=logging.INFO,
-    )  # Creates the log file
-    apla_tbx = parser.AutomatedPlanner(args.domain, args.problem)
-    logging.info("Starting the tool")
-    path, time = apla_tbx.depth_first_search(time_it=True)
-    logging.info(apla_tbx.get_actions_from_path(path))
-    logging.info("Computation time: %.2f seconds" % time)
-    logging.info("Tool finished")
-    # Output the log (to show something in the output screen)
-    logfile = open("logs/main.log", "r")
-    print(logfile.read())
-    logfile.close()
+    apla_tbx = AutomatedPlanner(args.domain, args.problem)
+    apla_tbx.logger.info("Starting the planning script")
+    apla_tbx.logger.debug("Available heuristics: " + str(apla_tbx.available_heuristics.keys()))
 
+    path, computation_time = apla_tbx.dijktra_best_first_search(time_it=True)
+    apla_tbx.logger.debug(apla_tbx.get_actions_from_path(path))
+
+    apla_tbx.logger.debug("Computation time: %.2f seconds" % computation_time)
+    apla_tbx.logger.info("Terminate with grace...")
 
 if __name__ == "__main__":
     main()

src/a_star.py

@@ -0,0 +1,78 @@
+from .heuristics import zero_heuristic
+from .node import Node
+import logging
+import math
+from time import time as now
+from datetime import datetime as timestamp
+
+
+class AStarBestFirstSearch:
+    def __init__(self, automated_planner, heuristic_function):
+        self.automated_planner = automated_planner
+        self.init = Node(
+            self.automated_planner.initial_state,
+            automated_planner,
+            is_closed=False,
+            is_open=True,
+            heuristic=heuristic_function,
+        )
+        self.heuristic_function = heuristic_function
+        self.open_nodes_n = 1
+        self.nodes = dict()
+        self.nodes[self.__hash(self.init)] = self.init
+
+    def __hash(self, node):
+        sep = ", Dict{Symbol,Any}"
+        string = str(node.state)
+        return string.split(sep, 1)[0] + ")"
+
+    def search(self):
+        time_start = now()
+        self.automated_planner.logger.debug("Search started at: " + str(timestamp.now()))
+        while self.open_nodes_n > 0:
+            current_key = min(
+                [n for n in self.nodes if self.nodes[n].is_open],
+                key=(lambda k: self.nodes[k].f_cost),
+            )
+            current_node = self.nodes[current_key]
+
+            if self.automated_planner.satisfies(
+                self.automated_planner.problem.goal, current_node.state
+            ):
+                computation_time = now() - time_start
+                self.automated_planner.logger.debug("Search finished at: " + str(timestamp.now()))
+                return current_node, computation_time
+
+            current_node.is_closed = True
+            current_node.is_open = False
+            self.open_nodes_n -= 1
+
+            actions = self.automated_planner.available_actions(current_node.state)
+            for act in actions:
+                child = Node(
+                    state=self.automated_planner.transition(current_node.state, act),
+                    automated_planner=self.automated_planner,
+                    parent_action=act,
+                    parent=current_node,
+                    heuristic=self.heuristic_function,
+                    is_closed=False,
+                    is_open=True,
+                )
+                child_hash = self.__hash(child)
+                if child_hash in self.nodes:
+                    if self.nodes[child_hash].is_closed:
+                        continue
+                    if not self.nodes[child_hash].is_open:
+                        self.nodes[child_hash] = child
+                        self.open_nodes_n += 1
+                    else:
+                        if child.g_cost < self.nodes[child_hash].g_cost:
+                            self.nodes[child_hash] = child
+                            self.open_nodes_n += 1
+
+                else:
+                    self.nodes[child_hash] = child
+                    self.open_nodes_n += 1
+        computation_time = now() - time_start
+        self.automated_planner.logger.warning("!!! No path found !!!")
+        return None, computation_time

src/astar.py

@@ -1,76 +1,54 @@
-from .modules import loading_bar_handler
 from .bfs import BreadthFirstSearch
 from .dfs import DepthFirstSearch
 from .dijkstra import DijkstraBestFirstSearch
-import logging
-
-UI = False
-
-if UI:
-    loading_bar_handler(False)
+from .a_star import AStarBestFirstSearch
+from .heuristics import goal_count_heuristic, zero_heuristic
+import coloredlogs, logging
 import julia
-
-_ = julia.Julia(compiled_modules=False)
-
-if UI:
-    loading_bar_handler(True)
-
+_ = julia.Julia(compiled_modules=False, debug=False)
 from julia import PDDL
 from time import time as now
-
-
+logging.getLogger("julia").setLevel(logging.WARNING)
 class AutomatedPlanner:
-    def __init__(self, domain_path, problem_path):
+    def __init__(self, domain_path, problem_path, log_level="DEBUG"):
+        # Planning Tool
         self.pddl = PDDL
         self.domain = self.pddl.load_domain(domain_path)
         self.problem = self.pddl.load_problem(problem_path)
         self.initial_state = self.pddl.initialize(self.problem)
         self.goals = self.__flatten_goal()
-
-    """
-    Transition from one state to the next using an action
-    """
+        self.available_heuristics = dict()
+        self.available_heuristics["goal_count"] = goal_count_heuristic
+        self.available_heuristics["zero"] = zero_heuristic
+
+        # Logging
+        logging.basicConfig(
+            filename="logs/main.log",
+            format="%(levelname)s:%(message)s",
+            filemode="w",
+            level=log_level,
+        )  # Creates the log file
+        self.logger = logging.getLogger("automated_planning")
+        coloredlogs.install(level=log_level)
 
     def transition(self, state, action):
         return self.pddl.transition(self.domain, state, action, check=False)
 
-    """
-    Returns all available actions from the given state 
-    """
-
     def available_actions(self, state):
         return self.pddl.available(state, self.domain)
 
-    """ 
-    Check if a vector of terms is satisfied by the given state
-    """
-
     def satisfies(self, asserted_state, state):
         return self.pddl.satisfy(asserted_state, state, self.domain)[0]
 
-    """
-    Check if the term is satisfied by the state
-    To do: compare if it's faster to compute the check on a vector of terms in julia or python 
-    """
-
     def state_has_term(self, state, term):
         if self.pddl.has_term_in_state(self.domain, state, term):
             return True
         else:
             return False
 
-    """
-    Flatten the goal to a vector of terms
-    To do: check if we can iterate over the jl vector 
-    """
-
     def __flatten_goal(self):
         return self.pddl.flatten_goal(self.problem)
 
-    """
-    Retrieves the linked list path 
-    """
-
     def __retrace_path(self, node):
         if not node:
             return []
@@ -81,13 +59,9 @@ def __retrace_path(self, node):
         path.reverse()
         return path
 
-    """
-    Returns all the actions operated to reach the goal 
-    """
-
     def get_actions_from_path(self, path):
         if not path:
-            logging.warning("Path is empty, can't operate...")
+            self.logger.warning("Path is empty, can't operate...")
             return []
         actions = []
         for node in path:
@@ -99,64 +73,45 @@ def get_actions_from_path(self, path):
         else:
             return (actions, cost)
 
-    """
-    Returns all the states that should be opened from start to goal 
-    """
-
     def get_state_def_from_path(self, path):
         if not path:
-            logging.warning("Path is empty, can't operate...")
+            self.logger.warning("Path is empty, can't operate...")
             return []
         trimmed_path = []
         for node in path:
             trimmed_path.append(node.state)
         return trimmed_path
 
-    """
-    Runs the BFS algorithm on the loaded domain/problem 
-    """
-
     def breadth_first_search(self, time_it=False):
-        if time_it:
-            start_time = now()
         bfs = BreadthFirstSearch(self)
-        last_node = bfs.search()
-        if time_it:
-            total_time = now() - start_time
+        last_node, total_time = bfs.search()
         path = self.__retrace_path(last_node)
         if time_it:
             return path, total_time
         else:
             return path, None
 
-    """
-    Runs the DFS algorithm on the domain/problem 
-    """
-
     def depth_first_search(self, time_it=False):
-        if time_it:
-            start_time = now()
         dfs = DepthFirstSearch(self)
-        last_node = dfs.search()
-        if time_it:
-            total_time = now() - start_time
+        last_node, total_time = dfs.search()
         path = self.__retrace_path(last_node)
         if time_it:
             return path, total_time
         else:
             return path, None
 
-    """
-    Runs the Dijkstra algorithm on the domain/problem 
-    """
-
     def dijktra_best_first_search(self, time_it=False):
-        if time_it:
-            start_time = now()
         dijkstra = DijkstraBestFirstSearch(self)
-        last_node = dijkstra.search()
+        last_node, total_time = dijkstra.search()
+        path = self.__retrace_path(last_node)
         if time_it:
-            total_time = now() - start_time
+            return path, total_time
+        else:
+            return path, None
+
+    def astar_best_first_search(self, time_it=False, heuristic=goal_count_heuristic):
+        astar = AStarBestFirstSearch(self, heuristic)
+        last_node, total_time = astar.search()
         path = self.__retrace_path(last_node)
         if time_it:
             return path, total_time

src/automated_planner.py

@@ -1,5 +1,6 @@
 from .node import Node
-import logging
+from time import time as now
+from datetime import datetime as timestamp
 
 
 class BreadthFirstSearch:
@@ -10,6 +11,8 @@ def __init__(self, automated_planner):
         self.queue = [self.init]
 
     def search(self):
+        time_start = now()
+        self.automated_planner.logger.debug("Search started at: " + str(timestamp.now()))
         while self.queue:
             current_node = self.queue.pop(0)
             if current_node not in self.visited:
@@ -18,7 +21,9 @@ def search(self):
                 if self.automated_planner.satisfies(
                     self.automated_planner.problem.goal, current_node.state
                 ):
-                    return current_node
+                    computation_time = now() - time_start
+                    self.automated_planner.logger.debug("Search finished at: " + str(timestamp.now()))
+                    return current_node, computation_time
 
                 actions = self.automated_planner.available_actions(current_node.state)
                 for act in actions:
@@ -33,5 +38,6 @@ def search(self):
                     if child in self.visited:
                         continue
                     self.queue.append(child)
-        logging.warning("!!! No path found !!!")
-        return None
+        computation_time = now() - time_start
+        self.automated_planner.logger.warning("!!! No path found !!!")
+        return None, computation_time

src/bfs.py

@@ -1,6 +1,6 @@
 from .node import Node
-import logging
-
+from datetime import datetime as timestamp
+from time import time as now
 
 class DepthFirstSearch:
     def __init__(self, automated_planner):
@@ -10,6 +10,8 @@ def __init__(self, automated_planner):
         self.stack = [self.init]
 
     def search(self):
+        time_start = now()
+        self.automated_planner.logger.debug("Search started at: " + str(timestamp.now()))
         while self.stack:
             current_node = self.stack.pop(0)
             if current_node not in self.visited:
@@ -18,7 +20,9 @@ def search(self):
                 if self.automated_planner.satisfies(
                     self.automated_planner.problem.goal, current_node.state
                 ):
-                    return current_node
+                    computation_time = now() - time_start
+                    self.automated_planner.logger.debug("Search finished at: " + str(timestamp.now()))
+                    return current_node, computation_time
 
                 actions = self.automated_planner.available_actions(current_node.state)
                 for act in actions:
@@ -33,5 +37,6 @@ def search(self):
                     if child in self.visited:
                         continue
                     self.stack.append(child)
-        logging.warning("!!! No path found !!!")
-        return None
+        computation_time = now() - time_start
+        self.automated_planner.logger.warning("!!! No path found !!!")
+        return None, computation_time