Skip to content

Commit

Permalink
feature(interpreted functions): added tests for plan validator - only…
Browse files Browse the repository at this point in the history 
… works with instantaneous actions
  • Loading branch information
Samuel Gobbi committed Sep 24, 2024
1 parent 68c0e75 commit 650e8cf
Show file tree
Hide file tree
Showing 3 changed files with 190 additions and 28 deletions.
167 changes: 139 additions & 28 deletions unified_planning/test/examples/minimals.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -551,71 +551,182 @@ def get_example_problems():
],
)

# interpreted functions ----------
problem = Problem("interpreted_functions_in_conditions")

def callableFunCon(inputone, inputtwo):
def i_f_simple_bool(inputone, inputtwo):
return (inputone * inputtwo) == 60

signatureConditionF = OrderedDict()
signatureConditionF["inputone"] = IntType()
signatureConditionF["inputtwo"] = IntType()
funx = InterpretedFunction("funx", BoolType(), signatureConditionF, callableFunCon)
eg = Fluent("eg")
funx = InterpretedFunction("funx", BoolType(), signatureConditionF, i_f_simple_bool)
end_goal = Fluent("end_goal")
ione = Fluent("ione", IntType(0, 20))
itwo = Fluent("itwo", IntType(0, 20))
awif = InstantaneousAction("awif")
awif.add_precondition(And(GE(ione, 10), Not(funx(itwo, itwo))))
awif.add_precondition((funx(ione, itwo)))
awif.add_effect(eg, True)
problem.add_fluent(eg)
instant_action_i_f_condition = InstantaneousAction("instant_action_i_f_condition")
instant_action_i_f_condition.add_precondition(
And(GE(ione, 10), Not(funx(itwo, itwo)))
)
instant_action_i_f_condition.add_precondition((funx(ione, itwo)))
instant_action_i_f_condition.add_effect(end_goal, Not(end_goal))
problem.add_fluent(end_goal)
problem.add_fluent(ione)
problem.add_fluent(itwo)
problem.add_action(awif)
problem.set_initial_value(eg, False)
problem.add_action(instant_action_i_f_condition)
problem.set_initial_value(end_goal, False)
problem.set_initial_value(ione, 12)
problem.set_initial_value(itwo, 5)
problem.add_goal(eg)
problem.add_goal(end_goal)
ifproblem = TestCase(
problem=problem,
solvable=True,
valid_plans=[up.plans.SequentialPlan([awif()])],
valid_plans=[
up.plans.SequentialPlan([instant_action_i_f_condition()]),
up.plans.SequentialPlan(
[
instant_action_i_f_condition(),
instant_action_i_f_condition(),
instant_action_i_f_condition(),
]
),
],
invalid_plans=[
up.plans.SequentialPlan([]),
up.plans.SequentialPlan(
[instant_action_i_f_condition(), instant_action_i_f_condition()]
),
up.plans.SequentialPlan(
[
instant_action_i_f_condition(),
instant_action_i_f_condition(),
instant_action_i_f_condition(),
instant_action_i_f_condition(),
]
),
],
) # solvable=True, valid_plans=[up.plans.SequentialPlan([awif])]
)
problems["interpreted_functions_in_conditions"] = ifproblem

signatureConditionF = OrderedDict()
signatureConditionF["inputone"] = IntType()
signatureConditionF["inputtwo"] = IntType()
funx = InterpretedFunction("funx", BoolType(), signatureConditionF, callableFunCon)
funx = InterpretedFunction("funx", BoolType(), signatureConditionF, i_f_simple_bool)

g = Fluent("g")
end_goal = Fluent("end_goal")
ione = Fluent("ione", IntType(0, 20))
itwo = Fluent("itwo", IntType(0, 20))

da = DurativeAction("a")
durative_action_i_f_condition = DurativeAction("durative_action_i_f_condition")

da.add_condition(EndTiming(), And(GE(ione, 10), Not(funx(itwo, itwo))))
da.add_condition(StartTiming(), (funx(ione, itwo)))
da.add_condition(StartTiming(), Not(And(GE(ione, 15), LE(itwo, 5))))
da.add_effect(EndTiming(), g, True)
da.set_fixed_duration(5)
durative_action_i_f_condition.add_condition(
EndTiming(), And(GE(ione, 10), Not(funx(itwo, itwo)))
)
durative_action_i_f_condition.add_condition(StartTiming(), (funx(ione, itwo)))
durative_action_i_f_condition.add_condition(
StartTiming(), Not(And(GE(ione, 15), LE(itwo, 5)))
)
durative_action_i_f_condition.add_effect(EndTiming(), end_goal, True)
durative_action_i_f_condition.set_fixed_duration(5)

problem = Problem("interpreted_functions_in_durative_conditions")
problem.add_fluent(g)
problem.add_fluent(end_goal)
problem.add_fluent(ione)
problem.add_fluent(itwo)
problem.add_action(da)
problem.set_initial_value(g, False)
problem.add_action(durative_action_i_f_condition)
problem.set_initial_value(end_goal, False)
problem.set_initial_value(ione, 12)
problem.set_initial_value(itwo, 5)
problem.add_goal(g)
problem.add_goal(end_goal)

ifproblem = TestCase(
problem=problem,
solvable=True,
valid_plans=[
up.plans.SequentialPlan([durative_action_i_f_condition()]),
],
invalid_plans=[
up.plans.SequentialPlan([]),
],
)
problems["interpreted_functions_in_durative_conditions"] = ifproblem

def i_f_go_home_duration(
israining, basetime
): # if it rains you will take longer to walk home
r = basetime
if israining:
r = basetime * 1.4
return r

Check warning on line 657 in unified_planning/test/examples/minimals.py

View check run for this annotation

Codecov / codecov/patch

unified_planning/test/examples/minimals.py#L654-L657 

Added lines #L654 - L657 were not covered by tests

gohomedurationsignature = OrderedDict()
gohomedurationsignature["israining"] = BoolType()
gohomedurationsignature["basetime"] = IntType()
gohomeduration = InterpretedFunction(
"gohomeduration", RealType(), gohomedurationsignature, i_f_go_home_duration
)

athome = Fluent("athome")
rain = Fluent("rain")
normaltime = Fluent("normaltime", IntType(0, 20))

gohome = DurativeAction("gohome")
gohome.add_condition(StartTiming(), Not(athome))
gohome.add_effect(EndTiming(), athome, True)
gohome.set_fixed_duration(gohomeduration(rain, normaltime))

problem = Problem("interpreted_functions_in_durations")
problem.add_fluent(athome)
problem.add_fluent(rain)
problem.add_fluent(normaltime)
problem.add_action(gohome)
problem.set_initial_value(athome, False)
problem.set_initial_value(rain, True)
problem.set_initial_value(normaltime, 10)
problem.add_goal(athome)
ifproblem = TestCase(
problem=problem,
solvable=True,
valid_plans=[
up.plans.SequentialPlan([gohome()]),
],
invalid_plans=[
up.plans.SequentialPlan([]),
],
)
problems["interpreted_functions_in_durations"] = ifproblem

funx = InterpretedFunction("funx", BoolType(), signatureConditionF, i_f_simple_bool)

ione = Fluent("ione", IntType(0, 20))
itwo = Fluent("itwo", IntType(0, 20))
end_goal = Fluent("end_goal")

apply_i_f_assignment = InstantaneousAction("apply_i_f_assignment")
apply_i_f_assignment.add_effect(end_goal, funx(ione, itwo))

increase_val = InstantaneousAction("increase_val")
increase_val.add_effect(ione, Plus(ione, 2))

problem = Problem("interpreted_functions_in_boolean_assignment")
problem.add_fluent(ione)
problem.add_fluent(itwo)
problem.add_fluent(end_goal)

problem.add_action(apply_i_f_assignment)
problem.add_action(increase_val)
problem.set_initial_value(ione, 3)
problem.set_initial_value(itwo, 12)
problem.set_initial_value(end_goal, False)
problem.add_goal(end_goal)

ifproblem = TestCase(
problem=problem,
solvable=True,
valid_plans=[
up.plans.SequentialPlan([increase_val(), apply_i_f_assignment()]),
],
invalid_plans=[
up.plans.SequentialPlan([apply_i_f_assignment(), increase_val()]),
],
)
problems["interpreted_functions_in_boolean_assignment"] = ifproblem

return problems
46 changes: 46 additions & 0 deletions unified_planning/test/test_plan_validator.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -148,3 +148,49 @@ def test_state_invariants(self):
problem.clear_trajectory_constraints()
validation_result = pv.validate(problem, bad_plan)
self.assertEqual(validation_result.status, ValidationResultStatus.VALID)

def test_with_interpreted_functions(self):
spv = SequentialPlanValidator(environment=get_environment())
spv.skip_checks = True # have to use this

p = self.problems["interpreted_functions_in_conditions"]
problem = p.problem
for plan in p.valid_plans:
validation_result = spv.validate(problem, plan)
self.assertEqual(validation_result.status, ValidationResultStatus.VALID)
for plan in p.invalid_plans:
validation_result = spv.validate(problem, plan)
self.assertEqual(validation_result.status, ValidationResultStatus.INVALID)

p = self.problems["interpreted_functions_in_boolean_assignment"]
problem = p.problem
for plan in p.valid_plans:
validation_result = spv.validate(problem, plan)
self.assertEqual(validation_result.status, ValidationResultStatus.VALID)
for plan in p.invalid_plans:
validation_result = spv.validate(problem, plan)
self.assertEqual(validation_result.status, ValidationResultStatus.INVALID)

# does not work with durations

# p = self.problems["interpreted_functions_in_durations"]
# problem = p.problem
# for plan in p.valid_plans:
# validation_result = spv.validate(problem, plan)
# self.assertEqual(validation_result.status, ValidationResultStatus.VALID)
# for plan in p.invalid_plans:
# validation_result = spv.validate(problem, plan)
# self.assertEqual(validation_result.status, ValidationResultStatus.INVALID)

# p = self.problems["interpreted_functions_in_durative_conditions"]
# problem = p.problem
# for plan in p.valid_plans:
# validation_result = spv.validate(problem, plan)
# self.assertEqual(
# validation_result.status, ValidationResultStatus.VALID
# )
# for plan in p.invalid_plans:
# validation_result = spv.validate(problem, plan)
# self.assertEqual(
# validation_result.status, ValidationResultStatus.INVALID
# )
5 changes: 5 additions & 0 deletions unified_planning/test/test_problem.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -611,6 +611,11 @@ def test_interpreted_functions_simple(self):
self.assertFalse(problem.kind.has_simple_numeric_planning())
problem = self.problems["interpreted_functions_in_durative_conditions"].problem
self.assertTrue(problem.kind.has_interpreted_functions_in_conditions())
problem = self.problems["interpreted_functions_in_durations"].problem
self.assertTrue(problem.kind.has_interpreted_functions_in_durations())
problem = self.problems["interpreted_functions_in_boolean_assignment"].problem
self.assertTrue(problem.kind.has_interpreted_functions_in_boolean_assignments())
self.assertFalse(problem.kind.has_simple_numeric_planning())
# should be changed to false once other checks are implemented in kind

# x = problem.fluent("x")
Expand Down

0 comments on commit 650e8cf

Please sign in to comment.