| 62 | |
| 63 | |
| 64 | class PythonExecutor: |
| 65 | def __init__( |
| 66 | self, |
| 67 | runtime: Optional[Any] = None, |
| 68 | get_answer_symbol: Optional[str] = None, |
| 69 | get_answer_expr: Optional[str] = None, |
| 70 | get_answer_from_stdout: bool = False, |
| 71 | timeout_length: int = 5, |
| 72 | ) -> None: |
| 73 | self.runtime = runtime if runtime else GenericRuntime() |
| 74 | self.answer_symbol = get_answer_symbol |
| 75 | self.answer_expr = get_answer_expr |
| 76 | self.get_answer_from_stdout = get_answer_from_stdout |
| 77 | self.pool = Pool(multiprocess.cpu_count()) |
| 78 | self.timeout_length = timeout_length |
| 79 | |
| 80 | def process_generation_to_code(self, gens: str): |
| 81 | return [g.strip().split('\n') for g in gens] |
| 82 | |
| 83 | @staticmethod |
| 84 | def execute( |
| 85 | code, |
| 86 | get_answer_from_stdout = None, |
| 87 | runtime = None, |
| 88 | answer_symbol = None, |
| 89 | answer_expr = None, |
| 90 | timeout_length = 10, |
| 91 | auto_mode=False |
| 92 | ): |
| 93 | try: |
| 94 | if auto_mode: |
| 95 | if "print(" in code[-1]: |
| 96 | program_io = io.StringIO() |
| 97 | with redirect_stdout(program_io): |
| 98 | timeout(timeout_length)(runtime.exec_code)('\n'.join(code)) |
| 99 | program_io.seek(0) |
| 100 | result = program_io.read() |
| 101 | else: |
| 102 | print(code) |
| 103 | timeout(timeout_length)(runtime.exec_code)('\n'.join(code[:-1])) |
| 104 | result = timeout(timeout_length)(runtime.eval_code)(code[-1]) |
| 105 | else: |
| 106 | if get_answer_from_stdout: |
| 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 | elif answer_symbol: |
| 113 | timeout(timeout_length)(runtime.exec_code)('\n'.join(code)) |
| 114 | result = runtime._global_vars[answer_symbol] |
| 115 | elif answer_expr: |
| 116 | timeout(timeout_length)(runtime.exec_code)('\n'.join(code)) |
| 117 | result = timeout(timeout_length)(runtime.eval_code)(answer_expr) |
| 118 | else: |
| 119 | timeout(timeout_length)(runtime.exec_code)('\n'.join(code[:-1])) |
| 120 | result = timeout(timeout_length)(runtime.eval_code)(code[-1]) |
| 121 | report = "Done" |