| 73 | |
| 74 | |
| 75 | class PythonExecutor: |
| 76 | def __init__( |
| 77 | self, |
| 78 | runtime: Optional[Any] = None, |
| 79 | get_answer_symbol: Optional[str] = None, |
| 80 | get_answer_expr: Optional[str] = None, |
| 81 | get_answer_from_stdout: bool = False, |
| 82 | timeout_length: int = 5, |
| 83 | ) -> None: |
| 84 | self.runtime = runtime if runtime else GenericRuntime() |
| 85 | self.answer_symbol = get_answer_symbol |
| 86 | self.answer_expr = get_answer_expr |
| 87 | self.get_answer_from_stdout = get_answer_from_stdout |
| 88 | self.pool = Pool(multiprocess.cpu_count()) |
| 89 | self.timeout_length = timeout_length |
| 90 | |
| 91 | def process_generation_to_code(self, gens: str): |
| 92 | return [g.strip().split('\n') for g in gens] |
| 93 | |
| 94 | @staticmethod |
| 95 | def execute( |
| 96 | code, |
| 97 | get_answer_from_stdout = None, |
| 98 | runtime = None, |
| 99 | answer_symbol = None, |
| 100 | answer_expr = None, |
| 101 | timeout_length = 10, |
| 102 | auto_mode=False |
| 103 | ): |
| 104 | try: |
| 105 | if auto_mode: |
| 106 | if "print(" in code[-1]: |
| 107 | program_io = io.StringIO() |
| 108 | with redirect_stdout(program_io): |
| 109 | timeout(timeout_length)(runtime.exec_code)('\n'.join(code)) |
| 110 | program_io.seek(0) |
| 111 | result = program_io.read() |
| 112 | else: |
| 113 | print(code) |
| 114 | timeout(timeout_length)(runtime.exec_code)('\n'.join(code[:-1])) |
| 115 | result = timeout(timeout_length)(runtime.eval_code)(code[-1]) |
| 116 | else: |
| 117 | if get_answer_from_stdout: |
| 118 | program_io = io.StringIO() |
| 119 | with redirect_stdout(program_io): |
| 120 | timeout(timeout_length)(runtime.exec_code)('\n'.join(code)) |
| 121 | program_io.seek(0) |
| 122 | result = program_io.read() |
| 123 | elif answer_symbol: |
| 124 | timeout(timeout_length)(runtime.exec_code)('\n'.join(code)) |
| 125 | result = runtime._global_vars[answer_symbol] |
| 126 | elif answer_expr: |
| 127 | timeout(timeout_length)(runtime.exec_code)('\n'.join(code)) |
| 128 | result = timeout(timeout_length)(runtime.eval_code)(answer_expr) |
| 129 | else: |
| 130 | timeout(timeout_length)(runtime.exec_code)('\n'.join(code[:-1])) |
| 131 | result = timeout(timeout_length)(runtime.eval_code)(code[-1]) |
| 132 | report = "Done" |