(self)
| 180 | return ActionResponse.model_validate(response) |
| 181 | |
| 182 | def step(self): |
| 183 | |
| 184 | # get next task/action from GAME API |
| 185 | action_response = self._get_action(self._session.function_result) |
| 186 | action_type = action_response.action_type |
| 187 | |
| 188 | print("#" * 50) |
| 189 | print("STEP") |
| 190 | print(f"Current Task: {action_response.agent_state.current_task}") |
| 191 | print(f"Action response: {action_response}") |
| 192 | print(f"Action type: {action_type}") |
| 193 | |
| 194 | # if new task is updated/generated |
| 195 | if ( |
| 196 | action_response.agent_state.hlp |
| 197 | and action_response.agent_state.hlp.change_indicator |
| 198 | ): |
| 199 | print("New task generated") |
| 200 | print(f"Task: {action_response.agent_state.current_task}") |
| 201 | |
| 202 | # execute action |
| 203 | if action_type in [ |
| 204 | ActionType.CALL_FUNCTION, |
| 205 | ActionType.CONTINUE_FUNCTION, |
| 206 | ]: |
| 207 | print(f"Action Selected: {action_response.action_args['fn_name']}") |
| 208 | print(f"Action Args: {action_response.action_args['args']}") |
| 209 | |
| 210 | if not action_response.action_args: |
| 211 | raise ValueError("No function information provided by GAME") |
| 212 | |
| 213 | self._session.function_result = ( |
| 214 | self.workers[self.current_worker_id] |
| 215 | .action_space[action_response.action_args["fn_name"]] |
| 216 | .execute(**action_response.action_args) |
| 217 | ) |
| 218 | |
| 219 | print(f"Function result: {self._session.function_result}") |
| 220 | |
| 221 | # update worker states |
| 222 | updated_worker_state = self.workers[self.current_worker_id].get_state_fn( |
| 223 | self._session.function_result, self.worker_states[self.current_worker_id]) |
| 224 | self.worker_states[self.current_worker_id] = updated_worker_state |
| 225 | |
| 226 | elif action_response.action_type == ActionType.WAIT: |
| 227 | print("Task ended completed or ended (not possible wiht current actions)") |
| 228 | |
| 229 | elif action_response.action_type == ActionType.GO_TO: |
| 230 | if not action_response.action_args: |
| 231 | raise ValueError("No location information provided by GAME") |
| 232 | |
| 233 | next_worker = action_response.action_args["location_id"] |
| 234 | print(f"Next worker selected: {next_worker}") |
| 235 | self.current_worker_id = next_worker |
| 236 | |
| 237 | else: |
| 238 | raise ValueError( |
| 239 | f"Unknown action type: {action_response.action_type}") |
no test coverage detected