from gym_minigrid.minigrid import * from gym_minigrid.register import register class Wizard(NPC): """ A simple NPC that knows who is telling the truth """ def __init__(self, color, name, env): super().__init__(color) self.name = name self.env = env self.npc_dir = 1 # NPC initially looks downward # todo: this should be id == name self.npc_type = 0 # this will be put into the encoding self.was_introduced_to = False def can_overlap(self): # If the NPC is hidden, agent can overlap on it return self.env.hidden_npc def encode(self, nb_dims=3): if self.env.hidden_npc: if nb_dims == 3: return (1, 0, 0) elif nb_dims == 4: return (1, 0, 0, 0) else: return super().encode(nb_dims=nb_dims) def listen(self, utterance): if self.env.hidden_npc: return None if self.was_introduced_to: if utterance == TalkItOutPoliteGrammar.construct_utterance([0, 1]): if self.env.nameless: return "Ask the {} guide.".format(self.env.true_guide.color) else: return "Ask {}.".format(self.env.true_guide.name) else: if utterance == TalkItOutPoliteGrammar.construct_utterance([3, 3]): self.was_introduced_to = True return "I am well." return None class Guide(NPC): """ A simple NPC that knows the correct door. """ def __init__(self, color, name, env, liar=False): super().__init__(color) self.name = name self.env = env self.liar = liar self.npc_dir = 1 # NPC initially looks downward # todo: this should be id == name self.npc_type = 1 # this will be put into the encoding self.was_introduced_to = False # Select a random target object as mission obj_idx = self.env._rand_int(0, len(self.env.door_pos)) self.target_pos = self.env.door_pos[obj_idx] self.target_color = self.env.door_colors[obj_idx] def can_overlap(self): # If the NPC is hidden, agent can overlap on it return self.env.hidden_npc def encode(self, nb_dims=3): if self.env.hidden_npc: if nb_dims == 3: return (1, 0, 0) elif nb_dims == 4: return (1, 0, 0, 0) else: return super().encode(nb_dims=nb_dims) def listen(self, utterance): if self.env.hidden_npc: return None if self.was_introduced_to: if utterance == TalkItOutPoliteGrammar.construct_utterance([0, 1]): if self.liar: fake_colors = [c for c in self.env.door_colors if c != self.env.target_color] fake_color = self.env._rand_elem(fake_colors) # Generate the mission string assert fake_color != self.env.target_color return 'go to the %s door' % fake_color else: return self.env.mission else: if utterance == TalkItOutPoliteGrammar.construct_utterance([3, 3]): self.was_introduced_to = True return "I am well." def render(self, img): c = COLORS[self.color] npc_shapes = [] # Draw eyes npc_shapes.append(point_in_circle(cx=0.70, cy=0.50, r=0.10)) npc_shapes.append(point_in_circle(cx=0.30, cy=0.50, r=0.10)) # Draw mouth npc_shapes.append(point_in_rect(0.20, 0.80, 0.72, 0.81)) # todo: move this to super function # todo: super.render should be able to take the npc_shapes and then rotate them if hasattr(self, "npc_dir"): # Pre-rotation to ensure npc_dir = 1 means NPC looks downwards npc_shapes = [rotate_fn(v, cx=0.5, cy=0.5, theta=-1*(math.pi / 2)) for v in npc_shapes] # Rotate npc based on its direction npc_shapes = [rotate_fn(v, cx=0.5, cy=0.5, theta=(math.pi/2) * self.npc_dir) for v in npc_shapes] # Draw shapes for v in npc_shapes: fill_coords(img, v, c) class TalkItOutPoliteGrammar(object): templates = ["Where is", "Open", "Close", "How are"] things = [ "sesame", "the exit", "the wall", "you", "the ceiling", "the window", "the entrance", "the closet", "the drawer", "the fridge", "the floor", "the lamp", "the trash can", "the chair", "the bed", "the sofa" ] assert len(templates)*len(things) == 64 print("language complexity {}:".format(len(templates)*len(things))) 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 TalkItOutPoliteEnv(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, hear_yourself=False, diminished_reward=True, step_penalty=False, nameless=False, max_steps=100, hidden_npc=False, ): assert size >= 5 self.empty_symbol = "NA \n" self.hear_yourself = hear_yourself self.diminished_reward = diminished_reward self.step_penalty = step_penalty self.nameless = nameless self.hidden_npc = hidden_npc if max_steps is None: max_steps = 5*size**2 super().__init__( grid_size=size, max_steps=max_steps, # Set this to True for maximum speed see_through_walls=True, actions=MiniGridEnv.Actions, action_space=spaces.MultiDiscrete([ len(MiniGridEnv.Actions), *TalkItOutPoliteGrammar.grammar_action_space.nvec ]), add_npc_direction=True ) print({ "size": size, "hear_yourself": hear_yourself, "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) # Generate the surrounding walls self.grid.wall_rect(0, 0, width, height) # Generate the surrounding walls self.grid.wall_rect(0, 0, width, height) # Generate the 4 doors at random positions self.door_pos = [] self.door_front_pos = [] # Remembers positions in front of door to avoid setting wizard here self.door_pos.append((self._rand_int(2, width-2), 0)) self.door_front_pos.append((self.door_pos[-1][0], self.door_pos[-1][1]+1)) self.door_pos.append((self._rand_int(2, width-2), height-1)) self.door_front_pos.append((self.door_pos[-1][0], self.door_pos[-1][1] - 1)) self.door_pos.append((0, self._rand_int(2, height-2))) self.door_front_pos.append((self.door_pos[-1][0] + 1, self.door_pos[-1][1])) self.door_pos.append((width-1, self._rand_int(2, height-2))) self.door_front_pos.append((self.door_pos[-1][0] - 1, self.door_pos[-1][1])) # Generate the door colors self.door_colors = [] while len(self.door_colors) < len(self.door_pos): color = self._rand_elem(COLOR_NAMES) if color in self.door_colors: continue self.door_colors.append(color) # Place the doors in the grid for idx, pos in enumerate(self.door_pos): color = self.door_colors[idx] self.grid.set(*pos, Door(color)) # Set a randomly coloured WIZARD at a random position color = self._rand_elem(COLOR_NAMES) self.wizard = Wizard(color, "Gandalf", self) # Place it randomly, omitting front of door positions self.place_obj(self.wizard, size=(width, height), reject_fn=lambda _, p: tuple(p) in self.door_front_pos) # add guides GUIDE_NAMES = ["John", "Jack"] # Set a randomly coloured TRUE GUIDE at a random position name = self._rand_elem(GUIDE_NAMES) color = self._rand_elem(COLOR_NAMES) self.true_guide = Guide(color, name, self, liar=False) # Place it randomly, omitting invalid positions self.place_obj(self.true_guide, size=(width, height), # reject_fn=lambda _, p: tuple(p) in self.door_front_pos) reject_fn=lambda _, p: tuple(p) in [*self.door_front_pos, tuple(self.wizard.cur_pos)]) # Set a randomly coloured FALSE GUIDE at a random position name = self._rand_elem([n for n in GUIDE_NAMES if n != self.true_guide.name]) color = self._rand_elem([c for c in COLOR_NAMES if c != self.true_guide.color]) self.false_guide = Guide(color, name, self, liar=True) # Place it randomly, omitting invalid positions self.place_obj(self.false_guide, size=(width, height), reject_fn=lambda _, p: tuple(p) in [ *self.door_front_pos, tuple(self.wizard.cur_pos), tuple(self.true_guide.cur_pos)]) assert self.true_guide.name != self.false_guide.name assert self.true_guide.color != self.false_guide.color # Randomize the agent's start position and orientation self.place_agent(size=(width, height)) # Select a random target door self.doorIdx = self._rand_int(0, len(self.door_pos)) self.target_pos = self.door_pos[self.doorIdx] self.target_color = self.door_colors[self.doorIdx] # Generate the mission string self.mission = 'go to the %s door' % self.target_color # 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 self.outcome_info = None def step(self, action): p_action = action[0] utterance_action = action[1:] # assert all nan or neither nan assert len(set(np.isnan(utterance_action))) == 1 speak_flag = not all(np.isnan(utterance_action)) obs, reward, done, info = super().step(p_action) if speak_flag: utterance = TalkItOutPoliteGrammar.construct_utterance(utterance_action) if self.hear_yourself: if self.nameless: self.utterance += "{} \n".format(utterance) else: self.utterance += "YOU: {} \n".format(utterance) self.conversation += "YOU: {} \n".format(utterance) # check if near wizard if self.wizard.is_near_agent(): reply = self.wizard.listen(utterance) if reply: if self.nameless: self.utterance += "{} \n".format(reply) else: self.utterance += "{}: {} \n".format(self.wizard.name, reply) self.conversation += "{}: {} \n".format(self.wizard.name, reply) if self.true_guide.is_near_agent(): reply = self.true_guide.listen(utterance) if reply: if self.nameless: self.utterance += "{} \n".format(reply) else: self.utterance += "{}: {} \n".format(self.true_guide.name, reply) self.conversation += "{}: {} \n".format(self.true_guide.name, reply) if self.false_guide.is_near_agent(): reply = self.false_guide.listen(utterance) if reply: if self.nameless: self.utterance += "{} \n".format(reply) else: self.utterance += "{}: {} \n".format(self.false_guide.name, reply) self.conversation += "{}: {} \n".format(self.false_guide.name, reply) if utterance == TalkItOutPoliteGrammar.construct_utterance([1, 0]): ax, ay = self.agent_pos tx, ty = self.target_pos if (ax == tx and abs(ay - ty) == 1) or (ay == ty and abs(ax - tx) == 1): reward = self._reward() for dx, dy in self.door_pos: if (ax == dx and abs(ay - dy) == 1) or (ay == dy and abs(ax - dx) == 1): # agent has chosen some door episode, regardless of if the door is correct the episode is over done = True # Don't let the agent open any of the doors if p_action == self.actions.toggle: done = True if p_action == self.actions.done: done = True # discount if self.step_penalty: reward = reward - 0.01 if self.hidden_npc: # all npc are hidden assert np.argwhere(obs['image'][:,:,0] == OBJECT_TO_IDX['npc']).size == 0 assert "{}:".format(self.wizard.name) not in self.utterance # assert "{}:".format(self.true_guide.name) not in self.utterance # assert "{}:".format(self.false_guide.name) not in self.utterance # fill observation with text self.append_existing_utterance_to_history() obs = self.add_utterance_to_observation(obs) self.reset_utterance() if done: if reward > 0: self.outcome_info = "SUCCESS: agent got {} reward \n".format(np.round(reward, 1)) else: self.outcome_info = "FAILURE: agent got {} reward \n".format(reward) 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) self.window.clear_text() # erase previous text self.window.set_caption(self.conversation, [ "Gandalf:", "Jack:", "John:", "Where is the exit", "Open sesame", ]) self.window.ax.set_title("correct door: {}".format(self.true_guide.target_color), loc="left", fontsize=10) if self.outcome_info: color = None if "SUCCESS" in self.outcome_info: color = "lime" elif "FAILURE" in self.outcome_info: color = "red" self.window.add_text(*(0.01, 0.85, self.outcome_info), **{'fontsize':15, 'color':color, 'weight':"bold"}) self.window.show_img(obs) # re-draw image to add changes to window return obs class TalkItOutPolite8x8Env(TalkItOutPoliteEnv): def __init__(self, **kwargs): super().__init__(size=8, max_steps=100, **kwargs) class TalkItOutPolite6x6Env(TalkItOutPoliteEnv): def __init__(self): super().__init__(size=6, max_steps=100) class TalkItOutPoliteNameless8x8Env(TalkItOutPoliteEnv): def __init__(self): super().__init__(size=8, max_steps=100, nameless=True) register( id='MiniGrid-TalkItOutPolite-5x5-v0', entry_point='gym_minigrid.envs:TalkItOutPoliteEnv' ) register( id='MiniGrid-TalkItOutPolite-6x6-v0', entry_point='gym_minigrid.envs:TalkItOutPolite6x6Env' ) register( id='MiniGrid-TalkItOutPolite-8x8-v0', entry_point='gym_minigrid.envs:TalkItOutPolite8x8Env' ) register( id='MiniGrid-TalkItOutPoliteNameless-8x8-v0', entry_point='gym_minigrid.envs:TalkItOutPoliteNameless8x8Env' )