Evaluate the A* heuristic lazily and simplify atom grounding

A* search:
- Only compute the heuristic (and create the child node) for a successor
  once we know it reaches its state more cheaply than any path seen so
  far. The heuristic value is only ever used when a node is enqueued, so
  computing it for non-improving successors was wasted work. This is
  behavior-preserving (identical expansions and plans) and roughly halves
  heuristic evaluations on transposition-heavy tasks (e.g. ~50% fewer on
  gripper/task03). It benefits astar, gbfs and wastar, which share this
  loop.

Grounding:
- Build the grounded argument list of an atom with a comprehension and
  dict.get instead of an explicit loop with membership tests. This is
  simpler and slightly faster in a function called for every atom of
  every grounded operator. Grounded operator counts are unchanged.

Author: jendrikseipp

pyperplan/grounding.py

@@ -313,12 +313,7 @@ def _ground_atom(atom, assignment):
     Return a string with the grounded representation of "atom" with respect
     to "assignment".
     """
-    names = []
-    for name, types in atom.signature:
-        if name in assignment:
-            names.append(assignment[name])
-        else:
-            names.append(name)
+    names = [assignment.get(name, name) for name, _ in atom.signature]
     return _get_grounded_string(atom.name, names)
 
 
@@ -329,7 +324,7 @@ def _ground_atoms(atoms, assignment):
 
 def _get_fact(atom):
     """Return the string representation of the grounded atom."""
-    args = [name for name, types in atom.signature]
+    args = [name for name, _ in atom.signature]
     return _get_grounded_string(atom.name, args)
 
 

pyperplan/search/a_star.py

@@ -135,20 +135,24 @@ def astar_search(
                     else:
                         logging.debug(f"keeping operator {op.name}")
 
+                # Only consider this successor if it reaches succ_state more
+                # cheaply than any path seen so far. The heuristic is expensive,
+                # so we avoid computing it for successors we would not enqueue.
+                succ_g = pop_node.g + 1
+                if succ_g >= state_cost.get(succ_state, float("inf")):
+                    continue
                 succ_node = searchspace.make_child_node(pop_node, op, succ_state)
                 h = heuristic(succ_node)
                 if h == float("inf"):
                     # Don't bother with states that can't reach the goal anyway.
                     continue
-                old_succ_g = state_cost.get(succ_state, float("inf"))
-                if succ_node.g < old_succ_g:
-                    # We either never saw succ_state before, or we found a
-                    # cheaper path to succ_state than previously.
-                    node_tiebreaker += 1
-                    heapq.heappush(
-                        open_list, make_open_entry(succ_node, h, node_tiebreaker)
-                    )
-                    state_cost[succ_state] = succ_node.g
+                # We either never saw succ_state before, or we found a cheaper
+                # path to succ_state than previously.
+                node_tiebreaker += 1
+                heapq.heappush(
+                    open_list, make_open_entry(succ_node, h, node_tiebreaker)
+                )
+                state_cost[succ_state] = succ_g
 
         counter += 1
     logging.info("No operators left. Task unsolvable.")