267 lines
8.7 KiB
Python
267 lines
8.7 KiB
Python
"""
|
||
Excel数据读取模块
|
||
|
||
该模块提供从Excel文件中读取8760小时负荷和发电曲线数据的功能。
|
||
|
||
作者: iFlow CLI
|
||
创建日期: 2025-12-25
|
||
"""
|
||
|
||
import pandas as pd
|
||
import numpy as np
|
||
from typing import Dict, List, Optional, Tuple
|
||
import os
|
||
|
||
|
||
def validate_excel_data(df: pd.DataFrame, data_type: str = "8760") -> bool:
|
||
"""
|
||
验证Excel数据格式是否正确
|
||
|
||
Args:
|
||
df: pandas DataFrame对象
|
||
data_type: 数据类型，"24"或"8760"
|
||
|
||
Returns:
|
||
bool: 验证是否通过
|
||
"""
|
||
expected_length = 8760 if data_type == "8760" else 24
|
||
|
||
# 检查行数
|
||
if len(df) != expected_length:
|
||
print(f"错误：数据行数应为{expected_length}，实际为{len(df)}")
|
||
return False
|
||
|
||
# 检查必需的列
|
||
required_columns = ['光伏出力(MW)', '风电出力(MW)', '火电出力(MW)', '负荷需求(MW)']
|
||
missing_columns = [col for col in required_columns if col not in df.columns]
|
||
|
||
if missing_columns:
|
||
print(f"错误：缺少必需的列：{missing_columns}")
|
||
return False
|
||
|
||
# 检查数据类型和非负值
|
||
for col in required_columns:
|
||
if not pd.api.types.is_numeric_dtype(df[col]):
|
||
print(f"错误：列'{col}'必须为数值类型")
|
||
return False
|
||
|
||
if (df[col] < 0).any():
|
||
print(f"错误：列'{col}'包含负值")
|
||
return False
|
||
|
||
return True
|
||
|
||
|
||
def read_excel_data(file_path: str, sheet_name: str = 0) -> Dict[str, List[float]]:
|
||
"""
|
||
从Excel文件读取8760小时数据
|
||
|
||
Args:
|
||
file_path: Excel文件路径
|
||
sheet_name: 工作表名称或索引，默认为第一个工作表
|
||
|
||
Returns:
|
||
包含所有数据的字典
|
||
|
||
Raises:
|
||
FileNotFoundError: 文件不存在
|
||
ValueError: 数据格式错误
|
||
"""
|
||
# 检查文件是否存在
|
||
if not os.path.exists(file_path):
|
||
raise FileNotFoundError(f"文件不存在：{file_path}")
|
||
|
||
try:
|
||
# 读取Excel文件
|
||
df = pd.read_excel(file_path, sheet_name=sheet_name)
|
||
|
||
# 自动检测数据类型
|
||
data_type = "8760" if len(df) >= 8760 else "24"
|
||
|
||
# 验证数据格式
|
||
if not validate_excel_data(df, data_type):
|
||
raise ValueError("Excel数据格式验证失败")
|
||
|
||
# 提取数据并转换为列表
|
||
solar_output = df['光伏出力(MW)'].tolist()
|
||
wind_output = df['风电出力(MW)'].tolist()
|
||
thermal_output = df['火电出力(MW)'].tolist()
|
||
load_demand = df['负荷需求(MW)'].tolist()
|
||
|
||
# 如果是24小时数据，扩展到8760小时（重复365天）
|
||
if data_type == "24" and len(df) == 24:
|
||
print("检测到24小时数据，自动扩展到8760小时（重复365天）")
|
||
solar_output = solar_output * 365
|
||
wind_output = wind_output * 365
|
||
thermal_output = thermal_output * 365
|
||
load_demand = load_demand * 365
|
||
|
||
return {
|
||
'solar_output': solar_output,
|
||
'wind_output': wind_output,
|
||
'thermal_output': thermal_output,
|
||
'load_demand': load_demand,
|
||
'data_type': data_type,
|
||
'original_length': len(df)
|
||
}
|
||
|
||
except Exception as e:
|
||
raise ValueError(f"读取Excel文件失败：{str(e)}")
|
||
|
||
|
||
def create_excel_template(file_path: str, data_type: str = "8760"):
|
||
"""
|
||
创建Excel数据模板文件
|
||
|
||
Args:
|
||
file_path: 保存路径
|
||
data_type: 数据类型，"24"或"8760"
|
||
"""
|
||
# 生成示例数据
|
||
if data_type == "24":
|
||
hours = 24
|
||
# 24小时典型日数据
|
||
solar = [0.0] * 6 + [1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 5.0, 4.0, 3.0, 2.0, 1.0, 0.0] + [0.0] * 6
|
||
wind = [2.0, 3.0, 4.0, 3.0, 2.0, 1.0] * 4
|
||
thermal = [5.0] * 24
|
||
load = [3.0, 4.0, 5.0, 6.0, 8.0, 10.0, 12.0, 14.0, 16.0, 18.0, 20.0, 18.0,
|
||
16.0, 14.0, 12.0, 10.0, 8.0, 6.0, 5.0, 4.0, 3.0, 2.0, 1.0, 2.0]
|
||
description = "24小时典型日数据模板"
|
||
else:
|
||
hours = 8760
|
||
# 生成8760小时的模拟数据（基于日模式加季节变化）
|
||
daily_solar = [0.0] * 6 + [1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 5.0, 4.0, 3.0, 2.0, 1.0, 0.0] + [0.0] * 6
|
||
daily_wind = [2.0, 3.0, 4.0, 3.0, 2.0, 1.0] * 4
|
||
daily_thermal = [5.0] * 24
|
||
daily_load = [3.0, 4.0, 5.0, 6.0, 8.0, 10.0, 12.0, 14.0, 16.0, 18.0, 20.0, 18.0,
|
||
16.0, 14.0, 12.0, 10.0, 8.0, 6.0, 5.0, 4.0, 3.0, 2.0, 1.0, 2.0]
|
||
|
||
solar = []
|
||
wind = []
|
||
thermal = []
|
||
load = []
|
||
|
||
np.random.seed(42) # 确保可重复性
|
||
|
||
for day in range(365):
|
||
# 季节性因子
|
||
season_factor = 1.0 + 0.3 * np.sin(2 * np.pi * day / 365)
|
||
|
||
for hour in range(24):
|
||
# 添加随机变化
|
||
solar_variation = 1.0 + 0.2 * (np.random.random() - 0.5)
|
||
wind_variation = 1.0 + 0.3 * (np.random.random() - 0.5)
|
||
load_variation = 1.0 + 0.1 * (np.random.random() - 0.5)
|
||
|
||
solar.append(daily_solar[hour] * season_factor * solar_variation)
|
||
wind.append(daily_wind[hour] * wind_variation)
|
||
thermal.append(daily_thermal[hour])
|
||
load.append(daily_load[hour] * (2.0 - season_factor) * load_variation)
|
||
|
||
description = "8760小时全年数据模板"
|
||
|
||
# 创建DataFrame
|
||
df = pd.DataFrame({
|
||
'小时': range(1, hours + 1),
|
||
'光伏出力(MW)': solar,
|
||
'风电出力(MW)': wind,
|
||
'火电出力(MW)': thermal,
|
||
'负荷需求(MW)': load
|
||
})
|
||
|
||
# 保存到Excel
|
||
with pd.ExcelWriter(file_path, engine='openpyxl') as writer:
|
||
df.to_excel(writer, sheet_name='数据', index=False)
|
||
|
||
# 添加说明工作表
|
||
description_df = pd.DataFrame({
|
||
'项目': ['数据说明', '数据类型', '时间范围', '单位', '注意事项'],
|
||
'内容': [
|
||
description,
|
||
f'{data_type}小时电力数据',
|
||
f'1-{hours}小时',
|
||
'MW (兆瓦)',
|
||
'所有数值必须为非负数'
|
||
]
|
||
})
|
||
description_df.to_excel(writer, sheet_name='说明', index=False)
|
||
|
||
print(f"Excel模板已创建：{file_path}")
|
||
|
||
|
||
def analyze_excel_data(file_path: str) -> Dict[str, float]:
|
||
"""
|
||
分析Excel数据的基本统计信息
|
||
|
||
Args:
|
||
file_path: Excel文件路径
|
||
|
||
Returns:
|
||
包含统计信息的字典
|
||
"""
|
||
try:
|
||
data = read_excel_data(file_path)
|
||
|
||
solar = data['solar_output']
|
||
wind = data['wind_output']
|
||
thermal = data['thermal_output']
|
||
load = data['load_demand']
|
||
|
||
return {
|
||
'data_length': len(solar),
|
||
'total_solar': sum(solar),
|
||
'total_wind': sum(wind),
|
||
'total_thermal': sum(thermal),
|
||
'total_generation': sum(solar) + sum(wind) + sum(thermal),
|
||
'total_load': sum(load),
|
||
'max_solar': max(solar),
|
||
'max_wind': max(wind),
|
||
'max_thermal': max(thermal),
|
||
'max_load': max(load),
|
||
'avg_solar': np.mean(solar),
|
||
'avg_wind': np.mean(wind),
|
||
'avg_thermal': np.mean(thermal),
|
||
'avg_load': np.mean(load)
|
||
}
|
||
except Exception as e:
|
||
print(f"分析数据失败：{str(e)}")
|
||
return {}
|
||
|
||
|
||
def main():
|
||
"""主函数，演示Excel数据读取功能"""
|
||
print("=== Excel数据读取模块演示 ===")
|
||
|
||
# 创建模板文件
|
||
template_8760 = "data_template_8760.xlsx"
|
||
template_24 = "data_template_24.xlsx"
|
||
|
||
print("\n1. 创建Excel模板文件...")
|
||
create_excel_template(template_8760, "8760")
|
||
create_excel_template(template_24, "24")
|
||
|
||
# 分析模板数据
|
||
print(f"\n2. 分析{template_8760}数据...")
|
||
stats = analyze_excel_data(template_8760)
|
||
if stats:
|
||
print("数据统计信息：")
|
||
for key, value in stats.items():
|
||
print(f" {key}: {value:.2f}")
|
||
|
||
print(f"\n3. 演示读取{template_24}数据...")
|
||
try:
|
||
data = read_excel_data(template_24)
|
||
print(f"成功读取数据，类型：{data['data_type']}")
|
||
print(f"光伏出力前10小时：{data['solar_output'][:10]}")
|
||
print(f"风电出力前10小时：{data['wind_output'][:10]}")
|
||
print(f"负荷需求前10小时：{data['load_demand'][:10]}")
|
||
except Exception as e:
|
||
print(f"读取失败：{str(e)}")
|
||
|
||
print("\n=== 演示完成 ===")
|
||
print("模板文件已创建，您可以根据实际数据修改Excel文件。")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|