feat: upload logreg data and main.py for new experiment

2024-10-24 11:51:20 +08:00
parent 8ee9f4a545
commit 6454c8199e
3 changed files with 588 additions and 0 deletions
@@ -0,0 +1,220 @@
+import importlib.util
+import os
+import requests
+import subprocess
+import time
+import traceback
+
+EXP_URL = "http://10.2.1.195:880"
+PROJECT_NAME = "experiment_logreg"
+LOW_ACCURACY_THRESHOLD = 0.3
+
+
+def load_user_function(file_path, function_name):
+    """
+    动态加载函数。
+    """
+    spec = importlib.util.spec_from_file_location("user_code", file_path)
+    user_code = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(user_code)
+    return getattr(user_code, function_name)
+
+
+def get_commit_hash_url():
+    """
+    获取最新的提交哈希值
+    """
+    try:
+        commit_hash = (
+            subprocess.check_output(["git", "rev-parse", "HEAD"])
+            .strip()
+            .decode("utf-8")
+        )
+        path = os.getcwd().replace("/workspace/", "")
+        url = f"https://git.intcplab.cn/{path}/commit/{commit_hash}"
+        return url
+    except subprocess.CalledProcessError as e:
+        print(f"Error: {e}")
+        return None
+
+
+def evaluate_logreg(
+    user_train_func,
+    user_test_func,
+    train_data,
+    train_labels,
+    test_data,
+    test_labels,
+    log_details,
+):
+    """
+    使用编写的训练 LogReg 函数训练权重，并用预测函数对测试集进行分类，并计算准确率。
+
+    Args:
+    - user_train_func (function): 编写的训练 LogReg 函数。
+    - user_test_func (function): 编写的预测函数。
+    - train_data (numpy array): 训练数据集。
+    - train_labels (numpy array): 训练标签集。
+    - test_data (numpy array): 测试数据集。
+    - test_labels (numpy array): 测试标签集。
+    - log_details (list): 日志记录列表。
+
+    Returns:
+    - float: 分类准确率。
+    """
+    correct = 0
+    total = len(test_data)
+    temp_log_details = []
+
+    try:
+        # 训练模型
+        time_start = time.time()
+        wegihts = user_train_func(train_data, train_labels)
+        time_end = time.time()
+        log_details.append(f"训练时间: {(time_end - time_start)*1000:.10f}ms")
+        print(f"训练时间: {(time_end - time_start)*1000:.10f}ms")
+
+        for i in range(total):
+            # 获取测试数据中的每个样本
+            result = user_test_func(test_data[i], wegihts)
+            if result == test_labels[i]:
+                correct += 1
+            else:
+                temp_log_details.append(
+                    f"第 {i+1} 次测试错误，数据 {test_data[i]} 被分类到 {result} 实际是 {test_labels[i]}"
+                )
+                print(
+                    f"第 {i+1} 次测试错误，数据 {test_data[i]} 被分类到 {result} 实际是 {test_labels[i]}"
+                )
+
+        accuracy = correct / total
+        if accuracy >= LOW_ACCURACY_THRESHOLD:
+            log_details.extend(temp_log_details)
+        else:
+            log_details.append("分类准确率较低，成绩中不记录错误日志")
+        log_details.append(f"分类准确率: {accuracy * 100:.2f}%")
+        return accuracy
+    except Exception as e:
+        log_details.append(f"评测期间发生错误: {e}")
+        print(f"评测期间发生错误: {e}")
+        log_details.append(traceback.format_exc())
+        print(traceback.format_exc())
+        log_details.append("成绩为零！")
+        print("成绩为零！")
+        return 0.0
+
+
+def upload_score(name, id, score_data):
+    """
+    上传成绩到服务器。
+    """
+    url = f"{EXP_URL}/upload_score"
+    try:
+        print("上传成绩中...")
+        # print(score_data)
+        response = requests.post(
+            url,
+            json={
+                "id": id,
+                "name": name,
+                "score": score_data["score"],
+                "running_time": score_data["running_time"],
+                "time": time.strftime(
+                    "%Y-%m-%d %H:%M:%S", time.gmtime(time.time() + 8 * 3600)
+                ),
+                "details": score_data["details"],
+                "project": PROJECT_NAME,
+                "hash": get_commit_hash_url(),
+            },
+        )
+        if response.status_code == 200:
+            print(response.json()["message"])
+        else:
+            print("上传成绩失败!\n原因:", response.json()["message"])
+    except Exception as e:
+        print(f"上传成绩出错: {e}")
+
+
+def main(name, id):
+    user_code_file = "function.py"  # 代码文件
+    user_train_function_name = "train_logreg"  # 训练 LogReg 函数名
+    user_test_function_name = "predict_logreg"  # 预测函数名
+    user_load_dataset_name = "load_data"  # 加载数据集函数名
+
+    # 加载自定义函数
+    log_details = []
+    try:
+        user_load_dataset = load_user_function(user_code_file, user_load_dataset_name)
+        user_train_func = load_user_function(user_code_file, user_train_function_name)
+        user_test_func = load_user_function(user_code_file, user_test_function_name)
+    except Exception as e:
+        log_details.append(f"加载自定义函数时出错: {e}")
+        print("\n".join(log_details))
+        return
+
+    load_start_time = time.time()
+
+    # 加载数据集
+    try:
+        train_data, train_labels = user_load_dataset("horse-colic.data")
+        test_data, test_labels = user_load_dataset("horse-colic.test")
+    except Exception as e:
+        print(f"加载数据集时出错: {e}")
+        print("请检查 load_dataset 函数是否正确！")
+        print("本次评测无效，记录不做提交！")
+        return
+
+    load_end_time = time.time()
+
+    # 记录评估时间
+    start_time = time.time()
+
+    # 评估分类准确率
+    accuracy = evaluate_logreg(
+        user_train_func,
+        user_test_func,
+        train_data,
+        train_labels,
+        test_data,
+        test_labels,
+        log_details,
+    )
+
+    # 计算运行时间
+    running_time = time.time() - start_time
+
+    # 输出结果
+    print("")
+    log_details.append(f"得分: {accuracy * 100:.2f}")
+    print(f"得分: {accuracy * 100:.2f}")
+    log_details.append(
+        f"加载数据集时间: {(load_end_time - load_start_time)*1000:.10f}ms"
+    )
+    print(f"加载数据集时间: {(load_end_time - load_start_time)*1000:.10f}ms")
+    log_details.append(f"运行时间: {running_time * 1000:.10f}ms")
+    print(f"运行时间: {running_time * 1000:.10f}ms")
+
+    # 上传成绩
+    score_data = {
+        "score": accuracy * 100,
+        "running_time": f"{running_time * 1000:.10f}ms",
+        "details": "<br>".join(log_details),
+    }
+
+    upload_score(name, id, score_data)
+
+
+if __name__ == "__main__":
+    try:
+        from function import name, id
+
+        print(f"正在评测 {name}@{id} 的代码...\n")
+        main(name, id)
+    except Exception as e:
+        log_details = [
+            f"评测出错: {e}",
+            traceback.format_exc(),
+            "请检查代码和返回值类型是否正确！",
+            "本次评测无效，记录不做提交！",
+        ]
+        print("\n".join(log_details))