| import numpy as np |
| from abc import ABC, abstractmethod |
|
|
| class AbstractEnvRunner(ABC): |
| def __init__(self, *, env, model, nsteps): |
| self.env = env |
| self.model = model |
| self.nenv = nenv = env.num_envs if hasattr(env, 'num_envs') else 1 |
| self.batch_ob_shape = (nenv*nsteps,) + env.observation_space.shape |
| self.obs = np.zeros((nenv,) + env.observation_space.shape, dtype=env.observation_space.dtype.name) |
| self.obs[:] = env.reset() |
| self.nsteps = nsteps |
| self.states = model.initial_state |
| self.dones = [False for _ in range(nenv)] |
|
|
| @abstractmethod |
| def run(self): |
| raise NotImplementedError |
|
|
|
|
