EEBD_AI
/
GreenTransPowerCalculate
11
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

add new api

This commit is contained in:
赵敬皓 2025-04-27 17:25:43 +08:00
parent c2ad1ebf80
commit 100838dee4
23 changed files with 1127 additions and 1159 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
Dockerfile Normal file
View File

@ -0,0 +1,9 @@
FROM python:3.10
WORKDIR /app
COPY . /app/
RUN pip install --upgrade pip -i https://pypi.tuna.tsinghua.edu.cn/simple --no-cache-dir
RUN pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple --no-cache-dir
CMD ["python3", "run.py"]

322
PV/PV_total2.py
View File

@ -1,322 +0,0 @@
import requests
import pandas as pd
import math
from scipy.optimize import fsolve
# 默认文件路径
PV_EXCEL_PATH = r".\pv_product.xlsx" # 光伏组件数值
def calculate_pv_metrics(component_name, electricity_price, pv_number, q, longitude, latitude,
is_fixed=True, optimize=True, peak_load_hour=16, cost_per_kw=3.4, E_S=1.0, K=0.8):
"""
计算光伏项目的各项指标，包括装机容量、年发电量、等效小时数、环境收益和内部收益率（IRR）。
参数：
component_name (str): 光伏组件名称，例如 "TWMHF-66HD715"
electricity_price (float): 电价（元/kWh）
pv_number (int): 光伏组件数量
q (float): 运维成本占初始投资成本的比例（例如 0.02 表示 2%）
longitude (float): 经度
latitude (float): 纬度
is_fixed (bool): 是否为固定式支架，True 为固定式，False 为跟踪式
optimize (bool): 是否优化倾角和方位角
peak_load_hour (int): 峰值负荷小时，默认 16
cost_per_kw (float): 每 kW 投资成本（元/kW），默认 3.4 元/kW
E_S (float): 标准辐射量，默认 1.0
K (float): 系统效率，默认 0.8
返回：
dict: 包含以下结果的字典：
- component_name: 组件名称
- tilt: 倾角 (度)
- azimuth: 方位角 (度)
- array_distance: 阵列间距 (米)
- max_power: 单组件最大功率 (Wp)
- capacity: 装机容量 (kW)
- peak_sunshine_hours: 峰值日照小时数 (小时/天)
- single_daily_energy: 一天单个组件发电量 (kWh)
- annual_energy: 年发电量 (kWh)
- equivalent_hours: 等效小时数 (小时)
- coal_reduction: 标准煤减排量 (kg)
- CO2_reduction: CO₂ 减排量 (kg)
- SO2_reduction: SO₂ 减排量 (kg)
- NOX_reduction: NOx artic_reduction量 (kg)
- IRR: 内部收益率 (%)
"""
# 1. 计算 PSH（峰值日照小时数）
def calculate_psh_average(lat, lon, start_year=2010, end_year=2023):
url = "https://power.larc.nasa.gov/api/temporal/monthly/point"
params = {
"parameters": "ALLSKY_SFC_SW_DWN", # 水平面全球辐照（GHI）
"community": "RE",
"longitude": lon,
"latitude": lat,
"format": "JSON",
"start": str(start_year),
"end": str(end_year)
}
try:
response = requests.get(url, params=params)
response.raise_for_status()
data = response.json()
ghi_data = data["properties"]["parameter"]["ALLSKY_SFC_SW_DWN"]
# 过滤掉年度平均值（例如 YYYY13）
ghi_data = {k: v for k, v in ghi_data.items() if not k.endswith("13")}
# 创建 DataFrame
df = pd.DataFrame.from_dict(ghi_data, orient="index", columns=["GHI (kWh/m²/day)"])
df.index = [f"{k[:4]}-{k[-2:]:02}" for k in df.index]
# 计算 PSH（PSH = GHI / 1 kW/m²）
df["PSH (hours/day)"] = df["GHI (kWh/m²/day)"]
# 计算每年的平均 PSH
df['Year'] = df.index.str[:4]
annual_avg = df.groupby('Year')['PSH (hours/day)'].mean()
# 计算总平均 PSH
return annual_avg.mean()
except requests.exceptions.RequestException as e:
print(f"NASA POWER API Error: {e}")
return None
except ValueError as e:
print(f"Data Error: {e}")
return None
# 2. 计算最佳倾角
def calculate_optimal_tilt(lat):
try:
# 将纬度从角度转换为弧度
lat_radians = math.radians(lat)
# 计算最佳倾角：倾角 = 纬度（弧度） × 0.86 + 24
optimal_tilt = lat_radians * 0.86 + 24
return optimal_tilt
except ValueError as e:
raise Exception(f"Tilt Calculation Error: {e}")
# 3. 计算倾角和方位角
def get_tilt_and_azimuth(is_fixed=True, optimize=True, longitude=116, latitude=None, peak_load_hour=16):
if optimize and latitude is None:
raise ValueError("优化模式下需提供纬度")
if is_fixed:
if optimize:
# 使用公式计算倾角
tilt = calculate_optimal_tilt(latitude)
if not (0 <= tilt <= 90):
raise ValueError(f"计算得到的倾角 {tilt:.2f}° 超出合理范围（0°-90°）")
azimuth = (peak_load_hour - 12) * 15 + (longitude - 116) # 方位角公式
azimuth = azimuth % 360 if azimuth >= 0 else azimuth + 360
else:
print("倾角：0°(水平)-90°(垂直) | 方位角：0°(正北)-180°(正南)，顺时针")
tilt = float(input("请输入倾角(度)："))
azimuth = float(input("请输入方位角(度)："))
if not (0 <= tilt <= 90) or not (0 <= azimuth <= 360):
raise ValueError("倾角需在0-90°，方位角需在0-360°")
else: # 跟踪式
azimuth = 180 # 固定朝南
if optimize:
tilt = calculate_optimal_tilt(latitude)
if not (0 <= tilt <= 90):
raise ValueError(f"计算得到的倾角 {tilt:.2f}° 超出合理范围（0°-90°）")
else:
print("倾角：0°(水平)-90°(垂直)")
tilt = float(input("请输入倾角(度)："))
if not (0 <= tilt <= 90):
raise ValueError("倾角需在0-90°")
return tilt, azimuth
# 4. 获取光伏组件信息
def get_pv_product_info(component_name, excel_path=PV_EXCEL_PATH):
try:
df = pd.read_excel(excel_path)
if len(df.columns) < 10:
raise ValueError("Excel文件需包含至少10列：组件名称、尺寸、功率等")
row = df[df.iloc[:, 1] == component_name]
if row.empty:
raise ValueError(f"未找到组件：{component_name}")
return {
"component_name": component_name,
"max_power": row.iloc[0, 5],
"efficiency": row.iloc[0, 9],
"pv_size": row.iloc[0, 3]
}
except FileNotFoundError:
raise FileNotFoundError(f"未找到Excel文件：{excel_path}")
except Exception as e:
raise Exception(f"读取Excel出错：{e}")
# 5. 计算光伏阵列间距
def calculate_array_distance(L, tilt, latitude):
beta_rad = math.radians(tilt)
phi_rad = math.radians(latitude)
return (L * math.cos(beta_rad) +
L * math.sin(beta_rad) * 0.707 * math.tan(phi_rad) +
0.4338 * math.tan(phi_rad))
# 6. 计算等效小时数
def calculate_equivalent_hours(P, P_r):
if P_r == 0:
raise ValueError("额定功率不能为 0")
h = P / P_r # 单位换算
return h
# 7. 计算装机容量
def calculate_installed_capacity(max_power, num_components):
if max_power < 0 or num_components < 0 or not isinstance(num_components, int):
raise ValueError("功率和数量需为非负数，数量需为整数")
return (max_power * num_components) / 1000 # 单位：kW
# 8. 计算年发电量
def calculate_annual_energy(peak_hours, capacity, E_S=E_S, K=K):
if any(x < 0 for x in [peak_hours, capacity]) or E_S <= 0 or not 0 <= K <= 1:
raise ValueError("输入参数需满足：辐射量、容量≥0，E_S>0，K∈[0,1]")
return peak_hours * 365 * (capacity / E_S) * K # 单位：kWh
# 9. 计算环境收益
def calculate_environmental_benefits(E_p_million_kwh):
if E_p_million_kwh < 0:
raise ValueError("年发电量需≥0")
return {
"coal_reduction": E_p_million_kwh * 0.404 * 10,
"CO2_reduction": E_p_million_kwh * 0.977 * 10,
"SO2_reduction": E_p_million_kwh * 0.03 * 10,
"NOX_reduction": E_p_million_kwh * 0.015 * 10
}
# 10. 计算净现值和内部收益率
def calculate_reference_yield(E_p, electricity_price, IC, q, n=25):
if E_p < 0 or electricity_price < 0 or IC <= 0 or not 0 <= q <= 1:
raise ValueError("发电量、电价≥0，投资成本>0，回收比例∈[0,1]")
def npv_equation(irr, p, w, ic, q_val, n=n):
term1 = (1 + irr) ** (-1)
term2 = irr * (1 + irr) ** (-1) if irr != 0 else float('inf')
pv_revenue = p * w * (term1 / term2) * (1 - (1 + irr) ** (-n))
pv_salvage = q_val * ic * (term1 / term2) * (1 - (1 + irr) ** (-n))
return pv_revenue - ic + pv_salvage
irr_guess = 0.1
irr = float(fsolve(npv_equation, irr_guess, args=(E_p, electricity_price, IC, q))[0])
if not 0 <= irr <= 1:
raise ValueError(f"IRR计算结果{irr:.4f}不合理，请检查输入")
npv = npv_equation(irr, E_p, electricity_price, IC, q)
return {"NPV": npv, "IRR": irr * 100}
# 主计算流程
try:
# 获取倾角和方位角
tilt, azimuth = get_tilt_and_azimuth(
is_fixed=is_fixed,
optimize=optimize,
longitude=longitude,
latitude=latitude,
peak_load_hour=peak_load_hour
)
# 获取组件信息
pv_info = get_pv_product_info(component_name)
width_mm = float(pv_info["pv_size"].split("×")[1])
L = (width_mm / 1000) * 4
array_distance = calculate_array_distance(L, tilt, latitude)
# 计算装机容量
max_power = pv_info["max_power"]
capacity = calculate_installed_capacity(max_power, pv_number) # 单位：kW
# 获取峰值日照小时数（使用 NASA POWER API）
peak_hours = calculate_psh_average(latitude, longitude)
# 计算一天单个组件发电量
single_daily_energy = peak_hours * (capacity / pv_number) * K # 单位：kWh
# 计算年发电量
E_p = calculate_annual_energy(peak_hours, capacity, E_S, K) # 单位：kWh
# 计算等效小时数
h = calculate_equivalent_hours(E_p, capacity) # P_r 单位为 kW，E_p 单位为 kWh
# 计算环境收益（转换为百万 kWh）
E_p_million_kwh = E_p / 1000000 # 转换为百万 kWh
env_benefits = calculate_environmental_benefits(E_p_million_kwh)
# 计算初始投资成本
IC = capacity * cost_per_kw * 1000 # 单位：元
# 计算 IRR
ref_yield = calculate_reference_yield(E_p, electricity_price, IC, q)
# 返回结果
return {
"component_name": component_name,
"tilt": tilt,
"azimuth": azimuth,
"array_distance": array_distance,
"max_power": max_power,
"capacity": capacity,
"peak_sunshine_hours": peak_hours,
"single_daily_energy": single_daily_energy,
"annual_energy": E_p,
"equivalent_hours": h,
"coal_reduction": env_benefits["coal_reduction"],
"CO2_reduction": env_benefits["CO2_reduction"],
"SO2_reduction": env_benefits["SO2_reduction"],
"NOX_reduction": env_benefits["NOX_reduction"],
"IRR": ref_yield["IRR"]
}
except Exception as e:
raise Exception(f"计算过程中发生错误: {str(e)}")
# 示例用法
if __name__ == "__main__":
try:
# 输入
# 输入并验证经纬度
latitude = float(input("请输入纬度（-90 到 90，例如 39.9042）："))
if not (-90 <= latitude <= 90):
raise ValueError("纬度必须在 -90 到 90 之间")
longitude = float(input("请输入经度（-180 到 180，例如 116.4074）："))
if not (-180 <= longitude <= 180):
raise ValueError("经度必须在 -180 到 180 之间")
component_name = input("请输入组件名称：")
electricity_price = float(input("请输入电价（元/kWh）："))
pv_number = 1200 # 固定组件数量
q = 0.02 # 运维成本占初始投资成本的比例
choice = input("选择光伏支架（固定式/跟踪式）：")
optimize = input("是否优化倾角和方位角（是/否）：")
# 调用主函数
result = calculate_pv_metrics(
component_name=component_name,
electricity_price=electricity_price,
pv_number=pv_number,
q=q,
is_fixed=(choice == "固定式"),
optimize=(optimize.lower() == "是"),
longitude=longitude,
latitude=latitude
)
# 打印结果
print("\n")
print(f"组件名称：{result['component_name']}")
print(f"倾角：{result['tilt']:.2f}° | 方位角：{result['azimuth']:.2f}°")
print(f"阵列间距：{result['array_distance']:.2f} 米")
print(f"单个组件最大功率（Wp）：{result['max_power']}")
print(f"装机容量：{result['capacity']:.2f} kW")
print(f"峰值日照小时数：{result['peak_sunshine_hours']:.2f} 小时/天")
print(f"一天单个组件发电量：{result['single_daily_energy']:.0f} kWh")
print(f"年发电量：{result['annual_energy']:,.0f} kWh")
print(f"等效小时数：{result['equivalent_hours']:.0f} 小时")
print("环境收益：")
print(f"标准煤减排量：{result['coal_reduction']:,.0f} kg")
print(f"CO₂减排量：{result['CO2_reduction']:,.0f} kg")
print(f"SO₂减排量：{result['SO2_reduction']:,.0f} kg")
print(f"NOx减排量：{result['NOX_reduction']:,.0f} kg")
print(f"内部收益率 IRR：{result['IRR']:.2f}%")
except Exception as e:
print(f"错误：{e}")

339
PV/PV_total3.py
View File

@ -1,339 +0,0 @@
import json
import pandas as pd
import math
from scipy.optimize import fsolve
from shapely.geometry import Point, shape
# 默认文件路径
PV_EXCEL_PATH = r"./pv_product.xlsx" # 光伏组件数值
TILT_EXCEL_PATH = r"./倾角_峰值小时数.xlsx" # 倾角和峰值小时数
GEOJSON_PATH = r"./中国_市.geojson" # GeoJSON 文件
def calculate_pv_metrics(component_name, electricity_price, pv_number, q, longitude, latitude,
is_fixed=True, optimize=True, peak_load_hour=16, cost_per_kw=3.4, E_S=1.0, K=0.8):
"""
计算光伏项目的各项指标，包括装机容量、年发电量、等效小时数、环境收益和内部收益率（IRR）。
参数：
component_name (str): 光伏组件名称，例如 "TWMHF-66HD715"
electricity_price (float): 电价（元/kWh）
pv_number (int): 光伏组件数量
q (float): 运维成本占初始投资成本的比例（例如 0.02 表示 2%）
longitude (float): 经度
latitude (float): 纬度
is_fixed (bool): 是否为固定式支架，True 为固定式，False 为跟踪式
optimize (bool): 是否优化倾角和方位角
peak_load_hour (int): 峰值负荷小时，默认 16
cost_per_kw (float): 每 kW 投资成本（元/kW），默认 3.4 元/kW
E_S (float): 标准辐射量，默认 1.0
K (float): 系统效率，默认 0.8
返回：
dict: 包含以下结果的字典：
- component_name: 组件名称
- tilt: 倾角 (度)
- azimuth: 方位角 (度)
- array_distance: 阵列间距 (米)
- max_power: 单组件最大功率 (Wp)
- capacity: 装机容量 (kW)
- peak_sunshine_hours: 峰值日照小时数 (小时/天)
- single_daily_energy: 一天单个组件发电量 (kWh)
- annual_energy: 年发电量 (kWh)
- equivalent_hours: 等效小时数 (小时)
- coal_reduction: 标准煤减排量 (kg)
- CO2_reduction: CO₂ 减排量 (kg)
- SO2_reduction: SO₂ 减排量 (kg)
- NOX_reduction: NOx 减排量 (kg)
- IRR: 内部收益率 (%)
"""
# 1. 加载 GeoJSON 数据
def load_geojson(file_path):
try:
with open(file_path, 'r', encoding='utf-8') as file:
data = json.load(file)
return data
except FileNotFoundError:
raise FileNotFoundError(f"未找到GeoJSON文件：{file_path}")
except Exception as e:
raise Exception(f"加载GeoJSON出错：{e}")
# 2. 根据经纬度查找城市
def find_city_by_coordinates(lat, lon, geojson_data):
point = Point(lon, lat)
for feature in geojson_data['features']:
polygon = shape(feature['geometry'])
if polygon.contains(point):
city_name = feature['properties'].get('name', '未知城市')
# 规范化城市名称（去除“市”或“地区”等后缀）
if city_name.endswith('市') or city_name.endswith('地区'):
city_name = city_name[:-1]
return city_name
return '未知城市'
# 3. 从 Excel 获取倾角和峰值日照小时数
def get_tilt_and_peak_hours(lat, lon, excel_path=TILT_EXCEL_PATH, geojson_path=GEOJSON_PATH):
try:
# 加载 GeoJSON 数据
geojson_data = load_geojson(geojson_path)
# 查找城市
city = find_city_by_coordinates(lat, lon, geojson_data)
# 读取 Excel 文件
df = pd.read_excel(excel_path)
if len(df.columns) < 5:
raise ValueError("Excel文件需包含至少5列：城市、倾角、峰值日照小时数等")
# 匹配城市名称（前两个字）
matched_row = df[df.iloc[:, 2].str.startswith(city, na=False)]
if matched_row.empty:
raise ValueError(f"未找到匹配的城市：{city}")
# 提取并转换为浮点数
tilt = float(matched_row.iloc[0, 3]) # 第四列：倾角
peak_hours = float(matched_row.iloc[0, 4]) # 第五列：PSH
return {
"tilt": tilt,
"peak_sunshine_hours": peak_hours
}
except FileNotFoundError:
raise FileNotFoundError(f"未找到Excel文件：{excel_path}")
except ValueError as e:
raise ValueError(f"Excel数据转换错误：{e}")
except Exception as e:
raise Exception(f"读取Excel或GeoJSON出错：{e}")
# 4. 计算倾角和方位角
def get_tilt_and_azimuth(is_fixed=True, optimize=True, longitude=116, latitude=None, peak_load_hour=16):
if optimize and latitude is None:
raise ValueError("优化模式下需提供纬度")
if is_fixed:
if optimize:
# 从 Excel 获取倾角
tilt_data = get_tilt_and_peak_hours(latitude, longitude)
tilt = tilt_data["tilt"]
if not (0 <= tilt <= 90):
raise ValueError(f"Excel中的倾角 {tilt:.2f}° 超出合理范围（0°-90°）")
azimuth = (peak_load_hour - 12) * 15 + (longitude - 116) # 方位角公式
azimuth = azimuth % 360 if azimuth >= 0 else azimuth + 360
else:
print("倾角：0°(水平)-90°(垂直) | 方位角：0°(正北)-180°(正南)，顺时针")
tilt = float(input("请输入倾角(度)："))
azimuth = float(input("请输入方位角(度)："))
if not (0 <= tilt <= 90) or not (0 <= azimuth <= 360):
raise ValueError("倾角需在0-90°，方位角需在0-360°")
else: # 跟踪式
azimuth = 180 # 固定朝南
if optimize:
tilt_data = get_tilt_and_peak_hours(latitude, longitude)
tilt = tilt_data["tilt"]
if not (0 <= tilt <= 90):
raise ValueError(f"Excel中的倾角 {tilt:.2f}° 超出合理范围（0°-90°）")
else:
print("倾角：0°(水平)-90°(垂直)")
tilt = float(input("请输入倾角(度)："))
if not (0 <= tilt <= 90):
raise ValueError("倾角需在0-90°")
return tilt, azimuth
# 5. 获取光伏组件信息
def get_pv_product_info(component_name, excel_path=PV_EXCEL_PATH):
try:
df = pd.read_excel(excel_path)
if len(df.columns) < 10:
raise ValueError("Excel文件需包含至少10列：组件名称、尺寸、功率等")
row = df[df.iloc[:, 1] == component_name]
if row.empty:
raise ValueError(f"未找到组件：{component_name}")
return {
"component_name": component_name,
"max_power": row.iloc[0, 5],
"efficiency": row.iloc[0, 9],
"pv_size": row.iloc[0, 3]
}
except FileNotFoundError:
raise FileNotFoundError(f"未找到Excel文件：{excel_path}")
except Exception as e:
raise Exception(f"读取Excel出错：{e}")
# 6. 计算光伏阵列间距
def calculate_array_distance(L, tilt, latitude):
beta_rad = math.radians(tilt)
phi_rad = math.radians(latitude)
return (L * math.cos(beta_rad) +
L * math.sin(beta_rad) * 0.707 * math.tan(phi_rad) +
0.4338 * math.tan(phi_rad))
# 7. 计算等效小时数
def calculate_equivalent_hours(P, P_r):
if P_r == 0:
raise ValueError("额定功率不能为 0")
h = P / P_r # 单位换算
return h
# 8. 计算装机容量
def calculate_installed_capacity(max_power, num_components):
if max_power < 0 or num_components < 0 or not isinstance(num_components, int):
raise ValueError("功率和数量需为非负数，数量需为整数")
return (max_power * num_components) / 1000 # 单位：kW
# 9. 计算年发电量
def calculate_annual_energy(peak_hours, capacity, E_S=E_S, K=K):
if any(x < 0 for x in [peak_hours, capacity]) or E_S <= 0 or not 0 <= K <= 1:
raise ValueError("输入参数需满足：辐射量、容量≥0，E_S>0，K∈[0,1]")
return peak_hours * 365 * (capacity / E_S) * K # 单位：kWh
# 10. 计算环境收益
def calculate_environmental_benefits(E_p_million_kwh):
if E_p_million_kwh < 0:
raise ValueError("年发电量需≥0")
return {
"coal_reduction": E_p_million_kwh * 0.404 * 10,
"CO2_reduction": E_p_million_kwh * 0.977 * 10,
"SO2_reduction": E_p_million_kwh * 0.03 * 10,
"NOX_reduction": E_p_million_kwh * 0.015 * 10
}
# 11. 计算净现值和内部收益率
def calculate_reference_yield(E_p, electricity_price, IC, q, n=25):
if E_p < 0 or electricity_price < 0 or IC <= 0 or not 0 <= q <= 1:
raise ValueError("发电量、电价≥0，投资成本>0，回收比例∈[0,1]")
def npv_equation(irr, p, w, ic, q_val, n=n):
term1 = (1 + irr) ** (-1)
term2 = irr * (1 + irr) ** (-1) if irr != 0 else float('inf')
pv_revenue = p * w * (term1 / term2) * (1 - (1 + irr) ** (-n))
pv_salvage = q_val * ic * (term1 / term2) * (1 - (1 + irr) ** (-n))
return pv_revenue - ic + pv_salvage
irr_guess = 0.1
irr = float(fsolve(npv_equation, irr_guess, args=(E_p, electricity_price, IC, q))[0])
if not 0 <= irr <= 1:
raise ValueError(f"IRR计算结果{irr:.4f}不合理，请检查输入")
npv = npv_equation(irr, E_p, electricity_price, IC, q)
return {"NPV": npv, "IRR": irr * 100}
# 主计算流程
try:
# 验证经纬度
if not (-90 <= latitude <= 90):
raise ValueError("纬度必须在 -90 到 90 之间")
if not (-180 <= longitude <= 180):
raise ValueError("经度必须在 -180 到 180 之间")
# 获取倾角和方位角
tilt, azimuth = get_tilt_and_azimuth(
is_fixed=is_fixed,
optimize=optimize,
longitude=longitude,
latitude=latitude,
peak_load_hour=peak_load_hour
)
# 获取峰值日照小时数
peak_hours = get_tilt_and_peak_hours(latitude, longitude)["peak_sunshine_hours"]
# 获取组件信息
pv_info = get_pv_product_info(component_name)
width_mm = float(pv_info["pv_size"].split("×")[1])
L = (width_mm / 1000) * 4
array_distance = calculate_array_distance(L, tilt, latitude)
# 计算装机容量
max_power = pv_info["max_power"]
capacity = calculate_installed_capacity(max_power, pv_number) # 单位：kW
# 计算一天单个组件发电量
single_daily_energy = peak_hours * (capacity / pv_number) * K # 单位：kWh
# 计算年发电量
E_p = calculate_annual_energy(peak_hours, capacity, E_S, K) # 单位：kWh
# 计算等效小时数
h = calculate_equivalent_hours(E_p, capacity) # P_r 单位为 kW，E_p 单位为 kWh
# 计算环境收益（转换为百万 kWh）
E_p_million_kwh = E_p / 1000000 # 转换为百万 kWh
env_benefits = calculate_environmental_benefits(E_p_million_kwh)
# 计算初始投资成本
IC = capacity * cost_per_kw * 1000 # 单位：元
# 计算 IRR
ref_yield = calculate_reference_yield(E_p, electricity_price, IC, q)
# 返回结果
return {
"component_name": component_name,
"tilt": tilt,
"azimuth": azimuth,
"array_distance": array_distance,
"max_power": max_power,
"capacity": capacity,
"peak_sunshine_hours": peak_hours,
"single_daily_energy": single_daily_energy,
"annual_energy": E_p,
"equivalent_hours": h,
"coal_reduction": env_benefits["coal_reduction"],
"CO2_reduction": env_benefits["CO2_reduction"],
"SO2_reduction": env_benefits["SO2_reduction"],
"NOX_reduction": env_benefits["NOX_reduction"],
"IRR": ref_yield["IRR"]
}
except Exception as e:
raise Exception(f"计算过程中发生错误: {str(e)}")
# 示例用法
if __name__ == "__main__":
try:
# 输入并验证经纬度
latitude = float(input("请输入纬度（-90 到 90，例如 39.9042）："))
if not (-90 <= latitude <= 90):
raise ValueError("纬度必须在 -90 到 90 之间")
longitude = float(input("请输入经度（-180 到 180，例如 116.4074）："))
if not (-180 <= longitude <= 180):
raise ValueError("经度必须在 -180 到 180 之间")
component_name = input("请输入组件名称：")
electricity_price = float(input("请输入电价（元/kWh）："))
pv_number = 1200 # 固定组件数量
q = 0.02 # 运维成本占初始投资成本的比例
choice = input("选择光伏支架（固定式/跟踪式）：")
optimize = input("是否优化倾角和方位角（是/否）：")
# 调用主函数
result = calculate_pv_metrics(
component_name=component_name,
electricity_price=electricity_price,
pv_number=pv_number,
q=q,
is_fixed=(choice == "固定式"),
optimize=(optimize.lower() == "是"),
longitude=longitude,
latitude=latitude
)
# 打印结果
print("\n")
print(f"组件名称：{result['component_name']}")
print(f"倾角：{result['tilt']:.2f}° | 方位角：{result['azimuth']:.2f}°")
print(f"阵列间距：{result['array_distance']:.2f} 米")
print(f"单个组件最大功率（Wp）：{result['max_power']}")
print(f"装机容量：{result['capacity']:.2f} kW")
print(f"峰值日照小时数：{result['peak_sunshine_hours']:.2f} 小时/天")
print(f"一天单个组件发电量：{result['single_daily_energy']:.0f} kWh")
print(f"年发电量：{result['annual_energy']:,.0f} kWh")
print(f"等效小时数：{result['equivalent_hours']:.0f} 小时")
print("环境收益：")
print(f"标准煤减排量：{result['coal_reduction']:,.0f} kg")
print(f"CO₂减排量：{result['CO2_reduction']:,.0f} kg")
print(f"SO₂减排量：{result['SO2_reduction']:,.0f} kg")
print(f"NOx减排量：{result['NOX_reduction']:,.0f} kg")
print(f"内部收益率 IRR：{result['IRR']:.2f}%")
except ValueError as e:
print(f"输入错误：{e}")
except Exception as e:
print(f"错误：{e}")

281
PV/pv_total.py
View File

@ -1,281 +0,0 @@
import pandas as pd
import math
from scipy.optimize import fsolve
# 默认文件路径
TILT_EXCEL_PATH = r"./peak_sunshine.xlsx" #各地区峰值小时数和倾角（注意城市名
PV_EXCEL_PATH = r"./pv_product.xlsx" #部分光伏组件数值
def calculate_pv_metrics(city, component_name, electricity_price, pv_number, q, longitude, latitude,
is_fixed=True, optimize=True, peak_load_hour=16, cost_per_kw=3.4, E_S=1.0, K=0.8):
"""
计算光伏项目的各项指标，包括装机容量、年发电量、等效小时数、环境收益和内部收益率（IRR）。
参数：
city (str): 城市名称，例如 "深圳"
component_name (str): 光伏组件名称，例如 "TWMHF-66HD715"
electricity_price (float): 电价（元/kWh）
pv_number (int): 光伏组件数量
q (float): 运维成本占初始投资成本的比例（例如 0.02 表示 2%）
is_fixed (bool): 是否为固定式支架，True 为固定式，False 为跟踪式
optimize (bool): 是否优化倾角和方位角
longitude (float): 经度
latitude (float): 纬度
peak_load_hour (int): 峰值负荷小时，默认 16
cost_per_kw (float): 每 kW 投资成本（元/kW），默认 3.4 元/kW
E_S (float): 标准辐射量，默认 1.0
K (float): 系统效率，默认 0.8
返回：
dict: 包含以下结果的字典：
- city: 城市名称
- component_name: 组件名称
- tilt: 倾角 (度)
- azimuth: 方位角 (度)
- array_distance: 阵列间距 (米)
- max_power: 单组件最大功率 (Wp)
- capacity: 装机容量 (kW)
- peak_sunshine_hours: 峰值日照小时数 (小时/天)
- single_daily_energy: 一天单个组件发电量 (kWh)
- annual_energy: 年发电量 (kWh)
- equivalent_hours: 等效小时数 (小时)
- coal_reduction: 标准煤减排量 (kg)
- CO2_reduction: CO₂ 减排量 (kg)
- SO2_reduction: SO₂ 减排量 (kg)
- NOX_reduction: NOx 减排量 (kg)
- IRR: 内部收益率 (%)
"""
# 1. 获取城市的倾角和峰值日照小时数
def get_tilt_and_peak_hours(city, excel_path=TILT_EXCEL_PATH):
"""从Excel获取城市的倾角和峰值日照小时数"""
try:
df = pd.read_excel(excel_path)
if len(df.columns) < 5:
raise ValueError("Excel文件需包含至少5列：城市、倾角、峰值日照小时数等")
row = df[df.iloc[:, 2] == city]
if row.empty:
raise ValueError(f"未找到城市：{city}")
return {"city": city, "tilt": row.iloc[0, 3], "peak_sunshine_hours": row.iloc[0, 4]}
except FileNotFoundError:
raise FileNotFoundError(f"未找到Excel文件：{excel_path}")
except Exception as e:
raise Exception(f"读取Excel出错：{e}")
# 2. 计算倾角和方位角
def get_tilt_and_azimuth(is_fixed=True, optimize=True, longitude=116, city=None, excel_path=TILT_EXCEL_PATH,
peak_load_hour=16):
"""计算光伏系统的倾角和方位角"""
if optimize and not city:
raise ValueError("优化模式下需提供城市名称")
if is_fixed:
if optimize:
tilt = get_tilt_and_peak_hours(city, excel_path)["tilt"]
azimuth = (peak_load_hour - 12) * 15 + (longitude - 116) # 方位角公式
azimuth = azimuth % 360 if azimuth >= 0 else azimuth + 360
else:
print("倾角：0°(水平)-90°(垂直) | 方位角：0°(正北)-180°(正南)，顺时针")
tilt = float(input("请输入倾角(度)："))
azimuth = float(input("请输入方位角(度)："))
if not (0 <= tilt <= 90) or not (0 <= azimuth <= 360):
raise ValueError("倾角需在0-90°，方位角需在0-360°")
else: # 跟踪式
azimuth = 180 # 固定朝南
if optimize:
tilt = get_tilt_and_peak_hours(city, excel_path)["tilt"]
else:
print("倾角：0°(水平)-90°(垂直)")
tilt = float(input("请输入倾角(度)："))
if not (0 <= tilt <= 90):
raise ValueError("倾角需在0-90°")
return tilt, azimuth
# 3. 获取光伏组件信息
def get_pv_product_info(component_name, excel_path=PV_EXCEL_PATH):
try:
df = pd.read_excel(excel_path)
if len(df.columns) < 10:
raise ValueError("Excel文件需包含至少10列：组件名称、尺寸、功率等")
row = df[df.iloc[:, 1] == component_name]
if row.empty:
raise ValueError(f"未找到组件：{component_name}")
return {
"component_name": component_name,
"max_power": row.iloc[0, 5],
"efficiency": row.iloc[0, 9],
"pv_size": row.iloc[0, 3]
}
except FileNotFoundError:
raise FileNotFoundError(f"未找到Excel文件：{excel_path}")
except Exception as e:
raise Exception(f"读取Excel出错：{e}")
# 4. 计算光伏阵列间距
def calculate_array_distance(L, tilt, latitude):
beta_rad = math.radians(tilt)
phi_rad = math.radians(latitude)
return (L * math.cos(beta_rad) +
L * math.sin(beta_rad) * 0.707 * math.tan(phi_rad) +
0.4338 * math.tan(phi_rad))
# 5. 计算等效小时数
def calculate_equivalent_hours(P, P_r):
if P_r == 0:
raise ValueError("额定功率不能为 0")
h = P / P_r # 单位换算
return h
# 6. 计算装机容量
def calculate_installed_capacity(max_power, num_components):
if max_power < 0 or num_components < 0 or not isinstance(num_components, int):
raise ValueError("功率和数量需为非负数，数量需为整数")
return (max_power * num_components) / 1000 # 单位：kW
# 7. 计算年发电量
def calculate_annual_energy(peak_hours, capacity, E_S=E_S, K=K):
if any(x < 0 for x in [peak_hours, capacity]) or E_S <= 0 or not 0 <= K <= 1:
raise ValueError("输入参数需满足：辐射量、容量≥0，E_S>0，K∈[0,1]")
return peak_hours * 365 * (capacity / E_S) * K # 单位：kWh
# 8. 计算环境收益
def calculate_environmental_benefits(E_p_million_kwh):
if E_p_million_kwh < 0:
raise ValueError("年发电量需≥0")
return {
"coal_reduction": E_p_million_kwh * 0.404 * 10,
"CO2_reduction": E_p_million_kwh * 0.977 * 10,
"SO2_reduction": E_p_million_kwh * 0.03 * 10,
"NOX_reduction": E_p_million_kwh * 0.015 * 10
}
# 9. 计算净现值和内部收益率
def calculate_reference_yield(E_p, electricity_price, IC, q, n=25):
if E_p < 0 or electricity_price < 0 or IC <= 0 or not 0 <= q <= 1:
raise ValueError("发电量、电价≥0，投资成本>0，回收比例∈[0,1]")
def npv_equation(irr, p, w, ic, q_val, n=n):
term1 = (1 + irr) ** (-1)
term2 = irr * (1 + irr) ** (-1) if irr != 0 else float('inf')
pv_revenue = p * w * (term1 / term2) * (1 - (1 + irr) ** (-n))
pv_salvage = q_val * ic * (term1 / term2) * (1 - (1 + irr) ** (-n))
return pv_revenue - ic + pv_salvage
irr_guess = 0.1
irr = float(fsolve(npv_equation, irr_guess, args=(E_p, electricity_price, IC, q))[0])
if not 0 <= irr <= 1:
raise ValueError(f"IRR计算结果{irr:.4f}不合理，请检查输入")
npv = npv_equation(irr, E_p, electricity_price, IC, q)
return {"NPV": npv, "IRR": irr * 100}
# 主计算流程
try:
# 获取倾角和方位角
tilt, azimuth = get_tilt_and_azimuth(is_fixed, optimize, longitude, city)
# 获取组件信息
pv_info = get_pv_product_info(component_name)
width_mm = float(pv_info["pv_size"].split("×")[1])
L = (width_mm / 1000) * 4
array_distance = calculate_array_distance(L, tilt, latitude)
# 计算装机容量
max_power = pv_info["max_power"]
capacity = calculate_installed_capacity(max_power, pv_number) # 单位：kW
# 获取峰值日照小时数
peak_hours = get_tilt_and_peak_hours(city)["peak_sunshine_hours"]
# 计算一天单个组件发电量
single_daily_energy = peak_hours * (capacity / pv_number) * K # 单位：kWh
# 计算年发电量
E_p = calculate_annual_energy(peak_hours, capacity, E_S, K) # 单位：kWh
# 计算等效小时数
h = calculate_equivalent_hours(E_p, capacity) # P_r 单位为 kW，E_p 单位为 kWh
# 计算环境收益（转换为百万 kWh）
E_p_million_kwh = E_p / 1000000 # 转换为百万 kWh
env_benefits = calculate_environmental_benefits(E_p_million_kwh)
# 计算初始投资成本
IC = capacity * cost_per_kw * 1000 # 单位：元
# 计算 IRR
ref_yield = calculate_reference_yield(E_p, electricity_price, IC, q)
# 返回结果
return {
"city": city,
"component_name": component_name,
"tilt": tilt,
"azimuth": azimuth,
"array_distance": array_distance,
"max_power": max_power,
"capacity": capacity,
"peak_sunshine_hours": peak_hours,
"single_daily_energy": single_daily_energy,
"annual_energy": E_p,
"equivalent_hours": h,
"coal_reduction": env_benefits["coal_reduction"],
"CO2_reduction": env_benefits["CO2_reduction"],
"SO2_reduction": env_benefits["SO2_reduction"],
"NOX_reduction": env_benefits["NOX_reduction"],
"IRR": ref_yield["IRR"]
}
except Exception as e:
raise Exception(f"计算过程中发生错误: {str(e)}")
# 示例用法
if __name__ == "__main__":
try:
# 输入
city = input("请输入城市：")
longitude = int(input("请城市经度："))
latitude = int(input("请城市纬度："))
component_name = input("请输入组件名称：")
electricity_price = float(input("请输入电价（元/kWh）："))
pv_number = 1200 # 固定组件数量
q = 0.02 # 运维成本占初始投资成本的比例
choice = input("选择光伏支架（固定式/跟踪式）：")
optimize = input("是否优化倾角和方位角（是/否）：")
# 调用主函数
result = calculate_pv_metrics(
city=city,
component_name=component_name,
electricity_price=electricity_price,
pv_number=pv_number,
q=q,
is_fixed=(choice == "固定式"),
optimize=(optimize.lower() == "是"),
longitude=longitude,
latitude=latitude
)
# 打印结果
print("\n")
print(f"城市：{result['city']}")
print(f"组件名称：{result['component_name']}")
print(f"倾角：{result['tilt']}° | 方位角：{result['azimuth']}°")
print(f"阵列间距：{result['array_distance']:.2f} 米")
print(f"单个组件最大功率（Wp）：{result['max_power']}")
print(f"装机容量：{result['capacity']:.2f} kW")
print(f"峰值日照小时数：{result['peak_sunshine_hours']:.2f} 小时/天")
print(f"一天单个组件发电量：{result['single_daily_energy']:.2f} kWh")
print(f"年发电量：{result['annual_energy']:,.2f} kWh")
print(f"等效小时数：{result['equivalent_hours']:.2f} 小时")
print("环境收益：")
print(f"标准煤减排量：{result['coal_reduction']:,.2f} kg")
print(f"CO₂减排量：{result['CO2_reduction']:,.2f} kg")
print(f"SO₂减排量：{result['SO2_reduction']:,.2f} kg")
print(f"NOx减排量：{result['NOX_reduction']:,.2f} kg")
print(f"内部收益率 IRR：{result['IRR']:.2f}%")
except Exception as e:
print(f"错误：{e}")

1
PV/中国_市.geojson
View File

File diff suppressed because one or more lines are too long

BIN
PV/倾角_峰值小时数.xlsx
View File

Binary file not shown.

12
Wind/temperature.txt
View File

@ -1,12 +0,0 @@
10
12
15
18
20
22
25
24
20
15
12
10

12
Wind/wind_speed.txt
View File

@ -1,12 +0,0 @@
3.5
4.0
4.2
3.8
5.0
5.5
6.0
5.8
5.2
4.5
4.0
3.7

192
Wind/wind_total.py
View File

@ -1,192 +0,0 @@
import pandas as pd
import math
from scipy.optimize import fsolve
def wind_farm_analysis(device_name, area_km2, electricity_price, file_path, velocity_path, T_path,
lateral_spacing_factor=5, longitudinal_spacing_factor=10, q=0.02, altitude=11,
hub_height=100, Cp=0.45, eta=0.8, cost_per_mw=5000):
"""
封装函数：分析风电场的风机数量及各项经济和技术指标
参数：
device_name (str): 设备名称
area_km2 (float): 风电场面积（平方公里）
electricity_price (float): 电价（元/kWh）
file_path (str): 风机参数 Excel 文件路径
velocity_path (str): 风速数据文件路径（12 个月平均风速）
T_path (str): 温度数据文件路径（12 个月平均温度）
lateral_spacing_factor (float): 横向间距因子（默认为 5D）
longitudinal_spacing_factor (float): 纵向间距因子（默认为 10D）
q (float): 运维成本占初始投资成本的比例（默认 0.02 表示 2%）
altitude (float): 海拔高度（m），默认 11m
hub_height (float): 轮毂高度（m），默认 100m
Cp (float): 风能利用系数，默认 0.45
eta (float): 总系统效率，默认 0.8
cost_per_mw (float): 每 MW 投资成本（万元/MW），默认 5000 万元/MW
返回：
dict: 包含风电场分析结果的字典
"""
def estimate_wind_turbine_count(area_km2, blade_diameter):
area_m2 = area_km2 * 1_000_000
lateral_spacing = lateral_spacing_factor * blade_diameter
longitudinal_spacing = longitudinal_spacing_factor * blade_diameter
turbine_area = lateral_spacing * longitudinal_spacing
turbine_count = int(area_m2 / turbine_area)
print(f"单台风机占地面积: {turbine_area:,} 平方米 "
f"(横向间距: {lateral_spacing} 米, 纵向间距: {longitudinal_spacing} 米)")
print(f"估算风机数量: {turbine_count} 台")
return turbine_count
def get_wind_turbine_specs(device_name, file_path):
try:
df = pd.read_excel(file_path)
match = df[df.iloc[:, 0] == device_name]
if not match.empty:
rated_power = match.iloc[0, 1] / 1000 # kW 转换为 MW
swept_area = match.iloc[0, 7] # 扫风面积
blade_diameter = match.iloc[0, 6] # 叶片直径
print(f"找到设备 '{device_name}'，额定功率: {rated_power} MW, "
f"扫风面积: {swept_area} m², 叶片直径: {blade_diameter} 米")
return rated_power, swept_area, blade_diameter
else:
raise ValueError(f"未找到设备名称: {device_name}")
except FileNotFoundError:
raise FileNotFoundError(f"文件未找到: {file_path}")
except Exception as e:
raise Exception(f"发生错误: {str(e)}")
def read_monthly_temperatures(file_path):
try:
with open(file_path, 'r', encoding='utf-8') as file:
temperatures = [float(line.strip()) for line in file.readlines()]
if len(temperatures) != 12:
raise ValueError(f"温度文件应包含 12 个月的数据，但实际有 {len(temperatures)} 条")
return temperatures
except Exception as e:
raise Exception(f"读取温度文件时出错: {str(e)}")
def air_density(altitude, hub_height, T0):
z = altitude + hub_height
LR = 0.0065
T = T0 - LR * z + 273.15
return (353.05 / T) * math.exp(-0.034 * (z / T))
def wind_power_density(densities, file_path):
try:
with open(file_path, 'r', encoding='utf-8') as file:
wind_speeds = [float(line.strip()) for line in file.readlines()]
if len(wind_speeds) != 12:
raise ValueError(f"风速文件应包含 12 个月的数据，但实际有 {len(wind_speeds)} 条")
sum_rho_v3 = sum(rho * (v ** 3) for rho, v in zip(densities, wind_speeds))
return (1 / (2 * 12)) * sum_rho_v3
except Exception as e:
raise Exception(f"读取风速文件时出错: {str(e)}")
def estimated_wind_power(num_turbines, rated_power):
if not isinstance(num_turbines, int) or num_turbines < 0:
raise ValueError("风机数量必须为非负整数")
return rated_power * num_turbines
def calculate_power_output(S, w, Cp, eta):
return w * S * Cp * 8760 * eta
def calculate_equivalent_hours(P, P_r):
if P_r == 0:
raise ValueError("额定功率不能为 0")
return (P / 1000) / (P_r * 1000)
def calculate_reference_yield(E_p, electricity_price, IC, q, n=20):
def npv_equation(irr, p, w, ic, q_val, n=n):
term1 = (1 + irr) ** (-1)
term2 = irr * (1 + irr) ** (-1) if irr != 0 else float('inf')
pv_revenue = p * w * (term1 / term2) * (1 - (1 + irr) ** (-n))
pv_salvage = q_val * ic * (term1 / term2) * (1 - (1 + irr) ** (-n))
return pv_revenue - ic + pv_salvage
irr_guess = 0.1
irr = float(fsolve(npv_equation, irr_guess, args=(E_p, electricity_price, IC, q))[0])
if not 0 <= irr <= 1:
raise ValueError(f"IRR计算结果{irr:.4f}不合理")
return irr * 100
try:
# 获取设备信息
rated_power, swept_area, blade_diameter = get_wind_turbine_specs(device_name, file_path)
# 估算风机数量
num_turbines = estimate_wind_turbine_count(area_km2, blade_diameter)
# 读取温度数据并计算空气密度
monthly_temps = read_monthly_temperatures(T_path)
densities = [air_density(altitude, hub_height, T0) for T0 in monthly_temps]
avg_density = sum(densities) / len(densities)
# 计算风功率密度
wpd = wind_power_density(densities, velocity_path)
# 计算装机容量
total_power = estimated_wind_power(num_turbines, rated_power)
# 计算初始投资成本
IC = total_power * cost_per_mw * 1000000
# 计算年发电量
P_test = calculate_power_output(swept_area, wpd, Cp, eta) * num_turbines
# 计算等效小时数
h = calculate_equivalent_hours(P_test, rated_power)
# 计算 IRR
irr = calculate_reference_yield(P_test, electricity_price, IC, q)
# 返回结果
return {
"device": device_name,
"rated_power": rated_power,
"swept_area": swept_area,
"blade_diameter": blade_diameter,
"num_turbines": num_turbines,
"avg_density": avg_density,
"wpd": wpd,
"total_power": total_power,
"annual_power_output": P_test / 10000000, # 万 kWh
"equivalent_hours": h,
"IRR": irr
}
except Exception as e:
raise Exception(f"风电场分析出错: {str(e)}")
# 主程序
if __name__ == "__main__":
file_path = r".\wind_product.xlsx"
velocity_path = r".\wind_speed.txt"
T_path = r".\temperature.txt"
device_name = input("请输入设备名称: ")
area_km2 = float(input("请输入风电场面积（平方公里）: "))
electricity_price = float(input("请输入电价（元/kWh）: "))
try:
result = wind_farm_analysis(
device_name=device_name,
area_km2=area_km2,
electricity_price=electricity_price,
file_path=file_path,
velocity_path=velocity_path,
T_path=T_path
)
print(f"\n设备: {result['device']}")
print(f"额定功率: {result['rated_power']:.2f} MW")
print(f"扫风面积: {result['swept_area']:.2f} m^2")
print(f"叶片直径: {result['blade_diameter']:.2f} m")
print(f"风机数量: {result['num_turbines']} 台")
print(f"平均空气密度: {result['avg_density']:.3f} kg/m^3")
print(f"风功率密度: {result['wpd']:.2f} W/m^2")
print(f"项目装机容量: {result['total_power']:.2f} MW")
print(f"年发电量: {result['annual_power_output']:.3f} 万 kWh")
print(f"等效小时数: {result['equivalent_hours']:.2f} 小时")
print(f"内部收益率 IRR: {result['IRR']:.2f}%")
except Exception as e:
print(f"错误: {str(e)}")

0
pv/__init__.py Normal file
View File

BIN
pv/__pycache__/__init__.cpython-310.pyc Normal file
View File

Binary file not shown.

BIN
pv/__pycache__/eva_pv.cpython-310.pyc Normal file
View File

Binary file not shown.

12
pv/config.json Normal file
View File

@ -0,0 +1,12 @@
{
"default_terrain_complexity": {
"耕地": 1.1,
"裸地": 1.1,
"草地": 1.2,
"灌木": 1.4,
"湿地": 1.65,
"林地": 1.65,
"建筑": 1.35,
"水域": 1.35
}
}

573
pv/eva_pv.py Normal file
View File

@ -0,0 +1,573 @@
import os
import sys
import pandas as pd
import math
import requests
from scipy.optimize import fsolve
# 获取当前文件的绝对路径
current_dir = os.path.dirname(os.path.abspath(__file__))
print(current_dir)
# 添加当前目录到sys.path
sys.path.append(current_dir)
# 默认文件路径
PV_EXCEL_PATH = f"{current_dir}/pv_product.xlsx" # 请确保此文件存在或更改为正确路径
# CONFIG_PATH = r"./config.json" # 配置文件路径
# 地形类型与复杂性因子范围
TERRAIN_COMPLEXITY_RANGES = {
"distributed": {
"耕地": (1.0, 1.2), "裸地": (1.0, 1.2), "草地": (1.1, 1.3),
"灌木": (1.3, 1.5), "湿地": (1.5, 1.8), "林地": (1.5, 1.8), "建筑": (1.2, 1.5)
},
"centralized": {
"耕地": (1.0, 1.2), "裸地": (1.0, 1.2), "草地": (1.1, 1.3),
"灌木": (1.3, 1.6), "湿地": (1.5, 1.8), "林地": (1.6, 2.0)
},
"floating": {"水域": (1.2, 1.5)}
}
# 地形类型与土地可用性、发电效率的映射
TERRAIN_ADJUSTMENTS = {
"耕地": {"land_availability": 0.85, "K": 0.8}, "裸地": {"land_availability": 0.85, "K": 0.8},
"草地": {"land_availability": 0.85, "K": 0.8}, "灌木": {"land_availability": 0.75, "K": 0.75},
"湿地": {"land_availability": 0.65, "K": 0.75}, "水域": {"land_availability": 0.85, "K": 0.8},
"林地": {"land_availability": 0.65, "K": 0.7}, "建筑": {"land_availability": 0.6, "K": 0.75}
}
# 光伏类型的装机容量上限 (MW/平方千米)
CAPACITY_LIMITS = {
"distributed": 25.0, "centralized": 50.0, "floating": 25.0
}
# 实际面板间距系数
PANEL_SPACING_FACTORS = {
"distributed": 1.5, "centralized": 1.2, "floating": 1.3
}
def get_slope_from_api(lat, lon):
"""
通过 OpenTopoData API 获取地形坡度（单位：度）。
返回：坡度（0-25°），失败时返回 None（由调用者处理）
"""
if not isinstance(lat, (int, float)) or not isinstance(lon, (int, float)):
print("警告：经纬度必须是数值")
return None
if not (-90 <= lat <= 90) or not (-180 <= lon <= 180):
print(f"警告：经纬度超出范围 (lat={lat}, lon={lon})")
return None
# 尝试多个数据集
datasets = ["srtm30m", "etopo1"]
step = 0.001 # 约100米
points = [
f"{lat:.6f},{lon:.6f}",
f"{lat + step:.6f},{lon:.6f}",
f"{lat - step:.6f},{lon:.6f}",
f"{lat:.6f},{lon + step:.6f}",
f"{lat:.6f},{lon - step:.6f}"
]
locations = "|".join(points)
for dataset in datasets:
url = f"https://api.opentopodata.org/v1/{dataset}"
params = {
"locations": locations,
"interpolation": "cubic"
}
print(f"发送请求：dataset={dataset}, locations={locations}")
try:
response = requests.get(url, params=params, timeout=10)
response.raise_for_status()
data = response.json()
if "results" not in data or len(data["results"]) != 5:
print(f"警告：{dataset} 返回无效数据: {data.get('error', '无错误信息')}")
continue
elevations = [result["elevation"] for result in data["results"]]
if any(elev is None for elev in elevations):
print(f"警告：{dataset} 高程数据包含空值")
continue
distance = 100
height_diffs = [
abs(elevations[1] - elevations[0]),
abs(elevations[2] - elevations[0]),
abs(elevations[3] - elevations[0]),
abs(elevations[4] - elevations[0])
]
avg_height_diff = sum(height_diffs) / len(height_diffs)
slope_rad = math.atan2(avg_height_diff, distance)
slope_deg = math.degrees(slope_rad)
slope_deg = min(max(slope_deg, 0), 25)
print(f"获取成功！坡度: {slope_deg:.2f}° (dataset={dataset})")
return slope_deg
except requests.exceptions.HTTPError as e:
print(f"警告：{dataset} 请求失败 (HTTP {e.response.status_code}): {e.response.text}")
continue
except requests.exceptions.RequestException as e:
print(f"警告：{dataset} 请求失败: {e}")
continue
except Exception as e:
print(f"警告：处理 {dataset} 数据出错: {e}")
continue
print("警告：所有数据集均失败")
return None
def calculate_psh_average(lat, lon, start_year=2010, end_year=2023):
"""
从 NASA POWER API 获取峰值日照小时数（PSH）。
返回：平均 PSH（小时/天），失败时返回默认值 4.0
"""
url = "https://power.larc.nasa.gov/api/temporal/monthly/point"
params = {
"parameters": "ALLSKY_SFC_SW_DWN",
"community": "RE",
"longitude": lon,
"latitude": lat,
"format": "JSON",
"start": str(start_year),
"end": str(end_year)
}
try:
response = requests.get(url, params=params, timeout=10)
response.raise_for_status()
data = response.json()
if "properties" not in data or "parameter" not in data["properties"]:
return 4.0
ghi_data = data["properties"]["parameter"].get("ALLSKY_SFC_SW_DWN", {})
if not ghi_data:
return 4.0
ghi_data = {k: v for k, v in ghi_data.items() if not k.endswith("13")}
if not ghi_data:
return 4.0
df = pd.DataFrame.from_dict(ghi_data, orient="index", columns=["GHI (kWh/m²/day)"])
new_index = [f"{k[:4]}-{k[-2:]:0>2}" for k in df.index]
df.index = new_index
if df.empty:
return 4.0
df["PSH (hours/day)"] = df["GHI (kWh/m²/day)"]
if df["PSH (hours/day)"].isna().any():
return 4.0
df['Year'] = df.index.str[:4]
annual_avg = df.groupby('Year')['PSH (hours/day)'].mean()
if annual_avg.empty:
return 4.0
psh = annual_avg.mean()
if math.isnan(psh):
return 4.0
print(f"获取成功！平均PSH: {psh:.2f} 小时/天")
return psh
except requests.exceptions.RequestException:
return 4.0
except Exception:
return 4.0
def calculate_optimal_tilt(lat):
"""根据纬度计算最佳倾角（单位：度）"""
try:
lat_abs = abs(lat)
if lat_abs < 25:
optimal_tilt = lat_abs * 0.87
elif lat_abs <= 50:
optimal_tilt = lat_abs * 0.76 + 3.1
else:
optimal_tilt = lat_abs * 0.5 + 16.3
return optimal_tilt
except ValueError as e:
raise Exception(f"倾角计算错误: {e}")
def pv_area(panel_capacity, slope_deg, shading_factor=0.1, land_compactness=1.0, terrain_complexity=1.0):
"""计算单块光伏组件占地面积"""
base_area = panel_capacity * 6
slope_factor = 1 + (slope_deg / 50) if slope_deg <= 15 else 1.5
shade_factor = 1 + shading_factor * 2
compact_factor = 1 / land_compactness if land_compactness > 0 else 1.5
terrain_factor = terrain_complexity
return base_area * slope_factor * shade_factor * compact_factor * terrain_factor
def calculate_pv_potential(available_area_sq_km, component_name, longitude, latitude, slope_deg=10,
shading_factor=0.1, land_compactness=0.8, terrain_complexity=1.2,
terrain_type="耕地", pv_type="centralized", land_availability=0.85,
min_irradiance=800, max_slope=25, electricity_price=0.65, q=0.02,
is_fixed=True, optimize=True, peak_load_hour=16, cost_per_kw=3.4,
E_S=1.0, K=0.8, project_lifetime=25, discount_rate=0.06):
"""计算最小和最大组件数量的光伏系统潜力"""
if available_area_sq_km <= 0:
raise ValueError("可用面积必须大于0")
if slope_deg < 0 or slope_deg > max_slope:
raise ValueError(f"坡度必须在0-{max_slope}度之间")
available_area_hectares = available_area_sq_km * 100
valid_terrains = TERRAIN_COMPLEXITY_RANGES.get(pv_type, {})
if terrain_type not in valid_terrains:
raise ValueError(f"{pv_type} 光伏不支持 {terrain_type} 地形。可选地形：{list(valid_terrains.keys())}")
terrain_adjustments = TERRAIN_ADJUSTMENTS.get(terrain_type, {"land_availability": 0.85, "K": 0.8})
adjusted_land_availability = terrain_adjustments["land_availability"] / max(1.0, terrain_complexity)
adjusted_K = terrain_adjustments["K"] / max(1.0, terrain_complexity)
pv_info = get_pv_product_info(component_name)
single_panel_capacity = pv_info["max_power"] / 1000
pv_size = pv_info["pv_size"].split("×")
panel_length = float(pv_size[0]) / 1000
panel_width = float(pv_size[1]) / 1000
panel_area_sqm = panel_length * panel_width
tilt, azimuth = get_tilt_and_azimuth(is_fixed, optimize, longitude, latitude, peak_load_hour)
array_distance = calculate_array_distance(panel_width * 1.1, tilt, latitude)
spacing_factor = PANEL_SPACING_FACTORS.get(pv_type, 1.2)
adjusted_array_distance = array_distance * spacing_factor
effective_area_hectares = available_area_hectares * adjusted_land_availability
effective_area_sqm = effective_area_hectares * 10000
area_per_mw = 10000 * (1 + slope_deg / 50 if slope_deg <= 15 else 1.5) * (
1 + shading_factor * 2) * terrain_complexity * spacing_factor
capacity_density_limit = CAPACITY_LIMITS.get(pv_type, 5.0) / 1000
max_capacity_by_density = effective_area_sqm * capacity_density_limit
row_spacing = panel_length * math.sin(math.radians(tilt)) + adjusted_array_distance
effective_panel_area = panel_area_sqm * (row_spacing / panel_length) * 1.2
min_area_per_panel = effective_panel_area * 0.8
max_area_per_panel = effective_panel_area * 1.5
max_panels = math.floor(effective_area_sqm / min_area_per_panel)
min_panels = math.floor(effective_area_sqm / max_area_per_panel)
max_capacity_raw = calculate_installed_capacity(pv_info["max_power"], max_panels)
min_capacity_raw = calculate_installed_capacity(pv_info["max_power"], min_panels)
max_capacity = min(max_capacity_raw, max_capacity_by_density)
min_capacity = min(min_capacity_raw, max_capacity_by_density * 0.8)
theoretical_max_capacity_mw = available_area_sq_km * CAPACITY_LIMITS.get(pv_type, 5.0)
if max_capacity / 1000 > theoretical_max_capacity_mw:
max_capacity = theoretical_max_capacity_mw * 1000
max_panels = math.floor(max_capacity * 1000 / pv_info["max_power"])
if min_capacity / 1000 > theoretical_max_capacity_mw * 0.8:
min_capacity = theoretical_max_capacity_mw * 1000 * 0.8
min_panels = math.floor(min_capacity * 1000 / pv_info["max_power"])
min_metrics = calculate_pv_metrics(
component_name=component_name, electricity_price=electricity_price, pv_number=min_panels,
q=q, longitude=longitude, latitude=latitude, is_fixed=is_fixed, optimize=optimize,
peak_load_hour=peak_load_hour, cost_per_kw=cost_per_kw * terrain_complexity, E_S=E_S, K=adjusted_K,
override_capacity=min_capacity
)
min_lcoe = calculate_lcoe(
capacity=min_metrics["capacity"], annual_energy=min_metrics["annual_energy"],
cost_per_kw=cost_per_kw * terrain_complexity, q=q, project_lifetime=project_lifetime,
discount_rate=discount_rate
)
max_metrics = calculate_pv_metrics(
component_name=component_name, electricity_price=electricity_price, pv_number=max_panels,
q=q, longitude=longitude, latitude=latitude, is_fixed=is_fixed, optimize=optimize,
peak_load_hour=peak_load_hour, cost_per_kw=cost_per_kw * terrain_complexity, E_S=E_S, K=adjusted_K,
override_capacity=max_capacity
)
max_lcoe = calculate_lcoe(
capacity=max_metrics["capacity"], annual_energy=max_metrics["annual_energy"],
cost_per_kw=cost_per_kw * terrain_complexity, q=q, project_lifetime=project_lifetime,
discount_rate=discount_rate
)
if min_panels == max_panels:
print(f"警告：最小和最大组件数量相同 ({min_panels})，请检查地形复杂性或面积是否过小")
if min_panels == 0 or max_panels == 0:
print(f"警告：组件数量为0，请检查输入参数")
out_metrics = {
"min_case": {
**min_metrics, "lcoe": min_lcoe, "actual_panels": min_panels,
"available_area_sq_km": available_area_sq_km, "available_area_hectares": available_area_hectares,
"effective_area_hectares": effective_area_hectares, "panel_area_sqm": max_area_per_panel,
"terrain_type": terrain_type, "pv_type": pv_type, "theoretical_max_capacity_mw": theoretical_max_capacity_mw
},
"max_case": {
**max_metrics, "lcoe": max_lcoe, "actual_panels": max_panels,
"available_area_sq_km": available_area_sq_km, "available_area_hectares": available_area_hectares,
"effective_area_hectares": effective_area_hectares, "panel_area_sqm": min_area_per_panel,
"terrain_type": terrain_type, "pv_type": pv_type, "theoretical_max_capacity_mw": theoretical_max_capacity_mw
}
}
return max_metrics
def calculate_lcoe(capacity, annual_energy, cost_per_kw, q, project_lifetime=25, discount_rate=0.06):
"""计算平准化度电成本(LCOE)"""
total_investment = capacity * cost_per_kw * 1000
annual_om_cost = total_investment * q
discount_factors = [(1 + discount_rate) ** -t for t in range(1, project_lifetime + 1)]
discounted_energy = sum(annual_energy * discount_factors[t] for t in range(project_lifetime))
discounted_costs = total_investment + sum(annual_om_cost * discount_factors[t] for t in range(project_lifetime))
if discounted_energy == 0:
return float('inf')
return discounted_costs / discounted_energy
def get_pv_product_info(component_name, excel_path=PV_EXCEL_PATH):
"""从Excel获取光伏组件信息"""
INDIAN_SITES = {
"Gujarat": {"latitude": 22.2587, "longitude": 71.1924},
"Rajasthan": {"latitude": 27.0238, "longitude": 74.2179},
"Tamil Nadu": {"latitude": 11.1271, "longitude": 78.6569}
}
try:
df = pd.read_excel(excel_path)
if len(df.columns) < 10:
raise ValueError("Excel文件需包含至少10列：组件名称、尺寸、功率等")
row = df[df.iloc[:, 1] == component_name]
if row.empty:
raise ValueError(f"未找到组件：{component_name}")
return {
"component_name": component_name,
"max_power": row.iloc[0, 5],
"efficiency": row.iloc[0, 9],
"pv_size": row.iloc[0, 3]
}
except FileNotFoundError:
raise FileNotFoundError(f"未找到Excel文件：{excel_path}")
except Exception as e:
raise Exception(f"读取Excel出错：{e}")
def get_tilt_and_azimuth(is_fixed=True, optimize=True, longitude=116, latitude=None, peak_load_hour=16):
"""计算光伏系统的倾角和方位角"""
if optimize and latitude is None:
raise ValueError("优化模式下需提供纬度")
if is_fixed:
if optimize:
tilt = calculate_optimal_tilt(latitude)
azimuth = (peak_load_hour - 12) * 15 + (longitude - 116)
azimuth = azimuth % 360 if azimuth >= 0 else azimuth + 360
else:
print("倾角：0°(水平)-90°(垂直) | 方位角：0°(正北)-180°(正南)，顺时针")
tilt = float(input("请输入倾角(度)："))
azimuth = float(input("请输入方位角(度)："))
if not (0 <= tilt <= 90) or not (0 <= azimuth <= 360):
raise ValueError("倾角需在0-90°，方位角需在0-360°")
else:
azimuth = 180
if optimize:
tilt = calculate_optimal_tilt(latitude)
else:
print("倾角：0°(水平)-90°(垂直)")
tilt = float(input("请输入倾角(度)："))
if not (0 <= tilt <= 90):
raise ValueError("倾角需在0-90°")
return tilt, azimuth
def calculate_array_distance(L, tilt, latitude):
"""计算阵列间距"""
beta_rad = math.radians(tilt)
phi_rad = math.radians(latitude)
return (L * math.cos(beta_rad) +
L * math.sin(beta_rad) * 0.707 * math.tan(phi_rad) +
0.4338 * math.tan(phi_rad))
def calculate_equivalent_hours(P, P_r):
"""计算等效小时数"""
if P_r == 0:
raise ValueError("额定功率不能为 0")
return P / P_r
def calculate_installed_capacity(max_power, num_components):
"""计算装机容量"""
if max_power < 0 or num_components < 0 or not isinstance(num_components, int):
raise ValueError("功率和数量需为非负数，数量需为整数")
return (max_power * num_components) / 1000
def calculate_annual_energy(peak_hours, capacity, E_S=1.0, K=0.8):
"""计算年发电量"""
if any(x < 0 for x in [peak_hours, capacity]) or E_S <= 0 or not 0 <= K <= 1:
raise ValueError("输入参数需满足：辐射量、容量≥0，E_S>0，K∈[0,1]")
return peak_hours * 365 * (capacity / E_S) * K
def calculate_environmental_benefits(E_p_million_kwh):
"""计算环境收益"""
if E_p_million_kwh < 0:
raise ValueError("年发电量需≥0")
return {
"coal_reduction": E_p_million_kwh * 0.404 * 10,
"CO2_reduction": E_p_million_kwh * 0.977 * 10,
"SO2_reduction": E_p_million_kwh * 0.03 * 10,
"NOX_reduction": E_p_million_kwh * 0.015 * 10
}
def calculate_reference_yield(E_p, electricity_price, IC, q, n=25):
"""计算净现值(NPV)和内部收益率(IRR)"""
if E_p < 0 or electricity_price < 0 or IC <= 0 or not 0 <= q <= 1:
raise ValueError("发电量、电价≥0，投资成本>0，回收比例∈[0,1]")
def npv_equation(irr, p, w, ic, q_val, n=n):
term1 = (1 + irr) ** (-1)
term2 = irr * (1 + irr) ** (-1) if irr != 0 else float('inf')
pv_revenue = p * w * (term1 / term2) * (1 - (1 + irr) ** (-n))
pv_salvage = q_val * ic * (term1 / term2) * (1 - (1 + irr) ** (-n))
return pv_revenue - ic + pv_salvage
irr_guess = 0.1
irr = float(fsolve(npv_equation, irr_guess, args=(E_p, electricity_price, IC, q))[0])
if not 0 <= irr <= 1:
raise ValueError(f"IRR计算结果{irr:.4f}不合理，请检查输入")
npv = npv_equation(irr, E_p, electricity_price, IC, q)
return {"NPV": npv, "IRR": irr * 100}
def calculate_pv_metrics(component_name, electricity_price, pv_number, q, longitude, latitude,
is_fixed=True, optimize=True, peak_load_hour=16, cost_per_kw=3.4, E_S=1.0, K=0.8,
override_capacity=None):
"""
__计算光伏项目的各项指标__
Raises:
Exception: _description_
Returns:
_type_: _description_
"""
try:
tilt, azimuth = get_tilt_and_azimuth(is_fixed, optimize, longitude, latitude, peak_load_hour)
pv_info = get_pv_product_info(component_name)
width_mm = float(pv_info["pv_size"].split("×")[1])
L = (width_mm / 1000) * 1.1
array_distance = calculate_array_distance(L, tilt, latitude)
max_power = pv_info["max_power"]
if override_capacity is not None:
capacity = override_capacity
else:
capacity = calculate_installed_capacity(max_power, pv_number)
peak_hours = calculate_psh_average(latitude, longitude)
single_daily_energy = peak_hours * (capacity / pv_number) * K if pv_number > 0 else 0
E_p = calculate_annual_energy(peak_hours, capacity, E_S, K)
h = calculate_equivalent_hours(E_p, capacity) if capacity > 0 else 0
E_p_million_kwh = E_p / 1000000
env_benefits = calculate_environmental_benefits(E_p_million_kwh)
IC = capacity * cost_per_kw * 1000
ref_yield = calculate_reference_yield(E_p, electricity_price, IC, q)
return {
"longitude": longitude,
"latitude": latitude,
"component_name": component_name,
"tilt": tilt,
"azimuth": azimuth,
"array_distance": array_distance,
"max_power": max_power,
"capacity": capacity,
"peak_sunshine_hours": peak_hours,
"single_daily_energy": single_daily_energy,
"annual_energy": E_p,
"equivalent_hours": h,
"coal_reduction": env_benefits["coal_reduction"],
"CO2_reduction": env_benefits["CO2_reduction"],
"SO2_reduction": env_benefits["SO2_reduction"],
"NOX_reduction": env_benefits["NOX_reduction"],
"IRR": ref_yield["IRR"]
}
except Exception as e:
raise Exception(f"计算过程中发生错误: {str(e)}")
# def print_result(min_case, max_case):
# """优化输出格式，使用表格展示结果"""
# headers = ["指标", "最小组件数量", "最大组件数量"]
# table_data = [
# ["经度", f"{min_case['longitude']:.2f}", f"{max_case['longitude']:.2f}"],
# ["纬度", f"{min_case['latitude']:.2f}", f"{max_case['latitude']:.2f}"],
# ["光伏类型", min_case["pv_type"], max_case["pv_type"]],
# ["地形类型", min_case["terrain_type"], max_case["terrain_type"]],
# ["组件型号", min_case["component_name"], max_case["component_name"]],
# ["识别面积 (平方千米)", f"{min_case['available_area_sq_km']:.2f}", f"{max_case['available_area_sq_km']:.2f}"],
# ["识别面积 (公顷)", f"{min_case['available_area_hectares']:.2f}", f"{max_case['available_area_hectares']:.2f}"],
# ["有效面积 (公顷)", f"{min_case['effective_area_hectares']:.2f}", f"{max_case['effective_area_hectares']:.2f}"],
# ["理论最大容量 (MW)", f"{min_case['theoretical_max_capacity_mw']:.2f}",
# f"{max_case['theoretical_max_capacity_mw']:.2f}"],
# ["单块组件占地 (m²)", f"{min_case['panel_area_sqm']:.2f}", f"{max_case['panel_area_sqm']:.2f}"],
# ["组件数量", f"{min_case['actual_panels']:,}", f"{max_case['actual_panels']:,}"],
# ["倾角 (度)", f"{min_case['tilt']:.2f}", f"{max_case['tilt']:.2f}"],
# ["方位角 (度)", f"{min_case['azimuth']:.2f}", f"{max_case['azimuth']:.2f}"],
# ["阵列间距 (m)", f"{min_case['array_distance']:.2f}", f"{max_case['array_distance']:.2f}"],
# ["单块功率 (Wp)", f"{min_case['max_power']}", f"{max_case['max_power']}"],
# ["装机容量 (MW)", f"{min_case['capacity'] / 1000:.2f}", f"{max_case['capacity'] / 1000:.2f}"],
# ["峰值日照 (小时/天)", f"{min_case['peak_sunshine_hours']:.2f}", f"{max_case['peak_sunshine_hours']:.2f}"],
# ["年发电量 (kWh)", f"{min_case['annual_energy']:,.0f}", f"{max_case['annual_energy']:,.0f}"],
# ["等效小时数", f"{min_case['equivalent_hours']:.2f}", f"{max_case['equivalent_hours']:.2f}"],
# ["LCOE (元/kWh)", f"{min_case['lcoe']:.4f}", f"{max_case['lcoe']:.4f}"],
# ["标准煤减排 (kg)", f"{min_case['coal_reduction']:,.0f}", f"{max_case['coal_reduction']:,.0f}"],
# ["CO₂减排 (kg)", f"{min_case['CO2_reduction']:,.0f}", f"{max_case['CO2_reduction']:,.0f}"],
# ["SO₂减排 (kg)", f"{min_case['SO2_reduction']:,.0f}", f"{max_case['SO2_reduction']:,.0f}"],
# ["NOx减排 (kg)", f"{min_case['NOX_reduction']:,.0f}", f"{max_case['NOX_reduction']:,.0f}"],
# ["IRR (%)", f"{min_case['IRR']:.2f}", f"{max_case['IRR']:.2f}"]
# ]
# print("\n===== 光伏系统潜力评估结果 =====")
# print(tabulate(table_data, headers=headers, tablefmt="grid"))
# 主程序
if __name__ == "__main__":
print("\n======= 光伏系统潜力评估 =======")
latitude = 45.3
longitude = 116.4
available_area_sq_km = 20
pv_type = "distributed"
terrain_type = "耕地"
terrain_config = {
"耕地": 1.1, "裸地": 1.1, "草地": 1.2, "灌木": 1.4,
"湿地": 1.65, "林地": 1.65, "建筑": 1.35, "水域": 1.35
}
component_name = "TWMND-72HD580"
electricity_price = 0.55
# 验证复杂性因子范围
terrain_complexity = terrain_config.get(terrain_type)
# 通过 API 获取坡度
slope_deg = get_slope_from_api(latitude, longitude)
print(f"使用参数：坡度={slope_deg:.2f}°，地形复杂性因子={terrain_complexity}")
# 计算光伏潜力
result = calculate_pv_potential(
available_area_sq_km=available_area_sq_km,
component_name=component_name,
longitude=longitude,
latitude=latitude,
slope_deg=slope_deg,
terrain_complexity=terrain_complexity,
pv_type=pv_type,
terrain_type=terrain_type,
electricity_price=electricity_price
)
print(result)
"""
{
"code": 200,
"data": {
"longitude": 123.2,
"latitude": 39,
"component_name": "TWMND-72HD580",
"tilt": 32.74,
"azimuth": 67.2,
"array_distance": 1.7867506003426947,
"max_power": 580,
"capacity": 849999.9999999999,
"peak_sunshine_hours": 3.9739880952380946,
"single_daily_energy": 0.5311220578210978,
"annual_energy": 896676222.9437226,
"equivalent_hours": 1054.9132034632032,
"coal_reduction": 3622.5719406926396,
"CO2_reduction": 8760.526698160169,
"SO2_reduction": 269.0028668831168,
"NOX_reduction": 134.5014334415584,
"IRR": 17.18080081007375
}
}
"""

0
PV/pv_product.xlsx → pv/pv_product.xlsx
View File

9
requirements.txt Normal file
View File

@ -0,0 +1,9 @@
numpy==1.22.0
pandas==1.5.3
scikit_learn==1.2.1
xlrd==2.0.1
logzero==1.7.0
scipy==1.11.4
flask==3.1.0
requests==2.31.0
openpyxl==3.1.2

90
run.py Normal file
View File

@ -0,0 +1,90 @@
# -*-coding:utf-8-*-
import os
import json
from flask import Flask, request, make_response, jsonify
from logzero import logger
current_path = os.path.dirname(os.path.abspath(__file__)) # for local
wind_product_path = f"{current_path}/wind/wind_product.xlsx"
# current_path = "/app" # for docker
logger.info(f"{current_path}")
app = Flask(__name__)
from pv.eva_pv import calculate_pv_potential, get_slope_from_api
from wind.wind_total import wind_farm_analysis
terrain_config = {
"耕地": 1.1, "裸地": 1.1, "草地": 1.2, "灌木": 1.4,
"湿地": 1.65, "林地": 1.65, "建筑": 1.35, "水域": 1.35
}
@app.route('/pv_power/', methods=["POST"])
def get_pv_potential():
resp_info = dict()
if request.method == "POST":
logger.info(request.get_json())
latitude = request.json.get('latitude')
longitude = request.json.get('longitude')
available_area_sq_km = float(request.json.get('available_area_sq_km'))
pv_type = request.json.get('pv_type')
terrain_type = request.json.get('terrain_type')
component_name = request.json.get('component_name')
electricity_price = float(request.json.get('electricity_price'))
terrain_complexity = terrain_config.get(terrain_type)
try:
# 通过 API 获取坡度
slope_deg = get_slope_from_api(latitude, longitude)
logger.info(f"使用参数：坡度={slope_deg:.2f}°，地形复杂性因子={terrain_complexity}")
pv_potential = calculate_pv_potential(
available_area_sq_km=available_area_sq_km,
component_name=component_name,
longitude=longitude,
latitude=latitude,
slope_deg=slope_deg,
terrain_complexity=terrain_complexity,
pv_type=pv_type,
terrain_type=terrain_type,
electricity_price=electricity_price
)
resp_info["code"] = 200
resp_info["data"] = pv_potential
except Exception as e:
logger.info(e)
resp_info["code"] = 406
resp_info["data"] = str(e)
resp = make_response(json.dumps(resp_info))
resp.status_code = 200
return resp
@app.route('/wind_power/', methods=["POST"])
def get_wind_potential():
resp_info = dict()
if request.method == "POST":
area_km2 = float(request.json.get('available_area_sq_km'))
device_name = request.json.get('component_name')
electricity_price = float(request.json.get('electricity_price'))
v_avg = float(request.json.get("velocity_avg"))
t_avg = float(request.json.get("temp_avg"))
try:
wind_potential = wind_farm_analysis(
device_name=device_name,
area_km2=area_km2,
electricity_price=electricity_price,
file_path=wind_product_path,
velocity_avg=v_avg,
T_avg=t_avg
)
resp_info["code"] = 200
resp_info["data"] = wind_potential
except Exception as e:
logger.info(e)
resp_info["code"] = 406
resp_info["data"] = str(e)
resp = make_response(json.dumps(resp_info))
resp.status_code = 200
return resp
if __name__ == '__main__':
app.run(port=12123, host="0.0.0.0", debug=False)

0
wind/__init__.py Normal file
View File

BIN
wind/__pycache__/__init__.cpython-310.pyc Normal file
View File

Binary file not shown.

BIN
wind/__pycache__/wind_total.cpython-310.pyc Normal file
View File

Binary file not shown.

230
wind/wind2.py Normal file
View File

@ -0,0 +1,230 @@
import pandas as pd
import math
from scipy.optimize import fsolve
import requests
import json
import os
def wind_farm_analysis(device_name, area_km2, electricity_price, file_path, latitude, longitude,
lateral_spacing_factor=5, longitudinal_spacing_factor=10, q=0.02, altitude=11,
hub_height=100, Cp=0.45, eta=0.8, cost_per_kw=5):
"""
封装函数：分析风电场的风机数量及各项经济和技术指标，使用 NASA POWER API 获取风速和温度数据
参数：
device_name (str): 设备名称
area_km2 (float): 风电场面积（平方公里）
electricity_price (float): 电价（元/kWh）
file_path (str): 风机参数 Excel 文件路径
latitude (float): 目标地点的纬度（度）
longitude (float): 目标地点的经度（度）
lateral_spacing_factor (float): 横向间距因子（默认为 5D）
longitudinal_spacing_factor (float): 纵向间距因子（默认为 10D）
q (float): 运维成本占初始投资成本的比例（默认 0.02 表示 2%）
altitude (float): 海拔高度（m），默认 11m
hub_height (float): 轮毂高度（m），默认 100m
Cp (float): 风能利用系数，默认 0.45
eta (float): 总系统效率，默认 0.8
cost_per_mw (float): 每 MW 投资成本（万元/MW），默认 5000 万元/MW
返回：
dict: 包含风电场分析结果的字典
"""
def estimate_wind_turbine_count(area_km2, blade_diameter):
area_m2 = area_km2 * 1_000_000
lateral_spacing = lateral_spacing_factor * blade_diameter
longitudinal_spacing = longitudinal_spacing_factor * blade_diameter
turbine_area = lateral_spacing * longitudinal_spacing
turbine_count = int(area_m2 / turbine_area)
print(f"单台风机占地面积: {turbine_area:,} 平方米 "
f"(横向间距: {lateral_spacing} 米, 纵向间距: {longitudinal_spacing} 米)")
print(f"估算风机数量: {turbine_count} 台")
return turbine_count
def get_wind_turbine_specs(device_name, file_path):
try:
df = pd.read_excel(file_path)
match = df[df.iloc[:, 0] == device_name]
if not match.empty:
rated_power = match.iloc[0, 1]
swept_area = match.iloc[0, 7] # 扫风面积
blade_diameter = match.iloc[0, 6] # 叶片直径
print(f"找到设备 '{device_name}'，额定功率: {rated_power} KW, "
f"扫风面积: {swept_area} m², 叶片直径: {blade_diameter} 米")
return rated_power, swept_area, blade_diameter
else:
raise ValueError(f"未找到设备名称: {device_name}")
except FileNotFoundError:
raise FileNotFoundError(f"文件未找到: {file_path}")
except Exception as e:
raise Exception(f"发生错误: {str(e)}")
def fetch_nasa_data(latitude, longitude, start_year="2023", end_year="2023"):
"""
从 NASA POWER API 获取 12 个月的平均气温和风速数据，不保存缓存
"""
try:
url = (f"https://power.larc.nasa.gov/api/temporal/monthly/point?"
f"parameters=T2M,WS10M&community=RE&longitude={longitude}&latitude={latitude}&"
f"start={start_year}&end={end_year}&format=JSON")
response = requests.get(url)
response.raise_for_status() # 检查请求是否成功
data = response.json()
# 提取气温和风速数据
year = start_year
temperatures = [data["properties"]["parameter"]["T2M"][f"{year}{str(i).zfill(2)}"] for i in range(1, 13)]
wind_speeds = [data["properties"]["parameter"]["WS10M"][f"{year}{str(i).zfill(2)}"] for i in range(1, 13)]
# 检查数据完整性
if len(temperatures) != 12 or len(wind_speeds) != 12:
raise ValueError("NASA 数据不完整，未包含 12 个月的数据")
return temperatures, wind_speeds
except requests.exceptions.HTTPError as http_err:
raise Exception(f"HTTP 错误: {str(http_err)}\n响应内容: {response.text}")
except Exception as e:
raise Exception(f"从 NASA POWER API 获取数据时出错: {str(e)}")
def adjust_wind_speed(v_10m, h_ref=10, h_hub=100, alpha=0.143):
"""根据风切变公式调整风速：v_hub = v_ref * (h_hub/h_ref)^alpha"""
return v_10m * (h_hub / h_ref) ** alpha
def air_density(altitude, hub_height, T0):
z = altitude + hub_height
LR = 0.0065
T = T0 - LR * z + 273.15
return (353.05 / T) * math.exp(-0.034 * (z / T))
def wind_power_density(densities, wind_speeds):
sum_rho_v3 = sum(rho * (v ** 3) for rho, v in zip(densities, wind_speeds))
return (1 / (2 * 12)) * sum_rho_v3
def estimated_wind_power(num_turbines, rated_power):
if not isinstance(num_turbines, int) or num_turbines < 0:
raise ValueError("风机数量必须为非负整数")
return rated_power * num_turbines
def calculate_power_output(S, w, Cp, eta,num_turbines):
return w * S * Cp * 8760 * eta *num_turbines
def calculate_equivalent_hours(P, P_r):
if P_r == 0:
raise ValueError("额定功率不能为 0")
#传入的P(发电量)wh,P_r(额定功率)Kw
return (P / 1000) / P_r
def calculate_environmental_benefits(E_p_million_kwh):
if E_p_million_kwh < 0:
raise ValueError("年发电量需≥0")
return {
"coal_reduction": E_p_million_kwh * 0.404 * 10,
"CO2_reduction": E_p_million_kwh * 0.977 * 10,
"SO2_reduction": E_p_million_kwh * 0.03 * 10,
"NOX_reduction": E_p_million_kwh * 0.015 * 10
}
def calculate_reference_yield(E_p, electricity_price, IC, q, n=20):
def npv_equation(irr, p, w, ic, q_val, n=n):
term1 = (1 + irr) ** (-1)
term2 = irr * (1 + irr) ** (-1) if irr != 0 else float('inf')
pv_revenue = p * w * (term1 / term2) * (1 - (1 + irr) ** (-n))
pv_salvage = q_val * ic * (term1 / term2) * (1 - (1 + irr) ** (-n))
return pv_revenue - ic + pv_salvage
irr_guess = 0.1
irr = float(fsolve(npv_equation, irr_guess, args=(E_p, electricity_price, IC, q))[0])
if not 0 <= irr <= 1:
raise ValueError(f"IRR计算结果{irr:.4f}不合理")
return irr * 100
# 获取设备信息
rated_power, swept_area, blade_diameter = get_wind_turbine_specs(device_name, file_path)
# 估算风机数量
num_turbines = estimate_wind_turbine_count(area_km2, blade_diameter)
# 从 NASA POWER API 获取气温和风速数据
monthly_temps, wind_speeds = fetch_nasa_data(latitude, longitude, start_year="2023", end_year="2023")
# 调整风速到轮毂高度
wind_speeds = [adjust_wind_speed(v) for v in wind_speeds]
# 计算空气密度
densities = [air_density(altitude, hub_height, T0) for T0 in monthly_temps]
avg_density = sum(densities) / len(densities)
# 计算风功率密度 w/m2
wpd = wind_power_density(densities, wind_speeds)
# 计算装机容量 KW
total_power = estimated_wind_power(num_turbines, rated_power)
# 计算初始投资成本
IC = total_power * cost_per_kw * 1000
# 计算年发电量 Wh
P_test = calculate_power_output(swept_area, wpd, Cp, eta,num_turbines)
# 计算等效小时数 年发电量(Wh)/额定功率(KW)
h = calculate_equivalent_hours(P_test, rated_power)
# 计算环境收益（转换为万 kWh）
E_p_million_kwh = P_test / 10000000 # 转换为 万 kWh
env_benefits = calculate_environmental_benefits(E_p_million_kwh)
# 计算 IRR
P_test_IRR = P_test/1000
irr = calculate_reference_yield(P_test_IRR, electricity_price, IC, q)
# 返回结果
return {
"device": device_name,
"rated_power": rated_power,
"swept_area": swept_area,
"blade_diameter": blade_diameter,
"num_turbines": num_turbines,
"avg_density": avg_density,
"wpd": wpd,
"total_power": total_power /1000, #变为了MW
"annual_power_output": P_test/10000000 , # 万 kWh
"equivalent_hours": h,
"coal_reduction": env_benefits["coal_reduction"],
"CO2_reduction": env_benefits["CO2_reduction"],
"SO2_reduction": env_benefits["SO2_reduction"],
"NOX_reduction": env_benefits["NOX_reduction"],
"IRR": irr
}
# 主程序
if __name__ == "__main__":
file_path = r"/home/zhaojh/workspace/GreenTransPowerCalculate/wind/wind_product.xlsx"
device_name = 'GW165-4.0'
area_km2 = 10
electricity_price = 0.6
latitude = 39
longitude = 116
result = wind_farm_analysis(
device_name=device_name,
area_km2=area_km2,
electricity_price=electricity_price,
file_path=file_path,
latitude=latitude,
longitude=longitude
)
print(f"\n设备: {result['device']}")
print(f"额定功率: {result['rated_power']:.2f} KW")
print(f"扫风面积: {result['swept_area']:.2f} m^2")
print(f"叶片直径: {result['blade_diameter']:.2f} m")
print(f"风机数量: {result['num_turbines']} 台")
print(f"平均空气密度: {result['avg_density']:.3f} kg/m^3")
print(f"风功率密度: {result['wpd']:.2f} W/m^2")
print(f"项目装机容量: {result['total_power']:.2f} MW")
print(f"年发电量: {result['annual_power_output']:.3f} 万 kWh")
print(f"等效小时数: {result['equivalent_hours']:.2f} 小时")
print(f"CO₂减排量：{result['CO2_reduction']:,.0f} kg")
print(f"SO₂减排量：{result['SO2_reduction']:,.0f} kg")
print(f"NOx减排量：{result['NOX_reduction']:,.0f} kg")
print(f"内部收益率 IRR: {result['IRR']:.2f}%")

0
Wind/wind_product.xlsx → wind/wind_product.xlsx
View File

204
wind/wind_total.py Normal file
View File

@ -0,0 +1,204 @@
import pandas as pd
import math
from scipy.optimize import fsolve
import os
import sys
# 获取当前文件的绝对路径
current_dir = os.path.dirname(os.path.abspath(__file__))
print(current_dir)
# 添加当前目录到sys.path
sys.path.append(current_dir)
def wind_farm_analysis(device_name, area_km2, electricity_price, file_path, velocity_avg, T_avg,
lateral_spacing_factor=5, longitudinal_spacing_factor=10, q=0.02, altitude=11,
hub_height=100, Cp=0.45, eta=0.8, cost_per_mw=5000):
"""
封装函数：分析风电场的风机数量及各项经济和技术指标
参数：
device_name (str): 设备名称
area_km2 (float): 风电场面积（平方公里）
electricity_price (float): 电价（元/kWh）
file_path (str): 风机参数 Excel 文件路径
velocity_avg (float): 全年平均风速
T_path (str): 全年平均温度
lateral_spacing_factor (float): 横向间距因子（默认为 5D）
longitudinal_spacing_factor (float): 纵向间距因子（默认为 10D）
q (float): 运维成本占初始投资成本的比例（默认 0.02 表示 2%）
altitude (float): 海拔高度（m），默认 11m
hub_height (float): 轮毂高度（m），默认 100m
Cp (float): 风能利用系数，默认 0.45
eta (float): 总系统效率，默认 0.8
cost_per_mw (float): 每 MW 投资成本（万元/MW），默认 5000 万元/MW
返回：
dict: 包含风电场分析结果的字典
"""
def estimate_wind_turbine_count(area_km2, blade_diameter):
area_m2 = area_km2 * 1_000_000
lateral_spacing = lateral_spacing_factor * blade_diameter
longitudinal_spacing = longitudinal_spacing_factor * blade_diameter
turbine_area = lateral_spacing * longitudinal_spacing
turbine_count = int(area_m2 / turbine_area)
print(f"单台风机占地面积: {turbine_area:,} 平方米 "
f"(横向间距: {lateral_spacing} 米, 纵向间距: {longitudinal_spacing} 米)")
print(f"估算风机数量: {turbine_count} 台")
return turbine_count
def get_wind_turbine_specs(device_name, file_path):
try:
df = pd.read_excel(file_path)
match = df[df.iloc[:, 0] == device_name]
if not match.empty:
rated_power = match.iloc[0, 1]
swept_area = match.iloc[0, 7] # 扫风面积
blade_diameter = match.iloc[0, 6] # 叶片直径
print(f"找到设备 '{device_name}'，额定功率: {rated_power} kW, "
f"扫风面积: {swept_area} m², 叶片直径: {blade_diameter} 米")
return rated_power, swept_area, blade_diameter
else:
raise ValueError(f"未找到设备名称: {device_name}")
except FileNotFoundError:
raise FileNotFoundError(f"文件未找到: {file_path}")
except Exception as e:
raise Exception(f"发生错误: {str(e)}")
def air_density(altitude, hub_height, T0):
z = altitude + hub_height
LR = 0.0065
T = T0 - LR * z + 273.15
return (353.05 / T) * math.exp(-0.034 * (z / T))
def wind_power_density(densities, velocity_avg):
rho_v3 = densities * velocity_avg
return 0.5 * rho_v3
def estimated_wind_power(num_turbines, rated_power):
if not isinstance(num_turbines, int) or num_turbines < 0:
raise ValueError("风机数量必须为非负整数")
return rated_power * num_turbines
def calculate_power_output(S, w, Cp, eta):
# 瓦时
return w * S * Cp * 8760 * eta
def calculate_equivalent_hours(P, P_r):
if P_r == 0:
raise ValueError("额定功率不能为 0")
return (P / 1000) / P_r
def calculate_environmental_benefits(E_p_million_kwh):
"""计算环境收益"""
if E_p_million_kwh < 0:
raise ValueError("年发电量需≥0")
return {
"coal_reduction": E_p_million_kwh * 0.404 * 10,
"CO2_reduction": E_p_million_kwh * 0.977 * 10,
"SO2_reduction": E_p_million_kwh * 0.03 * 10,
"NOX_reduction": E_p_million_kwh * 0.015 * 10
}
def calculate_reference_yield(E_p, electricity_price, IC, q, n=20):
def npv_equation(irr, p, w, ic, q_val, n=n):
term1 = (1 + irr) ** (-1)
term2 = irr * (1 + irr) ** (-1) if irr != 0 else float('inf')
pv_revenue = p * w * (term1 / term2) * (1 - (1 + irr) ** (-n))
pv_salvage = q_val * ic * (term1 / term2) * (1 - (1 + irr) ** (-n))
return pv_revenue - ic + pv_salvage
irr_guess = 0.1
irr = float(fsolve(npv_equation, irr_guess, args=(E_p, electricity_price, IC, q))[0])
if not 0 <= irr <= 1:
raise ValueError(f"IRR计算结果{irr:.4f}不合理")
return irr * 100
# 获取设备信息
rated_power, swept_area, blade_diameter = get_wind_turbine_specs(device_name, file_path)
# 估算风机数量
num_turbines = estimate_wind_turbine_count(area_km2, blade_diameter)
# 读取温度数据并计算空气密度
avg_density = air_density(altitude, hub_height, T_avg)
# 计算风功率密度
wpd = wind_power_density(avg_density, velocity_avg)
# 计算装机容量
total_power = estimated_wind_power(num_turbines, rated_power)
# 计算初始投资成本
IC = total_power * cost_per_mw * 1000000
# 计算年发电量 kwh
P_test = calculate_power_output(swept_area, wpd, Cp, eta) * num_turbines
# 计算等效小时数
h = calculate_equivalent_hours(P_test, rated_power)
# 计算 IRR
P_test_IRR = P_test/1000
irr = calculate_reference_yield(P_test_IRR, electricity_price, IC, q)
env_benefits = calculate_environmental_benefits((P_test / 10000000))
# 返回结果
out = {
"device": device_name,
"rated_power": rated_power,
"swept_area": swept_area,
"blade_diameter": blade_diameter,
"num_turbines": num_turbines,
"avg_density": avg_density,
"wpd": wpd,
"total_power": total_power,
"annual_power_output": P_test / 10000000, # 万 kWh
"equivalent_hours": h,
"IRR": irr
}
out.update(env_benefits)
return out
# 主程序
if __name__ == "__main__":
file_path = f"{current_dir}/wind_product.xlsx"
v_avg = 4.2
tavg = 15
device_name = "GW165-5.2"
area_km2 = 23.2
electricity_price = 0.45
result = wind_farm_analysis(
device_name=device_name,
area_km2=area_km2,
electricity_price=electricity_price,
file_path=file_path,
velocity_avg=v_avg,
T_avg=tavg
)
print(result)
"""
{
"code": 200,
"data": {
"device": "GW165-5.2",
"rated_power": 5.2,
"swept_area": 21382,
"blade_diameter": 165,
"num_turbines": 23,
"avg_density": 1.2118668826686871,
"wpd": 1.9995803564033336,
"total_power": 119.60000000000001,
"annual_power_output": 310.11418354861905,
"equivalent_hours": 596.3734299011904,
"IRR": 9.985793133871693,
"coal_reduction": 12528.61301536421,
"CO2_reduction": 30298.155732700077,
"SO2_reduction": 930.342550645857,
"NOX_reduction": 465.1712753229285
}
}
"""