(self, batch_size: Optional[int] = None)
| 139 | self.values[i] = values |
| 140 | |
| 141 | def get(self, batch_size: Optional[int] = None) -> Generator[PpoBufferSamples, None, None]: |
| 142 | assert self.full, '' |
| 143 | indices = np.random.permutation(self.buffer_size * self.n_envs) |
| 144 | # Prepare the data |
| 145 | for tensor in ['actions', 'values', 'log_probs', 'advantages', 'returns', |
| 146 | 'mus', 'sigmas', 'exploration_suggests']: |
| 147 | self.__dict__['flat_'+tensor] = self.flatten(self.__dict__[tensor]) |
| 148 | self.flat_observations = {} |
| 149 | for k in self.observations.keys(): |
| 150 | self.flat_observations[k] = self.flatten(self.observations[k]) |
| 151 | |
| 152 | # spinning up: the next two lines implement the advantage normalization trick |
| 153 | adv_mean = np.mean(self.advantages) |
| 154 | adv_std = np.std(self.advantages) + np.finfo(np.float32).eps |
| 155 | self.advantages = (self.advantages - adv_mean) / adv_std |
| 156 | |
| 157 | # Return everything, don't create minibatches |
| 158 | if batch_size is None: |
| 159 | batch_size = self.buffer_size * self.n_envs |
| 160 | |
| 161 | start_idx = 0 |
| 162 | while start_idx < self.buffer_size * self.n_envs: |
| 163 | yield self._get_samples(indices[start_idx:start_idx + batch_size]) |
| 164 | start_idx += batch_size |
| 165 | |
| 166 | def _get_samples(self, batch_inds: np.ndarray) -> PpoBufferSamples: |
| 167 | def to_torch(x): |
no test coverage detected