Wrapper around the OpenAI ``gym`` environment ``step()`` function. :param a: Action to take in the environment. :return: Observation, reward, done flag, and information dictionary.
(self, a: int)
| 122 | ), "Maximum spiking probability must be in (0, 1]." |
| 123 | |
| 124 | def step(self, a: int) -> Tuple[torch.Tensor, float, bool, Dict[Any, Any]]: |
| 125 | # language=rst |
| 126 | """ |
| 127 | Wrapper around the OpenAI ``gym`` environment ``step()`` function. |
| 128 | |
| 129 | :param a: Action to take in the environment. |
| 130 | :return: Observation, reward, done flag, and information dictionary. |
| 131 | """ |
| 132 | # Call gym's environment step function. |
| 133 | self.obs, self.reward, terminated, truncated, info = self.env.step(a) |
| 134 | self.done = terminated or truncated |
| 135 | |
| 136 | if self.clip_rewards: |
| 137 | self.reward = np.sign(self.reward) |
| 138 | |
| 139 | self.preprocess() |
| 140 | |
| 141 | # Add the raw observation from the gym environment into the info |
| 142 | # for debugging and display. |
| 143 | info["gym_obs"] = self.obs |
| 144 | |
| 145 | # Store frame of history and encode the inputs. |
| 146 | if len(self.history) > 0: |
| 147 | self.update_history() |
| 148 | self.update_index() |
| 149 | # Add the delta observation into the info for debugging and display. |
| 150 | info["delta_obs"] = self.obs |
| 151 | |
| 152 | # The new standard for images is BxTxCxHxW. |
| 153 | # The gym environment doesn't follow exactly the same protocol. |
| 154 | # |
| 155 | # 1D observations will be left as is before the encoder and will become BxTxL. |
| 156 | # 2D observations are assumed to be mono images will become BxTx1xHxW |
| 157 | # 3D observations will become BxTxCxHxW |
| 158 | if self.obs.dim() == 2 and self.add_channel_dim: |
| 159 | # We want CxHxW, it is currently HxW. |
| 160 | self.obs = self.obs.unsqueeze(0) |
| 161 | |
| 162 | # The encoder will add time - now Tx... |
| 163 | if self.encoder is not None: |
| 164 | self.obs = self.encoder(self.obs) |
| 165 | |
| 166 | # Add the batch - now BxTx... |
| 167 | self.obs = self.obs.unsqueeze(0) |
| 168 | |
| 169 | self.episode_step_count += 1 |
| 170 | |
| 171 | # Return converted observations and other information. |
| 172 | return self.obs, self.reward, self.done, info |
| 173 | |
| 174 | def reset(self, seed=None) -> torch.Tensor: |
| 175 | # language=rst |
nothing calls this directly
no test coverage detected