Critical Path Heuristic

jupyddl/automated_planner.py

@@ -4,7 +4,7 @@
 from .a_star import AStarBestFirstSearch
 from .greedy_best_first import GreedyBestFirstSearch
 from .metrics import Metrics
-from .heuristics import BasicHeuristic, DeleteRelaxationHeuristic
+from .heuristics import BasicHeuristic, DeleteRelaxationHeuristic, CriticalPathHeuristic
 import coloredlogs
 import logging
 import julia
@@ -27,10 +27,9 @@ def __init__(self, domain_path, problem_path, log_level="DEBUG"):
         self.initial_state = self.pddl.initialize(self.problem)
         self.goals = self.__flatten_goal()
         self.available_heuristics = [
-            "basic/zero",
-            "basic/goal_count",
-            "delete_relaxation/h_add",
-            "delete_relaxation/h_max",
+            "basic/zero", "basic/goal_count",
+            "delete_relaxation/h_add", "delete_relaxation/h_max",
+            "critical_path/1", "critical_path/2", "critical_path/3",
         ]
 
         # Logger
@@ -158,6 +157,8 @@ def astar_best_first_search(
             heuristic = BasicHeuristic(self, heuristic_key)
         elif "delete_relaxation" in heuristic_key:
             heuristic = DeleteRelaxationHeuristic(self, heuristic_key)
+        elif "critical_path" in heuristic_key:
+            heuristic = CriticalPathHeuristic(self, int(heuristic_key[-1]))
         else:
             logging.fatal("Not yet implemented")
             return [], Metrics()
@@ -178,6 +179,8 @@ def greedy_best_first_search(
             heuristic = BasicHeuristic(self, "basic/goal_count")
         elif "delete_relaxation" in heuristic_key:
             heuristic = DeleteRelaxationHeuristic(self, heuristic_key)
+        elif "critical_path" in heuristic_key:
+            heuristic = CriticalPathHeuristic(self, int(heuristic_key[-1]))
         else:
             logging.fatal("Not yet implemented")
             return [], Metrics()

jupyddl/heuristics.py

@@ -29,7 +29,6 @@ def __goal_count_heuristic(self, state):
                 count += 1
         return count
 
-
 class DeleteRelaxationHeuristic:
     def __init__(self, automated_planner, heuristic_key):
         class DRHCache:
@@ -127,3 +126,109 @@ def __facts_eq(self, facts_dict, facts_set):
             if not (str(f) in fact_costs_str.keys()):
                 return False
         return True
+
+class CriticalPathHeuristic:
+    def __init__(self, automated_planner, critical_path_level=1):
+        class CPCache:
+            def __init__(self, domain=None, axioms=None, preconds=None, additions=None):
+                self.domain = domain
+                self.axioms = axioms
+                self.preconds = preconds
+                self.additions = additions
+
+        self.automated_planner = automated_planner
+        self.cache = CPCache()
+        if critical_path_level > 3:
+            logging.warning("Critical Path level is only implemented until 3, forcing it to 3.")
+            self.critical_path_level = 3
+        if critical_path_level < 1:
+            logging.warning("Critical Path level has to be at least 1, forcing it to 1.")
+            self.critical_path_level = 1
+        else:
+            self.critical_path_level = critical_path_level
+        self.has_been_precomputed = False
+        self.__pre_compute()
+        # return self.heuristic_keys[self.current_h](state)
+
+    def compute(self, state):
+        if not self.has_been_precomputed:
+            self.__pre_compute()
+        domain = self.cache.domain
+        goals = self.automated_planner.goals
+        types = state.types
+        facts = state.facts
+        fact_costs = self.automated_planner.pddl.init_facts_costs(facts)
+        while True:
+            facts, state = self.automated_planner.pddl.get_facts_and_state(
+                fact_costs, types
+            )
+            if self.automated_planner.satisfies(goals, state):
+                costs = []
+                fact_costs_str = dict([(str(k), val) for k, val in fact_costs.items()])
+                if self.critical_path_level == 1:
+                    for g in goals:
+                        if str(g) in fact_costs_str:
+                            costs.append(fact_costs_str[str(g)])
+                if self.critical_path_level == 2:
+                    pairs_of_goals = [(g1, g2) for g1 in goals for g2 in goals if g1 != g2]
+                    for gs in pairs_of_goals:
+                        if str(gs[0]) in fact_costs_str and str(gs[1]) in fact_costs_str:
+                            costs.append(fact_costs_str[str(gs[0])] + fact_costs_str[str(gs[1])])
+                if self.critical_path_level == 3:
+                    triplets_of_goals = [(g1, g2, g3) for g1 in goals for g2 in goals for g3 in goals if g1 != g2 and g1 != g3 and g2 != g3]
+                    for gs in triplets_of_goals:
+                        if str(gs[0]) in fact_costs_str and str(gs[1]) in fact_costs_str and str(gs[2]) in fact_costs_str:
+                            costs.append(fact_costs_str[str(gs[0])] + fact_costs_str[str(gs[1])] + fact_costs_str[str(gs[2])])
+                costs.insert(0, 0)
+                return max(costs)
+
+            for ax in self.cache.axioms:
+                fact_costs = (
+                    self.automated_planner.pddl.compute_costs_one_step_derivation(
+                        facts, fact_costs, ax, "max"
+                    )
+                )
+
+            actions = self.automated_planner.available_actions(state)
+            for act in actions:
+                fact_costs = self.automated_planner.pddl.compute_cost_action_effect(
+                    fact_costs, act, domain, self.cache.additions, "max"
+                )
+
+            if len(fact_costs) == self.automated_planner.pddl.length(
+                facts
+            ) and self.__facts_eq(fact_costs, facts):
+                break
+
+        return float("inf")
+
+    def __pre_compute(self):
+        if self.has_been_precomputed:
+            return
+        domain = self.automated_planner.domain
+        domain, axioms = self.automated_planner.pddl.compute_hsp_axioms(domain)
+        # preconditions = dict()
+        additions = dict()
+        self.automated_planner.pddl.cache_global_preconditions(domain)
+        for name, definition in domain.actions.items():
+            additions[name] = self.automated_planner.pddl.effect_diff(
+                definition.effect
+            ).add
+        self.cache.additions = additions
+        self.cache.preconds = self.automated_planner.pddl.g_preconditions
+        self.cache.domain = domain
+        self.cache.axioms = axioms
+        self.has_been_precomputed = True
+
+    def __h_add(self, costs):
+        return sum(costs)
+
+    def __h_max(self, costs):
+        return max(costs)
+
+    def __facts_eq(self, facts_dict, facts_set):
+        fact_costs_str = dict([(str(k), val) for k, val in facts_dict.items()])
+        for f in facts_set:
+            if not (str(f) in fact_costs_str.keys()):
+                return False
+        return True