189 lines
5.8 KiB
Python
189 lines
5.8 KiB
Python
import importlib.util
|
|
import numpy as np
|
|
import requests
|
|
import time
|
|
import traceback
|
|
|
|
EXP_URL = "http://10.2.1.195:880"
|
|
PROJECT_NAME = "experiment_knn"
|
|
LOW_ACCURACY_THRESHOLD = 0.92
|
|
|
|
|
|
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 evaluate_knn(
|
|
user_func, train_data, train_labels, test_data, test_labels, log_details
|
|
):
|
|
"""
|
|
使用编写的 KNN 函数对测试集进行分类，并计算准确率。
|
|
|
|
Args:
|
|
- user_func (function): 编写的 KNN 函数。
|
|
- 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:
|
|
for i in range(total):
|
|
# 获取测试数据中的每个样本
|
|
result = user_func(test_data[i], train_data, train_labels)
|
|
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 load_data(file_path):
|
|
# """
|
|
# 加载 TSV 格式数据文件。
|
|
# """
|
|
# data = np.loadtxt(file_path, delimiter="\t")
|
|
# features = data[:, :-1]
|
|
# labels = data[:, -1].astype(int)
|
|
# return features, labels
|
|
|
|
|
|
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,
|
|
},
|
|
)
|
|
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_function_name = "classify" # KNN 函数名
|
|
user_load_dataset_name = "load_dataset" # 加载数据集函数名
|
|
|
|
# 加载自定义函数
|
|
log_details = []
|
|
try:
|
|
user_load_dataset = load_user_function(user_code_file, user_load_dataset_name)
|
|
user_func = load_user_function(user_code_file, user_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("trainingDigits")
|
|
test_data, test_labels = user_load_dataset("testDigits")
|
|
except Exception as e:
|
|
print(f"加载数据集时出错: {e}")
|
|
print("请检查 load_dataset 函数是否正确！")
|
|
print("本次评测无效，记录不做提交！")
|
|
return
|
|
|
|
load_end_time = time.time()
|
|
|
|
# 记录评估时间
|
|
start_time = time.time()
|
|
|
|
# 评估分类准确率
|
|
accuracy = evaluate_knn(
|
|
user_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))
|