add files

testing_util.py

@@ -0,0 +1,434 @@
+import json
+import sys
+import faulthandler
+
+# used for debugging to time steps
+from datetime import datetime
+
+# to run the solution files we're using a timing based approach
+import signal
+
+import numpy as np
+# for capturing the stdout
+from io import StringIO
+# used for testing the code that reads from input
+from unittest.mock import patch, mock_open
+
+from pyext import RuntimeModule
+
+from enum import Enum
+class CODE_TYPE(Enum):
+ call_based = 0
+ standard_input = 1
+
+# stuff for setting up signal timer
+class TimeoutException(Exception):
+ pass
+def timeout_handler(signum, frame):
+ print("alarm went off")
+ #return
+ raise TimeoutException
+signal.signal(signal.SIGALRM, timeout_handler)
+timeout = 4 # seconds
+
+# used to capture stdout as a list
+# from https://stackoverflow.com/a/16571630/6416660
+# alternative use redirect_stdout() from contextlib
+class Capturing(list):
+ def __enter__(self):
+ self._stdout = sys.stdout
+ sys.stdout = self._stringio = StringIO()
+ # Make closing the StringIO a no-op
+ self._stringio.close = lambda x: 1
+ return self
+ def __exit__(self, *args):
+ self.extend(self._stringio.getvalue().splitlines())
+ del self._stringio # free up some memory
+ sys.stdout = self._stdout
+
+
+def run_test(sample, test=None, debug=False):
+ """
+ if test(generated_code) is not None it'll try to run the code.
+ otherwise it'll just return an input and output pair.
+ """
+ if debug:
+ print(f"start = {datetime.now().time()}")
+
+ try:
+ in_outs = json.loads(sample["input_output"])
+ except ValueError:
+ in_outs = None
+ if in_outs:
+ #if debug:
+ # print(f"test cases json = {in_outs['inputs']} {in_outs['outputs']}")
+ if in_outs.get("fn_name") is None:
+ which_type = CODE_TYPE.standard_input # Standard input
+ method_name = None
+ else:
+ which_type = CODE_TYPE.call_based # Call-based
+ method_name = in_outs["fn_name"]
+
+ if debug:
+ print(f"loaded input_output = {datetime.now().time()}")
+ 
+ #else:
+ # continue
+ if test is None:
+ return in_outs
+ elif test is not None:
+ results = []
+ sol = "import sys\nimport time\nimport itertools\nfrom itertools import accumulate, product, permutations, combinations\nimport collections\nfrom collections import Counter, OrderedDict, deque, defaultdict, ChainMap\nfrom functools import lru_cache\nimport math\nfrom math import sqrt, sin, cos, tan, ceil, fabs, floor, gcd, exp, log, log2\nimport fractions\nfrom typing import List, Tuple\nimport numpy as np\nimport random\nimport heapq\nfrom heapq import *\n"
+ if debug:
+ print(f"loading test code = {datetime.now().time()}")
+ 
+ if which_type == CODE_TYPE.call_based:
+ sol += test
+ if debug:
+ print(f"sol = {sol}")
+ signal.alarm(timeout)
+ try:
+ tmp_sol = RuntimeModule.from_string("tmp_sol", "", sol)
+ if "class Solution" not in test:
+ tmp = tmp_sol
+ else:
+ tmp = tmp_sol.Solution()
+ signal.alarm(0)
+ except Exception as e:
+ signal.alarm(0)
+ print(f"type 0 compilation error = {e}")
+ results.append(-2)
+ return results
+ signal.alarm(0)
+
+ elif which_type == CODE_TYPE.standard_input:
+ # sol
+ tmp_test = test.split("\n")
+
+ new_test = []
+ for x in tmp_test:
+ if (not x.startswith("from ")) and (not x.startswith("import ")):
+ new_test.append("\t" + x + "\n")
+ else:
+ new_test.append(x + "\n")
+ tmp_test = new_test
+ 
+ new_test = ""
+ started = False
+ for i in tmp_test:
+ if i.startswith("\t") and not started:
+ new_test += "stdin = sys.stdin\nstdout = sys.stdout\n"
+ new_test += "def code():\n"
+ new_test += i
+ started = True
+ elif started and ((i.startswith("from ")) or (i.startswith("import "))): 
+ new_test += "\t" + i
+ else:
+ new_test += i
+ tmp_test = new_test
+
+ sol += tmp_test
+ if debug:
+ print(f"sol = {sol}")
+ method_name = "code"
+ signal.alarm(timeout)
+ try:
+ tmp_sol = RuntimeModule.from_string("tmp_sol", "", sol)
+ tmp = tmp_sol
+ signal.alarm(0)
+ except Exception as e:
+ signal.alarm(0)
+ print(f"type 1 compilation error = {e}")
+ results.append(-2)
+ return results
+ signal.alarm(0)
+ if debug:
+ print(f"get method = {datetime.now().time()}")
+ 
+ try:
+ method = getattr(tmp, method_name) # get_attr second arg must be str
+ except:
+ signal.alarm(0)
+ e = sys.exc_info()
+ print(f"unable to get function error = {e}")
+ return results
+
+ for index, inputs in enumerate(in_outs["inputs"]):
+ # JSON forces dictionaries to have string keys; this undoes this (assuming a singleton list)
+ try:
+ if isinstance(inputs[0], dict):
+ inputs = [{int(k): v for k,v in inputs[0].items()}]
+ except:
+ True
+ try:
+ if isinstance(in_outs["outputs"][index], dict):
+ in_outs["outputs"][index] = [{int(k): v for k,v in in_outs["outputs"][index].items()}]
+ except:
+ True
+ try:
+ if isinstance(in_outs["outputs"][index][0], dict):
+ in_outs["outputs"][index] = [{int(k): v for k,v in in_outs["outputs"][index][0].items()}]
+ except:
+ True
+
+ if debug:
+ print(f"time: {datetime.now().time()} testing index = {index} inputs = {inputs}, {type(inputs)}. type = {which_type}")
+ if which_type == CODE_TYPE.call_based: # Call-based
+ signal.alarm(timeout)
+ faulthandler.enable()
+ try:
+ output = method(*inputs)
+
+ # ground truth sequences are not tuples
+ if isinstance(output, tuple):
+ output = list(output)
+ 
+ tmp_result = output == in_outs["outputs"][index]
+ if isinstance(in_outs["outputs"][index], list) and in_outs["outputs"][index]:
+ tmp_result = tmp_result or (output == in_outs["outputs"][index][0])
+
+ # ground truth sequences are not tuples
+ try:
+ if isinstance(output[0], tuple):
+ tmp_result = tmp_result or ([list(x) for x in output] == in_outs["outputs"][index][0])
+ except:
+ True
+ results.append(tmp_result)
+
+ # reset the alarm
+ signal.alarm(0)
+ except Exception as e:
+ signal.alarm(0)
+ faulthandler.disable()
+ print(f"Standard input runtime error or time limit exceeded error = {e}")
+ results.append(-1)
+ continue
+ faulthandler.disable()
+ signal.alarm(0)
+ #if debug:
+ #print(f"outputs = {output}, test outputs = {in_outs['outputs'][index]}, inputs = {inputs}, {type(inputs)}, {output == [in_outs['outputs'][index]]}")
+ elif which_type == CODE_TYPE.standard_input: # Standard input
+ faulthandler.enable()
+ signal.alarm(timeout)
+ passed = False
+
+ if isinstance(inputs, list):
+ inputs = "\n".join(inputs)
+ if isinstance(in_outs['outputs'][index], list):
+ in_outs['outputs'][index] = "\n".join(in_outs['outputs'][index])
+ with Capturing() as output:
+ try:
+ print("doing call")
+ call_method(method, inputs)
+ print("call done")
+ # reset the alarm
+ signal.alarm(0)
+ passed = True
+ except Exception as e:
+ print("call not done we are in exception")
+ # runtime error or took too long
+ signal.alarm(0)
+ print(f"Call-based runtime error or time limit exceeded error = {repr(e)}{e}")
+ results.append(-1)
+ signal.alarm(0)
+
+ if not passed:
+ if debug:
+ nl = "\n"
+ if not isinstance(inputs, list):
+ print(f"not passed output = {output}, test outputs = {in_outs['outputs'][index]}, inputs = {inputs.replace(nl,' new-line ')}, {type(inputs)}, {output == [in_outs['outputs'][index]]}")
+ else:
+ print(f"not passed output = {output}, test outputs = {in_outs['outputs'][index]}, inputs = {inputs}, {type(inputs)}, {output == [in_outs['outputs'][index]]}")
+ continue
+
+ if passed and debug:
+ print(f"==> output = {output}, test outputs = {in_outs['outputs'][index]}")
+
+ if custom_compare_(output, in_outs['outputs'][index]):
+ tmp_result = True
+ results.append(tmp_result)
+ continue
+
+ # ground truth sequences are expressed as lists not tuples
+ if isinstance(output, tuple):
+ output = list(output)
+
+ tmp_result = False
+ try:
+ tmp_result = (output == [in_outs["outputs"][index]])
+ if isinstance(in_outs["outputs"][index], list):
+ tmp_result = tmp_result or (output == in_outs["outputs"][index])
+ if isinstance(output[0], str):
+ tmp_result = tmp_result or ([e.strip() for e in output] == in_outs["outputs"][index])
+ except Exception as e:
+ print(f"Failed check1 exception = {e}")
+ pass
+
+ if tmp_result == True: 
+ results.append(tmp_result)
+ continue
+
+ # try one more time without \n
+ if isinstance(in_outs["outputs"][index], list):
+ for tmp_index, i in enumerate(in_outs["outputs"][index]):
+ in_outs["outputs"][index][tmp_index] = i.split("\n")
+ in_outs["outputs"][index][tmp_index] = [x.strip() for x in in_outs["outputs"][index][tmp_index] if x]
+ else:
+ in_outs["outputs"][index] = in_outs["outputs"][index].split("\n")
+ in_outs["outputs"][index] = list(filter(len, in_outs["outputs"][index]))
+ in_outs["outputs"][index] = list(map(lambda x:x.strip(), in_outs["outputs"][index]))
+
+ try:
+ tmp_result = (output == [in_outs["outputs"][index]])
+ if isinstance(in_outs["outputs"][index], list):
+ tmp_result = tmp_result or (output == in_outs["outputs"][index])
+ except Exception as e:
+ print(f"Failed check2 exception = {e}")
+ pass
+
+ if tmp_result == True:
+ results.append(tmp_result)
+ continue
+
+ # try by converting the output into a split up list too
+ if isinstance(output, list):
+ output = list(filter(len, output))
+
+ if debug:
+ nl = "\n"
+ if not isinstance(inputs, list):
+ print(f"output = {output}, test outputs = {in_outs['outputs'][index]}, inputs = {inputs.replace(nl,' new-line ')}, {type(inputs)}, {output == [in_outs['outputs'][index]]}") 
+ else:
+ print(f"output = {output}, test outputs = {in_outs['outputs'][index]}, inputs = {inputs}, {type(inputs)}, {output == [in_outs['outputs'][index]]}") 
+ 
+ if tmp_result == True:
+ results.append(tmp_result)
+ continue
+
+ try:
+ tmp_result = (output == [in_outs["outputs"][index]])
+ if isinstance(in_outs["outputs"][index], list):
+ tmp_result = tmp_result or (output == in_outs["outputs"][index])
+ except Exception as e:
+ print(f"Failed check3 exception = {e}")
+ pass
+
+ try:
+ output_float = [float(e) for e in output]
+ gt_float = [float(e) for e in in_outs['outputs'][index]]
+ tmp_result = tmp_result or ((len(output_float) == len(gt_float)) and np.allclose(output_float, gt_float))
+ except Exception as e:
+ pass
+ try:
+ if isinstance(output[0], list):
+ output_float = [float(e) for e in output[0]]
+ gt_float = [float(e) for e in in_outs['outputs'][index][0]]
+ tmp_result = tmp_result or ((len(output_float) == len(gt_float)) and np.allclose(output_float, gt_float))
+ except Exception as e:
+ pass
+
+ if tmp_result == True:
+ results.append(tmp_result)
+ continue
+
+ # try by converting the stuff into split up list
+ if isinstance(in_outs["outputs"][index], list):
+ for tmp_index, i in enumerate(in_outs["outputs"][index]):
+ in_outs["outputs"][index][tmp_index] = set(i.split())
+ else:
+ in_outs["outputs"][index] = set(in_outs["outputs"][index].split())
+
+ try:
+ tmp_result = (output == in_outs["outputs"][index])
+ except Exception as e:
+ print(f"Failed check4 exception = {e}")
+ continue
+
+ if tmp_result == True:
+ results.append(tmp_result)
+ continue 
+
+ # try by converting the output into a split up list too
+ if isinstance(output, list):
+ for tmp_index, i in enumerate(output):
+ output[tmp_index] = i.split()
+ output = list(filter(len, output))
+ for tmp_index, i in enumerate(output):
+ output[tmp_index] = set(i) 
+ else:
+ output = output.split()
+ output = list(filter(len, output))
+ output = set(output)
+
+ try:
+ tmp_result = (set(frozenset(s) for s in output) == set(frozenset(s) for s in in_outs["outputs"][index]))
+ except Exception as e:
+ print(f"Failed check5 exception = {e}")
+
+
+ # if they are all numbers, round so that similar numbers are treated as identical
+ try:
+ tmp_result = tmp_result or (set(frozenset(round(float(t),3) for t in s) for s in output) ==\
+ set(frozenset(round(float(t),3) for t in s) for s in in_outs["outputs"][index]))
+ except Exception as e:
+ print(f"Failed check6 exception = {e}")
+ 
+ if tmp_result == True and debug:
+ print("PASSED")
+ 
+ results.append(tmp_result)
+ 
+ if debug:
+ nl = "\n"
+ if not isinstance(inputs, list):
+ print(f"output = {output}, test outputs = {in_outs['outputs'][index]}, inputs = {inputs.replace(nl,' new-line ')}, {type(inputs)}, {output == [in_outs['outputs'][index]]}")
+ else:
+ print(f"output = {output}, test outputs = {in_outs['outputs'][index]}, inputs = {inputs}, {type(inputs)}, {output == [in_outs['outputs'][index]]}") 
+ 
+
+ return results
+
+def custom_compare_(output, ground_truth):
+ 
+ if isinstance(output, list):
+ output_1 = "\n".join(output)
+ if stripped_string_compare(output_1, ground_truth):
+ return True
+
+ if isinstance(output, list):
+ output_2 = [o.lstrip().rstrip() for o in output]
+ output_2 = "\n".join(output_2)
+ if stripped_string_compare(output_2, ground_truth):
+ return True
+
+ return False
+
+def stripped_string_compare(s1, s2):
+ s1 = s1.lstrip().rstrip()
+ s2 = s2.lstrip().rstrip()
+ return s1 == s2
+
+def call_method(method, inputs):
+
+ if isinstance(inputs, list):
+ inputs = "\n".join(inputs)
+
+ inputs_line_iterator = iter(inputs.split("\n"))
+
+ # sys.setrecursionlimit(10000)
+
+ # @patch('builtins.input', side_effect=inputs.split("\n"))
+ @patch('builtins.open', mock_open(read_data=inputs))
+ @patch('sys.stdin', StringIO(inputs))
+ @patch('sys.stdin.readline', lambda *args: next(inputs_line_iterator))
+ @patch('sys.stdin.readlines', lambda *args: inputs.split("\n"))
+ @patch('sys.stdin.read', lambda *args: inputs)
+ # @patch('sys.stdout.write', print)
+ def _inner_call_method(_method):
+ try:
+ return _method()
+ except SystemExit as e:
+ pass
+ finally:
+ pass
+ return _inner_call_method(method) 

utils.py

@@ -0,0 +1,159 @@
+import itertools
+import numpy as np
+from typing import Dict
+from datasets import load_dataset
+import testing_util as test_util
+
+
+DATASET = "codeparrot/apps"
+
+
+def evaluate_generations(generations, level=["all"]):
+ """We take the list of code generations and try to compile them
+ and the run their corresponding unit tests which are retrieved from the APPS dataset.
+
+ Args:
+ generations: list of code generations, in the same order as APPS dataset samples
+ level: list of levels to evaluate, can be "all", "introductory", "interview" or "competition"
+
+ Returns:
+ results: dictionary of results, key is the problem index, value is a list of results for each generation
+ [-2] = compile error, [-1] = runtime error [False] = failed test case [True] = passed test case
+ """
+
+ # generations are code generations in the same order of the dataset
+ apps_eval = load_dataset(DATASET, split="test", difficulties=level)
+ gpt_codes = generations
+ results = {}
+ for index in range(len(generations)):
+ print(f"task {index}")
+ generated_code = gpt_codes[index]
+ sample = apps_eval[index]
+ res = []
+ # loop over the generations
+ for o_idx, o in enumerate(generated_code):
+ curr_res = [-2]
+ try:
+ print("Run test")
+ curr_res = test_util.run_test(sample, test=o, debug=False)
+ print("\nSuccessful compilation!")
+ fixed = []
+ for e in curr_res:
+ if isinstance(e, np.ndarray):
+ e = e.item(0)
+ if isinstance(e, np.bool_):
+ e = bool(e)
+ fixed.append(e)
+ curr_res = fixed
+ if not np.all(curr_res):
+ print(f"Results were not True for all test cases") #{curr_res}")
+ except Exception as e:
+ print(f"Compilation failed, test framework exception = {repr(e)}{e}\n")
+ break
+ finally:
+ assert isinstance(curr_res, list)
+ res.append(curr_res)
+ results[index] = res
+
+ return results
+
+
+def estimate_pass_at_k(num_samples, num_correct, k):
+ """Estimates pass@k of each problem and returns them in an array."""
+
+ def estimator(n: int, c: int, k: int) -> float:
+ """Calculates 1 - comb(n - c, k) / comb(n, k)."""
+ if n - c < k:
+ return 1.0
+ return 1.0 - np.prod(1.0 - k / np.arange(n - c + 1, n + 1))
+
+ if isinstance(num_samples, int):
+ num_samples_it = itertools.repeat(num_samples, len(num_correct))
+ else:
+ assert len(num_samples) == len(num_correct)
+ num_samples_it = iter(num_samples)
+
+ return np.array([estimator(int(n), int(c), k) for n, c in zip(num_samples_it, num_correct)])
+
+
+def get_results(results: Dict, count_errors: bool = False, k_list: list = [1, 10, 100]):
+ """
+ Given the results evaluated against the testcases we output some statistics.
+ For single generations:
+ >>> example_results = {"0": [[-2]],"1": [[False,False]],"2": [[True,True]],"3": [[False,True,False,True]], "4": [[-1,-1]]}
+ >>> get_results(example_results, count_errors=True)
+ number of compile errors = 1 avg = 0.2
+ number of runtime errors = 1 avg = 0.2
+ number of test cases run = 5
+ Test Case Average (average accuracy over problems) = 0.3
+ Strict Accuracy (all test cases passed / total problems) = 0.2
+
+ For multiple generations:
+ >>> example_results = {"0": [[-2], [True, True, True]],"1": [[-1,-1, -1], [True, False, True]]}
+ >>> get_results(example_results k_list=[1, 2])
+ {'pass@1': 0.25, 'pass@2': 0.5}
+ """
+
+ metrics = {"avg_accuracy": None, "strict_accuracy": None, "pass_at_k": None}
+
+ if len(results["0"]) == 1:
+ # for single generations we compute average accuracy and stric accuracy: original APPS metrics
+ print("Computing accuracy metrics...")
+ res = []
+ per_prob_res = []
+ all_correct = []
+ for index in results:
+ results[index] = np.array(results[index])
+ res.extend(results[index])
+ per_prob_res.append(np.mean(results[index]>0))
+ all_correct.append(np.all(results[index]>0))
+ # we count campilation and runtime errors once per pronlem
+ compile_errors = len([e for e in res if -2 in e])
+ runtime_errors = len([e for e in res if -1 in e])
+ total_testcases = len(res)
+ if count_errors:
+ print(f"number of compile errors = {compile_errors} avg = {compile_errors / total_testcases}")
+ print(f"number of runtime errors = {runtime_errors} avg = {runtime_errors / total_testcases}")
+ print(f"number of problems evaluated = {total_testcases}")
+
+ print(f"Test Case Average Accuracy (ver tests) = {np.mean(per_prob_res)}")
+ print(f"Strict Accuracy (over problems that pass all tests) = {np.mean(all_correct)}")
+ metrics["avg_accuracy"] = np.mean(per_prob_res)
+ metrics["strict_accuracy"] = np.mean(all_correct)
+
+ else:
+ # for multiple generations we use pass@k metric used in the HumanEval benchmark
+ # we use strict accuracy, a generation is valid if it has to pass all the tests
+ print("Computing pass@k metric for multiple generations...")
+ # total is list with nb generations per task (task=index)
+ # correct is number of generations that passed all tests per task
+ total = []
+ correct = [] 
+ for index in results:
+ all_correct = []
+ for generation in results[index]:
+ gen = np.array(generation)
+ all_correct.append(np.all(gen>0))
+ total.append(len(all_correct))
+ correct.append(sum(all_correct))
+ total = np.array(total)
+ correct = np.array(correct)
+ ks = k_list
+ pass_at_k = {f"pass@{k}": estimate_pass_at_k(total, correct, k).mean() for k in ks if (total >= k).all()}
+ print(pass_at_k)
+ metrics["pass_at_k"] = pass_at_k
+ return metrics
+
+def compute_metrics(generations, k_list=[1, 10, 100], count_errors=True, level=["all"]):
+ """Return metrics for the given generations.
+ Args:
+ generations: dict of generations, keyed by problem index
+ k_list: list of k values to compute pass@k when using multiple generations
+ count_errors: whether to count compilation and runtime errors when using single generations
+ level: which level difficulty in APPS dataset was used for the given generations
+ Returns:
+ metrics: dict of metrics 
+ """
+ results = evaluate_generations(generations, level=level)
+ metrics = get_results(results, count_errors=count_errors, k_list=k_list)
+ return metrics