Code for the Fundamentals of AI course
from copy import deepcopy
import numpy as np
from utils import Problem


################################################## RANDOM GRID PROBLEMS

W, H = 4, 3
P_EDGE = 0.7
MAX_STEPS = H * W 
SEED = 666
GRIDPOS = {n : ((n-1) % W, (n-1) // W) for n in range(1, H*W+1)}

grid1 = Problem.grid(W, H, p_edge=P_EDGE, seed=SEED, initial=1, goals={H*W})
grid2 = Problem.grid(W, H, p_edge=P_EDGE, seed=SEED, initial=3, goals={H*W})
grid3 = Problem.grid(W, H, p_edge=P_EDGE, seed=SEED+1, initial=1, goals={H*W}, cost=[1,10])
grid4 = Problem.grid(W, H, p_edge=P_EDGE, seed=SEED+2, initial=1, goals={H*W}, cost=[1,10])

# more exercises
grid5 = Problem.grid(W, H, p_edge=1.0, seed=SEED, cost=[1,10])
grid6 = Problem.grid(W, H, p_edge=1.0, seed=SEED+1, cost=[1,10])
grid7 = Problem.grid(W, H, p_edge=1.0, seed=SEED+2, cost=[1,10])
grid8 = Problem.grid(W, H, p_edge=1.0, seed=SEED+3, cost=[1,10])


################################################## TREE EXAMPLES


states = list(range(7))
postree = {0 : (3, 2),
           1 : (1, 1),
           2 : (5, 1),
           3 : (0, 0),
           4 : (2, 0),
           5 : (4, 0),
           6 : (6, 0)}

initial = 0

goals1 = {6}
tree1 = Problem(states, initial, goals1)
tree1.add_action(0, 1); tree1.add_action(0, 2)
tree1.add_action(1, 3); tree1.add_action(1, 4)
tree1.add_action(2, 5); tree1.add_action(2, 6)


tree2 = deepcopy(tree1)
tree2.goals = {2}

tree3 = deepcopy(tree1)
tree3.goals = {3}

tree4 = deepcopy(tree1)
tree4.goals = {4, 2}



################################################## PATH EXAMPLES

N = 5
states = list(range(N))
pospath = {n : (n, 1) for n in states}
pospath[N-1] = (N-2, 0)

initial, goal = states[0], states[-1]

path1 = Problem(states, initial, {goal})
for i in range(len(states) - 1):
    path1.add_action(states[i], states[i+1])

path1.add_action(initial, goal)

path2 = Problem(states, initial, {goal})
for i in range(len(states) - 1):
    path2.add_action(states[i], states[i+1], c=1)

path2.add_action(initial, goal, c=N-2)

path3 = Problem(states, initial, {goal})
for i in range(len(states) - 1):
    path3.add_action(states[i], states[i+1], c=1)

path3.add_action(initial, goal, c=N)

02/exercises.py
paolomorettin/FAI-code


import matplotlib.pyplot as plt
import networkx as nx
import os
from shutil import rmtree

from utils import Solution

def pyplot_printer(it, problem, frontier, reached, curr, solution, folder=None, pos=None, interactive=False):
    '''
    Use NetworkX + Matplotlib for displaying a snapshot of the search.

    '''
    node_size = 250
    arrow_size=15
    # '#264653'
    col_r = '#2a9d8f'
    col_f = '#e9c46a'
    col_g = '#f4a261'
    col_i = '#e76f51'
    col_w = '#ffffff'
    col_b = '#000000'

    if folder is None:
        folder = 'test'

    if it == 0:

        if not interactive:
            if os.path.isdir(folder):
                rmtree(folder)

            os.mkdir(folder)       

    G = problem.to_graph()

    if pos is None:
        pos = nx.spring_layout(G)

    initial_node = {problem.initial}
    goal_nodes = set(problem.goals)
    frontier_nodes = set(n.state for n in frontier)
    reached_nodes = set(reached) - frontier_nodes
    regular_nodes = set(G.nodes) - initial_node - goal_nodes

    if curr is not None:
        nx.draw_networkx_nodes(G, pos, nodelist=[curr.state], linewidths=2,
                               edgecolors=col_b,
                               node_color=col_w,
                               node_shape='s',
                               node_size=2*node_size)

    nx.draw_networkx_nodes(G, pos, nodelist=reached_nodes,
                           alpha=1, node_color=col_r, edgecolors=col_b, node_size=1.8*node_size)
    nx.draw_networkx_nodes(G, pos, nodelist=frontier_nodes,
                           alpha=1, node_color=col_f, edgecolors=col_b, node_size=1.8*node_size)
    nx.draw_networkx_nodes(G, pos, nodelist=goal_nodes,
                           alpha=1, node_color=col_g, edgecolors=col_b, node_size=1.0*node_size)
    nx.draw_networkx_nodes(G, pos, nodelist=initial_node,
                           alpha=1, node_color=col_i, edgecolors=col_b, node_size=1.0*node_size)
    nx.draw_networkx_nodes(G, pos, nodelist=regular_nodes,
                           alpha=1, node_color=col_w, edgecolors=col_b, node_size=1.0*node_size)

        
    
    nx.draw_networkx_labels(G, pos)
    nx.draw_networkx_edges(G, pos, arrowsize=arrow_size, connectionstyle=f'arc3, rad = 0.25')

    if solution is not None and isinstance(solution, Solution):
        edgelist = []
        for i in range(len(solution)-1):
            edgelist.append((solution[i].state, solution[i+1].state))
        
        nx.draw_networkx_edges(G, pos, edgelist=edgelist, width=1.5, edge_color=col_g,
                               arrowsize=arrow_size*1.5, connectionstyle=f'arc3, rad = 0.25')


    if problem.is_weighted(): # different costs
        cost_dict = {e : G.edges[e]['cost'] for e in G.edges}
        nx.draw_networkx_edge_labels(G, pos, cost_dict)


    if interactive:
        plt.show()
    else:
        plt.savefig(os.path.join(folder, f'{it}.png'), dpi=150, bbox_inches='tight', transparent=True)

    plt.clf()    

02/printers.py
paolomorettin/FAI-code

from utils import expand, Node, PriorityQueue, Solution


def best_first_search(problem, f, max_steps=100, printer=None):

    n = Node(problem.initial)
    frontier = PriorityQueue(f)
    reached = dict()
    solution = None
    
    if printer is not None:
        printer(0, problem, frontier, reached, None, solution)

    reached[n.state] = n.path_cost    
    frontier.push(n)
    
    if printer is not None:
        printer(1, problem, frontier, reached, None, solution)

    n_steps = 0
    while len(frontier) > 0 and n_steps < max_steps:

        n = frontier.pop()

        if problem.is_goal(n.state):
            
            solution = Solution.create(n)

            if printer is not None:
                printer(2 + n_steps, problem, frontier, reached, n, solution)

            return solution
        else:
            for child in expand(problem, n):
                if child.state not in reached or child.path_cost < reached[child.state]:
                    reached[child.state] = child.path_cost
                    frontier.push(child)

        if printer is not None:
            printer(2 + n_steps, problem, frontier, reached, n, solution)

        n_steps += 1

    return None


def bidirectional_path_cost_search(problem, max_steps=100, printer=None):

    def proceed(is_forward, problem, frontier, reached, reached2, solution):

        n = frontier.pop()

        for child in expand(problem, n, reverse=(not is_forward)):
            s = child.state
            if s not in reached or child.path_cost < reached[s].path_cost:
                reached[s] = child
                frontier.push(child)
                if s in reached2: 
                    solution2 = Solution.join(is_forward, child, reached2[s])
                    if not isinstance(solution, Solution) or solution2.cost < solution.cost:
                        solution = solution2

        return solution

    def terminated(solution, frontier1, frontier2):
        return solution == 'failure' or ((len(frontier1) == 0) and (len(frontier2) == 0))

    path_cost = lambda n : (n.path_cost, n.state)

    assert(len(problem.goals) == 1)

    frontier_f = PriorityQueue(path_cost)
    frontier_b = PriorityQueue(path_cost)

    reached_f = dict()
    reached_b = dict()

    solution = 'failure'

    if printer is not None:
        full_frontier = frontier_f.elements + frontier_b.elements
        full_reached = set(reached_f.keys()).union(set(reached_b.keys()))
        printer(0, problem, full_frontier, full_reached, None, solution)

    node_f = Node(problem.initial)
    node_b = Node(list(problem.goals)[0])

    frontier_f.push(node_f)
    frontier_b.push(node_b)

    reached_f[node_f.state] = node_f
    reached_b[node_b.state] = node_b

    it = 1
    while not terminated(solution, frontier_f, frontier_b):

        print('frontier_f', frontier_f)
        print('frontier_b', frontier_b)
        print('reached_f', reached_f.keys())
        print('reached_b', reached_b.keys())
        print('solution', solution)
        print('---')

        it += 1
        
        if len(frontier_f) == 0:
            backwards = False
            
        elif len(frontier_b) == 0 or frontier_f.top().path_cost >= frontier_b.top().path_cost:
            backwards = True

        else:
            backwards = False

        if printer is not None:
            full_frontier = frontier_f.elements + frontier_b.elements
            full_reached = set(reached_f.keys()).union(set(reached_b.keys()))
            current = frontier_b.top() if backwards else frontier_f.top()
            printer(it, problem, full_frontier, full_reached, current, solution)

        if backwards:
            solution = proceed(False, problem, frontier_b, reached_b, reached_f, solution)
            
        else:
            solution = proceed(True, problem, frontier_f, reached_f, reached_b, solution)
            


    if printer is not None:
        full_frontier = frontier_f.elements + frontier_b.elements
        full_reached = set(reached_f.keys()).union(set(reached_b.keys()))
        printer(it, problem, full_frontier, full_reached, None, solution)


    print('frontier_f', frontier_f)
    print('frontier_b', frontier_b)
    print('reached_f', reached_f.keys())
    print('reached_b', reached_b.keys())
    print('solution', solution)



    return solution



def breadth_first_search(problem, max_steps=100, printer=None):

    n = Node(problem.initial)
    frontier = list()
    reached = set()
    solution = None
    
    if printer is not None:
        printer(0, problem, frontier, reached, None, solution)

    reached.add(n.state)
    frontier.append(n)
    
    if printer is not None:
        printer(1, problem, frontier, reached, None, solution)

    n_steps = 0
    while len(frontier) > 0 and n_steps < max_steps:

        n = frontier.pop(0)
        for child in expand(problem, n):

            if problem.is_goal(child.state):

                solution = Solution.create(child)

                if printer is not None:
                    printer(2 + n_steps, problem, frontier, reached, n, solution)

                return solution
            
            if child.state not in reached:
                reached.add(child.state)
                frontier.append(child)

        if printer is not None:
            printer(2 + n_steps, problem, frontier, reached, n, solution)

        n_steps += 1

    return None



def depth_first_search(problem, max_steps=100, graph=False, printer=None, depth_limit=None, init_steps=0):

    depth = lambda n : 0 if n.parent is None else 1 + depth(n.parent)

    n = Node(problem.initial)
    frontier = list()
    reached = set()
    solution = 'failure'
    
    print("GO!\n")
    
    if printer is not None:
        printer(init_steps, problem, frontier, reached, None, solution)

    reached.add(n.state)
    frontier.append(n)
    
    if printer is not None:
        printer(init_steps + 1, problem, frontier, reached, None, solution)

    n_steps = init_steps + 2

    while len(frontier) > 0 and n_steps < max_steps:

        n = frontier.pop()
        if depth_limit is not None and depth(n) > depth_limit:
            solution = f'cutoff-{n_steps}'

        else:
            children = expand(problem, n)
            children.reverse()

            for child in children:
                if problem.is_goal(child.state):
                    solution = Solution.create(child)
                    if printer is not None:
                        print("frontier",  ", ".join(str(n) for n in frontier))
                        print("reached", reached)
                        print("n", n)
                        print("--------------------------------------------------")
                        print()
                        printer(n_steps, problem, frontier, reached, n, solution)
                        

                    return solution

                if not graph or (child.state not in reached):
                    frontier.append(child)

                reached.add(child.state)

            if printer is not None:
                print("frontier",  ", ".join(str(n) for n in frontier))
                print("reached", reached)
                print("n", n)
                print("--------------------------------------------------")
                print()

                printer(n_steps, problem, frontier, reached, n, solution)

        n_steps += 1

    return solution
            


def iterative_deepening_search(problem, max_steps=100, graph=True, printer=None):

    done = False
    depth = 0
    n_steps = 0
    while not done:

        res = depth_first_search(problem,
                                 max_steps=max_steps,
                                 graph=graph,
                                 printer=printer,
                                 depth_limit=depth,
                                 init_steps=n_steps)

        if not isinstance(res, str) or res == 'failure':
            done = True
        else:
            n_steps = int(res.partition('-')[-1])
            depth +=1



        
if __name__ == '__main__':
    import exercises
    from printers import pyplot_printer
    from utils import Problem
    
    f_cost = lambda x : (x.path_cost, x.state)

    problem = exercises.grid2
    pos = exercises.GRIDPOS

    interactive_printer = lambda i, p, f, r, c, s: pyplot_printer(i, p, f, r, c, s, pos=pos, interactive=True)
    #best_first_search(problem, f_cost, printer=interactive_printer, max_steps=100)
    #breadth_first_search(problem, printer=interactive_printer, max_steps=100)
    #depth_first_search(problem, printer=interactive_printer, max_steps=100)
    iterative_deepening_search(problem, printer=interactive_printer, max_steps=100)
    #bidirectional_path_cost_search(problem, printer=interactive_printer, max_steps=100)

02/search.py
paolomorettin/FAI-code

import networkx as nx
import numpy as np

class Node:
    '''
    Search node class.

    '''

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

    def __str__(self):
        str_parent = self.parent.state if self.parent is not None else 'None'
        str_succ = self.succ.state if self.succ is not None else 'None'
        return f"<{self.state}, {str_parent}, {str_succ}, {self.path_cost}>"


class Solution:

    def __init__(self):
        self.path = []
        self.cost = []

    def __getitem__(self, index):
        return self.path[index]

    def __len__(self):
        return len(self.path)

    def __str__(self):
        return "[" + ", ".join(str(n.state) for n in self.path) + f"] ({self.cost})"

    @staticmethod
    def create(final_node):

        solution = Solution()
        solution.cost = final_node.path_cost
        x = final_node
        
        while x is not None:
            solution.path = [x] + solution.path
            x = x.parent

        return solution

    @staticmethod
    def join(is_forward, node1, node2):

        assert(node1.state == node2.state)
        solution = Solution()

        if is_forward:
            forward_node, backward_node = node1, node2
        else:
            forward_node, backward_node = node2, node1

        solution.cost = forward_node.path_cost + backward_node.path_cost
        x = forward_node
        
        while x is not None:
            solution.path = [x] + solution.path
            x = x.parent

        x = backward_node.succ
        while x is not None:
            solution.path.append(x)
            x = x.succ

        return solution



class PriorityQueue:
    '''
    Fake 'priority queue' used in best-first search algorithms.

    Complexity: push O(nlogn), pop O(1)
    instead of: push O(logn), pop O(logn)

    Lazily implemented as a list that gets sorted at each insertion.
    Makes no difference for our purposes.
.

    '''
    def __init__(self, func):
        self.elements = []
        self.func = func

    def push(self, x):
        self.elements.append(x)
        self.elements.sort(key=self.func)

    def pop(self):
        return self.elements.pop(0)

    def top(self):
        if len(self) > 0:
            return self.elements[0]
        else:
            return None

    def __iter__(self):
        for x in self.elements:
            yield x

    def __len__(self):
        return len(self.elements)

    def __str__(self):
        return "[" + ", ".join(str(n) for n in self.elements) + "]"
        

class Problem:
    '''
    Problem class.

    '''

    def __init__(self, states, initial, goals):
        '''
        Initialize the problem instance with:
        states - a set of states
        initial - an initial state in 'states'
        goals - a set of goals (subset of 'states')

        The state space is implemented as dict-of-dicts:
        ((s0) x action) -> (s1, cost)

        '''
        assert(initial in states)
        assert(all([g in states for g in goals]))
        self.state_graph = {s : dict() for s in states}
        self.initial = initial
        self.goals = goals

    def add_action(self, s0, s1, a=None, c=1, undir=False):
        '''
        Add an action to the state space given:
          s0 - the start state in S
          s1 - the destination state in S
          a - the action resulting in (s0 -> s1)
          c - the cost of a (default: 1)
          undir - undirected (default: false)        

        '''
        assert(s0 in self.state_graph)
        assert(s1 in self.state_graph)
        if a is None:
            action = f'to_{s1}'

        self.state_graph[s0][action] = (s1, c)
        if undir:
            if a is None:
                action = f'to_{s0}'
            self.state_graph[s1][action] = (s0, c)

    def get_actions(self, s):
        '''
        Get the possible actions from state 's'

        '''
        assert(s in self.state_graph)
        return set(self.state_graph[s].keys())

    def is_goal(self, s):
        '''
        Return true if 's' is a goal

        '''
        assert(s in self.state_graph)
        return (s in self.goals)

    def is_weighted(self):
        '''
        Return true if actions have different costs

        '''
        costs = set.union(*[set([self.state_graph[s][a][1]
                                for a in self.state_graph[s]])
                           for s in self.state_graph])
        return len(costs) > 1

    

    def result(self, s, a):
        '''
        Return the destination state and the cost
        of using action 'a' in state 's'.

        '''
        assert(s in self.state_graph)
        assert(a in self.state_graph[s])
        return self.state_graph[s][a]

    def to_graph(self):
        '''
        Convert the prolem instance to a
        networkx directed multigraph.

        '''
        G = nx.DiGraph()
        for s0 in self.state_graph:
            G.add_node(s0)
            for a in self.state_graph[s0]:
                s1, c = self.state_graph[s0][a]
                G.add_edge(s0, s1, action=a, cost=c)

        return G
        

    @staticmethod
    def grid(w, h, initial=None, goals=None, p_edge=0.5, cost=None, seed=0):
        '''
        Generate a grid problem.

        '''
        states = set(range(1, h * w + 1))

        np.random.seed(seed)        
        initial = np.random.choice(list(states)) if initial is None else initial
        goals = {np.random.choice(list(states - {initial}))} if goals is None else goals

        problem = Problem(states, initial, goals)


        sxy = lambda x, y: 1 + x + w * y
        for y in range(h - 1):
            for x in range(w - 1):

                
                c = np.random.randint(cost[0], cost[1]+1) if cost is not None else 1
                if np.random.random() < p_edge:
                    problem.add_action(sxy(x, y), sxy(x + 1, y), c=c)
                if np.random.random() < p_edge:
                    problem.add_action(sxy(x + 1, y), sxy(x, y), c=c)


                c = np.random.randint(cost[0], cost[1]+1) if cost is not None else 1
                if np.random.random() < p_edge:
                    problem.add_action(sxy(x, y), sxy(x, y + 1), c=c)
                if np.random.random() < p_edge:
                    problem.add_action(sxy(x, y + 1), sxy(x, y), c=c)

            c = np.random.randint(cost[0], cost[1]+1) if cost is not None else 1
            if np.random.random() < p_edge:
                problem.add_action(sxy(x + 1, y), sxy(x + 1, y + 1), c=c)
            if np.random.random() < p_edge:
                problem.add_action(sxy(x + 1, y + 1), sxy(x + 1, y), c=c)

        for x in range(w - 1):
            c = np.random.randint(cost[0], cost[1]+1) if cost is not None else 1
            if np.random.random() < p_edge:
                problem.add_action(sxy(x, y + 1), sxy(x + 1, y + 1), c=c)
            if np.random.random() < p_edge:
                problem.add_action(sxy(x + 1, y + 1), sxy(x, y + 1), c=c)

        return problem
    


def expand(problem, node, reverse=False):

    s0 = node.state
    expanded_states = []
    
    if not reverse:
        for action in problem.get_actions(s0):
            s_next, cost = problem.result(s0, action)
            expanded_states.append((s_next, cost, action))

        expanded_states.sort(key=lambda x : x[0])
        return [Node(s, parent = node, path_cost = (node.path_cost + c), action=a)
                for s, c, a in expanded_states]

    else:
        for s_candidate in problem.state_graph:
            for action in problem.get_actions(s_candidate):
                s_next, cost = problem.result(s_candidate, action)
                if s_next == s0:
                    expanded_states.append((s_candidate, cost, action))

        expanded_states.sort(key=lambda x : x[0])
        return [Node(s, succ = node, path_cost = (node.path_cost + c), action=a)
                for s, c, a in expanded_states]
                
    

02/utils.py
paolomorettin/FAI-code

from utils import select_choice

feats = [['H', 'S', 'F'],
         ['L', 'T', 'O'],
         ['M', 'Y', 'G']]


def pp(p):
    return f"[{feats[0][p[0]]}-{feats[1][p[1]]}-{feats[2][p[2]]}]"

def select_parents(pop, obj, choices):

    i1, _ = select_choice([(i, obj(p)) for i, p in enumerate(pop)],
                       choices)
    i2, _ = select_choice([(i, obj(p)) for i, p in enumerate(pop)
                        if i != i1],
                       choices)
    return pop[i1], pop[i2]
    


def reproduce(p1, p2, choices):
    split, _ = select_choice([(1,1), (2,1)], choices)
    return p1[0:split] + p2[split:]


def mutate(p, p_mutation, choices):
    mutated = []
    for i in range(3):
        mf, _ = select_choice([(p[i], 1 - p_mutation),
                               ((p[i]+1) % 3, p_mutation / 2),
                               ((p[i]+2) % 3, p_mutation / 2)],
                              choices)
        mutated.append(mf)

    return tuple(mutated)


def genetic(obj, pop, choices, max_it=10, p_mutation=0.1):

    it = 0
    best = sorted(pop, key=lambda x : obj(x))[-1]
    while it < max_it:
        next_pop = []
        curr_best = None
        
        print()
        print("ITERATION:", it)
        print("Best:", pp(best), 'fitness:', obj(best))
        print("Population:")
        for p in pop: print(pp(p), 'fitness:', obj(p))
        print()
        
        for i in range(len(pop)):
            p1, p2 = select_parents(pop, obj, choices)
            child = reproduce(p1, p2, choices)
            mutated_child = mutate(child, p_mutation, choices)
            print("\t parent 1:", pp(p1), "parent 2:", pp(p2),
                  "made:", pp(child), "mutated:", pp(mutated_child))
            child = mutated_child

            next_pop.append(child)
            if curr_best is None or obj(curr_best) < obj(child):
                curr_best = child

        pop = next_pop
        if obj(curr_best) > obj(best):
            best = curr_best

        it += 1
        print()

    print("\n\nBEST:", pp(best), "fitness:", obj(best))
    return best


if __name__ == '__main__':

    choices = [0.2, 0.9, 0.5, 0.1]
    p_mutation = 0.2
    max_it = 5

    initial = [(0,0,0),
               (0,1,0),
               (0,0,1),
               (1,0,1)]

    obj = lambda p : sum(p)

    print("Initial population")
    for p in initial:
        print(pp(p), 'fitness:', obj(p))

    genetic(obj, initial, choices,
            max_it=max_it, p_mutation=p_mutation)
                


        

        
        

04/genetic.py
paolomorettin/FAI-code


from utils import all_neighbors, select_choice

hc_modes = ['steepest', 'stochastic (unweighted)', 'stochastic', 'first']




def hill_climbing(it, mode, obj, curr, choices, bounds=[0,10], printer=None):
    '''
    Recursive hill-climbing algorithm.

    mode -- one of 'hc_modes'
    obj -- objective function to be maximized
    curr -- current state
    choices -- predetermined list of choices (simulating stochasticity)
    bounds -- coordinate bounds of the state space

    '''

    print()
    print(f'Hill-climbing-{mode}, iteration: {it}, x: {curr}, f(x): {obj(curr)}')
    if printer is not None:
        printer(it, obj, bounds, curr)
    
    higher_neighbors = [(n, obj(n)) for n in all_neighbors(curr, bounds=bounds)
                        if obj(n) > obj(curr)]

    if len(higher_neighbors) == 0: # best in neighborhood!
        return curr, it

    else:

        if mode == 'steepest':
            nxt, _ = sorted(higher_neighbors, key=lambda x : (x[1],x[0]))[-1]

        elif mode == 'stochastic':
            nxt, _ = select_choice(higher_neighbors, choices)

        elif mode == 'stochastic (unweighted)':
            unweighted_neighbors = [(n, 1) for n, _ in higher_neighbors]
            nxt, _ = select_choice(unweighted_neighbors, choices)

        elif mode == 'first':
            nxt = None
            for nxt, fn in higher_neighbors:
                what_to_do, _ = select_choice([('break', 0.5), ('continue', 0.5)], choices)
                if what_to_do  == 'break':
                    break

        else:
            raise NotImplementedError(f'Mode {mode} not in {hc_modes}')

        return hill_climbing(it+1, mode, obj, nxt, choices, bounds=bounds, printer=printer)


if __name__ == '__main__':

    import numpy as np
    from utils import generate_nonconvex, plt_printer

    xbounds = [0,10] # bounds to input variables
    ybounds = [0,20] # bounds to the objective function
    seed = 666 # generation seed
    initial = (5,4) # initial state

    np.random.seed(seed)

    # predetermined non-deterministic choices
    n_choices = 4
    choices = [round(np.random.random(), 1) for _ in range(n_choices)]
    

    # generating the objective function
    obj = generate_nonconvex(seed, xbounds=xbounds, ybounds=ybounds)

    hill_climbing(1, 'steepest', obj, initial, choices, bounds=xbounds, printer=plt_printer)

    '''
    for mode in hc_modes:
        hill_climbing(1, mode, obj, initial, choices, bounds=xbounds, printer=plt_printer)

    '''

                


04/hill_climbing.py
paolomorettin/FAI-code

import matplotlib.pyplot as plt
import networkx as nx
import numpy as np

       

class Maze:

    def __init__(self, side):

        self.G = nx.Graph()
        self.side = side
        for x in range(side+1):
            for y in range(side+1):
                self.G.add_node((x,y))

        self.init = None
        self.goal = None


    def random(self, seed, init=None, goal=None):

        def neighbors(curr):
            x, y = curr
            neigh = []

            if y+1 <= self.side: neigh.append((x, y+1))
            if x+1 <= self.side: neigh.append((x+1, y))
            if y-1 >= 0: neigh.append((x, y-1))
            if x-1 >= 0: neigh.append((x-1, y))

            return neigh

        np.random.seed(seed)

        if init is not None:
            self.init = init
        else:
            candidates = list(self.G.nodes)
            self.init = candidates[np.random.randint(0, len(candidates))]


        if goal is not None:
            self.goal = goal
        else:
            candidates = list(set(self.G.nodes) - {self.init})
            self.goal = candidates[np.random.randint(0, len(candidates))]

        # generate edges
        reached = {self.init}
        while True:
            candidates = {(r, rn) for r in reached for rn in neighbors(r)
                          if rn not in reached}

            if len(candidates) == 0:
                break

            r, rn = list(candidates)[np.random.randint(len(candidates))]
            self.G.add_edge(r, rn)
            reached.add(rn)


    def actions(self, node):
        x, y = node
        adj = self.G[node]
        alist = []
        if (x, y+1) in adj: alist.append('N')
        if (x+1, y) in adj: alist.append('E')
        if (x, y-1) in adj: alist.append('S')
        if (x-1, y) in adj: alist.append('W')

        return alist


    def execute(self, node, action):
        x, y = node
        if action == 'N': nxt = (x, y+1)
        elif action == 'E': nxt = (x+1, y)
        elif action == 'S': nxt = (x, y-1)
        elif action == 'W': nxt = (x-1, y)

        assert((node, nxt) in self.G.edges)
        return nxt
            
        


    def plot(self, curr=None, path=None):

        n_size = 1200
        n_shape = 's'
        e_size=15
        # '#264653'
        col_r = '#2a9d8f'
        col_f = '#e9c46a'
        col_g = '#f4a261'
        col_i = '#e76f51'
        col_w = '#ffffff'
        col_b = '#000000'


        pos = {n : n for n in self.G.nodes}

        other_nodes = set(self.G.nodes) - {self.goal, self.init}

        if curr is not None:
            nx.draw_networkx_nodes(self.G, pos, nodelist=[curr],
                                   node_color=col_r, edgecolors=col_b,
                                   node_size=1.5*n_size, node_shape=n_shape)

        nx.draw_networkx_nodes(self.G, pos, nodelist=[self.goal],
                               node_color=col_g, edgecolors=col_b,
                               node_size=n_size, node_shape=n_shape)
            
        nx.draw_networkx_nodes(self.G, pos, nodelist=[self.init],
                               node_color=col_i, edgecolors=col_b,
                               node_size=n_size, node_shape=n_shape)

        nx.draw_networkx_nodes(self.G, pos, nodelist=other_nodes,
                               node_color=col_w, edgecolors=col_b,
                               node_size=n_size, node_shape=n_shape)

        strlabels = {n : f'{n[0]},{n[1]}' for n in self.G.nodes}
        nx.draw_networkx_labels(self.G, pos, labels=strlabels)
        nx.draw_networkx_edges(self.G, pos, width=e_size)

        plt.gca().set_aspect('equal')

        if path is None:
            plt.show()
        else:
            plt.savefig(path, dpi=150, bbox_inches='tight', transparent=True)

        plt.clf()


if __name__ == '__main__':

    side = 5
    m = Maze(side)
    m.random(666, init=(0,0), goal=(side, side))
    m.plot()
    #m.plot(path='big_maze.png')


            
                       
            
        



        

        

04/maze.py
paolomorettin/FAI-code

manhattan = lambda maze, s : abs(maze.goal[0] - s[0]) + abs(maze.goal[1] - s[1])

def odfs_printer(it, maze, s_curr, s_prev, a, result, untried, unbacktracked):

    restr = result[s_curr] if s_curr in result else '-'
    restr2 = result[s_prev] if s_prev is not None and s_prev in result else '-'
    unstr = untried[s_curr] if s_curr in untried else '-'
    ubstr = unbacktracked[s_curr] if s_curr in unbacktracked else '-'
    print("---")
    print("s':", s_curr, "s:", s_prev, "a:", a)
    print("result[s']:", restr)
    print("result[s]:", restr2)
    print("untried[s']:", unstr)
    print("unbacktracked[s']:", ubstr)
    print()

    maze.plot(s_curr)


def lrtastar_printer(it, maze, s_curr, s_prev, a, result, H):
    
    restr = result[s_curr] if s_curr in result else '-'
    restr2 = result[s_prev] if s_prev is not None and s_prev in result else '-'
    hstr = H[s_curr] if s_curr in H else '-'
    print("---")
    print("s':", s_curr, "s:", s_prev, "a:", a)
    print("result[s']:", restr)
    print("result[s]:", restr2)
    print("H[s']:", hstr)
    print()

    maze.plot(s_curr)



def online_dfs(maze, printer=None):

    s_prev, a = None, None
    result = dict()
    untried = dict()
    unbacktracked = dict()

    s_curr = maze.init
    it = 0

    if printer is not None:
        printer(it, maze, s_curr, s_prev, a, result,
                untried, unbacktracked)

    while True:

        it += 1

        if s_curr == maze.goal:
            if printer is not None:
                printer(it, maze, s_curr, s_prev, a, result,
                        untried, unbacktracked)
            return True

        if s_curr not in untried:
            # s_curr explored for the first time
            # aux data structures are initialized
            untried[s_curr] = maze.actions(s_curr)
            unbacktracked[s_curr] = []

        if s_prev is not None:
            # neither first step
            # nor a backtrack
            
            if s_prev not in result:
                result[s_prev] = {a : s_curr}
            else:
                result[s_prev][a] = s_curr

            unbacktracked[s_curr].append(s_prev)

        if printer is not None:

            printer(it, maze, s_curr, s_prev, a, result,
                    untried, unbacktracked)

        if len(untried[s_curr]) == 0:

            if len(unbacktracked[s_curr]) == 0:
                return False

            s_next = unbacktracked[s_curr].pop()
            a = None            
            for b in result[s_curr]:
                if result[s_curr][b] == s_next:
                    a = b
                    break

            s_prev = None # that dreaded line

        else:
            a = untried[s_curr].pop(0)
            s_prev = s_curr

        # execute
        s_curr = maze.execute(s_curr, a)


def lrtastar(maze, h, printer=None):

    def lrtas_cost(maze, s, a, result, H, h):

        s_next = None
        if s in result and a in result[s]:
            # uniform c(s, a, s') = 1 in our mazes
            return H[result[s][a]] + 1        
        else:
            return h(maze, s)


        

    s_prev, a = None, None
    result = dict()
    H = dict()

    s_curr = maze.init
    it = 0

    if printer is not None:
        printer(it, maze, s_curr, s_prev, a, result, H)

    while True:

        it += 1

        if s_curr == maze.goal:
            if printer is not None:
                printer(it, maze, s_curr, s_prev, a, result, H)

            return True

        if s_curr not in H:
            # s_curr explored for the first time
            # aux data structures are initialized
            H[s_curr] = h(maze, s_curr)

        if s_prev is not None:
            if s_prev not in result:
                result[s_prev] = {a : s_curr}
            else:
                result[s_prev][a] = s_curr

            H[s_prev] = min(lrtas_cost(maze, s_prev, b, result, H, h)
                            for b in maze.actions(s_prev))

            
        if printer is not None:
            printer(it, maze, s_curr, s_prev, a, result, H)


        a_costs = [(a, lrtas_cost(maze, s_curr, a, result, H, h))
                   for a in maze.actions(s_curr)]
        a_costs.sort(key=lambda x : (x[1], 'NESW'.index(x[0])))
        a = a_costs[0][0] # lower estimated cost action

        s_prev = s_curr
        s_curr = maze.execute(s_curr, a)
        


if __name__ == '__main__':

    from maze import Maze
    test_maze = Maze(2)
    test_maze.random(666)
    # uncomment this for ODFS
    #online_dfs(test_maze, printer=odfs_printer)
    lrtastar(test_maze, manhattan, printer=lrtastar_printer)

            
            

    
    
    

04/online_search.py
paolomorettin/FAI-code

import matplotlib.patches as patches
import matplotlib.pyplot as plt

import seaborn as sns

import numpy as np


''' Parametric univariate normal. Given mean and stdev, returns the
PDF f(x) s.t. x ~ N(mu, sigma).

'''
p_norm = lambda mu, sigma : (lambda x : np.exp((((x-mu)/sigma)**2)/-2)/(sigma*np.sqrt(2*np.pi)))


def select_choice(options, choices):
    '''Simulated stochastic process that deterministically pick an
    option given a pre-determined list of choices. Choices are
    cycled through. Options are weighted.

    '''
    choice = choices.pop(0) 
    choices.append(choice) # cycling

    w_sum = sum(w for _, w in options)
    p = 0
    for opt, w in options:
        p += w/w_sum
        if choice <= p:
            return opt, p


def all_neighbors(current_state, bounds=[0,10]):
    '''
    Returns the list all 8-neighbors in a 2D grid, in this order:

    SW, S, SE, W, -, E, NW, N, NE

    '''

    xc, yc = current_state
    return [(xn, yn)
            for yn in [yc-1, yc, yc+1]                  # the 9-tiles square centered in (xc, yc)
            for  xn in [xc-1, xc, xc+1]
            if ((xn != xc) or (yn != yc))               # minus (xc, yc) itself
            and xn >= bounds[0] and xn < bounds[1]
            and yn >= bounds[0] and yn < bounds[1]]     # and the states out of bounds


def generate_nonconvex(seed, xbounds=[0,10], ybounds=[0,99], n_modes=7):
    '''Generates a non-convex function by sampling a mixture of
    'n_modes' 2d-gaussians.

    '''
    np.random.seed(seed)
    rng = np.random.random
    xmin, xmax = xbounds

    params = []                             
    for _ in range(n_modes):
        mode = (rng() * xmax, rng() * xmax)
        stdev = xmax/n_modes #rng() * (xmax - xmin)
        weight = 1
        params.append((mode, stdev, weight))

    opt = lambda xy : sum(p3 \
                          * p_norm(p1[0],p2)(xy[0]) \
                          * p_norm(p1[1],p2)(xy[1])
                          for p1, p2, p3 in params)

    max_opt = max(opt((x,y)) for x in range(xmin, xmax)
                  for y in range(xmin, xmax))

    ymin, ymax = ybounds
    norm_opt = lambda xy : ymin + int(opt(xy)*(ymax-ymin)/max_opt)

    return norm_opt


def plt_printer(it, opt, bounds, curr, path=None):

    col_curr = '#00ff00'

    bmin, bmax = bounds
    xopt = np.array([[opt((x, y)) for x in range(bmin, bmax)]
                     for y in range(bmin, bmax)])
    
    ax = sns.heatmap(xopt, annot=True, cbar=False)


    if curr is not None:
        rect = patches.Rectangle(curr, 1, 1, linewidth=3, edgecolor=col_curr, facecolor='none')
        ax.add_patch(rect)

    ax.invert_yaxis()

    if path is not None:
        strpath = path(it)
        plt.savefig(strpath, dpi=150, bbox_inches='tight', transparent=True)
    else:
        plt.show()

    plt.clf()



if __name__ == '__main__':

    options = [(i, 1) for i in range(4)]
    choices = [0.2, 0.6, 0.3]

    print(select_choice(options, choices))
    print(select_choice(options, choices))
    print(select_choice(options, choices))
    print(select_choice(options, choices))


    

    

    

    
                          

    

04/utils.py
paolomorettin/FAI-code