Files
Evaluation-Backend/public/challenge/svm/main.py
T
2024-11-13 17:15:18 +08:00

236 lines
7.5 KiB
Python

import importlib.util
import os
import requests
import subprocess
import time
import traceback
EXP_URL = "http://10.2.1.195:880"
PROJECT_NAME = "challenge_svm"
LOW_ACCURACY_THRESHOLD = 0.6
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_svm(
user_train_func,
user_test_func,
train_data,
train_labels,
test_data,
test_labels,
log_details,
):
"""
使用编写的训练 SVM 函数训练权重,并用预测函数对测试集进行分类,并计算准确率。
Args:
- user_train_func (function): 编写的训练 SVM 函数。
- 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 = 67 # SVM 测试样本数
temp_log_details = []
if len(test_data) != total:
log_details.append("测试集样本数不匹配,请检查 load_data 函数!")
print("测试集样本数不匹配,请检查 load_data 函数!")
log_details.append("成绩为零!")
print("成绩为零!")
return 0.0
if len(set(test_labels)) != 3:
log_details.append(
"本次 SVM 挑战赛要求不合并死亡与安乐死(即使用 SVM 完成三分类问题),请检查 load_data 函数!"
)
print(
"本次 SVM 挑战赛要求不合并死亡与安乐死(即使用 SVM 完成三分类问题),请检查 load_data 函数!"
)
log_details.append("成绩为零!")
print("成绩为零!")
return 0.0
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(len(test_data)):
# 获取测试数据中的每个样本
result = user_test_func(test_data[i], wegihts)
if result == test_labels[i] and test_labels[i] != -1:
correct += 1
else:
print(f"第 {i+1} 个数据被分类到 {result},但实际是 {test_labels[i]}")
temp_log_details.append(f"{i+1}")
accuracy = correct / total
if accuracy >= LOW_ACCURACY_THRESHOLD:
log_details.append(f"以下测试次序的数据分类错误:")
log_details.append(", ".join(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_svm" # 训练 svm 函数名
user_test_function_name = "predict_svm" # 预测函数名
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_data 函数是否正确!")
print("本次评测无效,记录不做提交!")
return
load_end_time = time.time()
# 记录评估时间
start_time = time.time()
# 评估分类准确率
accuracy = evaluate_svm(
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))