Modified Grid World RL Problem (Policy Iteration Algorithm).ipynb

{
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "yXAf85FcFQjH"
      },
      "source": [
        "### Policy Iteration Algorithm\n",
        "\n",
        "*For performing the Policy Iteration, we follow the steps below*:\n",
        "1. Initialize a random policy\n",
        "2. Calculate the utility or value of all state given the current policy.\n",
        "3. With the new value(s) obtained for each state, compute the utility value for all possible actions i.e. up, right, down, left.\n",
        "4. Select the action with the maximum value.\n",
        "5. Compare this value with the value obtained from the initial policy.\n",
        "* 5a. If new value > old_value -> update the policy\n",
        "* 5b. Else if new value = old value -> Policy converged."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "SvBZNmglJAx7"
      },
      "outputs": [],
      "source": [
        "# Initialize random policies for the environment.\n",
        "# Policy Evaluation Phase\n",
        "\n",
        "def P_eval(initial_policy, v):\n",
        "  for k in range(1,16):\n",
        "    for row in range(n_rows):\n",
        "        for col in range(n_cols):\n",
        "            if (row, col) == terminals[0]:\n",
        "              v[row, col] = 1\n",
        "              continue\n",
        "\n",
        "            elif (row, col) == terminals[1]:\n",
        "              v[row, col] = -1\n",
        "              continue\n",
        "\n",
        "            elif (row, col) in walls:\n",
        "              continue\n",
        "\n",
        "            else:\n",
        "              p_next_states = [] # Initialize list for plausible next states\n",
        "              v_next_states = [] # Initialize list for the values of the plausible next states\n",
        "\n",
        "              for i, a in enumerate(actions):\n",
        "                next_row = row + row_change[i]\n",
        "                next_col = col + col_change[i]\n",
        "                if next_row < 0 or next_row >= n_rows or next_col < 0 or next_col >= n_cols or (next_row, next_col) in walls: # Checking to ensure boundaries of the environment is not violated.\n",
        "                  next_state = (row, col)\n",
        "                  p_next_states.append(next_state)\n",
        "                  v_next_states.append(v[next_state])\n",
        "                else:\n",
        "                  next_state = (next_row, next_col)\n",
        "                  p_next_states.append(next_state)\n",
        "                  v_next_states.append(v[next_state])\n",
        "\n",
        "              rand_action = initial_policy[row][col]\n",
        "\n",
        "              act_prob = list(motion_probs[rand_action].values())\n",
        "              temp_res = [act_prob[i] * v_next_states[i] for i in range(len(v_next_states))] # (P(s'|s,a)* V(s'))\n",
        "              sum_temp_res = sum(temp_res) # ∑(P(s'|s,a)* V(s')\n",
        "              v[row, col] = R[row, col] + gamma * sum_temp_res # R(s) + γ(∑(P(s'|s,a)* V(s'))\n",
        "  return v"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "ixs6QdMHjjbo"
      },
      "outputs": [],
      "source": [
        "def P_improv(initial_policy, v):\n",
        "  # Update the policy\n",
        "  for row in range(n_rows):\n",
        "    for col in range(n_cols):\n",
        "        if (row, col) == terminals[0]:\n",
        "          v[row, col] = 1\n",
        "          continue\n",
        "\n",
        "        elif (row, col) == terminals[1]:\n",
        "          v[row, col] = -1\n",
        "          continue\n",
        "\n",
        "        elif (row, col) in walls:\n",
        "          continue\n",
        "\n",
        "        else:\n",
        "          p_next_states = [] # Initialize list for plausible next states\n",
        "          v_next_states = [] # Initialize list for the values of the plausible next states\n",
        "\n",
        "          old_v = v[row, col]\n",
        "          old_policy = initial_policy[row][col]\n",
        "\n",
        "          for i, a in enumerate(actions):\n",
        "              next_row = row + row_change[i]\n",
        "              next_col = col + col_change[i]\n",
        "              if next_row < 0 or next_row >= n_rows or next_col < 0 or next_col >= n_cols or (next_row, next_col) in walls: # Checking to ensure boundaries of the environment is not violated.\n",
        "                  next_state = (row, col)\n",
        "                  p_next_states.append(next_state)\n",
        "                  v_next_states.append(v[next_state])\n",
        "              else:\n",
        "                  next_state = (next_row, next_col)\n",
        "                  p_next_states.append(next_state)\n",
        "                  v_next_states.append(v[next_state])\n",
        "\n",
        "          v_check = []  # Initialize empty list for the value of state after taking an action from up, right, down, left.\n",
        "          for a in actions:\n",
        "            act_prob = list(motion_probs[a].values())\n",
        "            temp_res = [act_prob[i] * v_next_states[i] for i in range(len(v_next_states))] # (P(s'|s,a)* V(s')\n",
        "            sum_temp_res = sum(temp_res) # ∑(P(s'|s,a)* V(s')\n",
        "            v_check.append(sum_temp_res)\n",
        "          v_selected = max(v_check)\n",
        "          v[row, col] = gamma * v_selected # γmax(∑(P(s'|s,a)* V(s'))\n",
        "\n",
        "          # Check previous policy against new values\n",
        "          if v[row, col] > old_v:\n",
        "            # Get the action that produces this.\n",
        "            max_index = np.argmax(v_check) # Extract the index of the highest value in that state: argmax(∑(P(s'|s,a)* V(s'))\n",
        "            initial_policy[row][col] = actions[max_index] # Update the policy\n",
        "          else:\n",
        "            initial_policy[row][col] = old_policy\n",
        "            continue\n",
        "  return initial_policy"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "8pUDSi3QAy_C"
      },
      "outputs": [],
      "source": [
        "import numpy as np\n",
        "\n",
        "# Define the rows and columns\n",
        "n_rows = 3\n",
        "n_cols = 5\n",
        "\n",
        "# Define the terminal states and the walls/impossible states\n",
        "terminals = [(0, 4), (2, 3)]\n",
        "walls = [(0, 3), (1, 1)]\n",
        "\n",
        "# Define the possible actions at each state\n",
        "actions = [\"UP\", \"RIGHT\", \"DOWN\", \"LEFT\"]\n",
        "\n",
        "# Define the row and column changes from each state to new states depending on actions taken\n",
        "row_change = [-1, 0, 1, 0] # up, right, down, left\n",
        "col_change = [0, 1, 0, -1] # up, right, down, left\n",
        "\n",
        "# Define the motion model\n",
        "motion_probs = {\n",
        "    'UP':    {'UP': 0.8, 'RIGHT': 0.1, 'DOWN': 0.0, 'LEFT': 0.1},\n",
        "    'RIGHT': {'UP': 0.1, 'RIGHT': 0.8, 'DOWN': 0.1, 'LEFT': 0.0},\n",
        "    'DOWN':  {'UP': 0.0, 'RIGHT': 0.1, 'DOWN': 0.8, 'LEFT': 0.1},\n",
        "    'LEFT':  {'UP': 0.1, 'RIGHT': 0.0, 'DOWN': 0.1, 'LEFT': 0.8}\n",
        "}\n",
        "\n",
        "# Definition and initialization of the reward function\n",
        "R = np.full((n_rows, n_cols), -0.1)\n",
        "R[0, 4] = 1 # Reward at terminal state I (Goal state)\n",
        "R[2, 3] = -1 # Reward at terminal state II (Bad state)\n",
        "\n",
        "# Define the discount factor\n",
        "gamma = 0.8\n",
        "\n",
        "threshold = 1e-3\n",
        "\n",
        "# Definition and initialization of the value for each state.\n",
        "v_original = np.zeros((n_rows, n_cols))\n",
        "\n",
        "v_real = v_original.copy()\n",
        "\n",
        "initial_policy = np.array([[\"LEFT\", \"UP\", \"DOWN\", \"NONE\", \"TERMINAL\"],\n",
        "                 [\"DOWN\", \"NONE\", \"DOWN\", \"DOWN\", \"RIGHT\"],\n",
        "                 [\"RIGHT\", \"UP\", \"RIGHT\", \"TERMINAL\", \"UP\"]])\n",
        "\n",
        "random_policy = initial_policy.copy()\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "GaA9NX0tEMGC",
        "outputId": "f1cb59fc-00b6-4bc1-b0df-a64ecf1236f2"
      },
      "outputs": [
        {
          "name": "stdout",
          "output_type": "stream",
          "text": [
            "Iteration: 1972\n",
            "Iteration: 1973\n",
            "Iteration: 1974\n",
            "Iteration: 1975\n",
            "Iteration: 1976\n",
            "Iteration: 1977\n",
            "Optimal policy reached\n",
            "\n"
          ]
        }
      ],
      "source": [
        "status = False\n",
        "\n",
        "Iter = 1\n",
        "\n",
        "while status == False:\n",
        "\n",
        "  print(f\"Iteration: {iter}\")\n",
        "\n",
        "  v_new = P_eval(initial_policy, v_original)\n",
        "\n",
        "  v_new_array = sum(v_new)\n",
        "\n",
        "  updated_policy = P_improv(initial_policy, v_new)\n",
        "\n",
        "  new_v = P_eval(updated_policy, v_original)\n",
        "\n",
        "  new_v_array = sum(new_v)\n",
        "\n",
        "  # Check if the absolute difference between each corresponding element is less than the threshold\n",
        "  if np.all(np.abs(v_new_array - new_v_array) < threshold):\n",
        "    print(\"Optimal policy reached\\n\")\n",
        "    updated_policy\n",
        "    status = True\n",
        "  else:\n",
        "    initial_policy = updated_policy\n",
        "    iter += 1\n",
        "    status = False"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "WI3iJ-kpX6_L",
        "outputId": "3523d94e-c087-4ae4-ee52-70aff1f54a88"
      },
      "outputs": [
        {
          "data": {
            "text/plain": [
              "array([['RIGHT', 'RIGHT', 'DOWN', 'NONE', 'TERMINAL'],\n",
              "       ['UP', 'NONE', 'RIGHT', 'RIGHT', 'UP'],\n",
              "       ['UP', 'LEFT', 'UP', 'TERMINAL', 'UP']], dtype='<U8')"
            ]
          },
          "execution_count": 235,
          "metadata": {},
          "output_type": "execute_result"
        }
      ],
      "source": [
        "updated_policy"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "lQGDcAfYZeET",
        "outputId": "aca1e725-adae-4895-f82f-91a0f9ee107f"
      },
      "outputs": [
        {
          "data": {
            "text/plain": [
              "array([['LEFT', 'UP', 'DOWN', 'NONE', 'TERMINAL'],\n",
              "       ['DOWN', 'NONE', 'DOWN', 'DOWN', 'RIGHT'],\n",
              "       ['RIGHT', 'UP', 'RIGHT', 'TERMINAL', 'UP']], dtype='<U8')"
            ]
          },
          "execution_count": 236,
          "metadata": {},
          "output_type": "execute_result"
        }
      ],
      "source": [
        "random_policy"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "wyMplXUfX8UR",
        "outputId": "eb0dfb8f-c479-4f39-bca2-cf843bcf078f"
      },
      "outputs": [
        {
          "data": {
            "text/plain": [
              "array([[-0.28014603, -0.20490052, -0.11268263,  0.        ,  1.        ],\n",
              "       [-0.3324915 ,  0.        ,  0.01988109,  0.22635815,  0.60663984],\n",
              "       [-0.37520335, -0.4049159 , -0.19966937, -1.        ,  0.22635815]])"
            ]
          },
          "execution_count": 237,
          "metadata": {},
          "output_type": "execute_result"
        }
      ],
      "source": [
        "new_v"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "ZTaTpLtwYATA",
        "outputId": "913cb669-4cf5-4fd1-9ee2-a80e63fa9822"
      },
      "outputs": [
        {
          "data": {
            "text/plain": [
              "array([[0., 0., 0., 0., 0.],\n",
              "       [0., 0., 0., 0., 0.],\n",
              "       [0., 0., 0., 0., 0.]])"
            ]
          },
          "execution_count": 238,
          "metadata": {},
          "output_type": "execute_result"
        }
      ],
      "source": [
        "v_real"
      ]
    }
  ],
  "metadata": {
    "colab": {
      "provenance": []
    },
    "kernelspec": {
      "display_name": "Python 3",
      "name": "python3"
    },
    "language_info": {
      "name": "python"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 0
}