"""The abstract class for a formal problem. You should subclass

def __init__(self, initial, goal=None):

"""The constructor specifies the initial state, and possibly a goal

self.initial = initial

self.goal = goal

def actions(self, state):

"""Return the actions that can be executed in the given

raise NotImplementedError

def result(self, state, action):

"""Return the state that results from executing the given

raise NotImplementedError

def goal_test(self, state):

"""Return True if the state is a goal. The default method compares the

if isinstance(self.goal, list):

return is_in(state, self.goal)

else:

return state == self.goal

def path_cost(self, c, state1, action, state2):

"""Return the cost of a solution path that arrives at state2 from

return c + 1

def value(self, state):

"""For optimization problems, each state has a value. Hill-climbing

"""A node in a search tree. Contains a pointer to the parent (the node

def __init__(self, state, parent=None, action=None, path_cost=0):

"""Create a search tree Node, derived from a parent by an action."""

self.state = state

self.parent = parent

self.action = action

self.path_cost = path_cost

self.depth = 0

if parent:

self.depth = parent.depth + 1

def __repr__(self):

return "<Node {}>".format(self.state)

def __lt__(self, node):

return self.state < node.state

def expand(self, problem):

"""List the nodes reachable in one step from this node."""

return [self.child_node(problem, action)

for action in problem.actions(self.state)]

def child_node(self, problem, action):

"""[Figure 3.10]"""

next = problem.result(self.state, action)

return Node(next, self, action,

problem.path_cost(self.path_cost, self.state,

action, next))

def solution(self):

"""Return the sequence of actions to go from the root to this node."""

return [node.action for node in self.path()[1:]]

def path(self):

"""Return a list of nodes forming the path from the root to this node."""

node, path_back = self, []

while node:

path_back.append(node)

node = node.parent

return list(reversed(path_back))

# We want for a queue of nodes in breadth_first_search or

# astar_search to have no duplicated states, so we treat nodes

# with the same state as equal. [Problem: this may not be what you

# want in other contexts.]

def __eq__(self, other):

return isinstance(other, Node) and self.state == other.state

def __hash__(self):

"""Abstract framework for a problem-solving agent. [Figure 3.1]"""

def __init__(self, initial_state=None):

"""State is an sbstract representation of the state

self.state = initial_state

self.seq = []

def __call__(self, percept):

"""[Figure 3.1] Formulate a goal and problem, then

self.state = self.update_state(self.state, percept)

if not self.seq:

goal = self.formulate_goal(self.state)

problem = self.formulate_problem(self.state, goal)

self.seq = self.search(problem)

if not self.seq:

return None

return self.seq.pop(0)

def update_state(self, percept):

raise NotImplementedError

def formulate_goal(self, state):

raise NotImplementedError

def formulate_problem(self, state, goal):

raise NotImplementedError

def search(self, problem):

"""Search through the successors of a problem to find a goal.

frontier.append(Node(problem.initial))

while frontier:

node = frontier.pop()

if problem.goal_test(node.state):

return node

frontier.extend(node.expand(problem))

"""Search through the successors of a problem to find a goal.

frontier.append(Node(problem.initial))

explored = set()

while frontier:

node = frontier.pop()

if problem.goal_test(node.state):

return node

explored.add(node.state)

frontier.extend(child for child in node.expand(problem)

if child.state not in explored and

child not in frontier)

"""[Figure 3.11]"""

node = Node(problem.initial)

if problem.goal_test(node.state):

return node

frontier = FIFOQueue()

frontier.append(node)

explored = set()

while frontier:

node = frontier.pop()

explored.add(node.state)

for child in node.expand(problem):

if child.state not in explored and child not in frontier:

if problem.goal_test(child.state):

return child

frontier.append(child)

"""Search the nodes with the lowest f scores first.

f = memoize(f, 'f')

node = Node(problem.initial)

if problem.goal_test(node.state):

return node

frontier = PriorityQueue(min, f)

frontier.append(node)

explored = set()

while frontier:

node = frontier.pop()

if problem.goal_test(node.state):

return node

explored.add(node.state)

for child in node.expand(problem):

if child.state not in explored and child not in frontier:

frontier.append(child)

elif child in frontier:

incumbent = frontier[child]

if f(child) < f(incumbent):

del frontier[incumbent]

frontier.append(child)

"""[Figure 3.17]"""

def recursive_dls(node, problem, limit):

if problem.goal_test(node.state):

return node

elif limit == 0:

return 'cutoff'

else:

cutoff_occurred = False

for child in node.expand(problem):

result = recursive_dls(child, problem, limit - 1)

if result == 'cutoff':

cutoff_occurred = True

elif result is not None:

return result

return 'cutoff' if cutoff_occurred else None

# Body of depth_limited_search:

"""[Figure 3.18]"""

for depth in range(sys.maxsize):

result = depth_limited_search(problem, depth)

if result != 'cutoff':

e = problem.find_min_edge()

gF, gB = {problem.initial : 0}, {problem.goal : 0}

openF, openB = [problem.initial], [problem.goal]

closedF, closedB = [], []

U = infinity

def extend(U, open_dir, open_other, g_dir, g_other, closed_dir):

"""Extend search in given direction"""

n = find_key(C, open_dir, g_dir)

open_dir.remove(n)

closed_dir.append(n)

for c in problem.actions(n):

if c in open_dir or c in closed_dir:

if g_dir[c] <= problem.path_cost(g_dir[n], n, None, c):

continue

open_dir.remove(c)

g_dir[c] = problem.path_cost(g_dir[n], n, None, c)

open_dir.append(c)

if c in open_other:

U = min(U, g_dir[c] + g_other[c])

return U, open_dir, closed_dir, g_dir

def find_min(open_dir, g):

"""Finds minimum priority, g and f values in open_dir"""

m, m_f = infinity, infinity

for n in open_dir:

f = g[n] + problem.h(n)

pr = max(f, 2*g[n])

m = min(m, pr)

m_f = min(m_f, f)

return m, m_f, min(g.values())

def find_key(pr_min, open_dir, g):

"""Finds key in open_dir with value equal to pr_min

m = infinity

state = -1

for n in open_dir:

pr = max(g[n] + problem.h(n), 2*g[n])

if pr == pr_min:

if g[n] < m:

m = g[n]

state = n

return state

while openF and openB:

pr_min_f, f_min_f, g_min_f = find_min(openF, gF)

pr_min_b, f_min_b, g_min_b = find_min(openB, gB)

C = min(pr_min_f, pr_min_b)

if U <= max(C, f_min_f, f_min_b, g_min_f + g_min_b + e):

return U

if C == pr_min_f:

# Extend forward

U, openF, closedF, gF = extend(U, openF, openB, gF, gB, closedF)

else:

# Extend backward

U, openB, closedB, gB = extend(U, openB, openF, gB, gF, closedB)

"""A* search is best-first graph search with f(n) = g(n)+h(n).

h = memoize(h or problem.h, 'h')

"""[Figure 3.26]"""

h = memoize(h or problem.h, 'h')

def RBFS(problem, node, flimit):

if problem.goal_test(node.state):

return node, 0 # (The second value is immaterial)

successors = node.expand(problem)

if len(successors) == 0:

return None, infinity

for s in successors:

s.f = max(s.path_cost + h(s), node.f)

while True:

# Order by lowest f value

successors.sort(key=lambda x: x.f)

best = successors[0]

if best.f > flimit:

return None, best.f

if len(successors) > 1:

alternative = successors[1].f

else:

alternative = infinity

result, best.f = RBFS(problem, best, min(flimit, alternative))

if result is not None:

return result, best.f

node = Node(problem.initial)

node.f = h(node)

result, bestf = RBFS(problem, node, infinity)

"""From the initial node, keep choosing the neighbor with highest value,

current = Node(problem.initial)

while True:

neighbors = current.expand(problem)

if not neighbors:

break

neighbor = argmax_random_tie(neighbors,

key=lambda node: problem.value(node.state))

if problem.value(neighbor.state) <= problem.value(current.state):

break

current = neighbor

"""[Figure 4.5] CAUTION: This differs from the pseudocode as it

current = Node(problem.initial)

for t in range(sys.maxsize):

T = schedule(t)

if T == 0:

return current.state

neighbors = current.expand(problem)

if not neighbors:

return current.state

next = random.choice(neighbors)

delta_e = problem.value(next.state) - problem.value(current.state)

if delta_e > 0 or probability(math.exp(delta_e / T)):

"""[Figure 4.11]Used when the environment is nondeterministic and completely observable.

# functions used by and_or_search

def or_search(state, problem, path):

"""returns a plan as a list of actions"""

if problem.goal_test(state):

return []

if state in path:

return None

for action in problem.actions(state):

plan = and_search(problem.result(state, action),

problem, path + [state, ])

if plan is not None:

return [action, plan]

def and_search(states, problem, path):

"""Returns plan in form of dictionary where we take action plan[s] if we reach state s."""

plan = {}

for s in states:

plan[s] = or_search(s, problem, path)

if plan[s] is None:

return None

return plan

# body of and or search

"""Problem of finding the highest peak in a limited grid"""

def __init__(self, initial, grid):

"""The grid is a 2 dimensional array/list whose state is specified by tuple of indices"""

Problem.__init__(self, initial)

self.grid = grid

self.n = len(grid)

assert self.n > 0

self.m = len(grid[0])

assert self.m > 0

def actions(self, state):

"""Allows movement in only 4 directions"""

# TODO: Add flag to allow diagonal motion

allowed_actions = []

if state[0] > 0:

allowed_actions.append('N')

if state[0] < self.n - 1:

allowed_actions.append('S')

if state[1] > 0:

allowed_actions.append('W')

if state[1] < self.m - 1:

allowed_actions.append('E')

return allowed_actions

def result(self, state, action):

"""Moves in the direction specified by action"""

x, y = state

x = x + (1 if action == 'S' else (-1 if action == 'N' else 0))

y = y + (1 if action == 'E' else (-1 if action == 'W' else 0))

return (x, y)

def value(self, state):

"""Value of a state is the value it is the index to"""

x, y = state

assert 0 <= x < self.n

assert 0 <= y < self.m

"""[Figure 4.21] The abstract class for an OnlineDFSAgent. Override

def __init__(self, problem):

self.problem = problem

self.s = None

self.a = None

self.untried = defaultdict(list)

self.unbacktracked = defaultdict(list)

self.result = {}

def __call__(self, percept):

s1 = self.update_state(percept)

if self.problem.goal_test(s1):

self.a = None

else:

if s1 not in self.untried.keys():

self.untried[s1] = self.problem.actions(s1)

if self.s is not None:

if s1 != self.result[(self.s, self.a)]:

self.result[(self.s, self.a)] = s1

self.unbacktracked[s1].insert(0, self.s)

if len(self.untried[s1]) == 0:

if len(self.unbacktracked[s1]) == 0:

self.a = None

else:

# else a <- an action b such that result[s', b] = POP(unbacktracked[s'])

unbacktracked_pop = self.unbacktracked[s1].pop(0)

for (s, b) in self.result.keys():

if self.result[(s, b)] == unbacktracked_pop:

self.a = b

break

else:

self.a = self.untried[s1].pop(0)

self.s = s1

return self.a

def update_state(self, percept):

"""To be overridden in most cases. The default case

"""

def __init__(self, initial, goal, graph):

self.initial = initial

self.goal = goal

self.graph = graph

def actions(self, state):

return self.graph.dict[state].keys()

def output(self, state, action):

return self.graph.dict[state][action]

def h(self, state):

"""Returns least possible cost to reach a goal for the given state."""

return self.graph.least_costs[state]

def c(self, s, a, s1):

"""Returns a cost estimate for an agent to move from state 's' to state 's1'."""

return 1

def update_state(self, percept):

raise NotImplementedError

def goal_test(self, state):

if state == self.goal:

return True

""" [Figure 4.24]

def __init__(self, problem):

self.problem = problem

# self.result = {} # no need as we are using problem.result

self.H = {}

self.s = None

self.a = None

def __call__(self, s1): # as of now s1 is a state rather than a percept

if self.problem.goal_test(s1):

self.a = None

return self.a

else:

if s1 not in self.H:

self.H[s1] = self.problem.h(s1)

if self.s is not None:

# self.result[(self.s, self.a)] = s1 # no need as we are using problem.output

# minimum cost for action b in problem.actions(s)

self.H[self.s] = min(self.LRTA_cost(self.s, b, self.problem.output(self.s, b),

self.H) for b in self.problem.actions(self.s))

# an action b in problem.actions(s1) that minimizes costs

self.a = argmin(self.problem.actions(s1),

key=lambda b: self.LRTA_cost(s1, b, self.problem.output(s1, b), self.H))

self.s = s1

return self.a

def LRTA_cost(self, s, a, s1, H):

"""Returns cost to move from state 's' to state 's1' plus

print(s, a, s1)

if s1 is None:

return self.problem.h(s)

else:

# sometimes we need to get H[s1] which we haven't yet added to H

# to replace this try, except: we can initialize H with values from problem.h

try:

return self.problem.c(s, a, s1) + self.H[s1]

except:

"""Call genetic_algorithm on the appropriate parts of a problem.

# NOTE: This is not tested and might not work.

# TODO: Use this function to make Problems work with genetic_algorithm.

s = problem.initial_state

states = [problem.result(s, a) for a in problem.actions(s)]

random.shuffle(states)

"""[Figure 4.8]"""

for i in range(ngen):

new_population = []

random_selection = selection_chances(fitness_fn, population)

for j in range(len(population)):

x = random_selection()

y = random_selection()

child = reproduce(x, y)

if random.uniform(0, 1) < pmut:

child = mutate(child, gene_pool)

new_population.append(child)

population = new_population

if f_thres:

fittest_individual = argmax(population, key=fitness_fn)

if fitness_fn(fittest_individual) >= f_thres:

return fittest_individual

"""Initializes population for genetic algorithm

g = len(gene_pool)

population = []

for i in range(pop_number):

new_individual = [gene_pool[random.randrange(0, g)] for j in range(state_length)]

population.append(new_individual)

n = len(x)

g = len(gene_pool)

c = random.randrange(0, n)

r = random.randrange(0, g)

new_gene = gene_pool[r]

"""A graph connects nodes (verticies) by edges (links). Each edge can also

def __init__(self, dict=None, directed=True):

self.dict = dict or {}

self.directed = directed

if not directed:

self.make_undirected()

def make_undirected(self):

"""Make a digraph into an undirected graph by adding symmetric edges."""

for a in list(self.dict.keys()):

for (b, dist) in self.dict[a].items():

self.connect1(b, a, dist)

def connect(self, A, B, distance=1):

"""Add a link from A and B of given distance, and also add the inverse

self.connect1(A, B, distance)

if not self.directed:

self.connect1(B, A, distance)

def connect1(self, A, B, distance):

"""Add a link from A to B of given distance, in one direction only."""

self.dict.setdefault(A, {})[B] = distance

def get(self, a, b=None):

"""Return a link distance or a dict of {node: distance} entries.

links = self.dict.setdefault(a, {})

if b is None:

return links

else:

return links.get(b)

def nodes(self):

"""Return a list of nodes in the graph."""

curvature=lambda: random.uniform(1.1, 1.5)):

"""Construct a random graph, with the specified nodes, and random links.

g = UndirectedGraph()

g.locations = {}

# Build the cities

for node in nodes:

g.locations[node] = (random.randrange(width), random.randrange(height))

# Build roads from each city to at least min_links nearest neighbors.

for i in range(min_links):

for node in nodes:

if len(g.get(node)) < min_links:

here = g.locations[node]

def distance_to_node(n):

if n is node or g.get(node, n):

return infinity

return distance(g.locations[n], here)

neighbor = argmin(nodes, key=distance_to_node)

d = distance(g.locations[neighbor], here) * curvature()

g.connect(node, neighbor, int(d))

Arad=dict(Zerind=75, Sibiu=140, Timisoara=118),

Bucharest=dict(Urziceni=85, Pitesti=101, Giurgiu=90, Fagaras=211),

Craiova=dict(Drobeta=120, Rimnicu=146, Pitesti=138),

Drobeta=dict(Mehadia=75),

Eforie=dict(Hirsova=86),

Fagaras=dict(Sibiu=99),

Hirsova=dict(Urziceni=98),

Iasi=dict(Vaslui=92, Neamt=87),

Lugoj=dict(Timisoara=111, Mehadia=70),

Oradea=dict(Zerind=71, Sibiu=151),

Pitesti=dict(Rimnicu=97),

Rimnicu=dict(Sibiu=80),

State_1=dict(Suck=['State_7', 'State_5'], Right=['State_2']),

State_2=dict(Suck=['State_8', 'State_4'], Left=['State_2']),

State_3=dict(Suck=['State_7'], Right=['State_4']),

State_4=dict(Suck=['State_4', 'State_2'], Left=['State_3']),

State_5=dict(Suck=['State_5', 'State_1'], Right=['State_6']),

State_6=dict(Suck=['State_8'], Left=['State_5']),

State_7=dict(Suck=['State_7', 'State_3'], Right=['State_8']),

State_8=dict(Suck=['State_8', 'State_6'], Left=['State_7'])

State_1=dict(Right='State_2'),

State_2=dict(Right='State_3', Left='State_1'),

State_3=dict(Right='State_4', Left='State_2'),

State_4=dict(Right='State_5', Left='State_3'),

State_5=dict(Right='State_6', Left='State_4'),

State_6=dict(Left='State_5')

T=dict(),

SA=dict(WA=1, NT=1, Q=1, NSW=1, V=1),

NT=dict(WA=1, Q=1),

"""The problem of searching a graph from one node to another."""

def __init__(self, initial, goal, graph):

Problem.__init__(self, initial, goal)

self.graph = graph

def actions(self, A):

"""The actions at a graph node are just its neighbors."""

return list(self.graph.get(A).keys())

def result(self, state, action):

"""The result of going to a neighbor is just that neighbor."""

return action

def path_cost(self, cost_so_far, A, action, B):

return cost_so_far + (self.graph.get(A, B) or infinity)

def find_min_edge(self):

"""Find minimum value of edges."""

m = infinity

for d in self.graph.dict.values():

local_min = min(d.values())

m = min(m, local_min)

return m

def h(self, node):

"""h function is straight-line distance from a node's state to goal."""

locs = getattr(self.graph, 'locations', None)

if locs:

if type(node) is str:

return int(distance(locs[node], locs[self.goal]))

return int(distance(locs[node.state], locs[self.goal]))

else:

"""

def result(self, state, action):

return self.graph.get(state, action)

def path_cost(self):

"""The problem of placing N queens on an NxN board with none attacking