New OM: ActionSelector

.mypy.ini

@@ -15,3 +15,6 @@ ignore_missing_imports = True
 
 [mypy-pyparsing.*]
 ignore_missing_imports = True
+
+[mypy-pysmt.*]
+ignore_missing_imports = True

docs/code_snippets/multi_agent_and_contingent.py

@@ -41,8 +41,7 @@
     problem.add_goal(Equals(Dot(robot, robot_position), locations[-1]))
 
 
-from unified_planning.model.contingent_problem import ContingentProblem
-from unified_planning.model.action import SensingAction
+from unified_planning.model import ContingentProblem, SensingAction
 
 problem = ContingentProblem("lost_packages")
 Package = UserType("Package")

docs/problem_representation.rst

@@ -55,7 +55,7 @@ Contingent Example
 A contingent planning problem represents an action-based problem in which the exact initial state is not entirely known and some of the actions produce “observations” upon execution. More specifically, some actions can be SensingActions, which indicate which fluents they observe and after the successful execution of such actions, the observed fluents become known to the executor. The inherent non-determinism in the initial state can therefore be “shrinked” by performing suitable SensingActions and a plan is then a strategy that prescribes what to execute based on the past observations.
 
 .. literalinclude:: ./code_snippets/multi_agent_and_contingent.py
-    :lines: 44-94
+    :lines: 44-93
 
 
 Hierarchical Example

requirements.txt

@@ -1,2 +1,3 @@
 pyparsing
 networkx
+pysmt

setup.py

@@ -20,7 +20,7 @@
     packages=find_packages(),
     include_package_data=True,
     python_requires=">=3.7",  # supported Python ranges
-    install_requires=["pyparsing", "networkx"],
+    install_requires=["pyparsing", "networkx", "pysmt"],
     extras_require={
         "dev": ["tarski[arithmetic]", "pytest", "pytest-cov", "mypy"],
         "grpc": ["grpcio", "grpcio-tools", "grpc-stubs"],

unified_planning/engines/engine.py

@@ -34,6 +34,7 @@ class OperationMode(Enum):
     SEQUENTIAL_SIMULATOR = "sequential_simulator"
     REPLANNER = "replanner"
     PLAN_REPAIRER = "plan_repairer"
+    ACTION_SELECTOR = "action_selector"
 
 
 class EngineMeta(ABCMeta):

unified_planning/engines/factory.py

@@ -36,6 +36,7 @@
 from unified_planning.engines.mixins.sequential_simulator import (
     SequentialSimulatorMixin,
 )
+from unified_planning.engines.mixins.action_selector import ActionSelectorMixin
 from unified_planning.engines.engine import OperationMode
 from typing import IO, Any, Dict, Tuple, Optional, List, Union, Type, Sequence, cast
 from pathlib import PurePath
@@ -659,14 +660,19 @@
                     msg = f"The problem has no quality metrics but the engine is required to be optimal!"
                     raise up.exceptions.UPUsageError(msg)
                 res = EngineClass(problem=problem, **params)
-            elif operation_mode == OperationMode.SEQUENTIAL_SIMULATOR:
+            elif (
+                operation_mode == OperationMode.SEQUENTIAL_SIMULATOR
+                or operation_mode == OperationMode.ACTION_SELECTOR
+            ):
                 assert problem is not None
                 res = EngineClass(
                     problem=problem,
                     error_on_failed_checks=error_failed_checks,
                     **params,
                 )
-                assert isinstance(res, SequentialSimulatorMixin)
+                assert isinstance(res, SequentialSimulatorMixin) or isinstance(
+                    res, ActionSelectorMixin
+                )
             elif operation_mode == OperationMode.COMPILER:
                 res = EngineClass(**params)
                 assert isinstance(res, CompilerMixin)
@@ -961,6 +967,31 @@
             optimality_guarantee=optimality_guarantee,
         )
 
+    def ActionSelector(
+        self,
+        problem: "up.model.AbstractProblem",
+        *,
+        name: Optional[str] = None,
+        params: Optional[Dict[str, str]] = None,
+    ) -> "up.engines.engine.Engine":
+        """
+        Returns an ActionSelector. There are two ways to call this method:
+
+        *   | using ``problem_kind`` through the problem field.
+            | e.g. ``ActionSelector(problem)``
+        *   | using ``name`` (the name of a specific action selector) and eventually some ``params``
+            | (engine dependent options).
+            | e.g. ``ActionSelector(problem, name='xxx')``
+        """
+        return self._get_engine(
+            OperationMode.ACTION_SELECTOR,
+            name,
+            None,
+            params,
+            problem.kind,
+            problem=problem,
+        )
+
     def PortfolioSelector(
         self,
         *,

unified_planning/engines/mixins/action_selector.py

@@ -0,0 +1,85 @@
+# Copyright 2023 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.
+#
+
+import unified_planning as up
+from typing import Dict
+from warnings import warn
+from unified_planning.exceptions import UPUsageError
+
+
+class ActionSelectorMixin:
+    """
+    This class defines the interface that an :class:`~unified_planning.engines.Engine`
+    that is also an `ActionSelector` must implement.
+
+    Important NOTE: The `AbstractProblem` instance is given at the constructor.
+    """
+
+    def __init__(self, problem: "up.model.AbstractProblem"):
+        self._problem = problem
+        self_class = type(self)
+        assert issubclass(
+            self_class, up.engines.engine.Engine
+        ), "ActionSelectorMixin does not implement the up.engines.Engine class"
+        assert isinstance(self, up.engines.engine.Engine)
+        if not self.skip_checks and not self_class.supports(problem.kind):
+            msg = f"We cannot establish whether {self.name} is able to handle this problem!"
+            if self.error_on_failed_checks:
+                raise UPUsageError(msg)
+            else:
+                warn(msg)
+
+    @staticmethod
+    def is_action_selector() -> bool:
+        """
+        Returns True if this engine is also an action selector, False otherwise.
+
+        :return: True if this engine is also an action selector, False otherwise.
+        """
+        return True
+
+    def get_action(self) -> "up.plans.ActionInstance":
+        """
+        Returns the next action to be taken in the current state of the problem.
+
+        :return: An instance of `ActionInstance` representing the next action to be taken.
+        """
+        return self._get_action()
+
+    def update(self, observation: Dict["up.model.FNode", "up.model.FNode"]):
+        """
+        Updates the internal state of the engine based on the given observation.
+
+        :param observation: A dictionary from observed fluents to their observed values.
+        """
+        self._update(observation)
+
+    def _get_action(self) -> "up.plans.ActionInstance":
+        """
+        Returns the next action to be taken in the current state of the problem. This method should be
+        implemented by subclasses.
+
+        :return: An instance of `ActionInstance` representing the next action to be taken.
+        """
+        raise NotImplementedError
+
+    def _update(self, observation: Dict["up.model.FNode", "up.model.FNode"]):
+        """
+        Updates the internal state of the engine based on the given observation. This method should be
+        implemented by subclasses.
+
+        :param observation: A dictionary from observed fluents to their observed values.
+        """
+        raise NotImplementedError

unified_planning/model/__init__.py

@@ -18,8 +18,8 @@
     Action,
     InstantaneousAction,
     DurativeAction,
-    SensingAction,
 )
+from unified_planning.model.contingent.sensing_action import SensingAction
 from unified_planning.model.effect import Effect, SimulatedEffect, EffectKind
 from unified_planning.model.expression import (
     BoolExpression,
@@ -35,7 +35,7 @@
 from unified_planning.model.parameter import Parameter
 from unified_planning.model.abstract_problem import AbstractProblem
 from unified_planning.model.problem import Problem
-from unified_planning.model.contingent_problem import ContingentProblem
+from unified_planning.model.contingent.contingent_problem import ContingentProblem
 from unified_planning.model.delta_stn import DeltaSimpleTemporalNetwork
 from unified_planning.model.problem_kind import ProblemKind
 from unified_planning.model.state import State, UPState

unified_planning/model/action.py

@@ -663,76 +663,3 @@ def is_conditional(self) -> bool:
         """Returns `True` if the `action` has `conditional effects`, `False` otherwise."""
         # re-implemenation needed for inheritance, delegate implementation.
         return TimedCondsEffs.is_conditional(self)
-
-
-class SensingAction(InstantaneousAction):
-    """This class represents a sensing action."""
-
-    def __init__(
-        self,
-        _name: str,
-        _parameters: Optional["OrderedDict[str, up.model.types.Type]"] = None,
-        _env: Optional[Environment] = None,
-        **kwargs: "up.model.types.Type",
-    ):
-        InstantaneousAction.__init__(self, _name, _parameters, _env, **kwargs)
-        self._observed_fluents: List["up.model.fnode.FNode"] = []
-
-    def __eq__(self, oth: object) -> bool:
-        if isinstance(oth, SensingAction):
-            return super().__eq__(oth) and set(self._observed_fluents) == set(
-                oth._observed_fluents
-            )
-        else:
-            return False
-
-    def __hash__(self) -> int:
-        res = super().__hash__()
-        for of in self._observed_fluents:
-            res += hash(of)
-        return res
-
-    def clone(self):
-        new_params = OrderedDict()
-        for param_name, param in self._parameters.items():
-            new_params[param_name] = param.type
-        new_sensing_action = SensingAction(self._name, new_params, self._environment)
-        new_sensing_action._preconditions = self._preconditions[:]
-        new_sensing_action._effects = [e.clone() for e in self._effects]
-        new_sensing_action._fluents_assigned = self._fluents_assigned.copy()
-        new_sensing_action._fluents_inc_dec = self._fluents_inc_dec.copy()
-        new_sensing_action._simulated_effect = self._simulated_effect
-        new_sensing_action._observed_fluents = self._observed_fluents.copy()
-        return new_sensing_action
-
-    def add_observed_fluents(self, observed_fluents: Iterable["up.model.fnode.FNode"]):
-        """
-        Adds the given list of observed fluents.
-
-        :param observed_fluents: The list of observed fluents that must be added.
-        """
-        for of in observed_fluents:
-            self.add_observed_fluent(of)
-
-    def add_observed_fluent(self, observed_fluent: "up.model.fnode.FNode"):
-        """
-        Adds the given observed fluent.
-
-        :param observed_fluent: The observed fluent that must be added.
-        """
-        self._observed_fluents.append(observed_fluent)
-
-    @property
-    def observed_fluents(self) -> List["up.model.fnode.FNode"]:
-        """Returns the `list` observed fluents."""
-        return self._observed_fluents
-
-    def __repr__(self) -> str:
-        b = InstantaneousAction.__repr__(self)[0:-3]
-        s = ["sensing-", b]
-        s.append("    observations = [\n")
-        for e in self._observed_fluents:
-            s.append(f"      {str(e)}\n")
-        s.append("    ]\n")
-        s.append("  }")
-        return "".join(s)

unified_planning/model/contingent/__init__.py

@@ -0,0 +1,20 @@
+# Copyright 2023 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 unified_planning.model.contingent.environment import (
+    Environment,
+    SimulatedEnvironment,
+)

unified_planning/model/contingent_problem.py → unified_planning/model/contingent/contingent_problem.py

@@ -156,7 +156,8 @@
         res = self._initial_value
         for f in self._fluents:
             for f_exp in get_all_fluent_exp(self, f):
-                res[f_exp] = self.initial_value(f_exp)
+                if f_exp not in self._hidden_fluents:
+                    res[f_exp] = self.initial_value(f_exp)
         return res
 
     @property