# Graphs and Graph Problems
import math
# Python modul vo koj se implementirani algoritmite za neinformirano i informirano prebaruvanje
# ______________________________________________________________________________________________
# Improtiranje na dopolnitelno potrebni paketi za funkcioniranje na kodovite
import sys
import bisect
infinity = float('inf') # sistemski definirana vrednost za beskonecnost
# ______________________________________________________________________________________________
# Definiranje na pomosni strukturi za cuvanje na listata na generirani, no neprovereni jazli
class Queue:
"""Queue is an abstract class/interface. There are three types:
Stack(): A Last In First Out Queue.
FIFOQueue(): A First In First Out Queue.
PriorityQueue(order, f): Queue in sorted order (default min-first).
Each type supports the following methods and functions:
q.append(item) -- add an item to the queue
q.extend(items) -- equivalent to: for item in items: q.append(item)
q.pop() -- return the top item from the queue
len(q) -- number of items in q (also q.__len())
item in q -- does q contain item?
Note that isinstance(Stack(), Queue) is false, because we implement stacks
as lists. If Python ever gets interfaces, Queue will be an interface."""
def __init__(self):
raise NotImplementedError
def extend(self, items):
for item in items:
self.append(item)
def Stack():
"""A Last-In-First-Out Queue."""
return []
class FIFOQueue(Queue):
"""A First-In-First-Out Queue."""
def __init__(self):
self.A = []
self.start = 0
def append(self, item):
self.A.append(item)
def __len__(self):
return len(self.A) - self.start
def extend(self, items):
self.A.extend(items)
def pop(self):
e = self.A[self.start]
self.start += 1
if self.start > 5 and self.start > len(self.A) / 2:
self.A = self.A[self.start:]
self.start = 0
return e
def __contains__(self, item):
return item in self.A[self.start:]
class PriorityQueue(Queue):
"""A queue in which the minimum (or maximum) element (as determined by f and
order) is returned first. If order is min, the item with minimum f(x) is
returned first; if order is max, then it is the item with maximum f(x).
Also supports dict-like lookup. This structure will be most useful in informed searches"""
def __init__(self, order=min, f=lambda x: x):
self.A = []
self.order = order
self.f = f
def append(self, item):
bisect.insort(self.A, (self.f(item), item))
def __len__(self):
return len(self.A)
def pop(self):
if self.order == min:
return self.A.pop(0)[1]
else:
return self.A.pop()[1]
def __contains__(self, item):
return any(item == pair[1] for pair in self.A)
def __getitem__(self, key):
for _, item in self.A:
if item == key:
return item
def __delitem__(self, key):
for i, (value, item) in enumerate(self.A):
if item == key:
self.A.pop(i)
# ______________________________________________________________________________________________
# Definiranje na klasa za strukturata na problemot koj ke go resavame so prebaruvanje
# Klasata Problem e apstraktna klasa od koja pravime nasleduvanje za definiranje na osnovnite karakteristiki
# na sekoj eden problem sto sakame da go resime
class Problem:
"""The abstract class for a formal problem. You should subclass this and
implement the method successor, and possibly __init__, goal_test, and
path_cost. Then you will create instances of your subclass and solve them
with the various search functions."""
def __init__(self, initial, goal=None):
"""The constructor specifies the initial state, and possibly a goal
state, if there is a unique goal. Your subclass's constructor can add
other arguments."""
self.initial = initial
self.goal = goal
def successor(self, state):
"""Given a state, return a dictionary of {action : state} pairs reachable
from this state. If there are many successors, consider an iterator
that yields the successors one at a time, rather than building them
all at once. Iterators will work fine within the framework. Yielding is not supported in Python 2.7"""
raise NotImplementedError
def actions(self, state):
"""Given a state, return a list of all actions possible from that state"""
raise NotImplementedError
def result(self, state, action):
"""Given a state and action, return the resulting state"""
raise NotImplementedError
def goal_test(self, state):
"""Return True if the state is a goal. The default method compares the
state to self.goal, as specified in the constructor. Implement this
method if checking against a single self.goal is not enough."""
return state == self.goal
def path_cost(self, c, state1, action, state2):
"""Return the cost of a solution path that arrives at state2 from
state1 via action, assuming cost c to get up to state1. If the problem
is such that the path doesn't matter, this function will only look at
state2. If the path does matter, it will consider c and maybe state1
and action. The default method costs 1 for every step in the path."""
return c + 1
def value(self):
"""For optimization problems, each state has a value. Hill-climbing
and related algorithms try to maximize this value."""
raise NotImplementedError
# ______________________________________________________________________________
# Definiranje na klasa za strukturata na jazel od prebaruvanje
# Klasata Node ne se nasleduva
class Node:
"""A node in a search tree. Contains a pointer to the parent (the node
that this is a successor of) and to the actual state for this node. Note
that if a state is arrived at by two paths, then there are two nodes with
the same state. Also includes the action that got us to this state, and
the total path_cost (also known as g) to reach the node. Other functions
may add an f and h value; see best_first_graph_search and astar_search for
an explanation of how the f and h values are handled. You will not need to
subclass this class."""
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 %s>" % (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):
"Return a child node from this node"
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 solve(self):
"Return the sequence of states to go from the root to this node."
return [node.state for node in self.path()[0:]]
def path(self):
"Return a list of nodes forming the path from the root to this node."
x, result = self, []
while x:
result.append(x)
x = x.parent
return list(reversed(result))
# 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):
return hash(self.state)
# ________________________________________________________________________________________________________
#Neinformirano prebaruvanje vo ramki na drvo
#Vo ramki na drvoto ne razresuvame jamki
def tree_search(problem, fringe):
"""Search through the successors of a problem to find a goal.
The argument fringe should be an empty queue."""
fringe.append(Node(problem.initial))
while fringe:
node = fringe.pop()
print (node.state)
if problem.goal_test(node.state):
return node
fringe.extend(node.expand(problem))
return None
def breadth_first_tree_search(problem):
"Search the shallowest nodes in the search tree first."
return tree_search(problem, FIFOQueue())
def depth_first_tree_search(problem):
"Search the deepest nodes in the search tree first."
return tree_search(problem, Stack())
# ________________________________________________________________________________________________________
#Neinformirano prebaruvanje vo ramki na graf
#Osnovnata razlika e vo toa sto ovde ne dozvoluvame jamki t.e. povtoruvanje na sostojbi
def graph_search(problem, fringe):
"""Search through the successors of a problem to find a goal.
The argument fringe should be an empty queue.
If two paths reach a state, only use the best one."""
closed = {}
fringe.append(Node(problem.initial))
while fringe:
node = fringe.pop()
if problem.goal_test(node.state):
return node
if node.state not in closed:
closed[node.state] = True
fringe.extend(node.expand(problem))
return None
def breadth_first_graph_search(problem):
"Search the shallowest nodes in the search tree first."
return graph_search(problem, FIFOQueue())
def depth_first_graph_search(problem):
"Search the deepest nodes in the search tree first."
return graph_search(problem, Stack())
def uniform_cost_search(problem):
"Search the nodes in the search tree with lowest cost first."
return graph_search(problem, PriorityQueue(lambda a, b: a.path_cost < b.path_cost))
def depth_limited_search(problem, limit=50):
"depth first search with limited depth"
def recursive_dls(node, problem, limit):
"helper function for depth limited"
cutoff_occurred = False
if problem.goal_test(node.state):
return node
elif node.depth == limit:
return 'cutoff'
else:
for successor in node.expand(problem):
result = recursive_dls(successor, problem, limit)
if result == 'cutoff':
cutoff_occurred = True
elif result != None:
return result
if cutoff_occurred:
return 'cutoff'
else:
return None
# Body of depth_limited_search:
return recursive_dls(Node(problem.initial), problem, limit)
def iterative_deepening_search(problem):
for depth in range(sys.maxint):
result = depth_limited_search(problem, depth)
if result is not 'cutoff':
return result
# ________________________________________________________________________________________________________
#Pomosna funkcija za informirani prebaruvanja
#So pomos na ovaa funkcija gi keshirame rezultatite od funkcijata na evaluacija
def memoize(fn, slot=None):
"""Memoize fn: make it remember the computed value for any argument list.
If slot is specified, store result in that slot of first argument.
If slot is false, store results in a dictionary."""
if slot:
def memoized_fn(obj, *args):
if hasattr(obj, slot):
return getattr(obj, slot)
else:
val = fn(obj, *args)
setattr(obj, slot, val)
return val
else:
def memoized_fn(*args):
if not memoized_fn.cache.has_key(args):
memoized_fn.cache[args] = fn(*args)
return memoized_fn.cache[args]
memoized_fn.cache = {}
return memoized_fn
# ________________________________________________________________________________________________________
#Informirano prebaruvanje vo ramki na graf
def best_first_graph_search(problem, f):
"""Search the nodes with the lowest f scores first.
You specify the function f(node) that you want to minimize; for example,
if f is a heuristic estimate to the goal, then we have greedy best
first search; if f is node.depth then we have breadth-first search.
There is a subtlety: the line "f = memoize(f, 'f')" means that the f
values will be cached on the nodes as they are computed. So after doing
a best first search you can examine the f values of the path returned."""
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)
return None
def greedy_best_first_graph_search(problem, h=None):
"Greedy best-first search is accomplished by specifying f(n) = h(n)"
h = memoize(h or problem.h, 'h')
return best_first_graph_search(problem, h)
def astar_search(problem, h=None):
"A* search is best-first graph search with f(n) = g(n)+h(n)."
h = memoize(h or problem.h, 'h')
return best_first_graph_search(problem, lambda n: n.path_cost + h(n))
# ________________________________________________________________________________________________________
#Dopolnitelni prebaruvanja
#Recursive_best_first_search e implementiran
#Kako zadaca za studentite da gi implementiraat SMA* i IDA*
def recursive_best_first_search(problem, h=None):
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)
return result
def distance(a, b):
"""The distance between two (x, y) points."""
return math.hypot((a[0] - b[0]), (a[1] - b[1]))
class Graph:
"""A graph connects nodes (verticies) by edges (links). Each edge can also
have a length associated with it. The constructor call is something like:
g = Graph({'A': {'B': 1, 'C': 2})
this makes a graph with 3 nodes, A, B, and C, with an edge of length 1 from
A to B, and an edge of length 2 from A to C. You can also do:
g = Graph({'A': {'B': 1, 'C': 2}, directed=False)
This makes an undirected graph, so inverse links are also added. The graph
stays undirected; if you add more links with g.connect('B', 'C', 3), then
inverse link is also added. You can use g.nodes() to get a list of nodes,
g.get('A') to get a dict of links out of A, and g.get('A', 'B') to get the
length of the link from A to B. 'Lengths' can actually be any object at
all, and nodes can be any hashable object."""
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
link if the graph is undirected."""
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.
.get(a,b) returns the distance or None;
.get(a) returns a dict of {node: distance} entries, possibly {}."""
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."""
return list(self.dict.keys())
def UndirectedGraph(dict=None):
"""Build a Graph where every edge (including future ones) goes both ways."""
return Graph(dict=dict, directed=False)
class GraphProblem(Problem):
"""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 h(self, node):
"""h function is straight-line distance from a node's state to goal."""
locs = getattr(self.graph, 'locations', None)
if locs:
return int(distance(locs[node.state], locs[self.goal]))
else:
return infinity
# Primer na definiranje na graf kako ekspliciten prostor na sostojbi vo koj ke se prebaruva
romania_map = UndirectedGraph(dict(
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),
Urziceni=dict(Vaslui=142)))
# Sledniot del ne e zadolzitelen da se definiranje
# Definicijata na klasata kako podrazbirana opcija za hevristicka funkcija koristi pravolinisko rastojanie
#
# Dokolku slednite informacii ne gi definirate ili ne mozete da gi definirate treba da ja promenite definicijata
# na funkcijata "h" vo ramki na klasata GraphProblem taka da na soodveten nacin presmetate hevristicka funkcija
romania_map.locations = dict(
Arad=(91, 492), Bucharest=(400, 327), Craiova=(253, 288),
Drobeta=(165, 299), Eforie=(562, 293), Fagaras=(305, 449),
Giurgiu=(375, 270), Hirsova=(534, 350), Iasi=(473, 506),
Lugoj=(165, 379), Mehadia=(168, 339), Neamt=(406, 537),
Oradea=(131, 571), Pitesti=(320, 368), Rimnicu=(233, 410),
Sibiu=(207, 457), Timisoara=(94, 410), Urziceni=(456, 350),
Vaslui=(509, 444), Zerind=(108, 531))
# Primeri na povici so koi na primer moze da se izminuva grafot za gradovite vo Romanija
#
romania_problem = GraphProblem('Arad', 'Bucharest', romania_map)
#
answer1=breadth_first_tree_search(romania_problem)
print (answer1.solve())
#
# answer2=breadth_first_search(romania_problem)
# print answer2.solve()
#
# answer3=uniform_cost_search(romania_problem)
# print answer3.solve()
#
# answer4=astar_search(romania_problem)
# print answer4.solve()