test_gym.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import gym\n",
    "import numpy as np\n",
    "from roundabout.env import Env\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "RLParams = {'desired_steps': 2400,\n",
    "            'warmup_steps': 800,\n",
    "            'Q_upperbound': 35,\n",
    "            'Q_lowerbound': 20,\n",
    "            'Q_randomness': 3}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "class RLRoundabout(gym.Env):\n",
    "    def __init__(self, envParams, initParams, RLParams):\n",
    "        self.env = Env(envParams)\n",
    "        self.initParams = initParams\n",
    "        self.desired_steps = RLParams['desired_steps']\n",
    "        self.warmup_steps = RLParams['warmup_steps']\n",
    "        self.Q_upperbound = RLParams['Q_upperbound']\n",
    "        self.Q_lowerbound = RLParams['Q_lowerbound']\n",
    "        self.Q_randomness = RLParams['Q_randomness']\n",
    "        self.action_space = gym.spaces.Box(low = 0.0, high = 1.0, shape = (4,), dtype=np.float32)\n",
    "        self.observation_space = gym.spaces.Dict({\"new_vehicles\": gym.spaces.MultiBinary([4]),\n",
    "                                                  \"approaching_vehicles\": gym.spaces.Box(low=0, high=100, shape = (4,), dtype=np.int64),\n",
    "                                                  \"slot_occupied_rate\": gym.spaces.Box(low=0, high=1, shape = (1,), dtype=np.float32),\n",
    "                                                  \"current_queue\": gym.spaces.Box(low = 0, high = 30, shape = (4,), dtype=np.int64)})\n",
    "        \n",
    "        self.env.initialize(self.initParams)\n",
    "    \n",
    "    def step(self, action):\n",
    "        \n",
    "        self.env.P = action\n",
    "        self.env.Q = np.random.uniform(self.Q_lowerbound, self.Q_upperbound, 4) + np.random.normal(0, self.Q_randomness, 4)\n",
    "        self.env.step()\n",
    "\n",
    "        state = dict()\n",
    "        state[\"approaching_vehicles\"] = np.array(list(map(len, self.env.approaching_vehicles)))\n",
    "        state[\"current_queue\"] = np.array(self.env.queue_length)\n",
    "        state[\"new_vehicles\"] = np.array(self.env.new_vehicles)\n",
    "        state[\"slot_occupied_rate\"] =[1 - sum(list(map(lambda slot: slot.virtual_vehicle is None, self.env.slots)))/self.env.numSlots]\n",
    "        \n",
    "        reward = - sum(np.square(self.env.queue_length))\n",
    "\n",
    "        done = (self.env.num_step >= self.desired_steps) \n",
    "        info = {}        \n",
    "        return state, reward, done, info\n",
    "    \n",
    "    def reset(self): \n",
    "        self.env.initialize(self.initParams)\n",
    "\n",
    "        for _ in range(self.warmup_steps):\n",
    "            state, _, _, _ = self.step(action = np.ones(4))\n",
    "        \n",
    "        # state = dict()\n",
    "        # state[\"new_vehicle\"] = self.env.new_vehicles\n",
    "        # state[\"approaching_vehicles\"] = np.array(list(map(len, self.env.approaching_vehicles)))\n",
    "        # state[\"current_queue\"] = np.array(self.env.queue_length)\n",
    "        \n",
    "        return state"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "envParams = {\n",
    "    \"a\": 3, \n",
    "    \"v\": 8,\n",
    "    \"C\": 3,\n",
    "    \"veh_length\": 5,\n",
    "    \"headway\": 3,\n",
    "    \"Q\": np.array([35, 35, 35, 35])*6/8,\n",
    "    \"step_size\": 0.05,\n",
    "    \"boundary\": 250,\n",
    "    \"eta\": np.array([[0.0, 0.0, 5/7, 2/7],\n",
    "                     [2/7, 0.0, 0.0, 5/7],\n",
    "                     [5/7, 2/7, 0.0, 0.0],\n",
    "                     [0.0, 5/7, 2/7, 0.0]]),\n",
    "    \"fifo\": True\n",
    "}\n",
    "\n",
    "initParams = {\n",
    "    \"l_0\": np.array([0, 0, 0, 0]),  \n",
    "    #\"l_0\": np.array([14, 15, 13, 14]),  \n",
    "}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{'approaching_vehicles': array([ 6,  6, 10,  7]),\n",
       " 'current_queue': array([0, 0, 0, 0]),\n",
       " 'new_vehicles': array([0, 1, 0, 1]),\n",
       " 'slot_occupied_rate': [0.8333333333333334]}"
      ]
     },
     "execution_count": 14,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "test_env = RLRoundabout(envParams, initParams, RLParams)\n",
    "test_env.reset()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "({'approaching_vehicles': array([ 8, 10,  9,  3]),\n",
       "  'current_queue': array([0, 1, 0, 0]),\n",
       "  'new_vehicles': [0, 0, 0, 0],\n",
       "  'slot_occupied_rate': [0.8333333333333334]},\n",
       " -1,\n",
       " False,\n",
       " {})"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "test_env.step(test_env.action_space.sample())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{'approaching_vehicles': array([5, 5, 4, 3]),\n",
       " 'current_queue': array([0, 0, 0, 0]),\n",
       " 'new_vehicles': [0, 0, 0, 0],\n",
       " 'slot_occupied_rate': [0.9166666666666666]}"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "test_env.reset()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "OrderedDict([('approaching_vehicles', array([ 64, 100,  41,  13])),\n",
       "             ('current_queue', array([20,  3, 30, 26])),\n",
       "             ('new_vehicles', array([[1, 0, 0, 1]], dtype=int8)),\n",
       "             ('slot_occupied_rate', array([0.5345858], dtype=float32))])"
      ]
     },
     "execution_count": 17,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "test_env.observation_space.sample()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "False"
      ]
     },
     "execution_count": 15,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "isinstance(OrderedDict(test_env.reset()), gym.spaces.Dict)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "OrderedDict([('approaching_vehicles', array([7, 8, 3, 4])),\n",
       "             ('current_queue', array([0, 0, 0, 0])),\n",
       "             ('new_vehicles', array([0, 0, 1, 0])),\n",
       "             ('slot_occupied_rate', [0.6666666666666667])])"
      ]
     },
     "execution_count": 16,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from collections import OrderedDict\n",
    "OrderedDict(test_env.reset())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "interpreter": {
   "hash": "7b00d0aff25ee417f6f37cd956ee31999fd0cef4334c125bb076ee6222a3f8af"
  },
  "kernelspec": {
   "display_name": "Python 3.9.5 64-bit ('chaojie': conda)",
   "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.5"
  },
  "orig_nbformat": 4
 },
 "nbformat": 4,
 "nbformat_minor": 2
}