Boolean Multiplexer environment

Author: Norbert Kozlowski

README.md

@@ -1,9 +1,14 @@
-# openai-maze-envs
+# Parrot Prediction OpenAI environments
+
+## Maze
 
 Initializing
 
     maze = gym.make('MazeF1-v0')
 
 Getting all possible transitions
 
-    transitions = maze.env.get_all_possible_transitions()
+    transitions = maze.env.get_all_possible_transitions()
+
+## Boolean Multiplexer
+Read blog [post](https://medium.com/parrot-prediction/boolean-multiplexer-in-practice-94e3236821b5) describing the usage.

ex.py

@@ -0,0 +1,33 @@
+import logging
+from random import choice
+
+import gym
+
+# noinspection PyUnresolvedReferences
+import gym_maze
+
+logging.basicConfig(level=logging.DEBUG)
+
+if __name__ == '__main__':
+    maze = gym.make('MazeF1-v0')
+
+    possible_actions = list(range(8))
+    transitions = maze.env.get_all_possible_transitions()
+
+    for i_episode in range(1):
+        observation = maze.reset()
+
+        for t in range(100):
+            logging.info("Time: [{}], observation: [{}]".format(t, observation))
+
+            action = choice(possible_actions)
+
+            logging.info("\t\tExecuted action: [{}]".format(action))
+            observation, reward, done, info = maze.step(action)
+
+            if done:
+                logging.info("Episode finished after {} timesteps.".format(t + 1))
+                logging.info("Last reward: {}".format(reward))
+                break
+
+    logging.info("Finished")

examples/maze.py

@@ -1,7 +1,3 @@
-import logging
-
-logger = logging.getLogger(__name__)
-
 PATH_MAPPING = 0
 WALL_MAPPING = 1
 REWARD_MAPPING = 9

gym_maze/Maze.py

@@ -1,11 +1,7 @@
-import logging
-
 from gym.envs.registration import register
 
-from gym_maze.Maze import Maze
-from gym_maze.Maze import PATH_MAPPING, WALL_MAPPING, REWARD_MAPPING
-
-logger = logging.getLogger(__name__)
+# noinspection PyUnresolvedReferences
+from gym_maze.Maze import Maze, PATH_MAPPING, WALL_MAPPING, REWARD_MAPPING
 
 ACTION_LOOKUP = {
     0: 'N',

gym_maze/__init__.py

@@ -1,17 +1,13 @@
-import gym
-from gym import error, spaces, utils
-from gym.utils import seeding
-
-from gym_maze import Maze, WALL_MAPPING, ACTION_LOOKUP
-from gym_maze.utils import get_all_possible_transitions
-
-import numpy as np
 import logging
 import random
 import sys
 
+import gym
+import numpy as np
+from gym import spaces, utils
 
-logger = logging.getLogger(__name__)
+from gym_maze import Maze, WALL_MAPPING, ACTION_LOOKUP
+from gym_maze.utils import get_all_possible_transitions
 
 ANIMAT_MARKER = 5
 
@@ -38,7 +34,7 @@ def _step(self, action):
         return observation, reward, episode_over, {}
 
     def _reset(self):
-        logger.debug("Resetting the environment")
+        logging.debug("Resetting the environment")
         self._insert_animat()
         return self._observe()
 

gym_maze/envs/AbstractMaze.py

@@ -1,4 +1,5 @@
 import gym
+# noinspection PyUnresolvedReferences
 import gym_maze
 
 

gym_maze/tests/test_utils.py

@@ -0,0 +1,27 @@
+from .boolean_multiplexer import BooleanMultiplexer
+
+from gym.envs.registration import register
+
+name = "boolean-multiplexer"
+max_episode_steps = 1
+
+register(
+    id='{}-3bit-v0'.format(name),
+    entry_point='gym_multiplexer:BooleanMultiplexer',
+    max_episode_steps=max_episode_steps,
+    kwargs={'control_bits': 1}
+)
+
+register(
+    id='{}-6bit-v0'.format(name),
+    entry_point='gym_multiplexer:BooleanMultiplexer',
+    max_episode_steps=max_episode_steps,
+    kwargs={'control_bits': 2}
+)
+
+register(
+    id='{}-11bit-v0'.format(name),
+    entry_point='gym_multiplexer:BooleanMultiplexer',
+    max_episode_steps=max_episode_steps,
+    kwargs={'control_bits': 3}
+)

gym_multiplexer/__init__.py

@@ -0,0 +1,52 @@
+import logging
+import random
+
+import gym
+from bitstring import BitArray
+from gym.spaces import Discrete
+
+
+class BooleanMultiplexer(gym.Env):
+
+    def __init__(self, control_bits=3) -> None:
+        self.metadata = {'render.modes': ['human']}
+        self.control_bits = control_bits
+        self.observation_space = Discrete(len(self._observation_string_length))
+        self.action_space = Discrete(2)
+
+    def _reset(self):
+        logging.debug("Resetting the environment")
+        bits = BitArray([random.randint(0, 1) for _ in
+                         self._observation_string_length])
+
+        self._ctrl_bits = bits[:self.control_bits]
+        self._data_bits = bits[self.control_bits:]
+
+    def _step(self, action):
+        state = self._observation()
+        reward = 0
+
+        if action == self._answer:
+            reward = 1
+
+        return state, reward, None, None
+
+    def _render(self, mode='human', close=False):
+        if close:
+            return
+
+        if mode == 'human':
+            return self.control_bits + self._data_bits
+        else:
+            super(BooleanMultiplexer, self).render(mode=mode)
+
+    def _observation(self):
+        return self.control_bits + self._data_bits
+
+    @property
+    def _observation_string_length(self):
+        return range(0, self.control_bits + pow(2, self.control_bits))
+
+    @property
+    def _answer(self):
+        return int(self._data_bits[self._ctrl_bits.uint])