(
self, action: Action
)
| 238 | self.context_manager.__exit__() |
| 239 | |
| 240 | def step( |
| 241 | self, action: Action |
| 242 | ) -> tuple[dict[str, Observation], float, bool, bool, dict[str, Any]]: |
| 243 | if not self.reset_finished: |
| 244 | raise RuntimeError("Call reset first before calling step.") |
| 245 | |
| 246 | success = False |
| 247 | fail_error = "" |
| 248 | try: |
| 249 | self.page = execute_action( |
| 250 | action, |
| 251 | self.page, |
| 252 | self.context, |
| 253 | self.observation_handler.action_processor, |
| 254 | ) |
| 255 | success = True |
| 256 | except Exception as e: |
| 257 | fail_error = str(e) |
| 258 | |
| 259 | # hard sleep TODO[shuyanzh] suboptimal, may need to check network |
| 260 | if self.sleep_after_execution > 0: |
| 261 | time.sleep(self.sleep_after_execution) |
| 262 | |
| 263 | observation = self._get_obs() |
| 264 | observation_metadata = self._get_obs_metadata() |
| 265 | |
| 266 | info = { |
| 267 | "page": DetachedPage(self.page.url, self.page.content()), |
| 268 | "fail_error": fail_error, |
| 269 | "observation_metadata": observation_metadata, |
| 270 | } |
| 271 | msg = ( |
| 272 | observation, |
| 273 | float(success), # reward |
| 274 | False, # terminated |
| 275 | False, # truncated |
| 276 | info, |
| 277 | ) |
| 278 | return msg |
no test coverage detected