<jupyter_start><jupyter_text>The TaskYou enter the maze in A and at the same time, the minotaur enters in B. The minotaur follows a random walk while staying within the limits of the maze. The minotaur’s walk goes through walls (which obviously you cannot do). At each step, you observe the position of the minotaur, and decide on a one-step move (up, down, right or left) or not to move. If the minotaur catches you, he will eat you. Your objective is to identify a strategy maximizing the probability of exiting the maze (reaching B) before time T.  MDP FormulationPart of this task is to define the Markov Decission Process surrounding this problem.States:- Gridworld of 7x8 = 56- Minotaur can be in either of the 56 cellsTotal states = 56*56 = 3136Actions:- (0) North, (1)South, (2)East, (3)WestTransition probability:- Every action will succesfully move the player to the adjacent cell, unless it hits a wall or a border, in those case the player states in the same cell.Rewards:- Arrive to B: T + 5. End- Hit wall: -5- Movement: -1- Same cell as the minotaur: -20. End- T steps without arriving: -10. EndEnd conditions:- Same cell as the minotaur, Reward: -20- Arrive to   Creating the environmentFirst we need to create the environment with which our agent is going to interact:<jupyter_code>import numpy as np
from copy import copy, deepcopy
import math
from collections import deque, defaultdict
import sys
def add(tuple_a, tuple_b):
    return (tuple_a[0] + tuple_b[0], tuple_a[1] + tuple_b[1] )
def in_limits(position, limits):
    return (position[0] >= 0 and position[0] < limits[0]) and (position[1] >= 0 and position[1] < limits[1])

mapGrid = np.array([
            [  0, 0, 1, 0, 0,  0, 0, 0],
            [  0, 0, 1, 0, 0,  1, 0, 0],
            [  0, 0, 1, 0, 0,  1, 1, 1],
            [  0, 0, 1, 0, 0,  1, 0, 0],
            [  0, 0, 0, 0, 0,  0, 0, 0],
            [  0, 1, 1, 1, 1,  1, 1, 0],
            [  0, 0, 0, 0, 1,  0, 0, 0]
        ])

translationDictionary = {
            0: (-1, 0),
            1: (1, 0),
            2: (0, 1),
            3: (0, -1)
        }

def generateMoveDict(grid, ignore_walls=True):
    moveDict = [[{0: (x, y), 1:(x, y), 2: (x, y), 3: (x, y)  } \
                 for y in range(grid.shape[1]) ] \
                for x in range(grid.shape[0])]
    
    for x in range(grid.shape[0]):
        for y in range(grid.shape[1]):
            moves = moveDict[x][y]
            for A in range(4):
                next_cell = add(moves[A], translationDictionary[A])
                if in_limits(next_cell, grid.shape):
                    if ignore_walls:
                        moves[A] = next_cell
                    else:
                        if grid[next_cell[0]][next_cell[1]] != 1:
                            moves[A] = next_cell
                            
            moveDict[x][y] = moves                
            
    return moveDict

def generateStateGrid(grid):
    states = [[ deepcopy(grid) for y in range(grid.shape[1]) ] for x in range(grid.shape[0])]
    for x in range(grid.shape[0]):
        for y in range(grid.shape[1]):
            states[x][y][x][y] = 5
    return states

class MinotaurEnvironment:
    def __init__(self, T=20):
        self.T = T
        self.playerInitialPosition = (0,0)
        self.minotaurInitialPosition = (6, 5)
        self.playerPos = self.playerInitialPosition
        self.minotaurPos = self.minotaurInitialPosition
        self.playerMovDict = generateMoveDict(mapGrid, ignore_walls=False)
        self.minotaurMovDict = generateMoveDict(mapGrid)
        self.stateGrid = generateStateGrid(mapGrid)
        self.state = self.__mapPosToState(self.playerPos)
        self.steps = 0
        self.done = False
        
    def reset(self):
        self.steps = 0
        self.done = False
        self.playerPos = self.playerInitialPosition
        self.minotaurPos = self.minotaurInitialPosition
        self.state = self.__mapPosToState(self.playerPos)
    
    def __mapPosToState(self, coordinate):
        posVariable = coordinate[0]*8 + coordinate[1]
        minotaurVariable = self.minotaurPos[0]*8 + self.minotaurPos[1]
        
        return minotaurVariable*7*8 + posVariable
    
    def __nextPlayerPosition(self, action):
        return self.playerMovDict[self.playerPos[0]][self.playerPos[1]][action]
    
    def __nextMinotaurPosition(self):
        return self.minotaurMovDict[self.minotaurPos[0]][self.minotaurPos[1]][np.random.choice(4, 1)[0]]
    
    def step(self, action):
        if (self.done):
            self.done = True
            return self.state, 0, True
        newPlayerPosition = self.__nextPlayerPosition(action)
        
        if (newPlayerPosition == self.minotaurPos):
            # Same cell as minotaur
            self.done = True
            self.state = self.__mapPosToState(newPlayerPosition)
            self.steps =+ 1
            return self.state, -20, True
        
        if (newPlayerPosition == self.minotaurInitialPosition):
            # Arrive to the exit
            self.done = True
            self.steps =+ 1
            self.state = self.__mapPosToState(newPlayerPosition)
            return self.state, self.T + 5, True
        
        self.minotaurPos = self.__nextMinotaurPosition()
        
        if (self.minotaurPos == newPlayerPosition):
            # Same cell as minotaur after minotaur moved
            self.done = True
            self.state = self.__mapPosToState(newPlayerPosition)
            self.steps =+ 1
            return self.state, -20, True
        
        self.steps =+ 1
        
        if (self.steps == self.T):
            self.done = True
            self.state = self.__mapPosToState(newPlayerPosition)
            return self.state, -self.T, True
        
        if (newPlayerPosition == self.playerPos):
            # Same position as before
            self.state = self.__mapPosToState(newPlayerPosition)
            return self.state, -5, False
        
        self.playerPos = newPlayerPosition
        self.state = self.__mapPosToState(newPlayerPosition)
        return self.state, -1, False

def getGridFromState(state):
    grid = deepcopy(mapGrid)
    
    playerState = state % (7*8)
    playerCoord = (math.floor(playerState/8), playerState % 8)
    
    minotaurState = math.floor(state/(7*8))
    minotaurCoord = (math.floor(minotaurState/8), minotaurState % 8)
    
    grid[playerCoord[0]][playerCoord[1]] = 42
    grid[minotaurCoord[0]][minotaurCoord[1]] = 99
    
    return grid
    
class QLearningAgent:

    def __init__(self, nA=6, epsilon=1.0, alpha=.05, gamma=1.0):
        """ Initialize agent.
        
        Params
        ======
        - nA: number of actions available to the agent
        - epsilon: parameter for epsilon-Greedy algorithm
        - alpha: "Learning rate"
        - gamma: Discount factor
        """
        self.nA = nA
        self.Q = defaultdict(lambda: np.zeros(self.nA))
        self.epsilon = epsilon
        self.alpha = alpha
        self.gamma = gamma


    def __update_Q(self, current_Q, next_Q, reward):
        """ Updates the action-value function estimate using the most recent time step """
        return current_Q + (self.alpha * (reward + (self.gamma * next_Q) - current_Q))

    def select_action(self, state):
        """ Given the state, select an action """
        policy_probabilities = np.ones(self.nA) * self.epsilon / self.nA
        policy_probabilities[np.argmax(self.Q[state])] = 1 - self.epsilon + (self.epsilon / self.nA)

        return np.random.choice(np.arange(self.nA), p=policy_probabilities)

    def step(self, state, action, reward, next_state, episode, done):
        """ Update the agent's knowledge, using the most recently sampled tuple """
        self.epsilon = 1.0 / episode
        
        self.Q[state][action] = self.__update_Q(self.Q[state][action], np.max(self.Q[next_state]), reward)      
def interact(env, agent, num_episodes=20000, window=100):
    """ Monitor agents performance
    
    Params
    ======
    - env: instance of OpenAI Gym's Taxi-v1 environment
    - agent: instance of class Agent (see Agent.py for details)
    - num_episodes: number of episodes of agent-environment interaction
    - window: number of episodes to consider when calculating average rewards

    Returns
    =======
    - avg_rewards: deque containing average rewards
    - best_avg_reward: largest value in the avg_rewards deque
    """
    avg_rewards = deque(maxlen=num_episodes)
    best_avg_reward = -math.inf
    samp_rewards = deque(maxlen=window)
    convergence_episode = 0
    
    for i_episode in range(1, num_episodes+1):
        state = env.reset()
        samp_reward = 0
        
        # Steps in episode
        while True:
            action = agent.select_action(state)
            next_state, reward, done = env.step(action)
            agent.step(state, action, reward, next_state, i_episode, done)
            
            # update the sampled reward
            samp_reward += reward

            state = next_state
            if done:
                # save final sampled reward
                samp_rewards.append(samp_reward)
                break
        if (i_episode >= window):
            # get average reward from last 100 episodes
            avg_reward = np.mean(samp_rewards)
            avg_rewards.append(avg_reward)

            if avg_reward > best_avg_reward:
                convergence_episode = i_episode
                best_avg_reward = avg_reward

        print("\rEpisode {}/{} || Best average reward {}".format(i_episode, num_episodes, best_avg_reward), end="")
        sys.stdout.flush()
        
        if i_episode == num_episodes: print('\n')
    return avg_rewards, best_avg_reward, convergence_episode
agent = QLearningAgent(nA=4, gamma=.9, alpha=.04, epsilon=0.9)
env = MinotaurEnvironment(T=30)
avg_rewards, best_avg_reward, episode_best_reward = interact(env, agent)
import time
env.reset()
steps = 0
states = []
# Steps in episode
while True:
    steps = steps + 1
    action = agent.select_action(state)
    next_state, reward, done = env.step(action)
    # update the sampled reward
    state = next_state
    states.append(getGridFromState(state))
    if done:
        break
def print_states(states):
    for state in states:
        print("{}\n".format(state), end="")
        time.sleep(1)
        sys.stdout.flush()        
print_states(states)<jupyter_output>[[42  0  1  0  0  0  0  0]
 [ 0  0  1  0  0  1  0  0]
 [ 0  0  1  0  0  1  1  1]
 [ 0  0  1  0  0  1  0  0]
 [ 0  0  0  0  0  0  0  0]
 [ 0  1  1  1  1 99  1  0]
 [ 0  0  0  0  1  0  0  0]]
[[ 0  0  1  0  0  0  0  0]
 [42  0  1  0  0  1  0  0]
 [ 0  0  1  0  0  1  1  1]
 [ 0  0  1  0  0  1  0  0]
 [ 0  0  0  0  0  0  0  0]
 [ 0  1  1  1  1  1 99  0]
 [ 0  0  0  0  1  0  0  0]]
[[ 0  0  1  0  0  0  0  0]
 [ 0  0  1  0  0  1  0  0]
 [42  0  1  0  0  1  1  1]
 [ 0  0  1  0  0  1  0  0]
 [ 0  0  0  0  0  0  0  0]
 [ 0  1  1  1  1  1  1  0]
 [ 0  0  0  0  1  0 99  0]]
[[ 0  0  1  0  0  0  0  0]
 [ 0  0  1  0  0  1  0  0]
 [ 0  0  1  0  0  1  1  1]
 [42  0  1  0  0  1  0  0]
 [ 0  0  0  0  0  0  0  0]
 [ 0  1  1  1  1  1  1  0]
 [ 0  0  0  0  1  0 99  0]]
[[ 0  0  1  0  0  0  0  0]
 [ 0  0  1  0  0  1  0  0]
 [ 0  0  1  0  0  1  1  1]
 [ 0 42  1  0  0  1  0  0]
 [ 0  0  0  0  0  0  0  0]
 [ 0  1  1  1  1  1 99  0]
 [ 0  0  0  0  1  0  0  0]]
[[ 0  0  1  0  0  0  0  0]
 [ 0  0  1  0  0  1  0 [...]
{'fraction_non_alphanumeric': 0.10262008733624454, 'fraction_numerical': 0.0451528384279476, 'mean_word_length': 1.9929430214323052, 'pattern_counts': {'":': 0, '<': 6, '<?xml version=': 0, '>': 8, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 1, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '1', 'dir': 'jupyter-structured-clean-dedup', 'id': '134807', 'license': 'Unlicense', 'n_tokens_mistral': 4058, 'n_tokens_neox': 3821, 'n_words': 1194, 'path': 'q-learning/Minotaur Maze.ipynb', 'repo_name': 'peoplenarthax/taxi-reinforcement-learning', 'size': '11450'}