cartpole.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "from sklearn.preprocessing import KBinsDiscretizer\n",
    "import numpy as np \n",
    "import time, math, random\n",
    "from typing import Tuple\n",
    "\n",
    "# import gym \n",
    "import gym"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# CartPole-v1"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "env = gym.make('CartPole-v1')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Visualise Enviroment\n",
    "Visualise the eniroment/simulation"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "policy = lambda obs: 1\n",
    "\n",
    "for _ in range(5):\n",
    "    obs = env.reset()\n",
    "    for _ in range(80):\n",
    "        actions = policy(*obs)\n",
    "        obs, reward, done, info = env.step(actions) \n",
    "        env.render()\n",
    "        time.sleep(0.05)\n",
    "\n",
    "env.close()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Look at the docstring."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "?env.env"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Hard Coded Policy "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "#Simple policy function \n",
    "policy = lambda _,__,___, tip_velocity : int( tip_velocity > 0 )"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Q-learning"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Convert Catpoles continues state space into discrete one. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "n_bins = ( 6 , 12 )\n",
    "lower_bounds = [ env.observation_space.low[2], -math.radians(50) ]\n",
    "upper_bounds = [ env.observation_space.high[2], math.radians(50) ]\n",
    "\n",
    "def discretizer( _ , __ , angle, pole_velocity ) -> Tuple[int,...]:\n",
    "    \"\"\"Convert continues state intro a discrete state\"\"\"\n",
    "    est = KBinsDiscretizer(n_bins=n_bins, encode='ordinal', strategy='uniform')\n",
    "    est.fit([lower_bounds, upper_bounds ])\n",
    "    return tuple(map(int,est.transform([[angle, pole_velocity]])[0]))\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Initialise the Q value table with zeros. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "(6, 12, 2)"
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "Q_table = np.zeros(n_bins + (env.action_space.n,))\n",
    "Q_table.shape"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Create a polict function,  uses the Q-table to and greedly selecting the highest Q value "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "def policy( state : tuple ):\n",
    "    \"\"\"Choosing action based on epsilon-greedy policy\"\"\"\n",
    "    return np.argmax(Q_table[state])\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Update function "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "def new_Q_value( reward : float ,  new_state : tuple , discount_factor=1 ) -> float:\n",
    "    \"\"\"Temperal diffrence for updating Q-value of state-action pair\"\"\"\n",
    "    future_optimal_value = np.max(Q_table[new_state])\n",
    "    learned_value = reward + discount_factor * future_optimal_value\n",
    "    return learned_value"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Decaying learning rate"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Adaptive learning of Learning Rate\n",
    "def learning_rate(n : int , min_rate=0.01 ) -> float  :\n",
    "    \"\"\"Decaying learning rate\"\"\"\n",
    "    return max(min_rate, min(1.0, 1.0 - math.log10((n + 1) / 25)))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Decaying exploration rate"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [],
   "source": [
    "def exploration_rate(n : int, min_rate= 0.1 ) -> float :\n",
    "    \"\"\"Decaying exploration rate\"\"\"\n",
    "    return max(min_rate, min(1, 1.0 - math.log10((n  + 1) / 25)))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Training"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "n_episodes = 10000 \n",
    "for e in range(n_episodes):\n",
    "    \n",
    "    # Siscretize state into buckets\n",
    "    current_state, done = discretizer(*env.reset()), False\n",
    "    \n",
    "    while done==False:\n",
    "        \n",
    "        # policy action \n",
    "        action = policy(current_state) # exploit\n",
    "        \n",
    "        # insert random action\n",
    "        if np.random.random() < exploration_rate(e) : \n",
    "            action = env.action_space.sample() # explore \n",
    "         \n",
    "        # increment enviroment\n",
    "        obs, reward, done, _ = env.step(action)\n",
    "        new_state = discretizer(*obs)\n",
    "        \n",
    "        # Update Q-Table\n",
    "        lr = learning_rate(e)\n",
    "        learnt_value = new_Q_value(reward , new_state )\n",
    "        old_value = Q_table[current_state][action]\n",
    "        Q_table[current_state][action] = (1-lr)*old_value + lr*learnt_value\n",
    "        \n",
    "        current_state = new_state\n",
    "        \n",
    "        # Render the cartpole environment\n",
    "        env.render()\n",
    "            \n",
    "\n",
    "            \n",
    "            "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "\n",
    "\n",
    "\n",
    "\n",
    "CREDIT :\n",
    "\n",
    "Sean Saito - vitoshacademy.com\n",
    "\n",
    "https://www.vitoshacademy.com/python-balancing-cartpole-with-machine-learning/ "
   ]
  }
 ],
 "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.7.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}