added misc

Author: sadanand1120

misc/benchmark_utils.py

@@ -0,0 +1,62 @@
+# here are all regex templates
+import re
+from typing import List
+
+# labels
+"""
+NOTE: there is often overlap between labels. However, you also often have 2 constraints
+with the same overlapping labels eg. constraint_A = (lable_AS, label_IT) and constraint_B = (label_AS, label_IT).
+Simply use your judgement to assign the best combination eg. constraint_A = (lable_AS) and constraint_B = (label_IT).
+
+Try to use single labels as much as possible
+"""
+label_AL = "Action Location"  # UMBRELLA label: checks for action at location eg. exists goto startloc
+label_OWK = "Open World Knowledge"
+label_M = "Manipulation at Location"
+label_EO = "Event Ordering"  # checks for before/after
+label_IT = "Correct Initial/Terminal"  # checks for the first or last event in a trace
+label_L = "Location"  # simple checks for robot location eg. exists goto startloc
+label_ES = "Exhaustive Search"  # "ForAll" style checks
+label_A = "Ask Statement at Location"
+label_S = "Say Statement at Location"
+label_CE = "Check Entity Statements at Location"
+
+labels = [
+    label_OWK,
+    label_M,
+    label_EO,
+    label_IT,
+    label_L,
+    label_ES,
+    label_AL,
+    label_A,
+    label_S,
+    label_CE,
+]
+"""
+providing (a|b|c) matches for subword
+providing a matches for whole word
+"""
+
+
+def contain_words(words: List[str]) -> str:
+    return r"^(?=.*" + r")(?=.*".join(words) + r").*$"
+
+
+def contain_words_and_exlude_words(
+    contained_words: List[str], excluded_words: List[str]
+) -> str:
+    excluded_pattern = r"|".join(
+        r"\b" + re.escape(word) + r"\b" for word in excluded_words
+    )
+    return (
+        r"^(?!.*"
+        + excluded_pattern
+        + r")(?=.*"
+        + r")(?=.*".join(contained_words)
+        + r").*$"
+    )
+
+
+def print_debug(check_name: str, check_output: bool, labels: list[str] = []):
+    print(f"{check_name}: {check_output}. Labels: {labels}")

misc/bounded_subprocess.py

@@ -0,0 +1,98 @@
+import os
+import signal
+import fcntl
+import time
+import subprocess
+from typing import List
+
+MAX_BYTES_PER_READ = 1024
+SLEEP_BETWEEN_READS = 0.1
+
+
+class Result:
+    timeout: int
+    exit_code: int
+    stdout: str
+    stderr: str
+
+    def __init__(self, timeout, exit_code, stdout, stderr):
+        self.timeout = timeout
+        self.exit_code = exit_code
+        self.stdout = stdout
+        self.stderr = stderr
+
+
+def set_nonblocking(reader):
+    fd = reader.fileno()
+    fl = fcntl.fcntl(fd, fcntl.F_GETFL)
+    fcntl.fcntl(fd, fcntl.F_SETFL, fl | os.O_NONBLOCK)
+
+
+def run(
+    args: List[str],
+    timeout_seconds: int = 15,
+    max_output_size: int = 2048,
+    env=None,
+) -> Result:
+    """
+    Runs the given program with arguments. After the timeout elapses, kills the process
+    and all other processes in the process group. Captures at most max_output_size bytes
+    of stdout and stderr each, and discards any output beyond that.
+    """
+    p = subprocess.Popen(
+        args,
+        env=env,
+        stdin=subprocess.DEVNULL,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        start_new_session=True,
+        bufsize=MAX_BYTES_PER_READ,
+    )
+    set_nonblocking(p.stdout)
+    set_nonblocking(p.stderr)
+
+    process_group_id = os.getpgid(p.pid)
+
+    # We sleep for 0.1 seconds in each iteration.
+    max_iterations = timeout_seconds * 10
+    stdout_saved_bytes = []
+    stderr_saved_bytes = []
+    stdout_bytes_read = 0
+    stderr_bytes_read = 0
+
+    for _ in range(max_iterations):
+        this_stdout_read = p.stdout.read(MAX_BYTES_PER_READ)
+        this_stderr_read = p.stderr.read(MAX_BYTES_PER_READ)
+        # this_stdout_read and this_stderr_read may be None if stdout or stderr
+        # are closed. Without these checks, test_close_output fails.
+        if this_stdout_read is not None and stdout_bytes_read < max_output_size:
+            stdout_saved_bytes.append(this_stdout_read)
+            stdout_bytes_read += len(this_stdout_read)
+        if this_stderr_read is not None and stderr_bytes_read < max_output_size:
+            stderr_saved_bytes.append(this_stderr_read)
+            stderr_bytes_read += len(this_stderr_read)
+
+        exit_code = p.poll()
+        if exit_code is not None:
+            # finish reading output
+            this_stdout_read = p.stdout.read(max_output_size - stdout_bytes_read)
+            this_stderr_read = p.stderr.read(max_output_size - stderr_bytes_read)
+            if this_stdout_read is not None:
+                stdout_saved_bytes.append(this_stdout_read)
+            if this_stderr_read is not None:
+                stderr_saved_bytes.append(this_stderr_read)
+            break
+
+        time.sleep(SLEEP_BETWEEN_READS)
+
+    try:
+        # Kills the process group. Without this line, test_fork_once fails.
+        os.killpg(process_group_id, signal.SIGKILL)
+    except ProcessLookupError:
+        pass
+
+    timeout = exit_code is None
+    exit_code = exit_code if exit_code is not None else -1
+    stdout = b"".join(stdout_saved_bytes).decode("utf-8", errors="ignore")
+    stderr = b"".join(stderr_saved_bytes).decode("utf-8", errors="ignore")
+    return Result(timeout=timeout, exit_code=exit_code, stdout=stdout, stderr=stderr)

misc/utils.py

@@ -0,0 +1,91 @@
+import importlib.util
+import sys
+import pandas as pd
+import glob
+from pathlib import Path
+import json
+
+
+def custom_serializer(obj):
+    if isinstance(obj, (list, tuple)):
+        return "__list__", obj
+    return obj
+
+
+def custom_sort_key(file_name):
+    # assume the format is in task_prompt_completion.py
+    file_name = file_name.split("/")[-1].split(".")[0]
+    name, seq1, seq2 = file_name.split("_")
+    return name, int(seq1), int(seq2)
+
+
+def check_file_exists(f: str):
+    return Path(f).exists()
+
+
+def load_module(module_name, module_path):
+    spec = importlib.util.spec_from_file_location(module_name, module_path)
+    foo = importlib.util.module_from_spec(spec)
+    sys.modules[module_name] = foo
+    spec.loader.exec_module(foo)
+    return foo
+
+
+def read_benchmark(p: Path, read_regex: str):
+    if not p.is_dir():
+        raise Exception(f"Path {p} is not a directory.")
+    python_files = glob.glob(str(p) + "/{}.py".format(read_regex))
+    print("benchmark python files: ", python_files)
+    benchmarks = []
+    for idx, f in enumerate(python_files):
+        program = load_module("p{}".format(idx), f)
+        benchmark = {
+            "name": program.name,
+            "prompts": program.prompts,
+            "tests": program.tests,
+        }
+        benchmarks.append(benchmark)
+    return pd.DataFrame(benchmarks)
+
+
+def read_completions(p: Path, read_regex: str):
+    # TODO: read completions from file
+    if not p.is_dir():
+        raise Exception(f"Path {p} is not a directory.")
+    python_files = sorted(
+        glob.glob(str(p) + "/{}.py".format(read_regex)), key=custom_sort_key
+    )
+
+    print("completion python files: ", python_files)
+    benchmarks = []
+    for idx, f in enumerate(python_files):
+        program = load_module("p{}".format(idx), f)
+        benchmark = {
+            "name": program.name,
+            "detailed_name": python_files[idx].split(".")[0].split("/")[-1],
+            "prompt": program.prompt,
+            "completion": program.completion,
+        }
+        benchmarks.append(benchmark)
+    return pd.DataFrame(benchmarks)
+
+
+def write_dict_to_python_file(variable_name, data_dict, filename, write_mode="w"):
+    # Convert the dictionary to a JSON-formatted string with custom serialization
+    json_str = json.dumps(data_dict, default=custom_serializer, indent=4)
+
+    # Create the content of the Python file with the formatted dictionary assignment
+    content = f"{variable_name} = {json_str}\n\n"
+
+    # Write the content to the specified filename
+    with open(filename, write_mode) as file:
+        file.write(content)
+
+
+def write_custom_to_python_file(message, filename, write_mode="w"):
+    # Create the content of the Python file with the formatted dictionary assignment
+    content = f"{message}\n\n"
+
+    # Write the content to the specified filename
+    with open(filename, write_mode) as file:
+        file.write(content)