File size: 3,845 Bytes
96abbd8 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 | # -*- coding: utf-8 -*-
"""Execute generated Python code and capture basic diagnostics."""
from __future__ import annotations
import os
import re
import subprocess
import sys
from dataclasses import dataclass
from typing import Optional
AUTO_OBJECTIVE_SNIPPET = """
# Auto-added snippet: attempt to print the objective value for downstream evaluation.
try:
candidate = None
for name in ("model", "m", "Model"):
if name in globals():
candidate = globals()[name]
break
if candidate is not None and hasattr(candidate, "objVal"):
print(f"OBJECTIVE_VALUE: {candidate.objVal}")
except Exception:
pass
""".strip()
@dataclass
class ExecutionResult:
status: str
stdout: str
stderr: str
objective_value: Optional[float]
returncode: Optional[int]
code_path: Optional[str]
def _ensure_directory(path: str) -> None:
os.makedirs(path, exist_ok=True)
def _append_objective_snippet(code: str) -> str:
if "OBJECTIVE_VALUE" in code:
return code if code.endswith("\n") else code + "\n"
return f"{code.rstrip()}\n\n{AUTO_OBJECTIVE_SNIPPET}\n"
def _normalize_output(value: object) -> str:
if value is None:
return ""
if isinstance(value, bytes):
return value.decode("utf-8", errors="replace")
return str(value)
def _extract_objective_value(output: str) -> Optional[float]:
if not output:
return None
patterns = [
r"OBJECTIVE_VALUE:\s*([+-]?\d+\.?\d*(?:[eE][+-]?\d+)?)",
r"Optimal\s+[Oo]bjective[:\s]+([+-]?\d+\.?\d*(?:[eE][+-]?\d+)?)",
r"Obj:\s*([+-]?\d+\.?\d*(?:[eE][+-]?\d+)?)",
r"Objective\s+value:\s*([+-]?\d+\.?\d*(?:[eE][+-]?\d+)?)",
]
for pattern in patterns:
match = re.search(pattern, output, re.IGNORECASE)
if not match:
continue
try:
return float(match.group(1))
except ValueError:
continue
return None
def execute_generated_code(
code: str,
problem_id: int,
output_dir: str,
timeout: int = 120,
) -> ExecutionResult:
"""Write code to disk, execute it, and capture the outcome."""
code_dir = os.path.join(output_dir, "code")
_ensure_directory(code_dir)
code_with_snippet = _append_objective_snippet(code)
code_file = os.path.join(code_dir, f"problem_{problem_id}.py")
with open(code_file, "w", encoding="utf-8") as fh:
fh.write(code_with_snippet)
try:
completed = subprocess.run(
[sys.executable, os.path.basename(code_file)],
cwd=code_dir,
capture_output=True,
text=True,
timeout=timeout,
)
except subprocess.TimeoutExpired as exc:
return ExecutionResult(
status="timeout",
stdout=_normalize_output(exc.stdout),
stderr=f"Execution timeout after {timeout} seconds",
objective_value=None,
returncode=None,
code_path=code_file,
)
except Exception as exc: # pragma: no cover - defensive
return ExecutionResult(
status="error",
stdout="",
stderr=str(exc),
objective_value=None,
returncode=None,
code_path=code_file,
)
stdout = _normalize_output(completed.stdout)
stderr = _normalize_output(completed.stderr)
returncode = completed.returncode
status = "success" if returncode == 0 else "execution_error"
objective_value = _extract_objective_value(stdout) if status == "success" else None
return ExecutionResult(
status=status,
stdout=stdout,
stderr=stderr,
objective_value=objective_value,
returncode=returncode,
code_path=code_file,
)
__all__ = ["ExecutionResult", "execute_generated_code"]
|