fix(interpreted functions): requested fixes

unified_planning/engines/interpreted_functions_planner.py

@@ -21,7 +21,10 @@
 )
 import unified_planning.engines.mixins as mixins
 from unified_planning.engines.mixins.compiler import CompilationKind
-from unified_planning.engines.plan_validator import SequentialPlanValidator
+from unified_planning.engines.plan_validator import (
+    SequentialPlanValidator,
+    TimeTriggeredPlanValidator,
+)
 import unified_planning.engines.results
 from unified_planning.environment import get_environment
 from unified_planning.model import ProblemKind
@@ -54,9 +57,6 @@ def __init__(self, *args, **kwargs):
         MetaEngine.__init__(self, *args, **kwargs)
         mixins.OneshotPlannerMixin.__init__(self)
         self._knowledge = OrderedDict()
-        self._use_old_compiler = False
-        self._times_called = 0
-        self._time_list = list()
 
     @property
     def knowledge(self):
@@ -148,82 +148,44 @@ def _solve(
     ) -> "PlanGenerationResult":
         assert isinstance(problem, up.model.Problem)
         assert isinstance(self.engine, mixins.OneshotPlannerMixin)
-        em = problem.environment.expression_manager
-        f = problem.environment.factory
-
-        with f.Compiler(
-            problem_kind=problem.kind,
-            compilation_kind=CompilationKind.INTERPRETED_FUNCTIONS_REMOVING,
-        ) as if_remover:
-            ifr = if_remover.compile(
-                problem, CompilationKind.INTERPRETED_FUNCTIONS_REMOVING
-            )
-        retval = _attempt_to_solve(
-            self, problem, ifr, heuristic, timeout, output_stream
-        )
-        return retval
-
-
-def _attempt_to_solve(
-    self,
-    problem: "up.model.AbstractProblem",
-    compilerresult,
-    heuristic: Optional[Callable[["up.model.state.State"], Optional[float]]] = None,
-    timeout: Optional[float] = None,
-    output_stream: Optional[IO[str]] = None,
-) -> "PlanGenerationResult":
-    cres = compilerresult
-    f = problem.environment.factory
-    start = time.time()
-    if self._skip_checks:
-        self.engine._skip_checks = True
-    found_solution = False
-    while not found_solution:
-        new_problem = cres.problem
-        res = self.engine.solve(new_problem, heuristic, timeout, output_stream)
-        if timeout is not None:
-            timeout -= min(timeout, time.time() - start)
-        if res.status in up.engines.results.POSITIVE_OUTCOMES:
-            # the planner found something
-            status = res.status
-            mapback = cres.map_back_action_instance
-            mappedbackplan = res.plan.replace_action_instances(mapback)
-            with f.PlanValidator(
-                problem_kind=problem.kind, plan_kind=mappedbackplan.kind
-            ) as validator:
-                validation_result = validator.validate(problem, mappedbackplan)
-            if validation_result.status == ValidationResultStatus.VALID:
-                # validator says ok, return this
-                retval = PlanGenerationResult(
-                    status,
-                    mappedbackplan,
-                    self.name,
-                    log_messages=res.log_messages,
+        start = time.time()
+        knowledge: dict[up.model.InterpretedFunction, up.model.FNode] = {}
+        if self._skip_checks:
+            self.engine._skip_checks = True
+        while True:
+            if timeout is not None:
+                timeout -= time.time() - start
+                if timeout <= 0:
+                    return PlanGenerationResult(
+                        PlanGenerationResultStatus.TIMEOUT, None, self.name
+                    )
+            with InterpretedFunctionsRemover(knowledge) as if_remover:
+                comp_res = if_remover.compile(problem)
+            res = self.engine.solve(comp_res.problem, heuristic, timeout, output_stream)
+            if res.status in up.engines.results.POSITIVE_OUTCOMES:
+                assert res.plan is not None
+                plan = res.plan.replace_action_instances(
+                    comp_res.map_back_action_instance
                 )
-                found_solution = True
+                validator: Optional[
+                    up.engines.plan_validator.SequentialPlanValidator
+                    | up.engines.plan_validator.TimeTriggeredPlanValidator
+                ] = None
+                if plan.kind == up.plans.PlanKind.SEQUENTIAL_PLAN:
+                    validator = SequentialPlanValidator()
+                elif plan.kind == up.plans.PlanKind.TIME_TRIGGERED_PLAN:
+                    validator = TimeTriggeredPlanValidator()
+                else:
+                    raise
+                validation_result = validator.validate(problem, plan)
+                if validation_result.status == ValidationResultStatus.VALID:
+                    return PlanGenerationResult(
+                        res.status, plan, self.name, log_messages=res.log_messages
+                    )
+                else:
+                    if validation_result.calculated_interpreted_functions is not None:
+                        knowledge.update(
+                            validation_result.calculated_interpreted_functions
+                        )
             else:
-                # validator says not ok, refine and retry
-                cres = _refine(self, problem, validation_result)
-
-        else:
-            # negative planner outcome, this is not solvable
-            retval = PlanGenerationResult(res.status, None, self.name)
-            found_solution = True
-    return retval
-
-
-def _refine(self, problem, validation_result):
-    newProb = None
-    if validation_result.calculated_interpreted_functions is None:
-        pass
-    elif len(validation_result.calculated_interpreted_functions) == 0:
-        pass
-    else:
-        for k in validation_result.calculated_interpreted_functions:
-            self.add_knowledge(k, validation_result.calculated_interpreted_functions[k])
-        with InterpretedFunctionsRemover(self.knowledge) as if_remover:
-            newProb = if_remover.compile(
-                problem,
-                CompilationKind.INTERPRETED_FUNCTIONS_REMOVING,
-            )
-    return newProb
+                return PlanGenerationResult(res.status, None, self.name)

unified_planning/engines/plan_validator.py

@@ -78,7 +78,6 @@ def __init__(self, **options):
                 options.get("environment", None)
             )
         )
-        self._calculated_interpreted_functions: dict = {}
 
     @property
     def name(self):
@@ -119,6 +118,7 @@ def _validate(
         assert isinstance(problem, Problem)
         metric = None
         hasif = False
+        _calculated_interpreted_functions: dict = {}
         if (
             problem.kind.has_interpreted_functions_in_conditions()
             or problem.kind.has_interpreted_functions_in_numeric_assignments()
@@ -161,9 +161,7 @@ def _validate(
                     msg = f"Preconditions {unsat_conds} of {str(i)}-th action instance {str(ai)} are not satisfied."
 
                     if hasif:
-                        self._calculated_interpreted_functions = (
-                            simulator.get_knowledge()
-                        )
+                        _calculated_interpreted_functions = simulator.get_knowledge()
                 else:
                     next_state = simulator.apply_unsafe(trace[-1], ai)
             except UPUsageError as e:
@@ -183,7 +181,7 @@ def _validate(
                     reason=FailedValidationReason.INAPPLICABLE_ACTION,
                     inapplicable_action=ai,
                     trace=trace,
-                    calculated_interpreted_functions=self._calculated_interpreted_functions,
+                    calculated_interpreted_functions=_calculated_interpreted_functions,
                 )
             assert next_state is not None
             if metric is not None:
@@ -241,7 +239,7 @@ def __init__(self, **options):
                 options.get("environment", None)
             )
         )
-        self._calculated_interpreted_functions: dict = {}
+        # self._calculated_interpreted_functions: dict = {}
 
     @property
     def name(self):
@@ -321,7 +319,6 @@ def _apply_effects(
                     else:
                         updates[f] = v
                         assigned[f] = ai
-        self.update_knowledge(se)
         return state.make_child(updated_values=updates)
 
     def _apply_effect(
@@ -356,7 +353,6 @@ def _apply_effect(
                     result[g_fluent] = se.evaluate(
                         em.Plus(f_value, g_value), state=state
                     )
-        self.update_knowledge(se)
         return result
 
     def _states_in_interval(
@@ -388,7 +384,6 @@ def _check_condition(
         self, state: State, se: StateEvaluator, condition: FNode
     ) -> bool:
         ret_val = se.evaluate(condition, state=state).bool_constant_value()
-        self.update_knowledge(se)
         return ret_val
 
     def _instantiate_timing(
@@ -655,7 +650,7 @@ def _validate(
                             reason=FailedValidationReason.INAPPLICABLE_ACTION,
                             inapplicable_action=opt_ai,
                             trace={k: v for k, v in trace.items() if k <= end},
-                            calculated_interpreted_functions=self._calculated_interpreted_functions,
+                            calculated_interpreted_functions=se.if_values,
                         )
                     else:
                         return ValidationResult(
@@ -687,8 +682,3 @@ def _validate(
             metric_evaluations=None,
             trace=trace,
         )
-
-    def update_knowledge(self, se):
-        for k in se.if_values.keys():
-            if k not in self._calculated_interpreted_functions.keys():
-                self._calculated_interpreted_functions[k] = se.if_values[k]

unified_planning/engines/results.py

@@ -279,7 +279,9 @@ class ValidationResult(Result):
     trace: Optional[
         Union[List[up.model.State], Dict[Fraction, up.model.State]]
     ] = field(default=None)
-    calculated_interpreted_functions: Optional[dict] = field(default=None)
+    calculated_interpreted_functions: Optional[
+        dict[up.model.InterpretedFunction, up.model.FNode]
+    ] = field(default=None)
 
     def __post_init__(self):
         assert (

unified_planning/engines/sequential_simulator.py

@@ -99,7 +99,6 @@ def __init__(
         Engine.__init__(self)
         SequentialSimulatorMixin.__init__(self, problem, error_on_failed_checks)
         pk = problem.kind
-        self._if_values: OrderedDict = OrderedDict()
         if not Grounder.supports(pk):
             msg = f"The Grounder used in the {type(self).__name__} does not support the given problem"
             if self.error_on_failed_checks:
@@ -352,9 +351,6 @@ def _evaluate_effect(
             evaluated_condition = (
                 not effect.is_conditional() or evaluate(effect.condition).is_true()
             )
-            for k, v in self._se.if_values.items():
-                if k not in self._if_values.keys():
-                    self._if_values[k] = v
         if evaluated_condition:
             new_value = evaluate(effect.value)
             if effect.is_assignment():
@@ -462,10 +458,6 @@ def get_unsatisfied_conditions(
         unsatisfied_conditions = []
         for c in g_action.preconditions:
             evaluated_cond = evaluate(c)
-            for k, v in self._se.if_values.items():
-                if k not in self._if_values.keys():
-                    self._if_values[k] = v
-
             if (
                 not evaluated_cond.is_bool_constant()
                 or not evaluated_cond.bool_constant_value()
@@ -664,7 +656,7 @@ def supports(problem_kind):
         return problem_kind <= UPSequentialSimulator.supported_kind()
 
     def get_knowledge(self):
-        return self._if_values
+        return self._se.if_values
 
 
 def evaluate_quality_metric(

unified_planning/model/walkers/quantifier_simplifier.py

@@ -43,7 +43,6 @@ def __init__(
         self._problem = problem
         self._assignments: Optional[Dict["Expression", "Expression"]] = None
         self._variable_assignments: Optional[Dict["Expression", "Expression"]] = None
-        self.if_values: dict = {}
 
     def qsimplify(
         self,
@@ -115,11 +114,6 @@ def _compute_node_result(self, expression: "FNode", **kwargs):
                     if not (af.type.is_compatible(av.type)):
                         compatible_substitution = False
                 result = self.memoization[key]
-                if compatible_substitution:
-                    temp_key = key.interpreted_function()
-                    new_key = temp_key(*args_values)
-                    if new_key not in self.if_values.keys():
-                        self.if_values[new_key] = result
 
     def _deep_subs_simplify(
         self,

unified_planning/model/walkers/simplifier.py

@@ -346,17 +346,12 @@ def walk_interpreted_function_exp(
         constantval = expression.interpreted_function().function(*newlist)
         if expression.interpreted_function().return_type.is_bool_type():
             constantval = self.manager.Bool((constantval))
-
         elif expression.interpreted_function().return_type.is_int_type():
-
             constantval = self.manager.Int((constantval))
         elif expression.interpreted_function().return_type.is_real_type():
-
             constantval = self.manager.Real((Fraction(constantval)))
-        elif (
-            expression.interpreted_function().return_type.is_user_type()
-        ):  # NOTE - not tested
-
+        elif expression.interpreted_function().return_type.is_user_type():
+            # NOTE - not tested
             constantval = self.manager.ObjectExp((constantval))
         else:
             return new_exp

unified_planning/model/walkers/state_evaluator.py

@@ -28,6 +28,7 @@ class StateEvaluator(QuantifierSimplifier):
 
     def __init__(self, problem: "up.model.problem.Problem"):
         QuantifierSimplifier.__init__(self, problem.environment, problem)
+        self.if_values: dict[up.model.InterpretedFunction, up.model.FNode] = {}
 
     def evaluate(
         self,
@@ -81,3 +82,14 @@ def walk_param_exp(self, expression: "FNode", args: List["FNode"]) -> "FNode":
         raise UPProblemDefinitionError(
             f"The StateEvaluator.evaluate should only be called on grounded expressions."
         )
+
+    def iter_walk(self, expression: FNode, **kwargs):
+        res = super(QuantifierSimplifier, self).iter_walk(expression, **kwargs)
+        for key, value in self.memoization.items():
+            if key.is_interpreted_function_exp():
+                args_values = [
+                    self.memoization[self._get_key(s, **kwargs)] for s in key.args
+                ]
+                new_key = key.interpreted_function()(*args_values)
+                self.if_values[new_key] = value
+        return res