Prepare a batch context that contains: game_lst: a list of game histories game_pos_lst: transition index in game (relative index) indices_lst: transition index in replay buffer weights_lst: the weight concering the priorit
(self, batch_size, beta)
| 77 | return game |
| 78 | |
| 79 | def prepare_batch_context(self, batch_size, beta): |
| 80 | """Prepare a batch context that contains: |
| 81 | game_lst: a list of game histories |
| 82 | game_pos_lst: transition index in game (relative index) |
| 83 | indices_lst: transition index in replay buffer |
| 84 | weights_lst: the weight concering the priority |
| 85 | make_time: the time the batch is made (for correctly updating replay buffer when data is deleted) |
| 86 | Parameters |
| 87 | ---------- |
| 88 | batch_size: int |
| 89 | batch size |
| 90 | beta: float |
| 91 | the parameter in PER for calculating the priority |
| 92 | """ |
| 93 | assert beta > 0 |
| 94 | |
| 95 | total = self.get_total_len() |
| 96 | |
| 97 | probs = self.priorities ** self._alpha |
| 98 | |
| 99 | probs /= probs.sum() |
| 100 | # sample data |
| 101 | indices_lst = np.random.choice(total, batch_size, p=probs, replace=False) |
| 102 | |
| 103 | weights_lst = (total * probs[indices_lst]) ** (-beta) |
| 104 | weights_lst /= weights_lst.max() |
| 105 | |
| 106 | game_lst = [] |
| 107 | game_pos_lst = [] |
| 108 | |
| 109 | for idx in indices_lst: |
| 110 | game_id, game_pos = self.game_look_up[idx] |
| 111 | game_id -= self.base_idx |
| 112 | game = self.buffer[game_id] |
| 113 | |
| 114 | game_lst.append(game) |
| 115 | game_pos_lst.append(game_pos) |
| 116 | |
| 117 | make_time = [time.time() for _ in range(len(indices_lst))] |
| 118 | |
| 119 | context = (game_lst, game_pos_lst, indices_lst, weights_lst, make_time) |
| 120 | return context |
| 121 | |
| 122 | def update_priorities(self, batch_indices, batch_priorities, make_time): |
| 123 | # update the priorities for data still in replay buffer |
nothing calls this directly
no test coverage detected