added more descriptive graph search algorithms

Author: hmp-anthony

polygon_obstacle_path_finder.ipynb

@@ -0,0 +1,277 @@
+import matplotlib.pyplot as plt
+import random
+import heapq
+import math
+import sys
+from collections import defaultdict, deque, Counter
+from itertools import combinations
+
+class Problem(object):
+    """The abstract class for a formal problem. A new domain subclasses this,
+    overriding `actions` and `results`, and perhaps other methods.
+    The default heuristic is 0 and the default action cost is 1 for all states.
+    When yiou create an instance of a subclass, specify `initial`, and `goal` states 
+    (or give an `is_goal` method) and perhaps other keyword args for the subclass."""
+
+    def __init__(self, initial=None, goal=None, **kwds): 
+        self.__dict__.update(initial=initial, goal=goal, **kwds) 
+        
+    def actions(self, state):        raise NotImplementedError
+    def result(self, state, action): raise NotImplementedError
+    def is_goal(self, state):        return state == self.goal
+    def action_cost(self, s, a, s1): return 1
+    def h(self, node):               return 0
+    
+    def __str__(self):
+        return '{}({!r}, {!r})'.format(
+            type(self).__name__, self.initial, self.goal)
+    
+
+class Node:
+    "A Node in a search tree."
+    def __init__(self, state, parent=None, action=None, path_cost=0):
+        self.__dict__.update(state=state, parent=parent, action=action, path_cost=path_cost)
+
+    def __repr__(self): return '<{}>'.format(self.state)
+    def __len__(self): return 0 if self.parent is None else (1 + len(self.parent))
+    def __lt__(self, other): return self.path_cost < other.path_cost
+    
+    
+failure = Node('failure', path_cost=math.inf) # Indicates an algorithm couldn't find a solution.
+cutoff  = Node('cutoff',  path_cost=math.inf) # Indicates iterative deepening search was cut off.
+    
+    
+def expand(problem, node):
+    "Expand a node, generating the children nodes."
+    s = node.state
+    for action in problem.actions(s):
+        s1 = problem.result(s, action)
+        cost = node.path_cost + problem.action_cost(s, action, s1)
+        yield Node(s1, node, action, cost)
+        
+
+def path_actions(node):
+    "The sequence of actions to get to this node."
+    if node.parent is None:
+        return []  
+    return path_actions(node.parent) + [node.action]
+
+
+def path_states(node):
+    "The sequence of states to get to this node."
+    if node in (cutoff, failure, None): 
+        return []
+    return path_states(node.parent) + [node.state]
+
+FIFOQueue = deque
+
+LIFOQueue = list
+
+class PriorityQueue:
+    """A queue in which the item with minimum f(item) is always popped first."""
+
+    def __init__(self, items=(), key=lambda x: x): 
+        self.key = key
+        self.items = [] # a heap of (score, item) pairs
+        for item in items:
+            self.add(item)
+         
+    def add(self, item):
+        """Add item to the queuez."""
+        pair = (self.key(item), item)
+        heapq.heappush(self.items, pair)
+
+    def pop(self):
+        """Pop and return the item with min f(item) value."""
+        return heapq.heappop(self.items)[1]
+    
+    def top(self): return self.items[0][1]
+
+    def __len__(self): return len(self.items)
+
+def best_first_search(problem, f):
+    "Search nodes with minimum f(node) value first."
+    node = Node(problem.initial)
+    frontier = PriorityQueue([node], key=f)
+    reached = {problem.initial: node}
+    while frontier:
+        node = frontier.pop()
+        if problem.is_goal(node.state):
+            return node
+        for child in expand(problem, node):
+            s = child.state
+            if s not in reached or child.path_cost < reached[s].path_cost:
+                reached[s] = child
+                frontier.add(child)
+    return failure
+
+
+def best_first_tree_search(problem, f):
+    "A version of best_first_search without the `reached` table."
+    frontier = PriorityQueue([Node(problem.initial)], key=f)
+    while frontier:
+        node = frontier.pop()
+        if problem.is_goal(node.state):
+            return node
+        for child in expand(problem, node):
+            if not is_cycle(child):
+                frontier.add(child)
+    return failure
+
+
+def g(n): return n.path_cost
+
+
+def astar_search(problem, h=None):
+    """Search nodes with minimum f(n) = g(n) + h(n)."""
+    h = h or problem.h
+    return best_first_search(problem, f=lambda n: g(n) + h(n))
+
+
+def astar_tree_search(problem, h=None):
+    """Search nodes with minimum f(n) = g(n) + h(n), with no `reached` table."""
+    h = h or problem.h
+    return best_first_tree_search(problem, f=lambda n: g(n) + h(n))
+
+
+def weighted_astar_search(problem, h=None, weight=1.4):
+    """Search nodes with minimum f(n) = g(n) + weight * h(n)."""
+    h = h or problem.h
+    return best_first_search(problem, f=lambda n: g(n) + weight * h(n))
+
+        
+def greedy_bfs(problem, h=None):
+    """Search nodes with minimum h(n)."""
+    h = h or problem.h
+    return best_first_search(problem, f=h)
+
+
+def uniform_cost_search(problem):
+    "Search nodes with minimum path cost first."
+    return best_first_search(problem, f=g)
+
+
+def breadth_first_bfs(problem):
+    "Search shallowest nodes in the search tree first; using best-first."
+    return best_first_search(problem, f=len)
+
+
+def depth_first_bfs(problem):
+    "Search deepest nodes in the search tree first; using best-first."
+    return best_first_search(problem, f=lambda n: -len(n))
+
+
+def is_cycle(node, k=30):
+    "Does this node form a cycle of length k or less?"
+    def find_cycle(ancestor, k):
+        return (ancestor is not None and k > 0 and
+                (ancestor.state == node.state or find_cycle(ancestor.parent, k - 1)))
+    return find_cycle(node.parent, k)
+
+def breadth_first_search(problem):
+    "Search shallowest nodes in the search tree first."
+    node = Node(problem.initial)
+    if problem.is_goal(problem.initial):
+        return node
+    frontier = FIFOQueue([node])
+    reached = {problem.initial}
+    while frontier:
+        node = frontier.pop()
+        for child in expand(problem, node):
+            s = child.state
+            if problem.is_goal(s):
+                return child
+            if s not in reached:
+                reached.add(s)
+                frontier.appendleft(child)
+    return failure
+
+
+def iterative_deepening_search(problem):
+    "Do depth-limited search with increasing depth limits."
+    for limit in range(1, sys.maxsize):
+        result = depth_limited_search(problem, limit)
+        if result != cutoff:
+            return result
+        
+        
+def depth_limited_search(problem, limit=10):
+    "Search deepest nodes in the search tree first."
+    frontier = LIFOQueue([Node(problem.initial)])
+    result = failure
+    while frontier:
+        node = frontier.pop()
+        if problem.is_goal(node.state):
+            return node
+        elif len(node) >= limit:
+            result = cutoff
+        elif not is_cycle(node):
+            for child in expand(problem, node):
+                frontier.append(child)
+    return result
+
+
+def depth_first_recursive_search(problem, node=None):
+    if node is None: 
+        node = Node(problem.initial)
+    if problem.is_goal(node.state):
+        return node
+    elif is_cycle(node):
+        return failure
+    else:
+        for child in expand(problem, node):
+            result = depth_first_recursive_search(problem, child)
+            if result:
+                return result
+        return failure
+
+class Map:
+    """A map of places in a 2D world: a graph with vertexes and links between them. 
+    In `Map(links, locations)`, `links` can be either [(v1, v2)...] pairs, 
+    or a {(v1, v2): distance...} dict. Optional `locations` can be {v1: (x, y)} 
+    If `directed=False` then for every (v1, v2) link, we add a (v2, v1) link."""
+
+    def __init__(self, links, locations=None, directed=False):
+        if not hasattr(links, 'items'): # Distances are 1 by default
+            links = {link: 1 for link in links}
+        if not directed:
+            for (v1, v2) in list(links):
+                links[v2, v1] = links[v1, v2]
+        self.distances = links
+        self.neighbors = multimap(links)
+        self.locations = locations or defaultdict(lambda: (0, 0))
+
+        
+def multimap(pairs) -> dict:
+    "Given (key, val) pairs, make a dict of {key: [val,...]}."
+    result = defaultdict(list)
+    for key, val in pairs:
+        result[key].append(val)
+    return result
+
+class RouteProblem(Problem):
+    """A problem to find a route between locations on a `Map`.
+    Create a problem with RouteProblem(start, goal, map=Map(...)}).
+    States are the vertexes in the Map graph; actions are destination states."""
+    
+    def actions(self, state): 
+        """The places neighboring `state`."""
+        return self.map.neighbors[state]
+    
+    def result(self, state, action):
+        """Go to the `action` place, if the map says that is possible."""
+        return action if action in self.map.neighbors[state] else state
+    
+    def action_cost(self, s, action, s1):
+        """The distance (cost) to go from s to s1."""
+        return self.map.distances[s, s1]
+    
+    def h(self, node):
+        "Straight-line distance between state and the goal."
+        locs = self.map.locations
+        return straight_line_distance(locs[node.state], locs[self.goal])
+    
+    
+def straight_line_distance(A, B):
+    "Straight-line distance between two points."
+    return sum(abs(a - b)**2 for (a, b) in zip(A, B)) ** 0.5