Find applicable operators with a successor generator

CHANGELOG.md

@@ -1,5 +1,6 @@
 # 2.2 (unreleased)
 
+* Find applicable operators with a Fast Downward-style successor generator instead of scanning all operators. Select the implementation with `--successor-generator {tree,naive}` (default: `tree`).
 * Support Python 3.10 to 3.14 and PyPy. Raise the minimum Python version to 3.10.
 * Add a `dev` dependency group and run continuous integration tests with uv.
 * Enable ruff's pyupgrade (`UP`) rules to keep the code on modern Python syntax.

dev/benchmark-search.py

@@ -58,10 +58,12 @@ def collect_tasks(benchmarks, per_domain):
     return tasks
 
 
-def run_config(search, heuristic, problem, env, timeout):
+def run_config(search, heuristic, problem, env, timeout, successor_generator):
     cmd = [sys.executable, "-m", "pyperplan", "-s", search]
     if heuristic is not None:
         cmd += ["-H", heuristic]
+    if successor_generator is not None:
+        cmd += ["--successor-generator", successor_generator]
     cmd.append(problem)  # The domain is guessed by pyperplan.
     try:
         proc = subprocess.run(
@@ -83,7 +85,7 @@ def run_config(search, heuristic, problem, env, timeout):
     return {"status": "error", "time": None}
 
 
-def benchmark(src, benchmarks, per_domain, timeout):
+def benchmark(src, benchmarks, per_domain, timeout, successor_generator):
     env = dict(os.environ, PYTHONPATH=src)
     tasks = collect_tasks(benchmarks, per_domain)
     results = {}
@@ -92,7 +94,9 @@ def benchmark(src, benchmarks, per_domain, timeout):
         config = f"{search}+{heuristic or 'none'}"
         rel = os.path.relpath(problem, benchmarks)
         key = f"{config} | {rel}"
-        result = run_config(search, heuristic, problem, env, timeout)
+        result = run_config(
+            search, heuristic, problem, env, timeout, successor_generator
+        )
         results[key] = result
         time = f"{result['time']:.3f}s" if result["time"] is not None else "-"
         print(f"[{i}/{len(runs)}] {key}: {result['status']} {time}")
@@ -124,18 +128,30 @@ def main():
         default=60.0,
         help="per-run timeout in seconds (default: 60)",
     )
+    parser.add_argument(
+        "--successor-generator",
+        choices=["naive", "tree"],
+        help="successor generator to pass to pyperplan (default: planner default)",
+    )
     parser.add_argument("--out", help="write the results to this JSON file")
     args = parser.parse_args()
 
     src = os.path.abspath(args.src)
     benchmarks = args.benchmarks or os.path.join(src, "benchmarks")
-    results = benchmark(src, benchmarks, args.tasks_per_domain, args.timeout)
+    results = benchmark(
+        src,
+        benchmarks,
+        args.tasks_per_domain,
+        args.timeout,
+        args.successor_generator,
+    )
 
     if args.out:
         payload = {
             "src": src,
             "timeout": args.timeout,
             "tasks_per_domain": args.tasks_per_domain,
+            "successor_generator": args.successor_generator,
             "results": results,
         }
         with open(args.out, "w") as f:

pyperplan/__main__.py

@@ -33,6 +33,7 @@
     validate_solution,
     write_solution,
 )
+from pyperplan.successor_generator import SUCCESSOR_GENERATORS
 
 
 def main():
@@ -68,6 +69,12 @@ def get_callable_names(callables, omit_string):
         help=f"Select a search algorithm from {search_names}",
         default="bfs",
     )
+    argparser.add_argument(
+        "--successor-generator",
+        choices=SUCCESSOR_GENERATORS.keys(),
+        help="Select how applicable operators are found",
+        default="tree",
+    )
     args = argparser.parse_args()
 
     logging.basicConfig(
@@ -107,6 +114,7 @@ def get_callable_names(callables, omit_string):
         search,
         heuristic,
         use_preferred_ops=use_preferred_ops,
+        successor_generator=args.successor_generator,
     )
 
     if solution is None:

pyperplan/planner.py

@@ -25,6 +25,7 @@
 
 from . import grounding, heuristics, search, tools
 from .pddl.parser import Parser
+from .successor_generator import create_successor_generator
 
 SEARCHES = {
     "astar": search.astar_search,
@@ -150,7 +151,12 @@ def write_solution(solution, filename):
 
 
 def search_plan(
-    domain_file, problem_file, search, heuristic_class, use_preferred_ops=False
+    domain_file,
+    problem_file,
+    search,
+    heuristic_class,
+    use_preferred_ops=False,
+    successor_generator="tree",
 ):
     """Parse the input files into a planning task and search for a solution.
 
@@ -160,13 +166,18 @@ def search_plan(
     search: A callable that performs a search on the task's search space.
     heuristic_class: A class implementing the heuristic_base.Heuristic
         interface.
+    successor_generator: The kind of successor generator to use ("tree" or
+        "naive").
 
     Returns a list of actions that solve the problem, or None if no solution
     exists.
     """
     overall_start_time = time.process_time()
     problem = _parse(domain_file, problem_file)
     task = _ground(problem)
+    task.set_successor_generator(
+        create_successor_generator(successor_generator, task.operators)
+    )
     heuristic = None
     if heuristic_class is not None:
         heuristic = heuristic_class(task)

pyperplan/successor_generator.py

@@ -0,0 +1,156 @@
+#
+# This file is part of pyperplan.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>
+#
+
+"""A decision-tree successor generator in the style of Fast Downward.
+
+The naive way to find the operators applicable in a state is to test every
+operator's preconditions against the state, which costs time proportional to the
+total number of operators. Fast Downward instead organizes the operators in a
+decision tree that branches on individual facts, so that only the operators
+whose preconditions can still be satisfied are visited.
+
+Each inner node tests one fact. Its ``match_child`` holds the operators that
+require this fact (and is only entered when the fact is true in the state),
+while its ``no_match_child`` holds the operators that do not mention the fact
+(and is always entered). Operators whose preconditions are all satisfied along
+the path from the root sit in a node's ``immediate`` list.
+"""
+
+
+class NaiveSuccessorGenerator:
+    """Find applicable operators by testing every operator in turn.
+
+    This is the straightforward baseline: applicability is checked for each
+    operator, so a query costs time proportional to the number of operators.
+    """
+
+    def __init__(self, operators):
+        self._operators = list(operators)
+
+    def get_applicable_operators(self, state):
+        """Return the operators applicable in ``state``, in their original order."""
+        return [op for op in self._operators if op.applicable(state)]
+
+
+class _Node:
+    """A node in the successor-generator decision tree.
+
+    A leaf has ``fact is None`` and no children; it only contributes its
+    ``immediate`` operators. An inner node branches on ``fact``.
+    """
+
+    __slots__ = ("fact", "immediate", "match_child", "no_match_child")
+
+    def __init__(self):
+        self.fact = None
+        self.immediate = []
+        self.match_child = None
+        self.no_match_child = None
+
+
+def _choose_fact(items):
+    """Return the fact occurring in the most of the given preconditions.
+
+    Branching on the most frequent fact keeps the tree shallow: it splits off as
+    many operators as possible into the ``match_child`` at every step.
+    """
+    counts = {}
+    for _, preconditions in items:
+        for fact in preconditions:
+            counts[fact] = counts.get(fact, 0) + 1
+    return max(counts, key=counts.get)
+
+
+class SuccessorGenerator:
+    """Index a set of operators for fast applicability queries."""
+
+    def __init__(self, operators):
+        items = [(op, op.preconditions) for op in operators]
+        self._root = self._build(items)
+
+    @staticmethod
+    def _build(items):
+        """Build the decision tree for ``items`` iteratively.
+
+        ``items`` is a list of ``(operator, remaining_preconditions)`` pairs,
+        where ``remaining_preconditions`` are the preconditions not yet tested on
+        the path to the current node. An explicit work stack is used instead of
+        recursion so that deep trees cannot exhaust the call stack.
+        """
+        root = _Node()
+        stack = [(root, items)]
+        while stack:
+            node, node_items = stack.pop()
+            remaining = []
+            for op, preconditions in node_items:
+                if preconditions:
+                    remaining.append((op, preconditions))
+                else:
+                    node.immediate.append(op)
+            if not remaining:
+                continue  # Leaf: all operators here are unconditionally applicable.
+            fact = _choose_fact(remaining)
+            node.fact = fact
+            match_items = []
+            no_match_items = []
+            for op, preconditions in remaining:
+                if fact in preconditions:
+                    match_items.append((op, preconditions - {fact}))
+                else:
+                    no_match_items.append((op, preconditions))
+            node.match_child = _Node()
+            node.no_match_child = _Node()
+            stack.append((node.match_child, match_items))
+            stack.append((node.no_match_child, no_match_items))
+        return root
+
+    def get_applicable_operators(self, state):
+        """Return the operators applicable in ``state``.
+
+        The operators come out in the tree's traversal order, which is
+        deterministic but generally differs from the order in which they were
+        passed in (use ``NaiveSuccessorGenerator`` to preserve that order).
+        """
+        found = []
+        stack = [self._root]
+        while stack:
+            node = stack.pop()
+            # The no-match child is always visited, so we follow that spine in a
+            # tight loop and only push the match child (taken when the fact is
+            # true) onto the stack. This keeps the stack small and avoids a
+            # pop/append pair for every node on the spine.
+            while node is not None:
+                if node.immediate:
+                    found.extend(node.immediate)
+                fact = node.fact
+                if fact is None:
+                    break
+                if fact in state:
+                    stack.append(node.match_child)
+                node = node.no_match_child
+        return found
+
+
+SUCCESSOR_GENERATORS = {
+    "naive": NaiveSuccessorGenerator,
+    "tree": SuccessorGenerator,
+}
+
+
+def create_successor_generator(name, operators):
+    """Return a successor generator of the given kind for ``operators``."""
+    return SUCCESSOR_GENERATORS[name](operators)

pyperplan/task.py

@@ -17,6 +17,8 @@
 
 """Classes for representing a STRIPS planning task."""
 
+from pyperplan.successor_generator import SuccessorGenerator
+
 
 class Operator:
     """An action that transforms one state into another.
@@ -93,6 +95,14 @@ def __init__(self, name, facts, initial_state, goals, operators):
         self.initial_state = initial_state
         self.goals = goals
         self.operators = operators
+        # The successor generator is set via ``set_successor_generator`` or built
+        # lazily on the first successor query, because not every consumer of a
+        # task searches (e.g. heuristics only read ``operators``).
+        self._successor_generator = None
+
+    def set_successor_generator(self, successor_generator):
+        """Use ``successor_generator`` to answer successor queries."""
+        self._successor_generator = successor_generator
 
     def goal_reached(self, state):
         """Return whether ``state`` satisfies all of the task's goals.
@@ -106,9 +116,15 @@ def get_successor_states(self, state):
         """Return the ``(operator, successor_state)`` pairs reachable from ``state``.
 
         Each pair consists of an operator applicable in ``state`` and the state
-        that results from applying it.
+        that results from applying it. A successor generator finds the applicable
+        operators without testing every operator individually.
         """
-        return [(op, op.apply(state)) for op in self.operators if op.applicable(state)]
+        if self._successor_generator is None:
+            self._successor_generator = SuccessorGenerator(self.operators)
+        return [
+            (op, op.apply(state))
+            for op in self._successor_generator.get_applicable_operators(state)
+        ]
 
     def __str__(self):
         operators = "\n".join(map(repr, self.operators))