Method
__init__
(self, buffer_size: int, observation_space: spaces.Space, action_space: spaces.Space,
gae_lambda: float = 1, gamma: float = 0.99, n_envs: int = 1)
Source from the content-addressed store, hash-verified
| 34 | |
| 35 | class PpoBuffer(): |
| 36 | def __init__(self, buffer_size: int, observation_space: spaces.Space, action_space: spaces.Space, |
| 37 | gae_lambda: float = 1, gamma: float = 0.99, n_envs: int = 1): |
| 38 | |
| 39 | self.buffer_size = buffer_size |
| 40 | self.observation_space = observation_space |
| 41 | self.action_space = action_space |
| 42 | self.gae_lambda = gae_lambda |
| 43 | self.gamma = gamma |
| 44 | self.n_envs = n_envs |
| 45 | self.reset() |
| 46 | |
| 47 | self.pos = 0 |
| 48 | self.full = False |
| 49 | if th.cuda.is_available(): |
| 50 | self.device = 'cuda' |
| 51 | else: |
| 52 | self.device = 'cpu' |
| 53 | |
| 54 | self.sample_queue = queue.Queue() |
| 55 | |
| 56 | def reset(self) -> None: |
| 57 | self.observations = {} |
Callers
nothing calls this directly
Tested by
no test coverage detected