Value Iteration.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "extended-philip",
   "metadata": {},
   "source": [
    "## Grid World Value Iteration\n",
    "\n",
    "<img src=\"img/bombs and gold numbers.png\" style=\"width: 320px;\" align=\"left\" caption=\"Figure 1\"/>\n",
    "\n",
    "**Actions available:** The agent has four possible actions in each grid square. These are *west*, *north*, *south*, and *east*. If the direction of movement is blocked by a wall (for example, if the agent executes action south in grid square 1), the agent remains in the same grid square. \n",
    "\n",
    "**Collecting gold:** On its first arrival at a grid square that contains gold (from a neighbouring grid square), the agent collects the gold. Note that, in order to collect the gold, the agent needs to transition into the grid square (containing the gold) from a different grid square.\n",
    "\n",
    "**Hitting the bomb:** On arrival at a grid square that contains a bomb (from a neighbouring grid square), the agent activates the bomb. \n",
    "\n",
    "**Terminal states:** The game terminates when **all** gold is collected or when the bomb is activated. In Exercises 1 and 2, you can define terminal states to be grid squares 18 and 23. In Exercise 3, you will need to define terminal state(s) differently.\n",
    "\n",
    "**Reward function:** $-1$ for each navigation action (including when the action results in hitting the wall), an additional $+10$ for collecting each piece of gold, and an additional $-10$ for activating the bomb. \n",
    "\n",
    "Discounting is not used (i.e. $\\gamma=1$)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "olive-money",
   "metadata": {},
   "outputs": [],
   "source": [
    "# environment\n",
    "import numpy as np\n",
    "\n",
    "class Gridworld:\n",
    "    def __init__(self):\n",
    "        self.num_rows = 5\n",
    "        self.num_cols = 5\n",
    "        self.num_cells = self.num_cols * self.num_rows\n",
    "        \n",
    "        # Choose starting position of the agent randomly among the first 5 cells\n",
    "        self.agent_position = np.random.randint(0, 5)\n",
    "        \n",
    "        # Choose position of the gold and bomb\n",
    "        self.bomb_positions = np.array([18])\n",
    "        self.gold_positions = np.array([23])\n",
    "       \n",
    "        # Specify rewards\n",
    "        self.rewards = np.zeros(self.num_cells)\n",
    "        self.rewards[self.bomb_positions] = -10\n",
    "        self.rewards[self.gold_positions] = 10\n",
    "        \n",
    "        # Specify available actions\n",
    "        self.actions = [\"UP\", \"RIGHT\", \"DOWN\", \"LEFT\"]\n",
    "        self.num_actions = len(self.actions)\n",
    "    \n",
    "    def setLocation(self, position):\n",
    "        self.agent_position = position\n",
    "    \n",
    "    def makeStep(self, action_index): \n",
    "        action = self.actions[action_index]\n",
    "\n",
    "        # Determine new position and check whether the agent hits a wall.\n",
    "        old_position = self.agent_position\n",
    "        new_position = self.agent_position\n",
    "        if action == \"DOWN\":\n",
    "            candidate_position = old_position + self.num_cols\n",
    "            if candidate_position < self.num_cells:\n",
    "                new_position = candidate_position\n",
    "        elif action == \"RIGHT\":\n",
    "            candidate_position = old_position + 1\n",
    "            if candidate_position % self.num_cols > 0:  # The %-operator denotes \"modulo\"-division.\n",
    "                new_position = candidate_position\n",
    "        elif action == \"UP\":\n",
    "            candidate_position = old_position - self.num_cols\n",
    "            if candidate_position >= 0:\n",
    "                new_position = candidate_position\n",
    "        elif action == \"LEFT\":  # \"LEFT\"\n",
    "            candidate_position = old_position - 1\n",
    "            if candidate_position % self.num_cols < self.num_cols - 1:\n",
    "                new_position = candidate_position\n",
    "        \n",
    "        # Calculate reward\n",
    "        reward = self.rewards[new_position]\n",
    "        reward -= 1\n",
    "        return reward, new_position"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "painful-antibody",
   "metadata": {},
   "source": [
    "This first task uses the Bellman equation for value iteration however assumes that an action is guaranteed to successfully take place. It outputs the optimal policy for the environment."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "lyric-bronze",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[[3. 4. 5. 4. 5.]\n",
      " [4. 5. 6. 5. 6.]\n",
      " [5. 6. 7. 6. 7.]\n",
      " [6. 7. 8. 0. 8.]\n",
      " [7. 8. 9. 0. 9.]]\n",
      "[['e' 'e' 's' 'e' 's']\n",
      " ['e' 'e' 's' 'e' 's']\n",
      " ['e' 'e' 's' 'e' 's']\n",
      " ['e' 'e' 's' '' 's']\n",
      " ['e' 'e' 'e' '' 'w']]\n"
     ]
    }
   ],
   "source": [
    "env = Gridworld()\n",
    "policy = np.empty(25, dtype = np.unicode_)\n",
    "v = np.zeros(25)\n",
    "theta = 1e-10\n",
    "valueAction = [0, 0, 0, 0]\n",
    "sigma = 1\n",
    "difference = theta + 1\n",
    "actions = ['n', 'e', 's', 'w']\n",
    "\n",
    "# loop thru \n",
    "while difference > theta:\n",
    "    difference = 0\n",
    "    for state in range (25):\n",
    "        # check if state is terminal, if terminal skip\n",
    "        if state == 18 or state == 23:\n",
    "            continue\n",
    "        currentStateValue = v[state]\n",
    "        # move agent to current state in environment\n",
    "        env.setLocation(state)\n",
    "        # for each action in the state, find valueAction\n",
    "        for action in range (4):\n",
    "            reward, position = env.makeStep(action)\n",
    "            valueAction[action] = reward + (sigma*v[position])\n",
    "        # find max valueAction\n",
    "        v[state] = max(valueAction[0], valueAction[1], valueAction[2], valueAction[3])\n",
    "        # policy improvement\n",
    "        policy[state] = actions[valueAction.index(v[state])]\n",
    "        difference = max(difference, abs(currentStateValue - v[state]))\n",
    "print(v.reshape((5, 5)))\n",
    "print(policy.reshape((5, 5)))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "covered-parcel",
   "metadata": {},
   "source": [
    "This second tasks continues using value iteration however the agent now fails its action 20% of the time."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "outdoor-census",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[[1.25 2.5  3.75 2.5  3.75]\n",
      " [2.5  3.75 5.   3.75 5.  ]\n",
      " [3.75 5.   6.25 5.   6.25]\n",
      " [5.   6.25 7.5  0.   7.5 ]\n",
      " [6.25 7.5  8.75 0.   8.75]]\n",
      "[['e' 'e' 's' 'e' 's']\n",
      " ['e' 'e' 's' 'e' 's']\n",
      " ['e' 'e' 's' 'e' 's']\n",
      " ['e' 'e' 's' '' 's']\n",
      " ['e' 'e' 'e' '' 'w']]\n"
     ]
    }
   ],
   "source": [
    "env = Gridworld()\n",
    "policy = np.empty(25, dtype = np.unicode_)\n",
    "v = np.zeros(25)\n",
    "theta = 1e-10\n",
    "valueAction = [0, 0, 0, 0]\n",
    "sigma = 1\n",
    "difference = theta + 1\n",
    "actions = ['n', 'e', 's', 'w']\n",
    "\n",
    "# loop thru \n",
    "while difference > theta:\n",
    "    difference = 0\n",
    "    for state in range (25):\n",
    "        # check if state is terminal, if terminal skip\n",
    "        if state == 18 or state == 23:\n",
    "            continue\n",
    "        currentStateValue = v[state]\n",
    "        # move agent to current state in environment\n",
    "        env.setLocation(state)\n",
    "        # for each action in the state, find valueAction\n",
    "        for action in range (4):\n",
    "            reward, position = env.makeStep(action)\n",
    "            #  80% chance of successful action, 20% chance that action fails\n",
    "            valueAction[action] = round((0.8 * (reward + (sigma*v[position]))) + (0.2 * (-1 + (sigma * v[state]))), 2)\n",
    "        # find max valueAction\n",
    "        v[state] = max(valueAction[0], valueAction[1], valueAction[2], valueAction[3])\n",
    "        # policy improvement\n",
    "        policy[state] = actions[valueAction.index(v[state])]\n",
    "        difference = max(difference, abs(currentStateValue - v[state]))\n",
    "\n",
    "\n",
    "print(v.reshape((5, 5)))\n",
    "print(policy.reshape((5, 5)))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "unique-queens",
   "metadata": {},
   "source": [
    "<img src=\"img/bomb and two gold.png\" style=\"width: 300px;\" align=\"left\" caption=\"Figure 1\"/>\n",
    "The final task uses a similar stochastic environment as the previous, however now includes two gold objects. A terminal state is only reached when both gold pieces are collected, or the bomb is activated."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "three-jefferson",
   "metadata": {},
   "outputs": [],
   "source": [
    "# environment\n",
    "\n",
    "class Ex3Gridworld:\n",
    "    def __init__(self):\n",
    "        self.num_rows = 5\n",
    "        self.num_cols = 5\n",
    "        self.num_cells = self.num_cols * self.num_rows\n",
    "        \n",
    "        # Choose starting position of the agent randomly among the first 5 cells\n",
    "        self.agent_position = np.random.randint(0, 5)\n",
    "        \n",
    "        # Choose position of the gold and bomb\n",
    "        self.bomb_positions = np.array([18])\n",
    "        self.gold_positions = np.array([23, 12])\n",
    "       \n",
    "        # Specify rewards\n",
    "        self.rewards = np.zeros(self.num_cells)\n",
    "        self.rewards[self.bomb_positions] = -10\n",
    "        self.rewards[self.gold_positions] = 10\n",
    "        \n",
    "        # Specify available actions\n",
    "        self.actions = [\"UP\", \"RIGHT\", \"DOWN\", \"LEFT\"]\n",
    "        self.num_actions = len(self.actions)\n",
    "    \n",
    "    def setLocation(self, position):\n",
    "        self.agent_position = position\n",
    "    \n",
    "    def makeStep(self, action_index): \n",
    "        action = self.actions[action_index]\n",
    "\n",
    "        # Determine new position and check whether the agent hits a wall.\n",
    "        old_position = self.agent_position\n",
    "        new_position = self.agent_position\n",
    "        if action == \"DOWN\":\n",
    "            candidate_position = old_position + self.num_cols\n",
    "            if candidate_position < self.num_cells:\n",
    "                new_position = candidate_position\n",
    "        elif action == \"RIGHT\":\n",
    "            candidate_position = old_position + 1\n",
    "            if candidate_position % self.num_cols > 0:  # The %-operator denotes \"modulo\"-division.\n",
    "                new_position = candidate_position\n",
    "        elif action == \"UP\":\n",
    "            candidate_position = old_position - self.num_cols\n",
    "            if candidate_position >= 0:\n",
    "                new_position = candidate_position\n",
    "        elif action == \"LEFT\":  # \"LEFT\"\n",
    "            candidate_position = old_position - 1\n",
    "            if candidate_position % self.num_cols < self.num_cols - 1:\n",
    "                new_position = candidate_position\n",
    "        \n",
    "        # Calculate reward\n",
    "        reward = self.rewards[new_position]\n",
    "        reward -= 1\n",
    "        return reward, new_position"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "affecting-mountain",
   "metadata": {},
   "outputs": [],
   "source": [
    "env = Ex3Gridworld()\n",
    "policy = np.empty(25, dtype = np.unicode_)\n",
    "v = np.zeros(25)\n",
    "theta = 1e-10\n",
    "valueAction = [0, 0, 0, 0]\n",
    "sigma = 1\n",
    "difference = theta + 1\n",
    "actions = ['n', 'e', 's', 'w']\n",
    "\n",
    "# loop thru \n",
    "while difference > theta:\n",
    "    difference = 0\n",
    "    for state in range (25):\n",
    "        # check if state is terminal, if terminal skip\n",
    "        if state == 18:\n",
    "            continue\n",
    "        # if both state 12 and 23 visited then continue ???\n",
    "        # TO DO\n",
    "        currentStateValue = v[state]\n",
    "        # move agent to current state in environment\n",
    "        env.setLocation(state)\n",
    "        # for each action in the state, find valueAction\n",
    "        for action in range (4):\n",
    "            reward, position = env.makeStep(action)\n",
    "            #  80% chance of successful action, 20% chance that action fails\n",
    "            valueAction[action] = round((0.8 * (reward + (sigma*v[position]))) + (0.2 * (-1 + (sigma * v[state]))), 2)\n",
    "        # find max valueAction\n",
    "        v[state] = max(valueAction[0], valueAction[1], valueAction[2], valueAction[3])\n",
    "        # policy improvement\n",
    "        policy[state] = actions[valueAction.index(v[state])]\n",
    "        difference = max(difference, abs(currentStateValue - v[state]))\n",
    "\n",
    "\n",
    "print(v.reshape((5, 5)))\n",
    "print(policy.reshape((5, 5)))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "worldwide-custom",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.9.0"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}