UP-BFGP interface v0.2.0, few-shot engine and BFGP++ generalized planner v0.1.0

setup.py

@@ -31,6 +31,7 @@
         "fmap": ["up-fmap==0.0.12"],
         "aries": ["up-aries>=0.3.2"],
         "symk": ["up-symk>=1.0.1"],
+        "bfgp": ["up-bfgp>=0.2.0"],
         "engines": [
             "tarski[arithmetic]",
             "up-pyperplan==1.0.0.6.dev1",
@@ -41,6 +42,7 @@
             "up-fmap==0.0.12",
             "up-aries>=0.3.2",
             "up-symk>=1.0.1",
+            "up-bfgp>=0.2.0",
         ],
         "plot": [
             "plotly",

unified_planning/engines/engine.py

@@ -28,6 +28,7 @@ class OperationMode(Enum):
     """
 
     ONESHOT_PLANNER = "oneshot_planner"
+    FEWSHOT_PLANNER = "fewshot_planner"
     ANYTIME_PLANNER = "anytime_planner"
     PLAN_VALIDATOR = "plan_validator"
     PORTFOLIO_SELECTOR = "portfolio_selector"

unified_planning/engines/factory.py

@@ -30,6 +30,7 @@
 from unified_planning.engines.mixins.anytime_planner import AnytimePlannerMixin
 from unified_planning.engines.mixins.compiler import CompilationKind, CompilerMixin
 from unified_planning.engines.mixins.oneshot_planner import OneshotPlannerMixin
+from unified_planning.engines.mixins.fewshot_planner import FewshotPlannerMixin
 from unified_planning.engines.mixins.plan_validator import PlanValidatorMixin
 from unified_planning.engines.mixins.portfolio import PortfolioSelectorMixin
 from unified_planning.engines.mixins.replanner import ReplannerMixin
@@ -47,6 +48,7 @@
     "fast-downward-opt": ("up_fast_downward", "FastDownwardOptimalPDDLPlanner"),
     "symk": ("up_symk", "SymKPDDLPlanner"),
     "symk-opt": ("up_symk", "SymKOptimalPDDLPlanner"),
+    "bfgp": ("up_bfgp", "BestFirstGeneralizedPlanner"),
     "pyperplan": ("up_pyperplan.engine", "EngineImpl"),
     "pyperplan-opt": ("up_pyperplan.engine", "OptEngineImpl"),
     "enhsp": ("up_enhsp.enhsp_planner", "ENHSPEngine"),
@@ -131,6 +133,7 @@
     "fast-downward-opt",
     "symk",
     "symk-opt",
+    "bfgp",
     "pyperplan",
     "pyperplan-opt",
     "enhsp",
@@ -431,11 +434,13 @@ def _engine_satisfies_conditions(
             return False
         if (
             operation_mode == OperationMode.ONESHOT_PLANNER
+            or operation_mode == OperationMode.FEWSHOT_PLANNER
             or operation_mode == OperationMode.REPLANNER
             or operation_mode == OperationMode.PORTFOLIO_SELECTOR
         ):
             assert (
                 issubclass(EngineClass, OneshotPlannerMixin)
+                or issubclass(EngineClass, FewshotPlannerMixin)
                 or issubclass(EngineClass, ReplannerMixin)
                 or issubclass(EngineClass, PortfolioSelectorMixin)
             )
@@ -696,12 +701,14 @@ def _get_engine(
                     res.default = compilation_kind
             elif (
                 operation_mode == OperationMode.ONESHOT_PLANNER
+                or operation_mode == OperationMode.FEWSHOT_PLANNER
                 or operation_mode == OperationMode.PLAN_REPAIRER
                 or operation_mode == OperationMode.PORTFOLIO_SELECTOR
             ):
                 res = EngineClass(**params)
                 assert (
                     isinstance(res, OneshotPlannerMixin)
+                    or issubclass(EngineClass, FewshotPlannerMixin)
                     or isinstance(res, PortfolioSelectorMixin)
                     or isinstance(res, PlanRepairerMixin)
                 )
@@ -760,6 +767,29 @@ def OneshotPlanner(
             optimality_guarantee,
         )
 
+
+    def FewshotPlanner(
+        self,
+        *,
+        name: Optional[str] = None,
+        names: Optional[Sequence[str]] = None,
+        params: Optional[Union[Dict[str, Any], Sequence[Dict[str, Any]]]] = None,
+        problem_kind: ProblemKind = ProblemKind(),
+    ) -> "up.engines.engine.Engine":
+        """
+        Returns a fewshot planner. The next is an example to call this method:
+        *   | using ``name`` (the name of a specific GP planner) and ``params`` (GP planner dependent options).
+            | e.g. ``FewshotPlanner(name='bfgp', params={'program_lines': 15})``
+        """
+        return self._get_engine(
+            OperationMode.FEWSHOT_PLANNER,
+            name,
+            names,
+            params,
+            problem_kind,
+        )
+
+
     def AnytimePlanner(
         self,
         *,

unified_planning/engines/mixins/__init__.py

@@ -18,6 +18,7 @@
     OneshotPlannerMixin,
     OptimalityGuarantee,
 )
+from unified_planning.engines.mixins.fewshot_planner import FewshotPlannerMixin
 from unified_planning.engines.mixins.anytime_planner import (
     AnytimePlannerMixin,
     AnytimeGuarantee,

unified_planning/engines/mixins/fewshot_planner.py

@@ -0,0 +1,73 @@
+# Copyright 2022 AIPlan4EU project
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+from warnings import warn
+import unified_planning as up
+from abc import ABC, abstractmethod
+from typing import IO, Optional, Callable, List
+
+
+class FewshotPlannerMixin(ABC):
+    """Base class to be extended by an :class:`~unified_planning.engines.Engine` that is also a `FewshotPlanner`."""
+
+    @staticmethod
+    def is_fewshot_planner() -> bool:
+        return True
+
+    def solve(
+        self,
+        problems: List["up.model.AbstractProblem"],
+        heuristic: Optional[Callable[["up.model.state.State"], Optional[float]]] = None,
+        timeout: Optional[float] = None,
+        output_stream: Optional[IO[str]] = None,
+    ) -> List["up.engines.results.PlanGenerationResult"]:
+        """
+        This method takes a list of `AbstractProblem`s and returns a list of `PlanGenerationResult`,
+        which contains information about the solution to the problems given by the generalized planner.
+
+        :param problems: is the list of `AbstractProblem`s to solve.
+        :param heuristic: is a function that given a state returns its heuristic value or `None` if the state is a
+                          dead-end, defaults to `None`.
+        :param timeout: is the time in seconds the generalized planner has to solve all problems, defaults to `None`.
+        :param output_stream: is a stream of strings where the planner writes his output (and also errors) while it is
+                              solving the problem; defaults to `None`.
+        :return: a list of `PlanGenerationResult` created by the generalized planner; a data structure containing the
+                 :class:`~unified_planning.plans.Plan` found and some additional information about it.
+
+        The only required parameter is `problems` but the generalized planner should warn the user if `heuristic`,
+        `timeout` or `output_stream` are not `None` and the generalized planner ignores them.
+        """
+        assert isinstance(self, up.engines.engine.Engine)
+        for problem in problems:
+            # Check whether the GP planner supports each input problem
+            if not self.skip_checks and not self.supports(problem.kind):
+                msg = f"We cannot establish whether {self.name} can solve problem {problem.name}!"
+                if self.error_on_failed_checks:
+                    raise up.exceptions.UPUsageError(msg)
+                else:
+                    warn(msg)
+        # Solve all problems
+        return self._solve(problems, heuristic, timeout, output_stream)
+
+    @abstractmethod
+    def _solve(
+        self,
+        problems: List["up.model.AbstractProblem"],
+        heuristic: Optional[Callable[["up.model.state.State"], Optional[float]]] = None,
+        timeout: Optional[float] = None,
+        output_stream: Optional[IO[str]] = None,
+    ) -> List["up.engines.results.PlanGenerationResult"]:
+        """Method called by the FewshotPlannerMixin.solve method."""
+        raise NotImplementedError

unified_planning/shortcuts.py

@@ -556,6 +556,26 @@ def OneshotPlanner(
     )
 
 
+def FewshotPlanner(
+    *,
+    name: Optional[str] = None,
+    names: Optional[Sequence[str]] = None,
+    params: Optional[Union[Dict[str, Any], Sequence[Dict[str, Any]]]] = None,
+    problem_kind: ProblemKind = ProblemKind(),
+) -> Engine:
+    """
+    Returns a fewshot planner. The next is an example to call this method:
+    *   | using ``name`` (the name of a specific GP planner) and ``params`` (GP planner dependent options).
+        | e.g. ``FewshotPlanner(name='bfgp', params={'program_lines': 15})``
+    """
+    return get_environment().factory.FewshotPlanner(
+        name=name,
+        names=names,
+        params=params,
+        problem_kind=problem_kind,
+    )
+
+
 def AnytimePlanner(
     *,
     name: Optional[str] = None,