Corridor and Grid enviornments (#15)

Author: khozzy

README.md

@@ -12,8 +12,10 @@ The repository contains environments used in LCS literature that are compliant w
 - Hand Eye
 - Checkerboard
 - Real-valued toy problems
+- 1D Corridor,
+- 2D Grid
 
-For usage examples look at [examples/](examples) directory.
+For some usage examples look at [examples/](examples) directory.
 
 ## Development
 

gym_corridor/__init__.py

@@ -0,0 +1,26 @@
+from gym.envs.registration import register
+
+from .corridor import Corridor
+
+max_episode_steps = 200
+
+register(
+    id='corridor-20-v0',
+    entry_point='gym_corridor:Corridor',
+    max_episode_steps=max_episode_steps,
+    kwargs={'size': 20}
+)
+
+register(
+    id='corridor-40-v0',
+    entry_point='gym_corridor:Corridor',
+    max_episode_steps=max_episode_steps,
+    kwargs={'size': 40}
+)
+
+register(
+    id='corridor-100-v0',
+    entry_point='gym_corridor:Corridor',
+    max_episode_steps=max_episode_steps,
+    kwargs={'size': 100}
+)

gym_corridor/corridor.py

@@ -0,0 +1,54 @@
+from random import randint
+
+import gym
+from gym.spaces import Discrete
+
+MOVE_LEFT = 0
+MOVE_RIGHT = 1
+
+
+class Corridor(gym.Env):
+    metadata = {'render.modes': ['human', 'ansi']}
+
+    REWARD = 1000
+
+    def __init__(self, size=20):
+        self._size = size
+        self._position = None
+
+        self.observation_space = Discrete(1)
+        self.action_space = Discrete(2)
+
+    def reset(self):
+        self._position = randint(1, self._size - 1)
+        return str(self._position)
+
+    def step(self, action):
+        if action == MOVE_LEFT:
+            self._position -= 1
+        elif action == MOVE_RIGHT:
+            self._position += 1
+        else:
+            raise ValueError("Illegal action passed")
+
+        if self._position == self._size:
+            return str(self._position), self.REWARD, True, None
+
+        if self._position == 0:
+            self._position = 1
+
+        return str(self._position), 0, False, None
+
+    def render(self, mode='human'):
+        if mode == 'human':
+            print(self._visualize())
+        elif mode == 'ansi':
+            return self._visualize()
+        else:
+            raise ValueError('Unknown visualisation mode')
+
+    def _visualize(self):
+        corridor = ["" for _ in range(0, self._size - 1)]
+        corridor[self._position - 1] = "X"
+        corridor[self._size - 2] = "$"
+        return "[" + ".".join(corridor) + "]"

gym_corridor/tests/__init__.py

@@ -0,0 +1,86 @@
+import logging
+import sys
+
+import gym
+
+# noinspection PyUnresolvedReferences
+import gym_corridor
+from gym_corridor.corridor import MOVE_LEFT, MOVE_RIGHT
+
+logging.basicConfig(level=logging.DEBUG, stream=sys.stdout)
+
+
+class TestCorridor:
+
+    def test_should_initialize(self):
+        # when
+        corr = gym.make('corridor-20-v0')
+
+        # then
+        assert corr is not None
+        assert 1 == corr.observation_space.n
+        assert 2 == corr.action_space.n
+
+    def test_should_visualize(self):
+        # given
+        corr = gym.make('corridor-20-v0')
+
+        # when
+        obs = corr.reset()
+        vis = corr.render(mode='ansi')
+
+        # then
+        assert 1 <= int(obs) < 20
+        assert len(vis) == 22
+        assert 1 == vis.count('X')
+        assert 1 == vis.count('$')
+        assert 18 == vis.count('.')
+
+    def test_should_hit_left_wall(self):
+        # given
+        corr = gym.make('corridor-20-v0')
+        reward = 0
+        done = False
+
+        # when
+        obs = corr.reset()
+
+        while not done:
+            obs, reward, done, _ = corr.step(MOVE_LEFT)
+
+        # then
+        assert obs == '1'
+        assert reward == 0
+        assert done is True
+
+    def test_should_get_reward(self):
+        # given
+        corr = gym.make('corridor-20-v0')
+        reward = 0
+        done = False
+
+        # when
+        obs = corr.reset()
+
+        while not done:
+            obs, reward, done, _ = corr.step(MOVE_RIGHT)
+
+        # then
+        assert obs == '20'
+        assert reward == 1000
+        assert done is True
+
+    def test_should_move_in_both_directions(self):
+        # given
+        corr = gym.make('corridor-20-v0')
+        p0 = corr.reset()
+
+        while p0 in ["1", "19"]:
+            p0 = corr.reset()
+
+        # when & then
+        p1, _, _, _ = corr.step(MOVE_LEFT)
+        assert int(p1) == int(p0) - 1
+
+        p2, _, _, _ = corr.step(MOVE_RIGHT)
+        assert int(p2) == int(p0)

gym_corridor/tests/test_corridor.py

@@ -0,0 +1,26 @@
+from gym.envs.registration import register
+
+from .grid import Grid
+
+max_episode_steps = 200
+
+register(
+    id='grid-20-v0',
+    entry_point='gym_grid:Grid',
+    max_episode_steps=max_episode_steps,
+    kwargs={'size': 20}
+)
+
+register(
+    id='grid-40-v0',
+    entry_point='gym_grid:Grid',
+    max_episode_steps=max_episode_steps,
+    kwargs={'size': 40}
+)
+
+register(
+    id='grid-100-v0',
+    entry_point='gym_grid:Grid',
+    max_episode_steps=max_episode_steps,
+    kwargs={'size': 100}
+)

gym_grid/__init__.py

@@ -0,0 +1,95 @@
+import gym
+import numpy as np
+from gym.spaces import Discrete
+
+MOVE_LEFT = 0
+MOVE_RIGHT = 1
+MOVE_UP = 3
+MOVE_DOWN = 4
+
+# Food located in [n, n]
+# Observation x,y in [1, n]
+
+
+class Grid(gym.Env):
+    metadata = {'render.modes': ['human', 'ansi']}
+
+    REWARD = 1000
+
+    def __init__(self, size=20):
+        self._size = size
+        self._pos_x = None
+        self._pos_y = None
+
+        self.observation_space = Discrete(2)
+        self.action_space = Discrete(4)
+
+    @property
+    def _state(self):
+        return str(self._pos_x), str(self._pos_y)
+
+    def reset(self):
+        (self._pos_x, self._pos_y) = np.random.randint(
+            1, self._size + 1, size=2)
+
+        if self._pos_x == self._size and self._pos_y == self._size:
+            self.reset()
+
+        return self._state
+
+    def step(self, action):
+        if action == MOVE_LEFT:
+            self._pos_x -= 1
+        elif action == MOVE_RIGHT:
+            self._pos_x += 1
+        elif action == MOVE_UP:
+            self._pos_y += 1
+        elif action == MOVE_DOWN:
+            self._pos_y -= 1
+        else:
+            raise ValueError("Illegal action passed")
+
+        # Handle reaching final state
+        if self._pos_x == self._size and self._pos_y == self._size:
+            return self._state, self.REWARD, True, None
+
+        # Handle leaving grid
+        if self._pos_x == 0:
+            self._pos_x = 1
+        elif self._pos_x == 21:
+            self._pos_x = 20
+
+        if self._pos_y == 0:
+            self._pos_y = 1
+        elif self._pos_y == 21:
+            self._pos_y = 20
+
+        # Return default observation
+        return self._state, 0, False, None
+
+    def render(self, mode='human'):
+        if mode == 'human':
+            print(self._visualize())
+        elif mode == 'ansi':
+            return self._visualize()
+        else:
+            raise ValueError('Unknown visualisation mode')
+
+    def _visualize(self):
+        print("")
+        print(self._state)
+        for y in reversed(range(0, self._size + 1)):
+            for x in range(0, self._size + 1):
+                if x == 0 and y == 0:
+                    print(f"{'':^3}", end='')
+                elif x == 0:
+                    print(f"{y:>3}", end='')
+                elif y == 0:
+                    print(f"{x:^3}", end='')
+                elif x == self._pos_x and y == self._pos_y:
+                    print(f"{'X':^3}", end='')
+                elif x == self._size and y == self._size:
+                    print(f"{'$':^3}", end='')
+                else:
+                    print(f"{'_':^3}", end='')
+            print("")

gym_grid/grid.py

@@ -0,0 +1,69 @@
+import logging
+import sys
+
+import gym
+import numpy as np
+
+# noinspection PyUnresolvedReferences
+import gym_grid
+from gym_grid.grid import MOVE_LEFT, MOVE_RIGHT, MOVE_UP, MOVE_DOWN
+
+logging.basicConfig(level=logging.DEBUG, stream=sys.stdout)
+
+
+class TestGrid:
+
+    def test_should_initialize(self):
+        # when
+        grid = gym.make('grid-20-v0')
+
+        # then
+        assert grid is not None
+        assert 2 == grid.observation_space.n
+        assert 4 == grid.action_space.n
+
+    def test_should_handle_hitting_boundaries(self):
+        # given
+        grid = gym.make('grid-20-v0')
+
+        # handle hitting upper bound
+        np.random.seed(42)
+        grid.reset()  # (x=7, y=20)
+        state, _, _, _ = grid.step(MOVE_UP)
+        assert state == ("7", "20")
+
+        # handle hitting right bound
+        np.random.seed(27)
+        grid.reset()  # (x=20, y=9)
+        state, _, _, _ = grid.step(MOVE_RIGHT)
+        assert state == ("20", "9")
+
+        # handle hitting lower bound
+        np.random.seed(50)
+        grid.reset()  # (x=17, y=1)
+        state, _, _, _ = grid.step(MOVE_DOWN)
+        assert state == ("17", "1")
+
+        # handle hitting left bound
+        np.random.seed(48)
+        grid.reset()  # (x=1, y=20)
+        state, _, _, _ = grid.step(MOVE_LEFT)
+        assert state == ("1", "20")
+
+    def test_should_get_reward(self):
+        # given
+        grid = gym.make('grid-20-v0')
+        reward = 0
+        done = False
+
+        # when
+        grid.reset()
+        for _ in range(0, 20):
+            grid.step(MOVE_RIGHT)
+        while not done:
+            obs, reward, done, _ = grid.step(MOVE_UP)
+
+        # then
+        assert obs == ('20', '20')
+        assert reward == 1000
+        assert done is True

gym_grid/tests/__init__.py

@@ -1,17 +1,17 @@
 from setuptools import setup, find_packages
 
 setup(name='parrotprediction-openai-envs',
-      version='2.0.4',
+      version='2.1.0',
       description='Custom environments for OpenAI Gym',
       keywords='acs lcs machine-learning reinforcement-learning openai',
       url='https://github.com/ParrotPrediction/openai-envs',
       author='Parrot Prediction Ltd.',
-      author_email='contact@parrotprediction.com',
+      author_email='nkozlowski@protonmail.com',
       license='MIT',
       packages=find_packages(),
       install_requires=[
           'numpy',
-          'gym>=0.10',
+          'gym==0.11',
           'networkx==2.0',
           'bitstring==3.1.5'
       ],