gym-minigrid/gym_minigrid/backup_envs/bobo.py

import numpy as np

from gym_minigrid.minigrid import *
from gym_minigrid.register import register
import time
from collections import deque


class Peer(NPC):
    """
    A dancing NPC that the agent has to copy
    """

    def __init__(self, color, name, env, knowledgeable=False):
        super().__init__(color)
        self.name = name
        self.npc_dir = 1  # NPC initially looks downward
        self.npc_type = 0
        self.env = env
        self.knowledgeable = knowledgeable 
        self.npc_actions = []
        self.dancing_step_idx = 0
        self.actions = MiniGridEnv.Actions
        self.add_npc_direction = True
        self.available_moves = [self.rotate_left, self.rotate_right, self.go_forward, self.toggle_action]
        self.exited = False

    def step(self):
        if self.exited:
            return

        if all(np.array(self.cur_pos) == np.array(self.env.door_pos)):
            # todo: disappear
            # todo: close door
            self.env.grid.set(*self.cur_pos, self.env.object)
            self.cur_pos = np.array([np.nan, np.nan])

            self.env.object.toggle(self.env, self.cur_pos)

            self.exited = True

        elif self.knowledgeable:

            if all(self.front_pos == self.env.door_pos):
                # in front of door
                if self.env.object.is_open:
                    self.go_forward()
                else:
                    self.toggle_action()

            else:
                if (self.cur_pos[0] == self.env.door_pos[0]) or (self.cur_pos[1] == self.env.door_pos[1]):
                    # is either in the correct row on in the correct column
                    next_wanted_position = self.env.door_pos
                else:
                    # choose the midpoint
                    for cand_x, cand_y in [
                        (self.cur_pos[0], self.env.door_pos[1]),
                        (self.env.door_pos[0], self.cur_pos[1])
                    ]:
                        if (
                                cand_x > 0 and cand_x < self.env.wall_x
                        ) and (
                                cand_y > 0 and cand_y < self.env.wall_y
                        ):
                            next_wanted_position = (cand_x, cand_y)

                if self.cur_pos[1] == next_wanted_position[1]:
                    # same y
                    if self.cur_pos[0] < next_wanted_position[0]:
                        wanted_dir = 0
                    else:
                        wanted_dir = 2
                    if self.npc_dir == wanted_dir:
                        self.go_forward()

                    else:
                        self.rotate_left()

                elif self.cur_pos[0] == next_wanted_position[0]:
                    # same x
                    if self.cur_pos[1] < next_wanted_position[1]:
                        wanted_dir = 1
                    else:
                        wanted_dir = 3


                    if self.npc_dir == wanted_dir:
                        self.go_forward()

                    else:
                        self.rotate_left()
                else:
                    raise ValueError("Something is wrong.")

        else:
            self.env._rand_elem(self.available_moves)()


class BoboGrammar(object):

    templates = ["Move your", "Shake your"]
    things = ["body", "head"]

    grammar_action_space = spaces.MultiDiscrete([len(templates), len(things)])

    @classmethod
    def construct_utterance(cls, action):
        return cls.templates[int(action[0])] + " " + cls.things[int(action[1])] + " "


class BoboEnv(MultiModalMiniGridEnv):
    """
    Environment in which the agent is instructed to go to a given object
    named using an English text string
    """

    def __init__(
        self,
        size=5,
        diminished_reward=True,
        step_penalty=False,
        knowledgeable=False,
    ):
        assert size >= 5
        self.empty_symbol = "NA \n"
        self.diminished_reward = diminished_reward
        self.step_penalty = step_penalty
        self.knowledgeable = knowledgeable

        super().__init__(
            grid_size=size,
            max_steps=5*size**2,
            # Set this to True for maximum speed
            see_through_walls=True,
            actions=MiniGridEnv.Actions,
            action_space=spaces.MultiDiscrete([
                len(MiniGridEnv.Actions),
                *BoboGrammar.grammar_action_space.nvec
            ]),
            add_npc_direction=True
        )

        print({
            "size": size,
            "diminished_reward": diminished_reward,
            "step_penalty": step_penalty,
        })

    def _gen_grid(self, width, height):
        # Create the grid
        self.grid = Grid(width, height, nb_obj_dims=4)

        # Randomly vary the room width and height
        width = self._rand_int(5, width+1)
        height = self._rand_int(5, height+1)

        self.wall_x = width - 1
        self.wall_y = height - 1

        # Generate the surrounding walls
        self.grid.wall_rect(0, 0, width, height)

        door_color = self._rand_elem(COLOR_NAMES)

        wall_for_door = self._rand_int(0, 4)

        if wall_for_door < 2:
            w = self._rand_int(1, width-1)
            h = height-1 if wall_for_door == 0 else 0
        else:
            w = width-1 if wall_for_door == 3 else 0
            h = self._rand_int(1, height-1)

        self.door_pos = (w, h)
        self.door = Door(door_color)
        self.grid.set(*self.door_pos, self.door)

        # Set a randomly coloured Dancer NPC
        color = self._rand_elem(COLOR_NAMES)
        self.peer = Peer(color, "Jim", self, knowledgeable=self.knowledgeable)

        # Place it on the middle left side of the room
        peer_pos = np.array((self._rand_int(1, width - 1), self._rand_int(1, height - 1)))

        self.grid.set(*peer_pos, self.peer)
        self.peer.init_pos = peer_pos
        self.peer.cur_pos = peer_pos

        # Randomize the agent's start position and orientation
        self.place_agent(size=(width, height))

        # Generate the mission string
        self.mission = 'watch dancer and repeat his moves afterwards'

        # Dummy beginning string
        self.beginning_string = "This is what you hear. \n"
        self.utterance = self.beginning_string

        # utterance appended at the end of each step
        self.utterance_history = ""

        # used for rendering
        self.conversation = self.utterance

    def step(self, action):
        p_action = action[0]
        utterance_action = action[1:]

        obs, reward, done, info = super().step(p_action)

        if np.isnan(p_action):
            pass

        if p_action == self.actions.done:
            done = True

        self.peer.step()

        if all(self.agent_pos == self.door_pos):
            reward = self._reward()
            done = True

        # discount
        if self.step_penalty:
            reward = reward - 0.01

        # fill observation with text
        self.append_existing_utterance_to_history()
        obs = self.add_utterance_to_observation(obs)
        self.reset_utterance()
        return obs, reward, done, info

    def _reward(self):
        if self.diminished_reward:
            return super()._reward()
        else:
            return 1.0

    def render(self, *args, **kwargs):
        obs = super().render(*args, **kwargs)
        print("conversation:\n", self.conversation)
        print("utterance_history:\n", self.utterance_history)
        self.window.set_caption(self.conversation, [self.peer.name])
        return obs


class Bobo8x8Env(BoboEnv):
    def __init__(self):
        super().__init__(size=8)


class Bobo6x6Env(BoboEnv):
    def __init__(self):
        super().__init__(size=6)


# knowledgeable
class BoboKnowledgeableEnv(BoboEnv):
    def __init__(self):
        super().__init__(size=5, knowledgeable=True)

class BoboKnowledgeable6x6Env(BoboEnv):
    def __init__(self):
        super().__init__(size=6, knowledgeable=True)

class BoboKnowledgeable8x8Env(BoboEnv):
    def __init__(self):
        super().__init__(size=8, knowledgeable=True)



register(
    id='MiniGrid-Bobo-5x5-v0',
    entry_point='gym_minigrid.envs:BoboEnv'
)

register(
    id='MiniGrid-Bobo-6x6-v0',
    entry_point='gym_minigrid.envs:Bobo6x6Env'
)

register(
    id='MiniGrid-Bobo-8x8-v0',
    entry_point='gym_minigrid.envs:Bobo8x8Env'
)

register(
    id='MiniGrid-BoboKnowledgeable-5x5-v0',
    entry_point='gym_minigrid.envs:BoboKnowledgeableEnv'
)

register(
    id='MiniGrid-BoboKnowledgeable-6x6-v0',
    entry_point='gym_minigrid.envs:BoboKnowledgeable6x6Env'
)

register(
    id='MiniGrid-BoboKnowledgeable-8x8-v0',
    entry_point='gym_minigrid.envs:BoboKnowledgeable8x8Env'
)