Upload 105 files

.gitattributes

@@ -33,3 +33,6 @@ saved_model/**/* filter=lfs diff=lfs merge=lfs -text
 *.zip filter=lfs diff=lfs merge=lfs -text
 *.zst filter=lfs diff=lfs merge=lfs -text
 *tfevents* filter=lfs diff=lfs merge=lfs -text
+data/CodeContest/Test.jsonl filter=lfs diff=lfs merge=lfs -text
+data/xCodeEval/problem_descriptions.jsonl filter=lfs diff=lfs merge=lfs -text
+data/xCodeEval/unittest_db.json filter=lfs diff=lfs merge=lfs -text

README.md

@@ -1,3 +1,23 @@
----
-license: mit
----
+## Running our project
+
+1. Clone our project
+```bash
+cd DebateCoder
+```
+
+2. Create a new conda or python virtual environment and run the following command
+```bash
+pip install -r requirements.txt
+```
+
+3. Set up the .env file by seeing the example.
+
+4. Run the following command to see the options of running this projects
+```bash
+python src/main.py --help
+```
+
+5. Finally run this project. An example is given below:
+```bash
+python src/main.py --model Pangu --dataset HumanEval --strategy DebateCoder
+```

data/CodeContest/Test.jsonl

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:8343e00e4b52160990576bcd2b2588466d2ac39b2751401beedd6606294423c9
+size 23356061

data/xCodeEval/problem_descriptions.jsonl

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:bda417d0e06ded34ee62255b84ac386e8535137b82f8c4f905adda0b93d9f78b
+size 18159029

data/xCodeEval/unittest_db.json

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:26b0b526ad96f7abbf11057b9547febf8237ce0bff00cad96fefc650e34374d5
+size 11346609

requirements.txt

@@ -0,0 +1,15 @@
+# python==3.11
+openai
+numpy
+pandas
+python-dotenv
+pyyml
+tenacity
+tiktoken
+tqdm
+gensim
+jsonlines
+astunparse
+pyarrow
+google-genai
+accelerate

src/constants/lang_mappings.py

@@ -0,0 +1,14 @@
+LANGUAGE_MAPPING = {
+    "Python": "Python 3",
+    "Python3": "Python 3",
+    "C#": "C# 10",
+    "NET-CORE": ".NET Core C#",
+    # "Node": "Node.js",
+    "Rust": "Rust",
+    # "Java":"Java 17",
+    "PHP": "PHP",
+    "Go": "Go",
+    "Ruby": "Ruby",
+    "C++": "GNU C++17",
+    "C": "GNU C"
+}

src/constants/paths.py

@@ -0,0 +1,135 @@
+import os
+from os.path import join, dirname
+
+# HumanEval Dataset
+HUMAN_DATA_DIR = join(
+    "data",
+    "HumanEval",
+)
+
+HUMAN_DATA_PATH = join(
+    HUMAN_DATA_DIR,
+    "HumanEval.jsonl"
+)
+
+HUMAN_WST_DATA_PATH = join(
+    HUMAN_DATA_DIR,
+    "HumanEvalWST.jsonl"
+)
+
+HUMAN_REFLXION_FILTERED_PATH = join(
+    HUMAN_DATA_DIR,
+    "humaneval-py.jsonl"
+)
+
+HUMAN_HARDSET_PATH = join(
+    HUMAN_DATA_DIR,
+    "humaneval-py_hardest50.jsonl"
+)
+
+HUMAN_ET_DATA_PATH = join(
+    HUMAN_DATA_DIR,
+    "HumanEvalET.jsonl"
+)
+
+HUMAN_SIMILAR_PROBLEMS_PATH = join(
+    HUMAN_DATA_DIR,
+    "similar_problems_solutions.jsonl"
+)
+
+
+# MBPP Dataset
+MBPP_DATA_DIR = join(
+    "data",
+    "MBPPEval",
+)
+
+MBPP_DATA_PATH = join(
+    MBPP_DATA_DIR,
+    "MBPP-py.jsonl"
+)
+
+MBPP_ET_DATA_PATH = join(
+    MBPP_DATA_DIR,
+    "MBPP_ET.jsonl"
+)
+
+MBPP_SANITIZED_DATA_PATH = join(
+    MBPP_DATA_DIR,
+    "MBPP_SANITIZED.json"
+)
+
+MBPP_SIMILAR_PROBLEMS_PATH = join(
+    MBPP_DATA_DIR,
+    "similar_problems_solutions.jsonl"
+)
+
+# XCodeEval Dataset
+XCODE_DATA_DIR = join(
+    "data",
+    "xCodeEval",
+)
+
+XCODE_VALIDATION_DATA_PATH = join(
+    XCODE_DATA_DIR,
+    "prog_syn_val.jsonl"
+)
+
+XCODE_TEST_DATA_PATH = join(
+    XCODE_DATA_DIR,
+    "prog_syn_test.jsonl"
+)
+
+XCODE_TRAIN_DATA_DIR_PATH = join(
+    XCODE_DATA_DIR,
+    "train"
+)
+
+XCODE_UNIT_TEST_PATH = join(
+    XCODE_DATA_DIR,
+    "unittest_db.json"
+)
+
+XCODE_PROBLEM_DESCRIPTION_PATH = join(
+    XCODE_DATA_DIR,
+    "problem_descriptions.jsonl"
+)
+
+XCODE_SIMILAR_SRC_UIDS_PATH = join(
+    XCODE_DATA_DIR,
+    "similar_src_uids.json"
+)
+
+XCODE_SIMILAR_PROBLEMS_PATH = join(
+    XCODE_DATA_DIR,
+    "similar_problems_solutions.json"
+)
+
+XCODE_PROBLEM_FILE_MAPPINGS_PATH = join(
+    XCODE_DATA_DIR,
+    "problem_file_mapping.json"
+)
+
+
+# Code Contest Dataset
+CODE_CONTEST_DATA_DIR = join(
+    "data",
+    "CodeContest",
+)
+
+CODE_CONTEST_DATA_PATH = join(
+    CODE_CONTEST_DATA_DIR,
+    "Test.jsonl"
+)
+
+
+# APPS Dataset
+APPS_DATA_DIR = join(
+    "data",
+    "APPS",
+)
+
+APPS_DATA_PATH = join(
+    APPS_DATA_DIR,
+    "selected150.jsonl"
+)

src/datasets/APPSDataset.py

@@ -0,0 +1,48 @@
+from .Dataset import Dataset
+from evaluations.evalute import contest_evaluate, contest_evaluate_public_tests
+from constants.paths import *
+
+
+class APPSDataset(Dataset):
+    def __init__(
+        self,
+        path: str = APPS_DATA_PATH,
+    ):
+        super().__init__(path)
+        self.id_key = "id"
+
+    def evaluate(
+        self,
+        item: dict,
+        cur_imp: str,
+        language: str,
+    ):
+        return contest_evaluate(
+            generated_code=cur_imp,
+            id=item["id"],
+            tests=item["test_list"],
+            lang=language
+        )
+
+    def evaluate_sample_io(
+        self,
+        item: dict,
+        cur_imp: str,
+        language: str,
+    ):
+        if len(item["sample_io"]) == 0:
+            return True, ""
+        return contest_evaluate_public_tests(
+            generated_code=cur_imp,
+            id=item["id"],
+            tests=item["sample_io"],
+            lang=language
+        )
+
+    @staticmethod
+    def get_prompt(item):
+        sample_io_format = ""
+        if len(item['sample_io']) > 0:
+            sample_io_format = f"Sample Input Format:\n{item['sample_io'][0]['input']}\nSample Output Format:\n{item['sample_io'][0]['output'][0]}\n\n-------\n"
+
+        return f"{item['description']}\n\n{sample_io_format}Important: You must follow the input output format. Input should be taken from standard input and output should be given to standard output.\nNote: If you are writing a function then after the function definition take input from using `input()` function, call the function with specified parameters and finally print the output of the function."

src/datasets/CodeContestDataset.py

@@ -0,0 +1,41 @@
+from .Dataset import Dataset
+from evaluations.evalute import contest_evaluate, contest_evaluate_public_tests
+from constants.paths import *
+
+class CodeContestDataset(Dataset):
+    def __init__(
+        self,
+        path: str=CODE_CONTEST_DATA_PATH,
+    ):
+        super().__init__(path)
+        self.id_key = "id"
+
+    def evaluate(
+        self,
+        item: dict,
+        cur_imp: str,
+        language: str,
+    ):
+        return contest_evaluate(
+            generated_code=cur_imp,
+            id=item["id"],
+            tests=item["test_list"],
+            lang=language
+        )
+    
+    def evaluate_sample_io(
+        self,
+        item: dict,
+        cur_imp: str,
+        language: str,
+    ):
+        return contest_evaluate_public_tests(
+            generated_code=cur_imp,
+            id=item["id"],
+            tests=item["sample_io"],
+            lang=language
+        )
+
+    @staticmethod
+    def get_prompt(item):
+        return f"{item['description']}\n\n-------\nImportant Note: You must follow the input output format. Input must be taken from standard input and output must be given to standard output. The code will be tested against multiple test cases and all the test cases must be passed."

src/datasets/Dataset.py

@@ -0,0 +1,33 @@
+from utils.jsonl import read_jsonl
+
+
+class Dataset(object):
+    def __init__(
+        self,
+        path: str,
+    ):
+        self.path = path
+        self.data = None
+        self.id_key = ""
+        self.load()
+
+    def load(self):
+        self.data = read_jsonl(self.path)
+
+    def __len__(self):
+        return len(self.data)
+
+    def __getitem__(self, idx):
+        return self.data[idx]
+
+    def evaluate(
+        self,
+        item: dict,
+        cur_imp: str,
+        language: str,
+    ):
+        raise NotImplementedError
+
+    @staticmethod
+    def get_prompt(item):
+        raise NotImplementedError

src/datasets/DatasetFactory.py

@@ -0,0 +1,25 @@
+from datasets.Dataset import Dataset
+from datasets.MBPPDataset import MBPPDataset
+from datasets.APPSDataset import APPSDataset
+from datasets.XCodeDataset import XCodeDataset
+from datasets.HumanEvalDataset import HumanDataset
+from datasets.CodeContestDataset import CodeContestDataset
+
+
+class DatasetFactory:
+    @staticmethod
+    def get_dataset_class(dataset_name):
+        if dataset_name == "APPS":
+            return APPSDataset
+        elif dataset_name == "MBPP":
+            return MBPPDataset
+        elif dataset_name == "XCode":
+            return XCodeDataset
+        elif dataset_name == "HumanEval":
+            return HumanDataset
+        elif dataset_name == "Human":
+            return HumanDataset
+        elif dataset_name == "CC":
+            return CodeContestDataset
+        else:
+            raise Exception(f"Unknown dataset name {dataset_name}")

src/datasets/HumanEvalDataset.py

@@ -0,0 +1,46 @@
+from .Dataset import Dataset
+from evaluations.func_evaluate import evaluate_functional_correctness, evaluate_io
+from constants.paths import *
+
+
+class HumanDataset(Dataset):
+    def __init__(
+        self,
+        path: str = HUMAN_WST_DATA_PATH,
+    ):
+        super().__init__(path)
+        self.id_key = "task_id"
+
+    def evaluate(
+        self,
+        item: dict,
+        cur_imp: str,
+        language: str,
+    ):
+        result = evaluate_functional_correctness(
+            problem=item,
+            completion=cur_imp
+        )
+        return result == "passed"
+
+    def evaluate_sample_io(
+        self,
+        item: dict,
+        cur_imp: str,
+        language: str,
+    ):
+        
+        return evaluate_io(
+            sample_io=item["sample_io"],
+            completion=cur_imp,            
+        )
+
+
+    @staticmethod
+    def get_prompt(item):
+        if "prompt" in item:
+            return f"{item['prompt']}"
+        elif "text" in item:
+            return f"{item['text']}"
+        else:
+            raise Exception("No prompt or text in item")

src/datasets/MBPPDataset.py

@@ -0,0 +1,47 @@
+from .Dataset import Dataset
+from evaluations.func_evaluate import evaluate_io, evaluate_functional_correctness
+from constants.paths import *
+
+
+class MBPPDataset(Dataset):
+    def __init__(
+        self,
+        path: str = MBPP_DATA_PATH,
+    ):
+        super().__init__(path)
+        self.id_key = "name"
+
+    def evaluate(
+        self,
+        item: dict,
+        cur_imp: str,
+        language: str,
+    ):
+        # result, _ = evaluate_io(item['test_list'],cur_imp,5,True)
+        # return result
+        result = evaluate_functional_correctness(
+            problem=item,
+            completion=cur_imp
+        )
+        return result == "passed"
+
+    def evaluate_sample_io(
+        self,
+        item: dict,
+        cur_imp: str,
+        language: str,
+    ):
+        if "sample_io" not in item:
+            return True, ""
+        if len(item["sample_io"]) == 0:
+            return True, ""
+        return evaluate_io(
+            sample_io=item["sample_io"],
+            completion=cur_imp,
+        )
+
+    @staticmethod
+    def get_prompt(item):
+        # function_signature = item['code'].split('\n')[0].strip()
+        # return f"{item['text']}\nFunction Signature: {function_signature}"
+        return item["prompt"]

src/datasets/XCodeDataset.py

@@ -0,0 +1,50 @@
+from .Dataset import Dataset
+from evaluations.evalute import xcode_evaluate, contest_evaluate_public_tests
+from constants.paths import *
+
+
+class XCodeDataset(Dataset):
+    def __init__(
+        self,
+        path: str = XCODE_VALIDATION_DATA_PATH,
+    ):
+        super().__init__(path)
+        self.id_key = "src_uid"
+
+    def evaluate_sample_io(
+        self,
+        item: dict,
+        cur_imp: str,
+        language: str,
+    ):
+        sample_io = []
+
+        for input, output in zip(item["sample_inputs"], item["sample_outputs"]):
+            sample_io.append({
+                "input": input,
+                "output": [output]
+            })
+
+        return contest_evaluate_public_tests(
+            generated_code=cur_imp,
+            id=item[self.id_key],
+            tests=sample_io,
+            lang=language
+        )
+
+
+    def evaluate(
+        self,
+        item: dict,
+        cur_imp: str,
+        language: str,
+    ):
+        return xcode_evaluate(
+            generated_code=cur_imp,
+            src_uid=item["src_uid"],
+            lang=language
+        )
+
+    @staticmethod
+    def get_prompt(item):
+        return f"Problem Description:\n{item['description']}\nInput Specification:\n{item['input_spec']}\nOutput Specification:\n{item['output_spec']}\nSample Inputs: {item['sample_inputs']}\nSample Outputs: {item['sample_outputs']}\nNote: {item['notes']}\nTake input from: {item['input_from']}\nGive output to: {item['output_to']}\nTime Limit: {item['time_limit']}\nMemory Limit: {item['memory_limit']}\n\nNote: If you are writing a function then after the function definition take input from using `input()` function, call the function with specified parameters and finally print the output of the function."

src/datasets/convert-apps-xcode.py

@@ -0,0 +1,129 @@
+import pandas as pd
+import json
+
+
+def read_jsonl(filename):
+    """Reads a jsonl file and yields each line as a dictionary"""
+    lines = []
+    # i = 0
+    with open(filename, "r", encoding="utf-8") as file:
+        for line in file:
+            lines.append(json.loads(line))
+            # i += 1
+            # print(i)
+    return lines
+
+# Write a python list of dictionaries into a jsonl file
+
+
+def write_jsonl(filename, lines):
+    """Writes a python list of dictionaries into a jsonl file"""
+    with open(filename, "w", encoding="utf-8") as file:
+        for line in lines:
+            file.write(json.dumps(line) + "\n")
+
+
+train_set = read_jsonl("./data/APPS/train.jsonl")
+test_set = read_jsonl("./data/APPS/train.jsonl")
+
+dataset = train_set + test_set
+
+print(len(dataset))
+
+dataset = pd.DataFrame(dataset)
+# dataset.columns
+
+print(dataset['difficulty'].unique())
+
+
+# Filter problems from codeforces with atleast 10 input and output
+filter_indices = [False] * len(dataset)
+for i in range(len(dataset)):
+    row = dataset.iloc[i]
+    if "codeforces" in row['url'] and row['input_output'] and len(json.loads(row['input_output'])["inputs"]) > 5:
+        filter_indices[i] = True
+
+codeforces_dataset = dataset[filter_indices]
+
+print(len(codeforces_dataset))
+
+# Randomly choose 50 problems
+codeforces_dataset_50 = codeforces_dataset.sample(n=min(50, len(codeforces_dataset)), random_state=1, replace=False)
+print(len(codeforces_dataset_50))
+
+codeforces_dataset_50.reset_index(drop=True, inplace=True)
+
+# Filter interview problems with atleast 10 input and output
+filter_indices = [False] * len(dataset)
+for i in range(len(dataset)):
+    row = dataset.iloc[i]
+    if "interview" == row['difficulty'] and row['input_output'] and len(row['input_output']) < 2000 and len(json.loads(row['input_output'])["inputs"]) > 5:
+        filter_indices[i] = True
+
+interview_dataset = dataset[filter_indices]
+
+print(len(interview_dataset))
+
+# Randomly choose 50 problems
+interview_dataset_50 = interview_dataset.sample(
+    n=min(50, len(interview_dataset)), random_state=1, replace=False)
+print(len(interview_dataset_50))
+
+interview_dataset_50.reset_index(drop=True, inplace=True)
+
+
+# Filter introductory problems with atleast 10 input and output
+filter_indices = [False] * len(dataset)
+for i in range(len(dataset)):
+    row = dataset.iloc[i]
+    if "introductory" == row['difficulty'] and len(row['input_output']) < 2000 and len(json.loads(row['input_output'])["inputs"]) > 5:
+        filter_indices[i] = True
+
+introductory_dataset = dataset[filter_indices]
+
+print(len(introductory_dataset))
+
+# Randomly choose 50 problems
+introductory_dataset_50 = introductory_dataset.sample(
+    n=min(50, len(introductory_dataset)), random_state=1, replace=False)
+print(len(introductory_dataset_50))
+
+introductory_dataset_50.reset_index(drop=True, inplace=True)
+
+selected_df = pd.concat([introductory_dataset_50, interview_dataset_50, codeforces_dataset_50], ignore_index=True)
+
+
+def get_test_cases(input, output):
+    return {
+        "input": "\n".join([str(x) for x in input]) if type(input) == list else input,
+        "output": output if type(output) == list else [output]
+    }
+
+
+selected_datasets = []
+
+for i in range(len(selected_df)):
+    row = selected_df.iloc[i]
+    test_cases = json.loads(row['input_output'])
+
+    public_test_cases = list(
+        map(get_test_cases, test_cases['inputs'][0:2], test_cases['outputs'][0:2]))
+    test_cases = list(
+        map(get_test_cases, test_cases['inputs'], test_cases['outputs']))
+
+    test = {
+        "name": str(row['id']),
+        "description": str(row['question']),
+        "difficulty": str(row['difficulty']),
+        "id": int(row['id']),
+        "sample_io": public_test_cases,
+        "test_list": test_cases,
+        "starter_code": str(row['starter_code']),
+    }
+
+    selected_datasets.append(test)
+
+
+write_jsonl("./data/APPS/selected150.jsonl", selected_datasets)
+
+

src/datasets/convert-cc-xcode.py

@@ -0,0 +1,66 @@
+# Using this python file we have converted the code contest dataset to the format of the xCodeEval dataset.
+
+import pandas as pd
+import json
+
+
+def read_jsonl(filename):
+    """Reads a jsonl file and yields each line as a dictionary"""
+    lines = []
+    # i = 0
+    with open(filename, "r", encoding="utf-8") as file:
+        for line in file:
+            lines.append(json.loads(line))
+            # i += 1
+            # print(i)
+    return lines
+
+# Write a python list of dictionaries into a jsonl file
+
+
+def write_jsonl(filename, lines):
+    """Writes a python list of dictionaries into a jsonl file"""
+    with open(filename, "w", encoding="utf-8") as file:
+        for line in lines:
+            file.write(json.dumps(line) + "\n")
+
+
+df = pd.read_parquet("./data/CodeContest/validation.parquet", engine='pyarrow')
+df = df[['name', 'cf_contest_id', 'cf_tags', 'difficulty',
+         'description', 'public_tests', 'private_tests', 'generated_tests']]
+
+
+def get_test_cases(input, output):
+    return {
+        "input": str(input),
+        "output": [str(output)]
+    }
+
+
+test_datasets = []
+
+for i in range(len(df)):
+    row = df.iloc[i]
+
+    public_test_cases = list(
+        map(get_test_cases, row['public_tests']['input'], row['public_tests']['output']))
+    test_cases = []
+    test_cases.extend(list(map(
+        get_test_cases, row['private_tests']['input'], row['private_tests']['output'])))
+    test_cases.extend(list(map(
+        get_test_cases, row['generated_tests']['input'], row['generated_tests']['output'])))
+
+    test = {
+        "name": str(row['name']),
+        "description": str(row['description']),
+        "tags": list(row['cf_tags']),
+        "difficulty": int(row['difficulty']),
+        "id": int(row['cf_contest_id']),
+        "sample_io": public_test_cases,
+        "test_list": test_cases
+    }
+
+    test_datasets.append(test)
+
+
+write_jsonl("./data/CodeContest/Val.jsonl", test_datasets)

src/evaluate-et-dataset.py

@@ -0,0 +1,104 @@
+from utils.jsonl import read_jsonl, write_jsonl
+from evaluations.func_evaluate import evaluate_io_et
+import os
+
+
+def generate_et_dataset(
+        NORMAL_RESULTS_PATH,
+        ET_RESULTS_PATH,
+        ET_DATA_PATH=".\\data\\HumanEval\\HumanEvalET.jsonl"
+):
+    dataset = read_jsonl(ET_DATA_PATH)
+    data_dict = {}
+    for item in dataset:
+        data_dict[item["task_id"]] = {"et_item": item}
+
+    results = read_jsonl(NORMAL_RESULTS_PATH)
+    for result in results:
+        data_dict[result["task_id"]]["result"] = result
+
+    correct_count = 0
+    et_results = []
+    for key, value in data_dict.items():
+        item = value["et_item"]
+        result = value["result"]
+        generated_code = result["source_codes"][0] if "source_codes" in result else result["solution"]
+
+        passed = evaluate_io_et(
+            item['test_case_list'],
+            generated_code,
+            prompt=item["prompt"]
+        )
+
+        if passed:
+            result["is_solved"] = True
+            correct_count += 1
+        else:
+            result["is_solved"] = False
+
+        et_results.append(result)
+        print(
+            f"Accuracy: {correct_count}/{len(et_results)} = {correct_count/len(et_results):.2f}")
+    # write_jsonl(ET_RESULTS_PATH, et_results)
+
+    et_results = sorted(
+        et_results,
+        key=lambda x: int(x["task_id"].split('/')[-1])
+    )
+
+    write_jsonl(ET_RESULTS_PATH, et_results)
+    print(
+        f"Accuracy: {correct_count}/{len(et_results)} = {correct_count/len(et_results):.2f}")
+
+
+def generate_et_dataset_mbpp(
+        NORMAL_RESULTS_PATH,
+        ET_RESULTS_PATH,
+        ET_DATA_PATH=".\\data\\MBPPEval\\MBPP_ET.jsonl"
+):
+    dataset = read_jsonl(ET_DATA_PATH)
+    data_dict = {}
+    for item in dataset:
+        data_dict[item["task_id"]] = {"et_item": item}
+
+    results = read_jsonl(NORMAL_RESULTS_PATH)
+    for result in results:
+        task_id = int(result["name"].split("_")[1])
+        data_dict[task_id]["result"] = result
+
+    correct_count = 0
+    et_results = []
+    for key, value in data_dict.items():
+        item = value["et_item"]
+        result = value.get("result", None)
+        if result is None:
+            continue
+
+        generated_code = result["source_codes"][0] if "source_codes" in result else result["solution"]
+
+        passed = evaluate_io_et(
+            item['test_list'],
+            generated_code
+        )
+
+        if passed:
+            result["is_solved"] = True
+            correct_count += 1
+        else:
+            result["is_solved"] = False
+
+        et_results.append(result)
+        print(
+            f"Accuracy: {correct_count}/{len(et_results)} = {correct_count/len(et_results):.2f}")
+    # write_jsonl(ET_RESULTS_PATH, et_results)
+
+    et_results = sorted(
+        et_results,
+        key=lambda x: int(x["name"].split("_")[1])
+    )
+
+    write_jsonl(ET_RESULTS_PATH, et_results)
+    print(
+        f"Accuracy: {correct_count}/{len(et_results)} = {correct_count/len(et_results):.2f}")
+
+

src/evaluations/api_comm.py

@@ -0,0 +1,115 @@
+from dataclasses import dataclass, field
+
+import requests
+from .exec_outcome import ExecOutcome
+
+@dataclass
+class ExtendedUnittest:
+    input: str
+    output: list[str] = field(default_factory=list)
+    result: str | None = None
+    exec_outcome: ExecOutcome | None = None
+
+    def json(self):
+        _json = self.__dict__
+        if self.exec_outcome is not None:
+            _json["exec_outcome"] = self.exec_outcome.name
+
+        return _json
+
+    @classmethod
+    def from_json(cls, _json):
+        return cls(
+            input=_json.get("input", ""),
+            output=_json.get("output", list()),
+            result=_json.get("result", None),
+            exec_outcome=_json.get("exec_outcome", None),
+        )
+
+
+class EmptyValueError(Exception):
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+
+
+class EmptyUnittestError(EmptyValueError):
+    pass
+
+
+class EmptyLanguageError(EmptyValueError):
+    pass
+
+
+class EmptySourceCodeError(EmptyValueError):
+    pass
+
+
+class APICommunication:
+    _session: requests.Session
+
+    def __init__(self, server_url: str = "http://localhost:5000"):
+        self._session = requests.Session()
+        self.execute_code_url = f"{server_url}/api/execute_code"
+        self.get_runtimes_url = f"{server_url}/api/all_runtimes"
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *args):
+        self._session.close()
+
+    def get_runtimes(self):
+        return self._session.get(self.get_runtimes_url).json()
+
+    def execute_code(
+        self,
+        language: str,
+        source_code: str,
+        unittests: list[dict],
+        limits: dict | None,
+        block_network: bool = True,
+        stop_on_first_fail: bool = True,
+        use_sanitizer: bool = False,
+        compiler_program_name: str | None = None,
+        compiler_flags: str | None = None,
+        interpreter_cmd: str | None = None,
+        interpreter_flags: str | None = None,
+        sample_id: int | None = None,
+        task_id: str | int | None = None,
+    ) -> tuple[list[ExtendedUnittest], int | None, str | int | None]:
+        if language is None:
+            raise EmptyLanguageError
+
+        if source_code is None:
+            raise EmptySourceCodeError
+
+        if unittests is None or len(unittests) == 0:
+            raise EmptyUnittestError
+
+        request_body = dict(
+            language=language,
+            source_code=source_code,
+            unittests=unittests,
+            limits=limits if isinstance(limits, dict) else dict(),
+            compile_cmd=compiler_program_name,
+            compile_flags=compiler_flags,
+            execute_cmd=interpreter_cmd,
+            execute_flags=interpreter_flags,
+            block_network=block_network,
+            stop_on_first_fail=stop_on_first_fail,
+            use_sanitizer=use_sanitizer,
+        )
+        json_response = self._session.post(
+            self.execute_code_url,
+            json=request_body,
+            headers={"Content-Type": "application/json"},
+        ).json()
+
+        if "data" not in json_response:
+            return "error", sample_id, task_id
+
+        return (
+            json_response["data"],
+            sample_id,
+            task_id,
+        )

src/evaluations/evalute.py

@@ -0,0 +1,168 @@
+import json
+import os
+import numpy as np
+import tqdm
+from yaml import safe_load
+from typing import List
+
+from .api_comm import APICommunication
+from .exec_outcome import ExecOutcome
+from constants.lang_mappings import LANGUAGE_MAPPING
+
+limits_by_lang_cfg_file = "./src/evaluations/limits_by_lang.yaml"
+
+assert os.path.exists(
+    limits_by_lang_cfg_file), "Need resource limit defaults for all runtimes, provide the path to default 'limits_by_lang.yaml' or to the modified one."
+
+with open(limits_by_lang_cfg_file) as limit_cfg_rp:
+    limits_by_lang = safe_load(limit_cfg_rp)
+
+unittest_file = "./data/xCodeEval/unittest_db.json"
+assert os.path.exists(unittest_file), "Unittest file not found."
+
+with open(unittest_file) as ut_rp:
+    unittest_db = json.load(ut_rp)
+
+
+api_comm = APICommunication()
+
+
+def xcode_evaluate(
+    generated_code: str,
+    src_uid: str,
+    lang: str
+):
+
+    assert src_uid in unittest_db, "Can not find the task id or source id"
+
+    assert lang in LANGUAGE_MAPPING, f"language must be inside the supported language list: {LANGUAGE_MAPPING.keys()}"
+
+    results, _, _ = api_comm.execute_code(
+        language=LANGUAGE_MAPPING[lang],
+        source_code=generated_code,
+        unittests=unittest_db[src_uid],
+        limits=limits_by_lang[LANGUAGE_MAPPING[lang]],
+        task_id=src_uid,
+    )
+
+    if results == "error":
+        return False
+
+    passed = True
+    for result in results:
+        if result['exec_outcome'] != ExecOutcome.PASSED.value:
+            passed = False
+            break
+
+    return passed
+
+
+def xcode_execute_internal_test(
+    generated_code: str,
+    tests: List[dict],
+    src_uid: str,
+    lang: str
+):
+    results, _, _ = api_comm.execute_code(
+        language=LANGUAGE_MAPPING[lang],
+        source_code=generated_code,
+        unittests=tests,
+        limits=limits_by_lang[LANGUAGE_MAPPING[lang]],
+        task_id=src_uid,
+        stop_on_first_fail=False
+    )
+
+    passed = True
+    passed_feedback = []
+    failed_feedback = []
+
+    idx = 0
+    try:
+        for idx, result in enumerate(results):
+            if result['exec_outcome'] == ExecOutcome.PASSED.value:
+                passed_feedback.append(tests[idx])
+            if result['exec_outcome'] != ExecOutcome.PASSED.value:
+                failed_feedback.append(tests[idx])
+                passed = False
+    except:
+        passed = False
+        failed_feedback.extend(tests[idx:])
+
+    feedback = f'Tested passed: \n{json.dumps(passed_feedback)}\n\nTests failed: \n{json.dumps(failed_feedback)}'
+
+    return passed, feedback
+
+
+def contest_evaluate(
+    generated_code: str,
+    lang: str,
+    id: int,
+    tests: List[dict],
+):
+    assert lang in LANGUAGE_MAPPING, f"language must be inside the supported language list: {LANGUAGE_MAPPING.keys()}"
+
+    results, _, _ = api_comm.execute_code(
+        language=LANGUAGE_MAPPING[lang],
+        source_code=generated_code,
+        unittests=tests,
+        limits=limits_by_lang[LANGUAGE_MAPPING[lang]],
+        task_id=id,
+    )
+
+    if results == "error":
+        return False
+
+    passed = True
+    for result in results:
+        if result['exec_outcome'] != ExecOutcome.PASSED.value:
+            passed = False
+            break
+
+    return passed
+
+
+def contest_evaluate_public_tests(
+    generated_code: str,
+    lang: str,
+    id: int,
+    tests: List[dict],
+):
+    results, _, _ = api_comm.execute_code(
+        language=LANGUAGE_MAPPING[lang],
+        source_code=generated_code,
+        unittests=tests,
+        limits=limits_by_lang[LANGUAGE_MAPPING[lang]],
+        task_id=id,
+        stop_on_first_fail=False
+    )
+
+    passed = True
+    passed_feedback = []
+    failed_feedback = []
+
+    idx = 0
+    try:
+        for idx, result in enumerate(results):
+            output = str(result['result'])
+            if len(output) > 500:
+                output = output[:500] + "..."
+            test_case = f"Input:\n{tests[idx]['input']}\nExpected Output:\n{tests[idx]['output'][0]}\nYour Output:\n{output}\n"
+            if result['exec_outcome'] == ExecOutcome.PASSED.value:
+                passed_feedback.append(test_case)
+            if result['exec_outcome'] != ExecOutcome.PASSED.value:
+                failed_feedback.append(test_case)
+                passed = False
+    except:
+        passed = False
+        test_cases = []
+        for i in range(idx, len(tests)):
+            test_case = f"Input:\n{tests[i]['input']}\nExpected Output:\n{tests[i]['output'][0]}\n"
+            test_cases.append(test_case)
+        
+        failed_feedback.extend(test_cases)
+
+    passed_feedback = '\n'.join(passed_feedback) if len(passed_feedback) > 0 else "No test cases passed."
+    failed_feedback = '\n'.join(failed_feedback)
+    feedback = f'## Tested passed:\n{passed_feedback}\n\n## Tests failed:\n{failed_feedback}'
+
+    return passed, feedback

src/evaluations/exec_outcome.py

@@ -0,0 +1,14 @@
+from enum import Enum
+
+
+class ExecOutcome(Enum):
+    PASSED = "PASSED"  # code executes and output matches expected output
+    WRONG_ANSWER = (
+        "WRONG_ANSWER"  # code executes and output does NOT matches expected output
+    )
+    TIME_LIMIT_EXCEEDED = "TIME_LIMIT_EXCEEDED"  # code executes and didn't exit in time, output is ignored in this case
+    RUNTIME_ERROR = "RUNTIME_ERROR"  # code failed to execute (crashed)
+    COMPILATION_ERROR = "COMPILATION_ERROR"  # code failed to compile
+    MEMORY_LIMIT_EXCEEDED = (
+        "MEMORY_LIMIT_EXCEEDED"  # code exceeded memory limit during execution
+    )

src/evaluations/executor_utils.py

@@ -0,0 +1,58 @@
+
+def timeout_handler(_, __):
+    raise TimeoutError()
+
+import os, json
+def to_jsonl(dict_data, file_path):
+    with open(file_path, 'a') as file:
+        json_line = json.dumps(dict_data)
+        file.write(json_line + os.linesep)
+
+from threading import Thread
+class PropagatingThread(Thread):
+    def run(self):
+        self.exc = None
+        try:
+            if hasattr(self, '_Thread__target'):
+                # Thread uses name mangling prior to Python 3.
+                self.ret = self._Thread__target(*self._Thread__args, **self._Thread__kwargs)
+            else:
+                self.ret = self._target(*self._args, **self._kwargs)
+        except BaseException as e:
+            self.exc = e
+
+    def join(self, timeout=None):
+        super(PropagatingThread, self).join(timeout)
+        if self.exc:
+            raise self.exc
+        return self.ret
+    
+
+def function_with_timeout(func, args, timeout):
+    result_container = []
+
+    def wrapper():
+        result_container.append(func(*args))
+
+    thread = PropagatingThread(target=wrapper)
+    thread.start()
+    thread.join(timeout)
+
+    if thread.is_alive():
+        raise TimeoutError()
+    else:
+        return result_container[0]
+    
+# Py tests
+
+# if __name__ == "__main__":
+#     formatter = PySubmissionFormatter()
+#     leetcode_1 = 'class Solution:\n    def solveSudoku(self, board: List[List[str]]) -> None:\n        """\n        Do not return anything, modify board in-place instead.\n        """\n        '
+#     humaneval_1 = 'def solveSudoku(self, board: List[List[str]]) -> None:\n        """\n        Do not return anything, modify board in-place instead.\n        """\n'
+
+#     assert leetcode_1 == formatter.to_leetcode(humaneval_1)
+#     assert humaneval_1 == formatter.to_humaneval(leetcode_1)
+
+
+
+