(self)
| 54 | self.sample_queue = queue.Queue() |
| 55 | |
| 56 | def reset(self) -> None: |
| 57 | self.observations = {} |
| 58 | for k, s in self.observation_space.spaces.items(): |
| 59 | self.observations[k] = np.zeros((self.buffer_size, self.n_envs,)+s.shape, dtype=s.dtype) |
| 60 | # int(np.prod(self.action_space.shape)) |
| 61 | self.actions = np.zeros((self.buffer_size, self.n_envs)+self.action_space.shape, dtype=np.float32) |
| 62 | self.rewards = np.zeros((self.buffer_size, self.n_envs), dtype=np.float32) |
| 63 | self.returns = np.zeros((self.buffer_size, self.n_envs), dtype=np.float32) |
| 64 | self.advantages = np.zeros((self.buffer_size, self.n_envs), dtype=np.float32) |
| 65 | self.dones = np.zeros((self.buffer_size, self.n_envs), dtype=np.float32) |
| 66 | self.values = np.zeros((self.buffer_size, self.n_envs), dtype=np.float32) |
| 67 | self.log_probs = np.zeros((self.buffer_size, self.n_envs), dtype=np.float32) |
| 68 | self.mus = np.zeros((self.buffer_size, self.n_envs)+self.action_space.shape, dtype=np.float32) |
| 69 | self.sigmas = np.zeros((self.buffer_size, self.n_envs)+self.action_space.shape, dtype=np.float32) |
| 70 | self.exploration_suggests = np.zeros((self.buffer_size, self.n_envs), dtype=[('acc', 'U10'), ('steer', 'U10')]) |
| 71 | |
| 72 | self.reward_debugs = [[] for i in range(self.n_envs)] |
| 73 | self.terminal_debugs = [[] for i in range(self.n_envs)] |
| 74 | |
| 75 | self.pos = 0 |
| 76 | self.full = False |
| 77 | |
| 78 | def compute_returns_and_advantage(self, last_value: th.Tensor, dones: np.ndarray) -> None: |
| 79 | last_gae_lam = 0 |
no outgoing calls
no test coverage detected