| 64 | |
| 65 | |
| 66 | class PythonExecutor: |
| 67 | def __init__( |
| 68 | self, |
| 69 | runtime: Optional[Any] = None, |
| 70 | get_answer_symbol: Optional[str] = None, |
| 71 | get_answer_expr: Optional[str] = None, |
| 72 | get_answer_from_stdout: bool = False, |
| 73 | timeout_length: int = 5, |
| 74 | ) -> None: |
| 75 | self.runtime = runtime if runtime else GenericRuntime() |
| 76 | self.answer_symbol = get_answer_symbol |
| 77 | self.answer_expr = get_answer_expr |
| 78 | self.get_answer_from_stdout = get_answer_from_stdout |
| 79 | self.timeout_length = timeout_length |
| 80 | |
| 81 | def process_generation_to_code(self, gens: str): |
| 82 | return [g.split('\n') for g in gens] |
| 83 | |
| 84 | @staticmethod |
| 85 | def execute( |
| 86 | code, |
| 87 | get_answer_from_stdout = None, |
| 88 | runtime = None, |
| 89 | answer_symbol = None, |
| 90 | answer_expr = None, |
| 91 | timeout_length = 10, |
| 92 | ): |
| 93 | try: |
| 94 | if get_answer_from_stdout: |
| 95 | program_io = io.StringIO() |
| 96 | with redirect_stdout(program_io): |
| 97 | timeout(timeout_length)(runtime.exec_code)('\n'.join(code)) |
| 98 | program_io.seek(0) |
| 99 | result = program_io.readlines()[-1] |
| 100 | elif answer_symbol: |
| 101 | timeout(timeout_length)(runtime.exec_code)('\n'.join(code)) |
| 102 | result = runtime._global_vars[answer_symbol] |
| 103 | elif answer_expr: |
| 104 | timeout(timeout_length)(runtime.exec_code)('\n'.join(code)) |
| 105 | result = timeout(timeout_length)(runtime.eval_code)(answer_expr) |
| 106 | else: |
| 107 | timeout(timeout_length)(runtime.exec_code)('\n'.join(code[:-1])) |
| 108 | result = timeout(timeout_length)(runtime.eval_code)(code[-1]) |
| 109 | exec_info = "Done" |
| 110 | str(result) |
| 111 | pickle.dumps(result) # serialization check |
| 112 | except: |
| 113 | result = '' |
| 114 | exec_info = traceback.format_exc().split('\n')[-2] |
| 115 | return result, exec_info |
| 116 | |
| 117 | def apply(self, code): |
| 118 | return self.batch_apply([code])[0] |
| 119 | |
| 120 | def batch_apply(self, batch_code): |
| 121 | all_code_snippets = self.process_generation_to_code(batch_code) |
| 122 | |
| 123 | timeout_cnt = 0 |