"""

def __init__(self, initial, goals, actions, domain=None):

self.initial = self.convert(initial) if domain is None else self.convert(initial) + self.convert(domain)

self.goals = self.convert(goals)

self.actions = actions

self.domain = domain

def convert(self, clauses):

"""Converts strings into exprs"""

if not isinstance(clauses, Expr):

if len(clauses) > 0:

clauses = expr(clauses)

else:

clauses = []

try:

clauses = conjuncts(clauses)

except AttributeError:

pass

new_clauses = []

for clause in clauses:

if clause.op == '~':

new_clauses.append(expr('Not' + str(clause.args[0])))

else:

new_clauses.append(clause)

return new_clauses

def expand_fluents(self, name=None):

kb = None

if self.domain:

kb = FolKB(self.convert(self.domain))

for action in self.actions:

if action.precond:

for fests in set(action.precond).union(action.effect).difference(self.convert(action.domain)):

if fests.op[:3] != 'Not':

kb.tell(expr(str(action.domain) + ' ==> ' + str(fests)))

objects = set(arg for clause in set(self.initial + self.goals) for arg in clause.args)

fluent_list = []

if name is not None:

for fluent in self.initial + self.goals:

if str(fluent) == name:

fluent_list.append(fluent)

break

else:

fluent_list = list(map(lambda fluent: Expr(fluent[0], *fluent[1]),

{fluent.op: fluent.args for fluent in self.initial + self.goals +

[clause for action in self.actions for clause in action.effect if

clause.op[:3] != 'Not']}.items()))

expansions = []

for fluent in fluent_list:

for permutation in itertools.permutations(objects, len(fluent.args)):

new_fluent = Expr(fluent.op, *permutation)

if (self.domain and kb.ask(new_fluent) is not False) or not self.domain:

expansions.append(new_fluent)

return expansions

def expand_actions(self, name=None):

"""Generate all possible actions with variable bindings for precondition selection heuristic"""

has_domains = all(action.domain for action in self.actions if action.precond)

kb = None

if has_domains:

kb = FolKB(self.initial)

for action in self.actions:

if action.precond:

kb.tell(expr(str(action.domain) + ' ==> ' + str(action)))

objects = set(arg for clause in self.initial for arg in clause.args)

expansions = []

action_list = []

if name is not None:

for action in self.actions:

if str(action.name) == name:

action_list.append(action)

break

else:

action_list = self.actions

for action in action_list:

for permutation in itertools.permutations(objects, len(action.args)):

bindings = unify_mm(Expr(action.name, *action.args), Expr(action.name, *permutation))

if bindings is not None:

new_args = []

for arg in action.args:

if arg in bindings:

new_args.append(bindings[arg])

else:

new_args.append(arg)

new_expr = Expr(str(action.name), *new_args)

if (has_domains and kb.ask(new_expr) is not False) or (

has_domains and not action.precond) or not has_domains:

new_preconds = []

for precond in action.precond:

new_precond_args = []

for arg in precond.args:

if arg in bindings:

new_precond_args.append(bindings[arg])

else:

new_precond_args.append(arg)

new_precond = Expr(str(precond.op), *new_precond_args)

new_preconds.append(new_precond)

new_effects = []

for effect in action.effect:

new_effect_args = []

for arg in effect.args:

if arg in bindings:

new_effect_args.append(bindings[arg])

else:

new_effect_args.append(arg)

new_effect = Expr(str(effect.op), *new_effect_args)

new_effects.append(new_effect)

expansions.append(Action(new_expr, new_preconds, new_effects))

return expansions

def is_strips(self):

"""

return (all(clause.op[:3] != 'Not' for clause in self.goals) and

all(clause.op[:3] != 'Not' for action in self.actions for clause in action.precond))

def goal_test(self):

"""Checks if the goals have been reached"""

return all(goal in self.initial for goal in self.goals)

def act(self, action):

"""

action_name = action.op

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))

if not list_action.check_precond(self.initial, args):

raise Exception("Action '{}' pre-conditions not satisfied".format(action))

"""

def __init__(self, action, precond, effect, domain=None):

if isinstance(action, str):

action = expr(action)

self.name = action.op

self.args = action.args

self.precond = self.convert(precond) if domain is None else self.convert(precond) + self.convert(domain)

self.effect = self.convert(effect)

self.domain = domain

def __call__(self, kb, args):

return self.act(kb, args)

def __repr__(self):

return '{}'.format(Expr(self.name, *self.args))

def convert(self, clauses):

"""Converts strings into Exprs"""

if isinstance(clauses, Expr):

clauses = conjuncts(clauses)

for i in range(len(clauses)):

if clauses[i].op == '~':

clauses[i] = expr('Not' + str(clauses[i].args[0]))

elif isinstance(clauses, str):

clauses = clauses.replace('~', 'Not')

if len(clauses) > 0:

clauses = expr(clauses)

try:

clauses = conjuncts(clauses)

except AttributeError:

pass

return clauses

def relaxed(self):

"""

return Action(Expr(self.name, *self.args), self.precond,

list(filter(lambda effect: effect.op[:3] != 'Not', self.effect)))

def substitute(self, e, args):

"""Replaces variables in expression with their respective Propositional symbol"""

new_args = list(e.args)

for num, x in enumerate(e.args):

for i, _ in enumerate(self.args):

if self.args[i] == x:

new_args[num] = args[i]

return Expr(e.op, *new_args)

def check_precond(self, kb, args):

"""Checks if the precondition is satisfied in the current state"""

if isinstance(kb, list):

kb = FolKB(kb)

for clause in self.precond:

if self.substitute(clause, args) not in kb.clauses:

return False

return True

def act(self, kb, args):

"""Executes the action on the state's knowledge base"""

if isinstance(kb, list):

kb = FolKB(kb)

if not self.check_precond(kb, args):

raise Exception('Action pre-conditions not satisfied')

for clause in self.effect:

kb.tell(self.substitute(clause, args))

if clause.op[:3] == 'Not':

new_clause = Expr(clause.op[3:], *clause.args)

if kb.ask(self.substitute(new_clause, args)) is not False:

kb.retract(self.substitute(new_clause, args))

else:

new_clause = Expr('Not' + clause.op, *clause.args)

if kb.ask(self.substitute(new_clause, args)) is not False:

kb.retract(self.substitute(new_clause, args))

"""Generic goal testing helper function"""

if isinstance(state, list):

kb = FolKB(state)

else:

kb = state

"""

return PlanningProblem(initial='At(C1, SFO) & At(C2, JFK) & At(P1, SFO) & At(P2, JFK)',

goals='At(C1, JFK) & At(C2, SFO)',

actions=[Action('Load(c, p, a)',

precond='At(c, a) & At(p, a)',

effect='In(c, p) & ~At(c, a)',

domain='Cargo(c) & Plane(p) & Airport(a)'),

Action('Unload(c, p, a)',

precond='In(c, p) & At(p, a)',

effect='At(c, a) & ~In(c, p)',

domain='Cargo(c) & Plane(p) & Airport(a)'),

Action('Fly(p, f, to)',

precond='At(p, f)',

effect='At(p, to) & ~At(p, f)',

domain='Plane(p) & Airport(f) & Airport(to)')],

"""

return PlanningProblem(initial='At(Flat, Axle) & At(Spare, Trunk)',

goals='At(Spare, Axle) & At(Flat, Ground)',

actions=[Action('Remove(obj, loc)',

precond='At(obj, loc)',

effect='At(obj, Ground) & ~At(obj, loc)',

domain='Tire(obj)'),

Action('PutOn(t, Axle)',

precond='At(t, Ground) & ~At(Flat, Axle)',

effect='At(t, Axle) & ~At(t, Ground)',

domain='Tire(t)'),

Action('LeaveOvernight',

precond='',

effect='~At(Spare, Ground) & ~At(Spare, Axle) & ~At(Spare, Trunk) & \

~At(Flat, Ground) & ~At(Flat, Axle) & ~At(Flat, Trunk)')],

"""

return PlanningProblem(initial='On(A, Table) & On(B, Table) & On(C, A) & Clear(B) & Clear(C)',

goals='On(A, B) & On(B, C)',

actions=[Action('Move(b, x, y)',

precond='On(b, x) & Clear(b) & Clear(y)',

effect='On(b, y) & Clear(x) & ~On(b, x) & ~Clear(y)',

domain='Block(b) & Block(y)'),

Action('MoveToTable(b, x)',

precond='On(b, x) & Clear(b)',

effect='On(b, Table) & Clear(x) & ~On(b, x)',

domain='Block(b) & Block(x)')],

"""

return PlanningProblem(initial='On(A, B) & Clear(A) & OnTable(B) & OnTable(C) & Clear(C)',

goals='On(B, A) & On(C, B)',

actions=[Action('ToTable(x, y)',

precond='On(x, y) & Clear(x)',

effect='~On(x, y) & Clear(y) & OnTable(x)'),

Action('FromTable(y, x)',

precond='OnTable(y) & Clear(y) & Clear(x)',

"""

return PlanningProblem(initial='Have(Cake)',

goals='Have(Cake) & Eaten(Cake)',

actions=[Action('Eat(Cake)',

precond='Have(Cake)',

effect='Eaten(Cake) & ~Have(Cake)'),

Action('Bake(Cake)',

precond='~Have(Cake)',

"""

return PlanningProblem(initial='At(Home) & Sells(SM, Milk) & Sells(SM, Banana) & Sells(HW, Drill)',

goals='Have(Milk) & Have(Banana) & Have(Drill)',

actions=[Action('Buy(x, store)',

precond='At(store) & Sells(store, x)',

effect='Have(x)',

domain='Store(store) & Item(x)'),

Action('Go(x, y)',

precond='At(x)',

effect='At(y) & ~At(x)',

domain='Place(x) & Place(y)')],

domain='Place(Home) & Place(SM) & Place(HW) & Store(SM) & Store(HW) & '

"""

return PlanningProblem(initial='',

goals='RightShoeOn & LeftShoeOn',

actions=[Action('RightShoe',

precond='RightSockOn',

effect='RightShoeOn'),

Action('RightSock',

precond='',

effect='RightSockOn'),

Action('LeftShoe',

precond='LeftSockOn',

effect='LeftShoeOn'),

Action('LeftSock',

precond='',

"""

return PlanningProblem(

initial='At(A, LeftBaseLine) & At(B, RightNet) & Approaching(Ball, RightBaseLine) & Partner(A, B) & Partner(B, A)',

goals='Returned(Ball) & At(a, LeftNet) & At(a, RightNet)',

actions=[Action('Hit(actor, Ball, loc)',

precond='Approaching(Ball, loc) & At(actor, loc)',

effect='Returned(Ball)'),

Action('Go(actor, to, loc)',

precond='At(actor, loc)',

"""

def __init__(self, planning_problem):

super().__init__(associate('&', planning_problem.initial), associate('&', planning_problem.goals))

self.planning_problem = planning_problem

self.expanded_actions = self.planning_problem.expand_actions()

def actions(self, state):

return [action for action in self.expanded_actions if all(pre in conjuncts(state) for pre in action.precond)]

def result(self, state, action):

return associate('&', action(conjuncts(state), action.args).clauses)

def goal_test(self, state):

return all(goal in conjuncts(state) for goal in self.planning_problem.goals)

def h(self, state):

"""

relaxed_planning_problem = PlanningProblem(initial=state.state,

goals=self.goal,

actions=[action.relaxed() for action in

self.planning_problem.actions])

try:

return len(linearize(GraphPlan(relaxed_planning_problem).execute()))

except:

"""

def __init__(self, planning_problem):

super().__init__(associate('&', planning_problem.goals), associate('&', planning_problem.initial))

self.planning_problem = planning_problem

self.expanded_actions = self.planning_problem.expand_actions()

def actions(self, subgoal):

"""

def negate_clause(clause):

return Expr(clause.op.replace('Not', ''), *clause.args) if clause.op[:3] == 'Not' else Expr(

'Not' + clause.op, *clause.args)

subgoal = conjuncts(subgoal)

return [action for action in self.expanded_actions if

(any(prop in action.effect for prop in subgoal) and

not any(negate_clause(prop) in subgoal for prop in action.effect) and

not any(negate_clause(prop) in subgoal and negate_clause(prop) not in action.effect

for prop in action.precond))]

def result(self, subgoal, action):

# g' = (g - effects(a)) + preconds(a)

return associate('&', set(set(conjuncts(subgoal)).difference(action.effect)).union(action.precond))

def goal_test(self, subgoal):

return all(goal in conjuncts(self.goal) for goal in conjuncts(subgoal))

def h(self, subgoal):

"""

relaxed_planning_problem = PlanningProblem(initial=self.goal,

goals=subgoal.state,

actions=[action.relaxed() for action in

self.planning_problem.actions])

try:

return len(linearize(GraphPlan(relaxed_planning_problem).execute()))

except:

"""

def st(var, stage):

"""Returns a string for the var-stage pair that can be used as a variable"""

return str(var) + "_" + str(stage)

def if_(v1, v2):

"""If the second argument is v2, the first argument must be v1"""

def if_fun(x1, x2):

return x1 == v1 if x2 == v2 else True

if_fun.__name__ = "if the second argument is " + str(v2) + " then the first argument is " + str(v1) + " "

return if_fun

def eq_if_not_in_(actset):

"""First and third arguments are equal if action is not in actset"""

def eq_if_not_in(x1, a, x2):

return x1 == x2 if a not in actset else True

eq_if_not_in.__name__ = "first and third arguments are equal if action is not in " + str(actset) + " "

return eq_if_not_in

expanded_actions = planning_problem.expand_actions()

fluent_values = planning_problem.expand_fluents()

for horizon in range(solution_length):

act_vars = [st('action', stage) for stage in range(horizon + 1)]

domains = {av: list(map(lambda action: expr(str(action)), expanded_actions)) for av in act_vars}

domains.update({st(var, stage): {True, False} for var in fluent_values for stage in range(horizon + 2)})

# initial state constraints

constraints = [Constraint((st(var, 0),), is_constraint(val))

for (var, val) in {expr(str(fluent).replace('Not', '')):

True if fluent.op[:3] != 'Not' else False

for fluent in planning_problem.initial}.items()]

constraints += [Constraint((st(var, 0),), is_constraint(False))

for var in {expr(str(fluent).replace('Not', ''))

for fluent in fluent_values if fluent not in planning_problem.initial}]

# goal state constraints

constraints += [Constraint((st(var, horizon + 1),), is_constraint(val))

for (var, val) in {expr(str(fluent).replace('Not', '')):

True if fluent.op[:3] != 'Not' else False

for fluent in planning_problem.goals}.items()]

# precondition constraints

constraints += [Constraint((st(var, stage), st('action', stage)), if_(val, act))

# st(var, stage) == val if st('action', stage) == act

for act, strps in {expr(str(action)): action for action in expanded_actions}.items()

for var, val in {expr(str(fluent).replace('Not', '')):

True if fluent.op[:3] != 'Not' else False

for fluent in strps.precond}.items()

for stage in range(horizon + 1)]

# effect constraints

constraints += [Constraint((st(var, stage + 1), st('action', stage)), if_(val, act))

# st(var, stage + 1) == val if st('action', stage) == act

for act, strps in {expr(str(action)): action for action in expanded_actions}.items()

for var, val in {expr(str(fluent).replace('Not', '')): True if fluent.op[:3] != 'Not' else False

for fluent in strps.effect}.items()

for stage in range(horizon + 1)]

# frame constraints

constraints += [Constraint((st(var, stage), st('action', stage), st(var, stage + 1)),

eq_if_not_in_(set(map(lambda action: expr(str(action)),

{act for act in expanded_actions if var in act.effect

or Expr('Not' + var.op, *var.args) in act.effect}))))

for var in fluent_values for stage in range(horizon + 1)]

csp = NaryCSP(domains, constraints)

sol = CSP_solver(csp, arc_heuristic=arc_heuristic)

if sol:

"""

def expand_transitions(state, actions):

state = sorted(conjuncts(state))

for action in filter(lambda act: act.check_precond(state, act.args), actions):

transition[associate('&', state)].update(

{Expr(action.name, *action.args):

associate('&', sorted(set(filter(lambda clause: clause.op[:3] != 'Not',

action(state, action.args).clauses))))

if planning_problem.is_strips()

else associate('&', sorted(set(action(state, action.args).clauses)))})

for state in transition[associate('&', state)].values():

if state not in transition:

expand_transitions(expr(state), actions)

transition = defaultdict(dict)

expand_transitions(associate('&', planning_problem.initial), planning_problem.expand_actions())

return SAT_plan(associate('&', sorted(planning_problem.initial)), transition,

"""

def __init__(self, kb):

"""Initializes variables to hold state and action details of a level"""

self.kb = kb

# current state

self.current_state = kb.clauses

# current action to state link

self.current_action_links = {}

# current state to action link

self.current_state_links = {}

# current action to next state link

self.next_action_links = {}

# next state to current action link

self.next_state_links = {}

# mutually exclusive actions

self.mutex = []

def __call__(self, actions, objects):

self.build(actions, objects)

self.find_mutex()

def separate(self, e):

"""Separates an iterable of elements into positive and negative parts"""

positive = []

negative = []

for clause in e:

if clause.op[:3] == 'Not':

negative.append(clause)

else:

positive.append(clause)

return positive, negative

def find_mutex(self):

"""Finds mutually exclusive actions"""

# Inconsistent effects

pos_nsl, neg_nsl = self.separate(self.next_state_links)

for negeff in neg_nsl:

new_negeff = Expr(negeff.op[3:], *negeff.args)

for poseff in pos_nsl:

if new_negeff == poseff:

for a in self.next_state_links[poseff]:

for b in self.next_state_links[negeff]:

if {a, b} not in self.mutex:

self.mutex.append({a, b})

# Interference will be calculated with the last step

pos_csl, neg_csl = self.separate(self.current_state_links)

# Competing needs

for pos_precond in pos_csl:

for neg_precond in neg_csl:

new_neg_precond = Expr(neg_precond.op[3:], *neg_precond.args)

if new_neg_precond == pos_precond:

for a in self.current_state_links[pos_precond]:

for b in self.current_state_links[neg_precond]:

if {a, b} not in self.mutex:

self.mutex.append({a, b})

# Inconsistent support

state_mutex = []

for pair in self.mutex:

next_state_0 = self.next_action_links[list(pair)[0]]

if len(pair) == 2:

next_state_1 = self.next_action_links[list(pair)[1]]

else:

next_state_1 = self.next_action_links[list(pair)[0]]

if (len(next_state_0) == 1) and (len(next_state_1) == 1):

state_mutex.append({next_state_0[0], next_state_1[0]})

self.mutex = self.mutex + state_mutex

def build(self, actions, objects):

"""Populates the lists and dictionaries containing the state action dependencies"""

for clause in self.current_state:

p_expr = Expr('P' + clause.op, *clause.args)

self.current_action_links[p_expr] = [clause]

self.next_action_links[p_expr] = [clause]

self.current_state_links[clause] = [p_expr]

self.next_state_links[clause] = [p_expr]

for a in actions:

num_args = len(a.args)

possible_args = tuple(itertools.permutations(objects, num_args))

for arg in possible_args:

if a.check_precond(self.kb, arg):

for num, symbol in enumerate(a.args):

if not symbol.op.islower():

arg = list(arg)

arg[num] = symbol

arg = tuple(arg)

new_action = a.substitute(Expr(a.name, *a.args), arg)

self.current_action_links[new_action] = []

for clause in a.precond:

new_clause = a.substitute(clause, arg)

self.current_action_links[new_action].append(new_clause)

if new_clause in self.current_state_links:

self.current_state_links[new_clause].append(new_action)

else:

self.current_state_links[new_clause] = [new_action]

self.next_action_links[new_action] = []

for clause in a.effect:

new_clause = a.substitute(clause, arg)

self.next_action_links[new_action].append(new_clause)

if new_clause in self.next_state_links:

self.next_state_links[new_clause].append(new_action)

else:

self.next_state_links[new_clause] = [new_action]

def perform_actions(self):

"""Performs the necessary actions and returns a new Level"""

new_kb = FolKB(list(set(self.next_state_links.keys())))

"""

def __init__(self, planning_problem):

self.planning_problem = planning_problem

self.kb = FolKB(planning_problem.initial)

self.levels = [Level(self.kb)]

self.objects = set(arg for clause in self.kb.clauses for arg in clause.args)

def __call__(self):

self.expand_graph()

def expand_graph(self):

"""Expands the graph by a level"""

last_level = self.levels[-1]

last_level(self.planning_problem.actions, self.objects)

self.levels.append(last_level.perform_actions())

def non_mutex_goals(self, goals, index):

"""Checks whether the goals are mutually exclusive"""

goal_perm = itertools.combinations(goals, 2)

for g in goal_perm:

if set(g) in self.levels[index].mutex:

return False

"""

def __init__(self, planning_problem):

self.graph = Graph(planning_problem)

self.no_goods = []

self.solution = []

def check_leveloff(self):

"""Checks if the graph has levelled off"""

check = (set(self.graph.levels[-1].current_state) == set(self.graph.levels[-2].current_state))

if check:

return True

def extract_solution(self, goals, index):

"""Extracts the solution"""

level = self.graph.levels[index]

if not self.graph.non_mutex_goals(goals, index):

self.no_goods.append((level, goals))

return

level = self.graph.levels[index - 1]

# Create all combinations of actions that satisfy the goal

actions = []

for goal in goals:

actions.append(level.next_state_links[goal])

all_actions = list(itertools.product(*actions))

# Filter out non-mutex actions

non_mutex_actions = []

for action_tuple in all_actions:

action_pairs = itertools.combinations(list(set(action_tuple)), 2)

non_mutex_actions.append(list(set(action_tuple)))

for pair in action_pairs:

if set(pair) in level.mutex:

non_mutex_actions.pop(-1)

break

# Recursion

for action_list in non_mutex_actions:

if [action_list, index] not in self.solution:

self.solution.append([action_list, index])

new_goals = []

for act in set(action_list):

if act in level.current_action_links:

new_goals = new_goals + level.current_action_links[act]

if abs(index) + 1 == len(self.graph.levels):

return

elif (level, new_goals) in self.no_goods:

return

else:

self.extract_solution(new_goals, index - 1)

# Level-Order multiple solutions

solution = []

for item in self.solution:

if item[1] == -1:

solution.append([])

solution[-1].append(item[0])

else:

solution[-1].append(item[0])

for num, item in enumerate(solution):

item.reverse()

solution[num] = item

return solution

def goal_test(self, kb):

return all(kb.ask(q) is not False for q in self.graph.planning_problem.goals)

def execute(self):

"""Executes the GraphPlan algorithm for the given problem"""

while True: