(self, input_data, timestamp)
| 34 | self._wrapper_kwargs = cfg['env_wrapper']['kwargs'] |
| 35 | |
| 36 | def run_step(self, input_data, timestamp): |
| 37 | input_data = copy.deepcopy(input_data) |
| 38 | |
| 39 | policy_input = self._wrapper_class.process_obs(input_data, self._wrapper_kwargs['input_states'], train=False) |
| 40 | |
| 41 | actions, values, log_probs, mu, sigma, features = self._policy.forward( |
| 42 | policy_input, deterministic=True, clip_action=True) |
| 43 | control = self._wrapper_class.process_act(actions, self._wrapper_kwargs['acc_as_action'], train=False) |
| 44 | self.supervision_dict = { |
| 45 | 'action': np.array([control.throttle, control.steer, control.brake], dtype=np.float32), |
| 46 | 'value': values[0], |
| 47 | 'action_mu': mu[0], |
| 48 | 'action_sigma': sigma[0], |
| 49 | 'features': features[0], |
| 50 | 'speed': input_data['speed']['forward_speed'] |
| 51 | } |
| 52 | self.supervision_dict = copy.deepcopy(self.supervision_dict) |
| 53 | |
| 54 | return control |
| 55 | |
| 56 | @property |
| 57 | def obs_configs(self): |
nothing calls this directly
no test coverage detected