GridWorld_Q_Learning.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "name": "GridWorld Q Learning",
      "provenance": []
    },
    "kernelspec": {
      "name": "python3",
      "display_name": "Python 3"
    }
  },
  "cells": [
    {
      "cell_type": "code",
      "metadata": {
        "id": "0SnSpSNWS_0f"
      },
      "source": [
        "import numpy as np"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "S9RcLe1pCDa2"
      },
      "source": [
        "class GridWorld:\n",
        "\n",
        "  def __init__(self, rows, cols, start, goal):\n",
        "\n",
        "    self.env = np.zeros((rows, cols))\n",
        "\n",
        "    self.start = start\n",
        "\n",
        "    self.goal = goal\n",
        "\n",
        "    self.env[goal] = 0\n",
        "\n",
        "    self.env[start] = 0\n",
        "\n",
        "    # Rmat[0:n-1,1,:],Rmat[1:,3,:] = -100,-100\n",
        "\n",
        "    # self.\n",
        "\n",
        "\n",
        "    self.rows = rows\n",
        "\n",
        "    self.cols = cols\n",
        "\n",
        "    self.cur_values = np.zeros((rows,cols))\n",
        "\n",
        "    self.qvalues = np.random.uniform(0,1,(4,rows,cols))\n",
        "\n",
        "    \n",
        "\n",
        "\n",
        "  def nextState(self, s, a):\n",
        "\n",
        "    if a == 0:\n",
        "\n",
        "      if s[1] == 0:\n",
        "        return (s[0],s[1])\n",
        "\n",
        "      return (s[0], s[1] - 1)\n",
        "\n",
        "    elif a == 1:\n",
        "\n",
        "      if s[1] == self.cols - 1:\n",
        "        return (s[0], s[1])\n",
        "\n",
        "      return (s[0], s[1] + 1)\n",
        "\n",
        "    elif a == 2:\n",
        "\n",
        "      if s[0] == 0:\n",
        "        return (s[0], s[1])\n",
        "\n",
        "      return (s[0] - 1, s[1])\n",
        "\n",
        "    elif a == 3:\n",
        "\n",
        "      if s[0] == self.rows - 1:\n",
        "        return (s[0], s[1])\n",
        "\n",
        "      return (s[0] + 1, s[1])\n",
        "\n",
        "  def reward(self, s, a):\n",
        "\n",
        "    if a == 0:\n",
        "\n",
        "      if s[1] == 0:\n",
        "        # return self.env[(s[0],s[1])] - 1\n",
        "        return -1\n",
        "\n",
        "\n",
        "      # return self.env[(s[0], s[1] - 1)]\n",
        "      return -1\n",
        "\n",
        "    elif a == 1:\n",
        "\n",
        "      if s[1] == self.cols - 1:\n",
        "        # return self.env[(s[0], s[1])] - 1\n",
        "        return -1\n",
        "\n",
        "\n",
        "      # return self.env[(s[0], s[1] + 1)]\n",
        "      return -1\n",
        "\n",
        "    elif a == 2:\n",
        "\n",
        "      if s[0] == 0:\n",
        "        return -1\n",
        "        # return self.env[(s[0], s[1])] - 1\n",
        "\n",
        "      # return self.env[(s[0] - 1, s[1])]\n",
        "      return -1\n",
        "\n",
        "    elif a == 3:\n",
        "\n",
        "      if s[0] == self.rows - 1:\n",
        "        # return self.env[(s[0], s[1])] - 1\n",
        "        return -1\n",
        "\n",
        "\n",
        "      # return self.env[(s[0] + 1, s[1])]\n",
        "      return -1\n",
        "\n",
        "    \n"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "_mJOoWPPCDa3"
      },
      "source": [
        "class Agent:\n",
        "\n",
        "  def __init__(self, rows, cols, start):\n",
        "\n",
        "    self.actions = [\"left\", \"right\", \"up\", \"down\"]\n",
        "\n",
        "    self.policy_values = np.full((4,rows,cols),0.25)\n",
        "\n",
        "    self.start = start\n",
        "\n",
        "    self.cur_state = start\n",
        "\n",
        "    self.epsilon = np.random.uniform(0,1)\n",
        "\n",
        "\n"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "kXtHQ4WpCDa4"
      },
      "source": [
        "class PolicyIteration:\n",
        "\n",
        "  def __init__(self, agent, env):\n",
        "\n",
        "    self.df = 1\n",
        "\n",
        "    self.step_size = np.random.uniform(0,1)\n",
        "    \n",
        "    self.agent = agent\n",
        "\n",
        "    self.env = env\n",
        "\n",
        "    self.epno = 0\n",
        "\n",
        "  def evaluate_policy(self):\n",
        "\n",
        "    s = self.env.start\n",
        "\n",
        "    count = 0\n",
        "\n",
        "    while s != self.env.goal:\n",
        "\n",
        "      a = np.random.choice([0,1,2,3] ,p=self.agent.policy_values[:,s[0],s[1]])\n",
        "      \n",
        "      s1 = self.env.nextState(s,a)\n",
        "\n",
        "      r =  self.env.reward(s,a)\n",
        "\n",
        "      a1 = np.argmax(self.agent.policy_values[:,s1[0],s1[1]])\n",
        "\n",
        "      count += 1\n",
        "\n",
        "      self.env.qvalues[a,s[0],s[1]] +=  self.step_size * (r + (self.df*self.env.qvalues[a1,s1[0],s1[1]]) - self.env.qvalues[a,s[0],s[1]])\n",
        "\n",
        "      s = s1\n",
        "\n",
        "      a = a1\n",
        "\n",
        "      if(count == 100):\n",
        "        break\n",
        "\n",
        "\n",
        "    \n",
        "  \n",
        "\n",
        "\n",
        "  def findMax(self,s):\n",
        "\n",
        "    return self.env.qvalues[:,s[0],s[1]]\n",
        "\n",
        "    # if s[0] >= 0 and s[1] - 1 >= 0:\n",
        "\n",
        "    #   vals.append(self.env.qvalues[a,s[0],s[1]])\n",
        "\n",
        "    # else:\n",
        "\n",
        "    #   vals.append(self.env.cur_values[s[0],s[1]])\n",
        "    #   # vals.append(0)\n",
        "\n",
        "\n",
        "    # if s[0] >=0 and s[1] + 1 <= self.env.cols - 1:\n",
        "\n",
        "    #   vals.append(self.env.cur_values[s[0],s[1]+1])\n",
        "\n",
        "    # else:\n",
        "\n",
        "    #   vals.append(self.env.cur_values[s[0],s[1]])\n",
        "    #   # vals.append(0)\n",
        "\n",
        "\n",
        "\n",
        "    # if s[0] - 1 >= 0 and s[1] >= 0:\n",
        "\n",
        "    #   vals.append(self.env.cur_values[s[0] - 1,s[1]])\n",
        "\n",
        "    # else:\n",
        "\n",
        "    #   vals.append(self.env.cur_values[s[0],s[1]])\n",
        "    #   # vals.append(0)\n",
        "\n",
        "\n",
        "\n",
        "    # if s[0] + 1 <= self.env.rows - 1 and s[1] >= 0:\n",
        "\n",
        "    #   vals.append(self.env.cur_values[s[0] + 1,s[1]])\n",
        "\n",
        "    # else:\n",
        "\n",
        "    #   vals.append(self.env.cur_values[s[0],s[1]])\n",
        "    #   # vals.append(0)\n",
        "\n",
        "    # return vals\n",
        "\n",
        "\n",
        "  def improve_policy(self):\n",
        "\n",
        "    for i in range(self.env.rows):\n",
        "\n",
        "      for j in range(self.env.cols):\n",
        "\n",
        "        s = (i,j)\n",
        "\n",
        "        values = self.findMax(s)\n",
        "\n",
        "        c = np.argmax(values)\n",
        "\n",
        "        p = [self.agent.epsilon/4,self.agent.epsilon/4,self.agent.epsilon/4,self.agent.epsilon/4]\n",
        "\n",
        "        p[c] = self.agent.epsilon/4 + 1 - self.agent.epsilon\n",
        "\n",
        "        self.agent.policy_values[:,s[0],s[1]] = p\n",
        "\n",
        "\n",
        "  def printPolicy(self):\n",
        "\n",
        "    grid = np.zeros((self.env.rows,self.env.cols))    \n",
        "\n",
        "    for i in range(self.env.rows):\n",
        "\n",
        "      for j in range(self.env.cols):\n",
        "\n",
        "        s = (i,j)\n",
        "\n",
        "        grid[s] = np.argmax(self.agent.policy_values[:,s[0],s[1]])\n",
        "\n",
        "\n",
        "    print(grid)\n",
        "\n",
        "            \n",
        "\n"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "S3LtN_Wpc20E"
      },
      "source": [
        "e = GridWorld(4,4,(0,0),(3,3))"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "VXbZkYwtdAoc"
      },
      "source": [
        "ag = Agent(4,4,(0,0))"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "37o6QFO2116F"
      },
      "source": [
        "p = PolicyIteration(ag, e)"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "4dIc3XuH1-Fd"
      },
      "source": [
        "for i in range(100):\n",
        "\n",
        "  p.evaluate_policy()\n",
        "  p.improve_policy()\n",
        "  # print(p.env.env)\n",
        "  # print(\"end\")\n",
        "  # break"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "dOqI-eOZ4Igq",
        "outputId": "bab749e9-ce34-4446-a404-828bf55ac35e"
      },
      "source": [
        "p.printPolicy()"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "[[1. 1. 1. 3.]\n",
            " [1. 1. 1. 3.]\n",
            " [1. 1. 1. 3.]\n",
            " [1. 1. 1. 2.]]\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "yC_uDoTnVXXR"
      },
      "source": [
        ""
      ],
      "execution_count": null,
      "outputs": []
    }
  ]
}