Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
62 changes: 37 additions & 25 deletions examples/circle_packing/evaluator.py
Original file line number Diff line number Diff line change
Expand Up @@ -79,41 +79,36 @@ def validate_packing(centers, radii):
return True


def run_with_timeout(program_path, timeout_seconds=20):
def _subprocess_script(program_path, results_path):
"""
Run the program in a separate process with timeout
using a simple subprocess approach
Build the runner executed in the child process.

Args:
program_path: Path to the program file
timeout_seconds: Maximum execution time in seconds

Returns:
centers, radii, sum_radii tuple from the program
Paths are inserted with repr(). A quoted Windows path such as
'C:\\Users\\...' is not valid Python: \\U starts a unicode escape and the
child exits with SyntaxError before the program runs.
"""
# Create a temporary file to execute
with tempfile.NamedTemporaryFile(suffix=".py", delete=False) as temp_file:
# Write a script that executes the program and saves results
script = f"""
program_lit = repr(os.fspath(program_path))
results_lit = repr(os.fspath(results_path))
return f"""
import sys
import numpy as np
import os
import pickle
import traceback

# Add the directory to sys.path
sys.path.insert(0, os.path.dirname('{program_path}'))
sys.path.insert(0, os.path.dirname({program_lit}))

# Debugging info
print(f"Running in subprocess, Python version: {{sys.version}}")
print(f"Program path: {program_path}")
print("Program path: " + {program_lit})

try:
# Import the program
spec = __import__('importlib.util').util.spec_from_file_location("program", '{program_path}')
spec = __import__('importlib.util').util.spec_from_file_location("program", {program_lit})
program = __import__('importlib.util').util.module_from_spec(spec)
spec.loader.exec_module(program)

# Run the packing function
print("Calling run_packing()...")
centers, radii, sum_radii = program.run_packing()
Expand All @@ -126,22 +121,39 @@ def run_with_timeout(program_path, timeout_seconds=20):
'sum_radii': sum_radii
}}

with open('{temp_file.name}.results', 'wb') as f:
with open({results_lit}, 'wb') as f:
pickle.dump(results, f)
print(f"Results saved to {temp_file.name}.results")
print("Results saved to " + {results_lit})

except Exception as e:
# If an error occurs, save the error instead
print(f"Error in subprocess: {{str(e)}}")
traceback.print_exc()
with open('{temp_file.name}.results', 'wb') as f:
with open({results_lit}, 'wb') as f:
pickle.dump({{'error': str(e)}}, f)
print(f"Error saved to {temp_file.name}.results")
print("Error saved to " + {results_lit})
"""
temp_file.write(script.encode())
temp_file_path = temp_file.name

results_path = f"{temp_file_path}.results"

def run_with_timeout(program_path, timeout_seconds=20):
"""
Run the program in a separate process with timeout
using a simple subprocess approach

Args:
program_path: Path to the program file
timeout_seconds: Maximum execution time in seconds

Returns:
centers, radii, sum_radii tuple from the program
"""
# Create a temporary file to execute
with tempfile.NamedTemporaryFile(suffix=".py", delete=False) as temp_file:
temp_file_path = temp_file.name
results_path = f"{temp_file_path}.results"
# Write a script that executes the program and saves results
script = _subprocess_script(program_path, results_path)
temp_file.write(script.encode())

try:
# Run the script with timeout
Expand Down
62 changes: 37 additions & 25 deletions examples/circle_packing_with_artifacts/evaluator.py
Original file line number Diff line number Diff line change
Expand Up @@ -94,41 +94,36 @@ def validate_packing(centers, radii):
return is_valid, validation_details


def run_with_timeout(program_path, timeout_seconds=20):
def _subprocess_script(program_path, results_path):
"""
Run the program in a separate process with timeout
using a simple subprocess approach
Build the runner executed in the child process.

Args:
program_path: Path to the program file
timeout_seconds: Maximum execution time in seconds

Returns:
centers, radii, sum_radii tuple from the program
Paths are inserted with repr(). A quoted Windows path such as
'C:\\Users\\...' is not valid Python: \\U starts a unicode escape and the
child exits with SyntaxError before the program runs.
"""
# Create a temporary file to execute
with tempfile.NamedTemporaryFile(suffix=".py", delete=False) as temp_file:
# Write a script that executes the program and saves results
script = f"""
program_lit = repr(os.fspath(program_path))
results_lit = repr(os.fspath(results_path))
return f"""
import sys
import numpy as np
import os
import pickle
import traceback

# Add the directory to sys.path
sys.path.insert(0, os.path.dirname('{program_path}'))
sys.path.insert(0, os.path.dirname({program_lit}))

# Debugging info
print(f"Running in subprocess, Python version: {{sys.version}}")
print(f"Program path: {program_path}")
print("Program path: " + {program_lit})

try:
# Import the program
spec = __import__('importlib.util').util.spec_from_file_location("program", '{program_path}')
spec = __import__('importlib.util').util.spec_from_file_location("program", {program_lit})
program = __import__('importlib.util').util.module_from_spec(spec)
spec.loader.exec_module(program)

# Run the packing function
print("Calling run_packing()...")
centers, radii, sum_radii = program.run_packing()
Expand All @@ -141,22 +136,39 @@ def run_with_timeout(program_path, timeout_seconds=20):
'sum_radii': sum_radii
}}

with open('{temp_file.name}.results', 'wb') as f:
with open({results_lit}, 'wb') as f:
pickle.dump(results, f)
print(f"Results saved to {temp_file.name}.results")
print("Results saved to " + {results_lit})

except Exception as e:
# If an error occurs, save the error instead
print(f"Error in subprocess: {{str(e)}}")
traceback.print_exc()
with open('{temp_file.name}.results', 'wb') as f:
with open({results_lit}, 'wb') as f:
pickle.dump({{'error': str(e)}}, f)
print(f"Error saved to {temp_file.name}.results")
print("Error saved to " + {results_lit})
"""
temp_file.write(script.encode())
temp_file_path = temp_file.name

results_path = f"{temp_file_path}.results"

def run_with_timeout(program_path, timeout_seconds=20):
"""
Run the program in a separate process with timeout
using a simple subprocess approach

Args:
program_path: Path to the program file
timeout_seconds: Maximum execution time in seconds

Returns:
centers, radii, sum_radii tuple from the program
"""
# Create a temporary file to execute
with tempfile.NamedTemporaryFile(suffix=".py", delete=False) as temp_file:
temp_file_path = temp_file.name
results_path = f"{temp_file_path}.results"
# Write a script that executes the program and saves results
script = _subprocess_script(program_path, results_path)
temp_file.write(script.encode())

try:
# Run the script with timeout
Expand Down
85 changes: 85 additions & 0 deletions tests/test_circle_packing_subprocess_paths.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,85 @@
"""
Regression test for circle-packing evaluators embedding paths in generated Python.

A Windows temp path contains \\Users. Putting that path inside single quotes makes
\\U a unicode escape, so the child process dies with SyntaxError and every
evaluation is scored as a failure.
"""

import importlib.util
import os
import tempfile
import textwrap
import unittest

REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
EVALUATOR_PATHS = [
os.path.join(REPO_ROOT, "examples", "circle_packing", "evaluator.py"),
os.path.join(REPO_ROOT, "examples", "circle_packing_with_artifacts", "evaluator.py"),
]


def load_evaluator(path):
"""Load an example evaluator by path (examples are not part of the package)"""
name = "circle_packing_eval_" + os.path.splitext(os.path.basename(os.path.dirname(path)))[0]
spec = importlib.util.spec_from_file_location(name, path)
evaluator = importlib.util.module_from_spec(spec)
spec.loader.exec_module(evaluator)
return evaluator


class TestCirclePackingSubprocessPaths(unittest.TestCase):
"""Generated runner source must be valid Python for Windows paths"""

@classmethod
def setUpClass(cls):
cls.evaluators = [load_evaluator(path) for path in EVALUATOR_PATHS]

def test_windows_user_temp_path_compiles(self):
"""C:\\Users\\... must not be parsed as a unicode escape"""
program_path = r"C:\Users\alice\AppData\Local\Temp\prog.py"
results_path = program_path + ".results"
for evaluator in self.evaluators:
script = evaluator._subprocess_script(program_path, results_path)
compile(script, "<circle-packing-runner>", "exec")
self.assertIn(repr(program_path), script)
self.assertIn(repr(results_path), script)

def test_apostrophe_in_path_compiles(self):
"""A quote in a directory name must stay inside the string literal"""
program_path = r"C:\Users\o'brien\prog.py"
results_path = program_path + ".results"
for evaluator in self.evaluators:
script = evaluator._subprocess_script(program_path, results_path)
compile(script, "<circle-packing-runner>", "exec")

def test_run_with_timeout_executes_program_from_user_temp(self):
"""The documented subprocess path must return the program's packing"""
program = textwrap.dedent("""
import numpy as np

def run_packing():
centers = np.array([[0.5, 0.5]])
radii = np.array([0.1])
return centers, radii, float(radii.sum())
""")
with tempfile.NamedTemporaryFile(
mode="w", suffix=".py", delete=False, encoding="utf-8"
) as handle:
handle.write(program)
program_path = handle.name

try:
for evaluator in self.evaluators:
centers, radii, reported_sum = evaluator.run_with_timeout(
program_path, timeout_seconds=30
)
self.assertEqual(centers.shape, (1, 2))
self.assertEqual(float(radii[0]), 0.1)
self.assertAlmostEqual(float(reported_sum), 0.1)
finally:
os.unlink(program_path)


if __name__ == "__main__":
unittest.main()