Spaces:
Sleeping
Sleeping
File size: 3,699 Bytes
dbc6675 | 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 | import os
import random
import subprocess
import tempfile
from pathlib import Path
import numpy as np
import torch
def set_seed(seed: int):
"""
Sets the seed for generating random numbers to ensure reproducibility.
"""
random.seed(seed)
np.random.seed(seed)
torch.manual_seed(seed)
torch.cuda.manual_seed(seed)
torch.cuda.manual_seed_all(seed) # if using multi-GPU.
# Ensure deterministic behavior in CuDNN
torch.backends.cudnn.deterministic = True
torch.backends.cudnn.benchmark = False
# Enforce strict deterministic algorithms
# Note: This might throw errors if an operation doesn't have a deterministic implementation,
# but for LSTM/Linear it is supported.
torch.use_deterministic_algorithms(True)
# Set Python hash seed
os.environ["PYTHONHASHSEED"] = str(seed)
# Set CUBLAS workspace config for deterministic LSTM on CUDA >= 10.2
os.environ["CUBLAS_WORKSPACE_CONFIG"] = ":4096:8"
print(f"Global seed set to: {seed}")
def worker_init_fn(worker_id):
"""
Function to ensure DataLoader workers are seeded deterministically.
"""
worker_seed = torch.initial_seed() % 2**32
np.random.seed(worker_seed)
random.seed(worker_seed)
def validate_plan(domain_path, problem_path, plan_actions, val_path):
"""
Writes the plan to a temp file and runs VAL.
Returns:
- is_solved (bool): Goal reached (VAL: "Plan valid")
- is_executable (bool): All actions applied validly (VAL: "Plan executed successfully")
"""
# 0. Pre-checks
if not plan_actions:
print("Empty plan provided for validation.")
return False, False
val_bin = Path(val_path)
# Check existence and permissions of VAL binary
if not val_bin.exists() or not os.access(val_bin, os.X_OK):
print(f"VAL binary not found or not executable at: {val_path}")
return False, False
# Ensure domain/prob paths are absolute
abs_domain = Path(domain_path).resolve()
abs_problem = Path(problem_path).resolve()
# 1. Write plan to temporary file
# VAL expects actions on separate lines: (action arg1 arg2)
with tempfile.NamedTemporaryFile(mode="w", delete=False, suffix=".plan") as tmp:
for action in plan_actions:
# Clean up action string. Pyperplan might give "(action a b)" or "action a b"
act_str = str(action).strip()
# Ensure lowercase
act_str = act_str.lower()
if not act_str.startswith("("):
act_str = f"({act_str})"
line = f"{act_str}\n"
tmp.write(line)
tmp_plan_path = Path(tmp.name).resolve()
# 2. Run VAL
# Command: Validate -v domain.pddl problem.pddl plan.plan
cmd = [str(val_bin), "-v", str(abs_domain), str(abs_problem), str(tmp_plan_path)]
try:
# Capture both stdout and stderr
result = subprocess.run(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
check=False,
)
output = result.stdout
# 3. Parse Output
# "Plan valid" implies both executable AND goal reached.
# "Plan executed successfully" implies executable, but goal might not be reached.
is_solved = "Plan valid" in output
is_executable = is_solved or "Plan executed successfully" in output
except Exception as e:
print(f"Error running VAL: {e}")
is_solved = False
is_executable = False
finally:
# Cleanup temp file
if os.path.exists(tmp_plan_path):
os.remove(tmp_plan_path)
return is_solved, is_executable
|