306 lines
12 KiB
Python
306 lines
12 KiB
Python
"""
|
||
测试程序 - 验证Excel数据输入和储能容量优化
|
||
|
||
该程序使用data_template_24.xlsx和data_template_8760.xlsx作为输入，
|
||
测试储能容量优化系统的功能和错误处理。
|
||
|
||
作者: iFlow CLI
|
||
创建日期: 2025-12-25
|
||
"""
|
||
|
||
import sys
|
||
import os
|
||
sys.path.append(os.path.join(os.path.dirname(__file__), '..', 'src'))
|
||
|
||
import numpy as np
|
||
import pandas as pd
|
||
import matplotlib.pyplot as plt
|
||
from typing import Dict, Any
|
||
from excel_reader import read_excel_data, analyze_excel_data, read_system_parameters, validate_system_parameters
|
||
from storage_optimization import optimize_storage_capacity, SystemParameters
|
||
from advanced_visualization import create_comprehensive_plot, create_time_series_plot
|
||
|
||
|
||
def test_excel_file(file_path: str, test_name: str) -> Dict[str, Any]:
|
||
"""
|
||
测试单个Excel文件
|
||
|
||
Args:
|
||
file_path: Excel文件路径
|
||
test_name: 测试名称
|
||
|
||
Returns:
|
||
测试结果字典
|
||
"""
|
||
print(f"\n{'='*60}")
|
||
print(f"测试：{test_name}")
|
||
print(f"文件：{file_path}")
|
||
print(f"{'='*60}")
|
||
|
||
result = {
|
||
'test_name': test_name,
|
||
'file_path': file_path,
|
||
'success': False,
|
||
'error': None,
|
||
'data_stats': {},
|
||
'optimization_result': {}
|
||
}
|
||
|
||
try:
|
||
# 1. 检查文件是否存在
|
||
if not os.path.exists(file_path):
|
||
raise FileNotFoundError(f"文件不存在：{file_path}")
|
||
print("[OK] 文件存在检查通过")
|
||
|
||
# 2. 读取Excel数据
|
||
print("正在读取Excel数据...")
|
||
data = read_excel_data(file_path, include_parameters=True)
|
||
print(f"[OK] 数据读取成功，类型：{data['data_type']}")
|
||
print(f" 原始数据长度：{data['original_length']}")
|
||
print(f" 处理后数据长度：{len(data['solar_output'])}")
|
||
|
||
# 2.1 测试参数读取
|
||
if 'system_parameters' in data:
|
||
params = data['system_parameters']
|
||
print(f"[OK] 系统参数读取成功")
|
||
print(f" 最大弃风率: {params.max_curtailment_wind}")
|
||
print(f" 最大弃光率: {params.max_curtailment_solar}")
|
||
print(f" 最大上网电量比例: {params.max_grid_ratio}")
|
||
print(f" 储能效率: {params.storage_efficiency}")
|
||
print(f" 放电倍率: {params.discharge_rate}")
|
||
print(f" 充电倍率: {params.charge_rate}")
|
||
print(f" 最大储能容量: {params.max_storage_capacity}")
|
||
|
||
# 验证参数
|
||
validation = validate_system_parameters(params)
|
||
if validation['valid']:
|
||
print("[OK] 系统参数验证通过")
|
||
else:
|
||
print("[ERROR] 系统参数验证失败:")
|
||
for error in validation['errors']:
|
||
print(f" - {error}")
|
||
raise ValueError(f"系统参数验证失败: {validation['errors']}")
|
||
|
||
if validation['warnings']:
|
||
print("[WARNING] 系统参数警告:")
|
||
for warning in validation['warnings']:
|
||
print(f" - {warning}")
|
||
else:
|
||
print("[WARNING] 未找到系统参数，使用默认参数")
|
||
params = SystemParameters()
|
||
|
||
# 3. 分析数据统计信息
|
||
print("正在分析数据统计信息...")
|
||
stats = analyze_excel_data(file_path)
|
||
result['data_stats'] = stats
|
||
print("[OK] 数据统计分析完成")
|
||
|
||
# 4. 验证数据完整性
|
||
print("正在验证数据完整性...")
|
||
solar_output = data['solar_output']
|
||
wind_output = data['wind_output']
|
||
thermal_output = data['thermal_output']
|
||
load_demand = data['load_demand']
|
||
|
||
# 检查数据长度一致性
|
||
if not (len(solar_output) == len(wind_output) == len(thermal_output) == len(load_demand)):
|
||
raise ValueError("数据长度不一致")
|
||
print("[OK] 数据长度一致性检查通过")
|
||
|
||
# 检查非负值
|
||
for name, values in [
|
||
("光伏出力", solar_output), ("风电出力", wind_output),
|
||
("火电出力", thermal_output), ("负荷需求", load_demand)
|
||
]:
|
||
if any(v < 0 for v in values):
|
||
raise ValueError(f"{name}包含负值")
|
||
print("[OK] 非负值检查通过")
|
||
|
||
# 5. 储能容量优化测试
|
||
print("正在进行储能容量优化计算...")
|
||
|
||
# 使用Excel中的参数和不同的测试配置
|
||
excel_params = params # 从Excel中读取的参数
|
||
test_configs = [
|
||
{
|
||
'name': 'Excel配置',
|
||
'params': excel_params
|
||
},
|
||
{
|
||
'name': '基础配置',
|
||
'params': SystemParameters(
|
||
max_curtailment_wind=0.1,
|
||
max_curtailment_solar=0.1,
|
||
max_grid_ratio=0.2,
|
||
storage_efficiency=0.9,
|
||
discharge_rate=1.0,
|
||
charge_rate=1.0
|
||
)
|
||
},
|
||
{
|
||
'name': '高弃风弃光配置',
|
||
'params': SystemParameters(
|
||
max_curtailment_wind=0.2,
|
||
max_curtailment_solar=0.2,
|
||
max_grid_ratio=0.3,
|
||
storage_efficiency=0.85,
|
||
discharge_rate=1.5,
|
||
charge_rate=1.5
|
||
)
|
||
},
|
||
{
|
||
'name': '低储能效率配置',
|
||
'params': SystemParameters(
|
||
max_curtailment_wind=0.05,
|
||
max_curtailment_solar=0.05,
|
||
max_grid_ratio=0.1,
|
||
storage_efficiency=0.75,
|
||
discharge_rate=0.8,
|
||
charge_rate=0.8
|
||
)
|
||
}
|
||
]
|
||
|
||
optimization_results = []
|
||
|
||
for config in test_configs:
|
||
print(f" 测试配置：{config['name']}")
|
||
opt_result = optimize_storage_capacity(
|
||
solar_output, wind_output, thermal_output, load_demand, config['params']
|
||
)
|
||
|
||
print(f" 所需储能容量：{opt_result['required_storage_capacity']:.2f} MWh")
|
||
print(f" 弃风率：{opt_result['total_curtailment_wind_ratio']:.3f}")
|
||
print(f" 弃光率：{opt_result['total_curtailment_solar_ratio']:.3f}")
|
||
print(f" 上网电量比例：{opt_result['total_grid_feed_in_ratio']:.3f}")
|
||
print(f" 能量平衡校验：{'通过' if opt_result['energy_balance_check'] else '未通过'}")
|
||
|
||
optimization_results.append({
|
||
'config_name': config['name'],
|
||
'result': opt_result
|
||
})
|
||
|
||
result['optimization_result'] = optimization_results
|
||
print("[OK] 储能容量优化计算完成")
|
||
|
||
# 6. 可视化测试（仅对24小时数据进行，避免8760小时数据生成过大图像）
|
||
if data['data_type'] == '24':
|
||
print("正在生成可视化图表...")
|
||
try:
|
||
# 使用基础配置的结果生成图表
|
||
base_result = optimization_results[0]['result']
|
||
base_params = test_configs[0]['params']
|
||
|
||
# 生成基础图表（截取前24小时数据）
|
||
solar_24 = solar_output[:24] if len(solar_output) > 24 else solar_output
|
||
wind_24 = wind_output[:24] if len(wind_output) > 24 else wind_output
|
||
thermal_24 = thermal_output[:24] if len(thermal_output) > 24 else thermal_output
|
||
load_24 = load_demand[:24] if len(load_demand) > 24 else load_demand
|
||
|
||
# 截取结果的24小时数据
|
||
result_24 = {
|
||
'required_storage_capacity': base_result['required_storage_capacity'],
|
||
'charge_profile': base_result['charge_profile'][:24] if len(base_result['charge_profile']) > 24 else base_result['charge_profile'],
|
||
'discharge_profile': base_result['discharge_profile'][:24] if len(base_result['discharge_profile']) > 24 else base_result['discharge_profile'],
|
||
'curtailed_wind': base_result['curtailed_wind'][:24] if len(base_result['curtailed_wind']) > 24 else base_result['curtailed_wind'],
|
||
'curtailed_solar': base_result['curtailed_solar'][:24] if len(base_result['curtailed_solar']) > 24 else base_result['curtailed_solar'],
|
||
'grid_feed_in': base_result['grid_feed_in'][:24] if len(base_result['grid_feed_in']) > 24 else base_result['grid_feed_in'],
|
||
'storage_profile': base_result['storage_profile'][:24] if len(base_result['storage_profile']) > 24 else base_result['storage_profile'],
|
||
'total_curtailment_wind_ratio': base_result['total_curtailment_wind_ratio'],
|
||
'total_curtailment_solar_ratio': base_result['total_curtailment_solar_ratio'],
|
||
'total_grid_feed_in_ratio': base_result['total_grid_feed_in_ratio'],
|
||
'energy_balance_check': base_result['energy_balance_check']
|
||
}
|
||
|
||
create_comprehensive_plot(solar_24, wind_24, thermal_24, load_24, result_24, base_params)
|
||
create_time_series_plot(solar_24, wind_24, thermal_24, load_24, result_24)
|
||
print("[OK] 可视化图表生成完成")
|
||
|
||
except Exception as e:
|
||
print(f"⚠ 可视化图表生成失败：{str(e)}")
|
||
print(" 这不是严重错误，可能是字体或显示问题")
|
||
|
||
# 标记测试成功
|
||
result['success'] = True
|
||
print(f"\n[OK] 测试 '{test_name}' 成功完成！")
|
||
|
||
except Exception as e:
|
||
result['error'] = str(e)
|
||
result['traceback'] = traceback.format_exc()
|
||
print(f"\n[ERROR] 测试 '{test_name}' 失败：{str(e)}")
|
||
print("详细错误信息：")
|
||
traceback.print_exc()
|
||
|
||
return result
|
||
|
||
|
||
def run_comprehensive_tests():
|
||
"""运行综合测试"""
|
||
print("开始运行Excel数据输入综合测试...")
|
||
print(f"当前工作目录：{os.getcwd()}")
|
||
|
||
# 测试文件列表
|
||
test_files = [
|
||
{
|
||
'path': 'data_template_24.xlsx',
|
||
'name': '24小时数据模板测试'
|
||
},
|
||
{
|
||
'path': 'data_template_8760.xlsx',
|
||
'name': '8760小时数据模板测试'
|
||
}
|
||
]
|
||
|
||
# 运行所有测试
|
||
results = []
|
||
for test_file in test_files:
|
||
result = test_excel_file(test_file['path'], test_file['name'])
|
||
results.append(result)
|
||
|
||
# 生成测试报告
|
||
print(f"\n{'='*80}")
|
||
print("测试报告总结")
|
||
print(f"{'='*80}")
|
||
|
||
total_tests = len(results)
|
||
successful_tests = sum(1 for r in results if r['success'])
|
||
failed_tests = total_tests - successful_tests
|
||
|
||
print(f"总测试数：{total_tests}")
|
||
print(f"成功测试：{successful_tests}")
|
||
print(f"失败测试：{failed_tests}")
|
||
print(f"成功率：{successful_tests/total_tests*100:.1f}%")
|
||
|
||
print("\n详细结果：")
|
||
for i, result in enumerate(results, 1):
|
||
status = "[OK] 成功" if result['success'] else "[ERROR] 失败"
|
||
print(f"{i}. {result['test_name']}: {status}")
|
||
if not result['success']:
|
||
print(f" 错误：{result['error']}")
|
||
|
||
# 如果有失败的测试，返回非零退出码
|
||
if failed_tests > 0:
|
||
print(f"\n有{failed_tests}个测试失败，请检查上述错误信息")
|
||
return False
|
||
else:
|
||
print("\n所有测试通过！系统运行正常。")
|
||
return True
|
||
|
||
|
||
def main():
|
||
"""主函数"""
|
||
try:
|
||
success = run_comprehensive_tests()
|
||
sys.exit(0 if success else 1)
|
||
except KeyboardInterrupt:
|
||
print("\n测试被用户中断")
|
||
sys.exit(1)
|
||
except Exception as e:
|
||
print(f"\n测试过程中发生未预期的错误：{str(e)}")
|
||
traceback.print_exc()
|
||
sys.exit(1)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|