Add initial project files for Soil Water Balance Model including Docker setup, deployment scripts, API implementation, and data files
This commit is contained in:
106
test_api.py
Normal file
106
test_api.py
Normal file
@@ -0,0 +1,106 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
测试Soil Water Balance Model API
|
||||
"""
|
||||
|
||||
import requests
|
||||
import json
|
||||
import pandas as pd
|
||||
import os
|
||||
|
||||
def test_health_check():
|
||||
"""测试健康检查端点"""
|
||||
try:
|
||||
response = requests.get("http://localhost:8000/health")
|
||||
if response.status_code == 200:
|
||||
print("✓ 健康检查通过")
|
||||
return True
|
||||
else:
|
||||
print("✗ 健康检查失败")
|
||||
return False
|
||||
except Exception as e:
|
||||
print(f"✗ 健康检查错误: {e}")
|
||||
return False
|
||||
|
||||
def load_test_data():
|
||||
"""加载测试数据"""
|
||||
base_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), os.pardir))
|
||||
|
||||
# 加载参数文件
|
||||
with open(os.path.join(base_dir, 'data', 'management.json')) as f:
|
||||
management = json.load(f)
|
||||
|
||||
with open(os.path.join(base_dir, 'data', 'SOILPROP.json')) as f:
|
||||
soil_params = json.load(f)
|
||||
|
||||
with open(os.path.join(base_dir, 'data', 'N_organ.json')) as f:
|
||||
n_params = json.load(f)
|
||||
|
||||
with open(os.path.join(base_dir, 'data', 'plant_para.json')) as f:
|
||||
plant_params = json.load(f)
|
||||
|
||||
with open(os.path.join(base_dir, 'data', 'para.json')) as f:
|
||||
cultivar_params = json.load(f)
|
||||
|
||||
# 加载气象数据
|
||||
weather_data = pd.read_csv(
|
||||
os.path.join(base_dir, 'data', '57278_2024.csv'),
|
||||
index_col='date',
|
||||
parse_dates=True
|
||||
)
|
||||
|
||||
# 转换为字典列表格式
|
||||
weather_list = []
|
||||
for date, row in weather_data.iterrows():
|
||||
weather_dict = {'date': date.strftime('%m/%d/%Y')}
|
||||
weather_dict.update(row.to_dict())
|
||||
weather_list.append(weather_dict)
|
||||
|
||||
return {
|
||||
"start_date": "10/9/2023",
|
||||
"end_date": "10/15/2023", # 缩短测试时间
|
||||
"cultivar_params": cultivar_params,
|
||||
"plant_params": plant_params,
|
||||
"soil_params": soil_params,
|
||||
"management": management,
|
||||
"n_params": n_params,
|
||||
"weather_data": weather_list
|
||||
}
|
||||
|
||||
def test_simulation():
|
||||
"""测试模拟计算"""
|
||||
try:
|
||||
test_data = load_test_data()
|
||||
|
||||
response = requests.post(
|
||||
"http://localhost:8000/simulate",
|
||||
json=test_data,
|
||||
headers={"Content-Type": "application/json"}
|
||||
)
|
||||
|
||||
if response.status_code == 200:
|
||||
result = response.json()
|
||||
print("✓ 模拟计算成功")
|
||||
print(f"结果包含 {len(result.get('results', {}))} 天的计算结果")
|
||||
return True
|
||||
else:
|
||||
print(f"✗ 模拟计算失败: {response.status_code}")
|
||||
print(response.text)
|
||||
return False
|
||||
except Exception as e:
|
||||
print(f"✗ 模拟计算错误: {e}")
|
||||
return False
|
||||
|
||||
if __name__ == "__main__":
|
||||
print("开始测试Soil Water Balance Model API...")
|
||||
|
||||
# 测试健康检查
|
||||
if not test_health_check():
|
||||
print("API服务未运行,请先启动API服务")
|
||||
exit(1)
|
||||
|
||||
# 测试模拟计算
|
||||
if test_simulation():
|
||||
print("所有测试通过!")
|
||||
else:
|
||||
print("测试失败")
|
||||
Reference in New Issue
Block a user