Updates the observations inside history by performing subtraction from most recent observation and the sum of previous observations. If there are not enough observations to take a difference from, simply store the observation without any differencing.
(self)
| 220 | self.obs = torch.from_numpy(self.obs).float() |
| 221 | |
| 222 | def update_history(self) -> None: |
| 223 | # language=rst |
| 224 | """ |
| 225 | Updates the observations inside history by performing subtraction from most |
| 226 | recent observation and the sum of previous observations. If there are not enough |
| 227 | observations to take a difference from, simply store the observation without any |
| 228 | differencing. |
| 229 | """ |
| 230 | # Recording initial observations. |
| 231 | if self.episode_step_count < len(self.history) * self.delta: |
| 232 | # Store observation based on delta value. |
| 233 | if self.episode_step_count % self.delta == 0: |
| 234 | self.history[self.history_index] = self.obs |
| 235 | else: |
| 236 | # Take difference between stored frames and current frame. |
| 237 | temp = torch.clamp(self.obs - sum(self.history.values()), 0, 1) |
| 238 | |
| 239 | # Store observation based on delta value. |
| 240 | if self.episode_step_count % self.delta == 0: |
| 241 | self.history[self.history_index] = self.obs |
| 242 | |
| 243 | assert ( |
| 244 | len(self.history) == self.history_length |
| 245 | ), "History size is out of bounds" |
| 246 | self.obs = temp |
| 247 | |
| 248 | def update_index(self) -> None: |
| 249 | # language=rst |