Save a game history block Parameters ---------- game: Any a game history block end_tag: bool True -> the game is finished. (always True) gap_steps: int if the game is not finished, we only save the transitions that can be co
(self, game, end_tag, gap_steps, priorities=None)
| 36 | self.save_game(game, True, gap_step, priorities) |
| 37 | |
| 38 | def save_game(self, game, end_tag, gap_steps, priorities=None): |
| 39 | """Save a game history block |
| 40 | Parameters |
| 41 | ---------- |
| 42 | game: Any |
| 43 | a game history block |
| 44 | end_tag: bool |
| 45 | True -> the game is finished. (always True) |
| 46 | gap_steps: int |
| 47 | if the game is not finished, we only save the transitions that can be computed |
| 48 | priorities: list |
| 49 | the priorities corresponding to the transitions in the game history |
| 50 | """ |
| 51 | if self.get_total_len() >= self.config.total_transitions: |
| 52 | return |
| 53 | |
| 54 | if end_tag: |
| 55 | self._eps_collected += 1 |
| 56 | valid_len = len(game) |
| 57 | else: |
| 58 | valid_len = len(game) - gap_steps |
| 59 | |
| 60 | if priorities is None: |
| 61 | max_prio = self.priorities.max() if self.buffer else 1 |
| 62 | self.priorities = np.concatenate((self.priorities, [max_prio for _ in range(valid_len)] + [0. for _ in range(valid_len, len(game))])) |
| 63 | else: |
| 64 | assert len(game) == len(priorities), " priorities should be of same length as the game steps" |
| 65 | priorities = priorities.copy().reshape(-1) |
| 66 | # priorities[valid_len:len(game)] = 0. |
| 67 | self.priorities = np.concatenate((self.priorities, priorities)) |
| 68 | |
| 69 | self.buffer.append(game) |
| 70 | self.game_look_up += [(self.base_idx + len(self.buffer) - 1, step_pos) for step_pos in range(len(game))] |
| 71 | |
| 72 | def get_game(self, idx): |
| 73 | # return a game |
no test coverage detected