code/common/utils.py

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