Newer
Older
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
act1 = None
for element in self.agenda:
if element[0] == number[0]:
act1 = element[1]
break
if number[0] in self.expanded_actions:
self.expanded_actions.remove(number[0])
return number[0], act1, actions_for_precondition[number[0]]
def find_action_for_precondition(self, oprec):
"""Find action for a given precondition"""
# either
# choose act0 E Actions such that act0 achieves G
for action in self.actions:
for effect in action.effect:
if effect == oprec:
return action, 0
# or
# choose act0 E Actions such that act0 achieves G
Donato Meoli
a validé
for action in self.planning_problem.actions:
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
for effect in action.effect:
if effect.op == oprec.op:
bindings = unify(effect, oprec)
if bindings is None:
break
return action, bindings
def generate_expr(self, clause, bindings):
"""Generate atomic expression from generic expression given variable bindings"""
new_args = []
for arg in clause.args:
if arg in bindings:
new_args.append(bindings[arg])
else:
new_args.append(arg)
try:
return Expr(str(clause.name), *new_args)
except:
return Expr(str(clause.op), *new_args)
Donato Meoli
a validé
def generate_action_object(self, action, bindings):
Donato Meoli
a validé
"""Generate action object given a generic action and variable bindings"""
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
# if bindings is 0, it means the action already exists in self.actions
if bindings == 0:
return action
# bindings cannot be None
else:
new_expr = self.generate_expr(action, bindings)
new_preconds = []
for precond in action.precond:
new_precond = self.generate_expr(precond, bindings)
new_preconds.append(new_precond)
new_effects = []
for effect in action.effect:
new_effect = self.generate_expr(effect, bindings)
new_effects.append(new_effect)
return Action(new_expr, new_preconds, new_effects)
def cyclic(self, graph):
"""Check cyclicity of a directed graph"""
new_graph = dict()
for element in graph:
if element[0] in new_graph:
new_graph[element[0]].append(element[1])
else:
new_graph[element[0]] = [element[1]]
path = set()
def visit(vertex):
path.add(vertex)
for neighbor in new_graph.get(vertex, ()):
if neighbor in path or visit(neighbor):
return True
path.remove(vertex)
return False
value = any(visit(v) for v in new_graph)
return value
def add_const(self, constraint, constraints):
"""Add the constraint to constraints if the resulting graph is acyclic"""
if constraint[0] == self.finish or constraint[1] == self.start:
return constraints
new_constraints = set(constraints)
new_constraints.add(constraint)
if self.cyclic(new_constraints):
return constraints
return new_constraints
def is_a_threat(self, precondition, effect):
"""Check if effect is a threat to precondition"""
if (str(effect.op) == 'Not' + str(precondition.op)) or ('Not' + str(effect.op) == str(precondition.op)):
if effect.args == precondition.args:
return True
return False
def protect(self, causal_link, action, constraints):
"""Check and resolve threats by promotion or demotion"""
threat = False
for effect in action.effect:
if self.is_a_threat(causal_link[1], effect):
threat = True
break
if action != causal_link[0] and action != causal_link[2] and threat:
# try promotion
new_constraints = set(constraints)
new_constraints.add((action, causal_link[0]))
if not self.cyclic(new_constraints):
constraints = self.add_const((action, causal_link[0]), constraints)
else:
# try demotion
new_constraints = set(constraints)
new_constraints.add((causal_link[2], action))
if not self.cyclic(new_constraints):
constraints = self.add_const((causal_link[2], action), constraints)
else:
# both promotion and demotion fail
print('Unable to resolve a threat caused by', action, 'onto', causal_link)
return
return constraints
def convert(self, constraints):
"""Convert constraints into a dict of Action to set orderings"""
graph = dict()
for constraint in constraints:
if constraint[0] in graph:
graph[constraint[0]].add(constraint[1])
else:
graph[constraint[0]] = set()
graph[constraint[0]].add(constraint[1])
return graph
def toposort(self, graph):
"""Generate topological ordering of constraints"""
if len(graph) == 0:
return
graph = graph.copy()
for k, v in graph.items():
v.discard(k)
extra_elements_in_dependencies = _reduce(set.union, graph.values()) - set(graph.keys())
Donato Meoli
a validé
graph.update({element: set() for element in extra_elements_in_dependencies})
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
1175
1176
1177
1178
1179
1180
1181
1182
1183
1184
1185
1186
1187
1188
1189
1190
1191
1192
1193
1194
1195
1196
1197
1198
1199
1200
1201
1202
1203
1204
1205
1206
1207
1208
1209
1210
1211
1212
1213
1214
1215
1216
1217
1218
1219
1220
1221
1222
1223
1224
1225
1226
1227
1228
1229
1230
1231
1232
1233
1234
1235
1236
while True:
ordered = set(element for element, dependency in graph.items() if len(dependency) == 0)
if not ordered:
break
yield ordered
graph = {element: (dependency - ordered) for element, dependency in graph.items() if element not in ordered}
if len(graph) != 0:
raise ValueError('The graph is not acyclic and cannot be linearly ordered')
def display_plan(self):
"""Display causal links, constraints and the plan"""
print('Causal Links')
for causal_link in self.causal_links:
print(causal_link)
print('\nConstraints')
for constraint in self.constraints:
print(constraint[0], '<', constraint[1])
print('\nPartial Order Plan')
print(list(reversed(list(self.toposort(self.convert(self.constraints))))))
def execute(self, display=True):
"""Execute the algorithm"""
step = 1
while len(self.agenda) > 0:
step += 1
# select <G, act1> from Agenda
try:
G, act1, possible_actions = self.find_open_precondition()
except IndexError:
print('Probably Wrong')
break
act0 = possible_actions[0]
# remove <G, act1> from Agenda
self.agenda.remove((G, act1))
# For actions with variable number of arguments, use least commitment principle
# act0_temp, bindings = self.find_action_for_precondition(G)
# act0 = self.generate_action_object(act0_temp, bindings)
# Actions = Actions U {act0}
self.actions.add(act0)
# Constraints = add_const(start < act0, Constraints)
self.constraints = self.add_const((self.start, act0), self.constraints)
# for each CL E CausalLinks do
# Constraints = protect(CL, act0, Constraints)
for causal_link in self.causal_links:
self.constraints = self.protect(causal_link, act0, self.constraints)
# Agenda = Agenda U {<P, act0>: P is a precondition of act0}
for precondition in act0.precond:
self.agenda.add((precondition, act0))
# Constraints = add_const(act0 < act1, Constraints)
self.constraints = self.add_const((act0, act1), self.constraints)
# CausalLinks U {<act0, G, act1>}
if (act0, G, act1) not in self.causal_links:
self.causal_links.append((act0, G, act1))
# for each A E Actions do
# Constraints = protect(<act0, G, act1>, A, Constraints)
for action in self.actions:
self.constraints = self.protect((act0, G, act1), action, self.constraints)
if step > 200:
Donato Meoli
a validé
print("Couldn't find a solution")
return None, None
if display:
self.display_plan()
else:
Donato Meoli
a validé
return self.constraints, self.causal_links
Donato Meoli
a validé
def spare_tire_graphPlan():
"""Solves the spare tire problem using GraphPlan"""
return GraphPlan(spare_tire()).execute()
Donato Meoli
a validé
def three_block_tower_graphPlan():
"""Solves the Sussman Anomaly problem using GraphPlan"""
return GraphPlan(three_block_tower()).execute()
Donato Meoli
a validé
def air_cargo_graphPlan():
"""Solves the air cargo problem using GraphPlan"""
return GraphPlan(air_cargo()).execute()
Donato Meoli
a validé
def have_cake_and_eat_cake_too_graphPlan():
"""Solves the cake problem using GraphPlan"""
return [GraphPlan(have_cake_and_eat_cake_too()).execute()[1]]
Donato Meoli
a validé
def shopping_graphPlan():
"""Solves the shopping problem using GraphPlan"""
return GraphPlan(shopping_problem()).execute()
Donato Meoli
a validé
def socks_and_shoes_graphPlan():
"""Solves the socks and shoes problem using GraphPlan"""
Donato Meoli
a validé
def simple_blocks_world_graphPlan():
"""Solves the simple blocks world problem"""
return GraphPlan(simple_blocks_world()).execute()
class HLA(Action):
"""
Define Actions for the real-world (that may be refined further), and satisfy resource
constraints.
"""
unique_group = 1
def __init__(self, action, precond=None, effect=None, duration=0,
consume=None, use=None):
"""
As opposed to actions, to define HLA, we have added constraints.
duration holds the amount of time required to execute the task
consumes holds a dictionary representing the resources the task consumes
uses holds a dictionary representing the resources the task uses
"""
precond = precond or [None]
effect = effect or [None]
super().__init__(action, precond, effect)
self.duration = duration
self.consumes = consume or {}
self.uses = use or {}
self.completed = False
# self.priority = -1 # must be assigned in relation to other HLAs
# self.job_group = -1 # must be assigned in relation to other HLAs
def do_action(self, job_order, available_resources, kb, args):
"""
An HLA based version of act - along with knowledge base updation, it handles
resource checks, and ensures the actions are executed in the correct order.
"""
# print(self.name)
if not self.has_usable_resource(available_resources):
raise Exception('Not enough usable resources to execute {}'.format(self.name))
if not self.has_consumable_resource(available_resources):
raise Exception('Not enough consumable resources to execute {}'.format(self.name))
if not self.inorder(job_order):
raise Exception("Can't execute {} - execute prerequisite actions first".
format(self.name))
kb = super().act(kb, args) # update knowledge base
for resource in self.consumes: # remove consumed resources
available_resources[resource] -= self.consumes[resource]
self.completed = True # set the task status to complete
1324
1325
1326
1327
1328
1329
1330
1331
1332
1333
1334
1335
1336
1337
1338
1339
1340
1341
1342
1343
1344
1345
1346
1347
1348
1349
1350
1351
1352
1353
1354
1355
1356
1357
1358
1359
1360
1361
def has_consumable_resource(self, available_resources):
"""
Ensure there are enough consumable resources for this action to execute.
"""
for resource in self.consumes:
if available_resources.get(resource) is None:
return False
if available_resources[resource] < self.consumes[resource]:
return False
return True
def has_usable_resource(self, available_resources):
"""
Ensure there are enough usable resources for this action to execute.
"""
for resource in self.uses:
if available_resources.get(resource) is None:
return False
if available_resources[resource] < self.uses[resource]:
return False
return True
def inorder(self, job_order):
"""
Ensure that all the jobs that had to be executed before the current one have been
successfully executed.
"""
for jobs in job_order:
if self in jobs:
for job in jobs:
if job is self:
return True
if not job.completed:
return False
return True
Donato Meoli
a validé
class RealWorldPlanningProblem(PlanningProblem):
"""
Define real-world problems by aggregating resources as numerical quantities instead of
named entities.
Donato Meoli
a validé
This class is identical to PDDL, except that it overloads the act function to handle
resource and ordering conditions imposed by HLA as opposed to Action.
"""
Donato Meoli
a validé
def __init__(self, initial, goals, actions, jobs=None, resources=None):
super().__init__(initial, goals, actions)
self.resources = resources or {}
def act(self, action):
"""
Performs the HLA given as argument.
Note that this is different from the superclass action - where the parameter was an
Expression. For real world problems, an Expr object isn't enough to capture all the
detail required for executing the action - resources, preconditions, etc need to be
checked for too.
"""
args = action.args
list_action = first(a for a in self.actions if a.name == action.name)
if list_action is None:
raise Exception("Action '{}' not found".format(action.name))
Donato Meoli
a validé
self.initial = list_action.do_action(self.jobs, self.resources, self.initial, args).clauses
Donato Meoli
a validé
def refinements(hla, library): # refinements may be (multiple) HLA themselves ...
"""
state is a Problem, containing the current state kb
library is a dictionary containing details for every possible refinement. eg:
{
'HLA': [
'Go(Home, SFO)',
'Go(Home, SFO)',
'Drive(Home, SFOLongTermParking)',
'Shuttle(SFOLongTermParking, SFO)',
'Taxi(Home, SFO)'
],
'steps': [
['Drive(Home, SFOLongTermParking)', 'Shuttle(SFOLongTermParking, SFO)'],
['Taxi(Home, SFO)'],
[],
[],
[]
],
# empty refinements indicate a primitive action
'precond': [
['At(SFOLongTermParking)'],
['At(Home)']
],
'effect': [
['At(SFO) & ~At(Home)'],
['At(SFO) & ~At(Home)'],
['At(SFOLongTermParking) & ~At(Home)'],
['At(SFO) & ~At(SFOLongTermParking)'],
['At(SFO) & ~At(Home)']
indices = [i for i, x in enumerate(library['HLA']) if expr(x).op == hla.name]
actions = []
for j in range(len(library['steps'][i])):
Donato Meoli
a validé
# find the index of the step [j] of the HLA
index_step = [k for k, x in enumerate(library['HLA']) if x == library['steps'][i][j]][0]
precond = library['precond'][index_step][0] # preconditions of step [j]
effect = library['effect'][index_step][0] # effect of step [j]
actions.append(HLA(library['steps'][i][j], precond, effect))
yield actions
def hierarchical_search(problem, hierarchy):
"""
[Figure 11.5] 'Hierarchical Search, a Breadth First Search implementation of Hierarchical
Forward Planning Search'
The problem is a real-world problem defined by the problem class, and the hierarchy is
a dictionary of HLA - refinements (see refinements generator for details)
"""
Donato Meoli
a validé
act = Node(problem.initial, None, [problem.actions[0]])
frontier = deque()
plan = frontier.popleft()
Donato Meoli
a validé
# finds the first non primitive hla in plan actions
(hla, index) = RealWorldPlanningProblem.find_hla(plan, hierarchy)
Donato Meoli
a validé
outcome = RealWorldPlanningProblem(RealWorldPlanningProblem.result(problem.initial, prefix), problem.goals,
problem.actions)
suffix = plan.action[index + 1:]
if not hla: # hla is None and plan is primitive
if outcome.goal_test():
Donato Meoli
a validé
for sequence in RealWorldPlanningProblem.refinements(hla, hierarchy): # find refinements
frontier.append(Node(outcome.initial, plan, prefix + sequence + suffix))
"""The outcome of applying an action to the current problem"""
Donato Meoli
a validé
for a in actions:
if a.check_precond(state, a.args):
state = a(state, a.args).clauses
return state
def angelic_search(problem, hierarchy, initialPlan):
"""
Donato Meoli
a validé
[Figure 11.8] A hierarchical planning algorithm that uses angelic semantics to identify and
commit to high-level plans that work while avoiding high-level plans that don’t.
The predicate MAKING-PROGRESS checks to make sure that we aren’t stuck in an infinite regression
of refinements.
At top level, call ANGELIC-SEARCH with [Act ] as the initialPlan.
Donato Meoli
a validé
InitialPlan contains a sequence of HLA's with angelic semantics
Donato Meoli
a validé
The possible effects of an angelic HLA in initialPlan are :
~ : effect remove
$+: effect possibly add
$-: effect possibly remove
$$: possibly add or remove
Donato Meoli
a validé
"""
Donato Meoli
a validé
while True:
Donato Meoli
a validé
plan = frontier.popleft() # sequence of HLA/Angelic HLA's
opt_reachable_set = RealWorldPlanningProblem.reach_opt(problem.initial, plan)
pes_reachable_set = RealWorldPlanningProblem.reach_pes(problem.initial, plan)
if problem.intersects_goal(opt_reachable_set):
if RealWorldPlanningProblem.is_primitive(plan, hierarchy):
return [x for x in plan.action]
guaranteed = problem.intersects_goal(pes_reachable_set)
if guaranteed and RealWorldPlanningProblem.making_progress(plan, initialPlan):
final_state = guaranteed[0] # any element of guaranteed
return RealWorldPlanningProblem.decompose(hierarchy, final_state, pes_reachable_set)
# there should be at least one HLA/Angelic_HLA, otherwise plan would be primitive
hla, index = RealWorldPlanningProblem.find_hla(plan, hierarchy)
Donato Meoli
a validé
suffix = plan.action[index + 1:]
outcome = RealWorldPlanningProblem(RealWorldPlanningProblem.result(problem.initial, prefix),
problem.goals, problem.actions)
for sequence in RealWorldPlanningProblem.refinements(hla, hierarchy): # find refinements
frontier.append(
AngelicNode(outcome.initial, plan, prefix + sequence + suffix, prefix + sequence + suffix))
def intersects_goal(problem, reachable_set):
"""
Find the intersection of the reachable states and the goal
"""
Donato Meoli
a validé
return [y for x in list(reachable_set.keys()) for y in reachable_set[x] if
all(goal in y for goal in problem.goals)]
Donato Meoli
a validé
def is_primitive(plan, library):
Donato Meoli
a validé
checks if the hla is primitive action
Donato Meoli
a validé
for hla in plan.action:
indices = [i for i, x in enumerate(library['HLA']) if expr(x).op == hla.name]
for i in indices:
Donato Meoli
a validé
if library["steps"][i]:
Donato Meoli
a validé
def reach_opt(init, plan):
Donato Meoli
a validé
Finds the optimistic reachable set of the sequence of actions in plan
Donato Meoli
a validé
optimistic_description = plan.action # list of angelic actions with optimistic description
return RealWorldPlanningProblem.find_reachable_set(reachable_set, optimistic_description)
Donato Meoli
a validé
def reach_pes(init, plan):
"""
Finds the pessimistic reachable set of the sequence of actions in plan
"""
reachable_set = {0: [init]}
Donato Meoli
a validé
pessimistic_description = plan.action_pes # list of angelic actions with pessimistic description
return RealWorldPlanningProblem.find_reachable_set(reachable_set, pessimistic_description)
def find_reachable_set(reachable_set, action_description):
"""
Donato Meoli
a validé
Finds the reachable states of the action_description when applied in each state of reachable set.
"""
Donato Meoli
a validé
reachable_set[i + 1] = []
if type(action_description[i]) is AngelicHLA:
possible_actions = action_description[i].angelic_action()
Donato Meoli
a validé
else:
possible_actions = action_description
for action in possible_actions:
for state in reachable_set[i]:
Donato Meoli
a validé
if action.check_precond(state, action.args):
if action.effect[0]:
Donato Meoli
a validé
reachable_set[i + 1].append(new_state)
else:
reachable_set[i + 1].append(state)
return reachable_set
def find_hla(plan, hierarchy):
"""
Finds the the first HLA action in plan.action, which is not primitive
and its corresponding index in plan.action
"""
hla = None
index = len(plan.action)
Donato Meoli
a validé
for i in range(len(plan.action)): # find the first HLA in plan, that is not primitive
if not RealWorldPlanningProblem.is_primitive(Node(plan.state, plan.parent, [plan.action[i]]), hierarchy):
hla = plan.action[i]
MariannaSpyrakou
a validé
return hla, index
Donato Meoli
a validé
"""
Prevents from infinite regression of refinements
Donato Meoli
a validé
(infinite regression of refinements happens when the algorithm finds a plan that
its pessimistic reachable set intersects the goal inside a call to decompose on
the same plan, in the same circumstances)
Donato Meoli
a validé
if plan == initialPlan[i]:
Donato Meoli
a validé
return True
Donato Meoli
a validé
def decompose(hierarchy, plan, s_f, reachable_set):
solution = []
Donato Meoli
a validé
while plan.action_pes:
Donato Meoli
a validé
if i == 0:
Donato Meoli
a validé
s_i = RealWorldPlanningProblem.find_previous_state(s_f, reachable_set, i, action)
problem = RealWorldPlanningProblem(s_i, s_f, plan.action)
angelic_call = RealWorldPlanningProblem.angelic_search(problem, hierarchy,
[AngelicNode(s_i, Node(None), [action], [action])])
Donato Meoli
a validé
for x in angelic_call:
solution.insert(0, x)
else:
Donato Meoli
a validé
i -= 1
return solution
def find_previous_state(s_f, reachable_set, i, action):
"""
Donato Meoli
a validé
Given a final state s_f and an action finds a state s_i in reachable_set
such that when action is applied to state s_i returns s_f.
Donato Meoli
a validé
s_i = reachable_set[i - 1][0]
for state in reachable_set[i - 1]:
if s_f in [x for x in
RealWorldPlanningProblem.reach_pes(state, AngelicNode(state, None, [action], [action]))[1]]:
s_i = state
def job_shop_problem():
"""
A job-shop scheduling problem for assembling two cars,
with resource and ordering constraints.
Example:
>>> from planning import *
>>> p = job_shop_problem()
>>> p.goal_test()
False
>>> p.act(p.jobs[1][0])
>>> p.act(p.jobs[1][1])
>>> p.act(p.jobs[1][2])
>>> p.act(p.jobs[0][0])
>>> p.act(p.jobs[0][1])
>>> p.goal_test()
False
>>> p.act(p.jobs[0][2])
>>> p.goal_test()
True
>>>
"""
resources = {'EngineHoists': 1, 'WheelStations': 2, 'Inspectors': 2, 'LugNuts': 500}
add_engine1 = HLA('AddEngine1', precond='~Has(C1, E1)', effect='Has(C1, E1)', duration=30, use={'EngineHoists': 1})
add_engine2 = HLA('AddEngine2', precond='~Has(C2, E2)', effect='Has(C2, E2)', duration=60, use={'EngineHoists': 1})
Donato Meoli
a validé
add_wheels1 = HLA('AddWheels1', precond='~Has(C1, W1)', effect='Has(C1, W1)', duration=30, use={'WheelStations': 1},
consume={'LugNuts': 20})
add_wheels2 = HLA('AddWheels2', precond='~Has(C2, W2)', effect='Has(C2, W2)', duration=15, use={'WheelStations': 1},
consume={'LugNuts': 20})
inspect1 = HLA('Inspect1', precond='~Inspected(C1)', effect='Inspected(C1)', duration=10, use={'Inspectors': 1})
inspect2 = HLA('Inspect2', precond='~Inspected(C2)', effect='Inspected(C2)', duration=10, use={'Inspectors': 1})
actions = [add_engine1, add_engine2, add_wheels1, add_wheels2, inspect1, inspect2]
job_group1 = [add_engine1, add_wheels1, inspect1]
job_group2 = [add_engine2, add_wheels2, inspect2]
Donato Meoli
a validé
return RealWorldPlanningProblem(
initial='Car(C1) & Car(C2) & Wheels(W1) & Wheels(W2) & Engine(E2) & Engine(E2) & ~Has(C1, E1) & ~Has(C2, '
'E2) & ~Has(C1, W1) & ~Has(C2, W2) & ~Inspected(C1) & ~Inspected(C2)',
goals='Has(C1, W1) & Has(C1, E1) & Inspected(C1) & Has(C2, W2) & Has(C2, E2) & Inspected(C2)',
actions=actions,
jobs=[job_group1, job_group2],
resources=resources)
def go_to_sfo():
"""Go to SFO Problem"""
go_home_sfo1 = HLA('Go(Home, SFO)', precond='At(Home) & Have(Car)', effect='At(SFO) & ~At(Home)')
go_home_sfo2 = HLA('Go(Home, SFO)', precond='At(Home)', effect='At(SFO) & ~At(Home)')
Donato Meoli
a validé
drive_home_sfoltp = HLA('Drive(Home, SFOLongTermParking)', precond='At(Home) & Have(Car)',
effect='At(SFOLongTermParking) & ~At(Home)')
shuttle_sfoltp_sfo = HLA('Shuttle(SFOLongTermParking, SFO)', precond='At(SFOLongTermParking)',
effect='At(SFO) & ~At(SFOLongTermParking)')
taxi_home_sfo = HLA('Taxi(Home, SFO)', precond='At(Home)', effect='At(SFO) & ~At(Home)')
actions = [go_home_sfo1, go_home_sfo2, drive_home_sfoltp, shuttle_sfoltp_sfo, taxi_home_sfo]
library = {
'HLA': [
'Go(Home, SFO)',
'Go(Home, SFO)',
'Drive(Home, SFOLongTermParking)',
'Shuttle(SFOLongTermParking, SFO)',
'Taxi(Home, SFO)'
],
'steps': [
['Drive(Home, SFOLongTermParking)', 'Shuttle(SFOLongTermParking, SFO)'],
['Taxi(Home, SFO)'],
[],
[],
[]
],
'precond': [
['At(SFOLongTermParking)'],
['At(Home)']
],
'effect': [
['At(SFO) & ~At(Home)'],
['At(SFO) & ~At(Home)'],
['At(SFOLongTermParking) & ~At(Home)'],
['At(SFO) & ~At(SFOLongTermParking)'],
['At(SFO) & ~At(Home)']
Donato Meoli
a validé
return RealWorldPlanningProblem(initial='At(Home)', goals='At(SFO)', actions=actions), library
Donato Meoli
a validé
class AngelicHLA(HLA):
"""
Define Actions for the real-world (that may be refined further), under angelic semantics
"""
Donato Meoli
a validé
def __init__(self, action, precond, effect, duration=0, consume=None, use=None):
super().__init__(action, precond, effect, duration, consume, use)
def convert(self, clauses):
"""
Converts strings into Exprs
Donato Meoli
a validé
An HLA with angelic semantics can achieve the effects of simple HLA's (add / remove a variable)
and furthermore can have following effects on the variables:
Possibly add variable ( $+ )
Possibly remove variable ( $- )
Possibly add or remove a variable ( $$ )
Overrides HLA.convert function
Donato Meoli
a validé
"""
lib = {'~': 'Not',
'$+': 'PosYes',
Donato Meoli
a validé
'$$': 'PosYesNot'}
if isinstance(clauses, Expr):
clauses = conjuncts(clauses)
for i in range(len(clauses)):
for ch in lib.keys():
if clauses[i].op == ch:
Donato Meoli
a validé
clauses[i] = expr(lib[ch] + str(clauses[i].args[0]))
elif isinstance(clauses, str):
for ch in lib.keys():
clauses = clauses.replace(ch, lib[ch])
if len(clauses) > 0:
clauses = expr(clauses)
try:
clauses = conjuncts(clauses)
except AttributeError:
pass
return clauses
def angelic_action(self):
"""
Donato Meoli
a validé
Converts a high level action (HLA) with angelic semantics into all of its corresponding high level actions (HLA).
An HLA with angelic semantics can achieve the effects of simple HLA's (add / remove a variable)
and furthermore can have following effects for each variable:
Donato Meoli
a validé
Possibly add variable ( $+: 'PosYes' ) --> corresponds to two HLAs:
HLA_1: add variable
HLA_2: leave variable unchanged
Possibly remove variable ( $-: 'PosNot' ) --> corresponds to two HLAs:
HLA_1: remove variable
HLA_2: leave variable unchanged
Donato Meoli
a validé
Possibly add / remove a variable ( $$: 'PosYesNot' ) --> corresponds to three HLAs:
HLA_1: add variable
HLA_2: remove variable
Donato Meoli
a validé
HLA_3: leave variable unchanged
example: the angelic action with effects possibly add A and possibly add or remove B corresponds to the
following 6 effects of HLAs:
'$+A & $$B': HLA_1: 'A & B' (add A and add B)
HLA_2: 'A & ~B' (add A and remove B)
HLA_3: 'A' (add A)
HLA_4: 'B' (add B)
HLA_5: '~B' (remove B)
Donato Meoli
a validé
HLA_6: ' ' (no effect)
Donato Meoli
a validé
effects = [[]]
Donato Meoli
a validé
(n, w) = AngelicHLA.compute_parameters(clause)
effects = effects * n # create n copies of effects
it = range(1)
if len(effects) != 0:
# split effects into n sublists (separate n copies created in compute_parameters)
it = range(len(effects) // n)
Donato Meoli
a validé
if clause.args:
effects[i] = expr(str(effects[i]) + '&' + str(
Expr(clause.op[w:], clause.args[0]))) # make changes in the ith part of effects
if n == 3:
effects[i + len(effects) // 3] = expr(
str(effects[i + len(effects) // 3]) + '&' + str(Expr(clause.op[6:], clause.args[0])))
else:
effects[i] = expr(
str(effects[i]) + '&' + str(expr(clause.op[w:]))) # make changes in the ith part of effects
if n == 3:
effects[i + len(effects) // 3] = expr(
str(effects[i + len(effects) // 3]) + '&' + str(expr(clause.op[6:])))
else:
if clause.args:
effects[i] = Expr(clause.op[w:], clause.args[0]) # make changes in the ith part of effects
if n == 3:
effects[i + len(effects) // 3] = Expr(clause.op[6:], clause.args[0])
else:
effects[i] = expr(clause.op[w:]) # make changes in the ith part of effects
Donato Meoli
a validé
if n == 3:
effects[i + len(effects) // 3] = expr(clause.op[6:])
# print('effects', effects)
Donato Meoli
a validé
return [HLA(Expr(self.name, self.args), self.precond, effects[i]) for i in range(len(effects))]
Donato Meoli
a validé
def compute_parameters(clause):
"""
computes n,w
Donato Meoli
a validé
n = number of HLA effects that the angelic HLA corresponds to
w = length of representation of angelic HLA effect
n = 1, if effect is add
n = 1, if effect is remove
n = 2, if effect is possibly add
n = 2, if effect is possibly remove
n = 3, if effect is possibly add or remove
"""
if clause.op[:9] == 'PosYesNot':
Donato Meoli
a validé
# possibly add/remove variable: three possible effects for the variable
n = 3
w = 9
elif clause.op[:6] == 'PosYes': # possibly add variable: two possible effects for the variable
n = 2
w = 6
elif clause.op[:6] == 'PosNot': # possibly remove variable: two possible effects for the variable
n = 2
w = 3 # We want to keep 'Not' from 'PosNot' when adding action
else: # variable or ~variable
n = 1
w = 0
return n, w
class AngelicNode(Node):
"""
Extends the class Node.
self.action: contains the optimistic description of an angelic HLA
self.action_pes: contains the pessimistic description of an angelic HLA
"""
Donato Meoli
a validé
def __init__(self, state, parent=None, action_opt=None, action_pes=None, path_cost=0):
super().__init__(state, parent, action_opt, path_cost)
self.action_pes = action_pes