| 65 | |
| 66 | |
| 67 | class ReplayMemory: |
| 68 | def __init__(self, size=MAX_EXPERIENCES, frame_height=IM_SIZE, frame_width=IM_SIZE, |
| 69 | agent_history_length=4, batch_size=32): |
| 70 | """ |
| 71 | Args: |
| 72 | size: Integer, Number of stored transitions |
| 73 | frame_height: Integer, Height of a frame of an Atari game |
| 74 | frame_width: Integer, Width of a frame of an Atari game |
| 75 | agent_history_length: Integer, Number of frames stacked together to create a state |
| 76 | batch_size: Integer, Number of transitions returned in a minibatch |
| 77 | """ |
| 78 | self.size = size |
| 79 | self.frame_height = frame_height |
| 80 | self.frame_width = frame_width |
| 81 | self.agent_history_length = agent_history_length |
| 82 | self.batch_size = batch_size |
| 83 | self.count = 0 |
| 84 | self.current = 0 |
| 85 | |
| 86 | # Pre-allocate memory |
| 87 | self.actions = np.empty(self.size, dtype=np.int32) |
| 88 | self.rewards = np.empty(self.size, dtype=np.float32) |
| 89 | self.frames = np.empty((self.size, self.frame_height, self.frame_width), dtype=np.uint8) |
| 90 | self.terminal_flags = np.empty(self.size, dtype=np.bool) |
| 91 | |
| 92 | # Pre-allocate memory for the states and new_states in a minibatch |
| 93 | self.states = np.empty((self.batch_size, self.agent_history_length, |
| 94 | self.frame_height, self.frame_width), dtype=np.uint8) |
| 95 | self.new_states = np.empty((self.batch_size, self.agent_history_length, |
| 96 | self.frame_height, self.frame_width), dtype=np.uint8) |
| 97 | self.indices = np.empty(self.batch_size, dtype=np.int32) |
| 98 | |
| 99 | def add_experience(self, action, frame, reward, terminal): |
| 100 | """ |
| 101 | Args: |
| 102 | action: An integer-encoded action |
| 103 | frame: One grayscale frame of the game |
| 104 | reward: reward the agend received for performing an action |
| 105 | terminal: A bool stating whether the episode terminated |
| 106 | """ |
| 107 | if frame.shape != (self.frame_height, self.frame_width): |
| 108 | raise ValueError('Dimension of frame is wrong!') |
| 109 | self.actions[self.current] = action |
| 110 | self.frames[self.current, ...] = frame |
| 111 | self.rewards[self.current] = reward |
| 112 | self.terminal_flags[self.current] = terminal |
| 113 | self.count = max(self.count, self.current+1) |
| 114 | self.current = (self.current + 1) % self.size |
| 115 | |
| 116 | def _get_state(self, index): |
| 117 | if self.count is 0: |
| 118 | raise ValueError("The replay memory is empty!") |
| 119 | if index < self.agent_history_length - 1: |
| 120 | raise ValueError("Index must be min 3") |
| 121 | return self.frames[index-self.agent_history_length+1:index+1, ...] |
| 122 | |
| 123 | def _get_valid_indices(self): |
| 124 | for i in range(self.batch_size): |