import pandas as pd
from bc_combined_modelling import utils
from pathlib import Path
from bc_combined_modelling.attributes_parser import AttributesParser
[docs]
class FutureTechInvestments:
def __init__(self,
year,
scenario_name:str,
storage_algorithm:str='Kotzur',
timeslices:int=8,
solver_name:str='gurobi'):
self.year = year
self.scenario_name=scenario_name
self.storage_alogorithm=storage_algorithm
self.timeslices=timeslices
self.solver_name:str=solver_name
self.aparser=AttributesParser()
self.clewb_cfg = self.aparser.clewsb_config
self.get_data_paths()
self.clews_total_capacity=self.get_scenario_investment(timeslices)
self.PWR_capacity_year = self.clews_total_capacity[self.clews_total_capacity['TECHNOLOGY'].str.startswith('PWR') &
~self.clews_total_capacity['TECHNOLOGY'].str.contains('PWRTRN') &
(self.clews_total_capacity['YEAR'] == self.year)]
self.techs_wnd = {k: v for k, v in self.clewb_cfg['TECHNOLOGIES'].items() if 'WND' in k}
self.techs_sol = {k: v for k, v in self.clewb_cfg['TECHNOLOGIES'].items() if 'SOL' in k}
self.techs_nuc = {k: v for k, v in self.clewb_cfg['TECHNOLOGIES'].items() if 'URN' in k}
[docs]
def get_data_paths(self):
(
self.resource_options_wind,
self.resource_options_solar,
self.committed_wind,
self.committed_solar,
self.resource_options_wind_ts,
self.resource_options_solar_ts,
self.committed_wind_ts,
self.committed_solar_ts,
)=self.aparser.get_data_paths()
self.pypsa_resources_solar_year=Path('data/processed_data/solar/potential')
self.pypsa_resources_wind_year=Path('data/processed_data/wind/potential')
self.pypsa_committed_solar_year=Path('data/processed_data/solar/committed')
self.pypsa_committed_wind_year=Path('data/processed_data/wind/committed')
self.pypsa_resources_nuc_year=Path('data/processed_data/nuclear/potential')
self.pypsa_resources_nuc_year.mkdir(parents=True, exist_ok=True)
[docs]
def get_scenario_investment(self,timeslices):
self.clews_results_dir=self.aparser.get_bcnexus_scenario_results_path(scenario=self.scenario_name,
storage_algorithm=self.storage_alogorithm)
scenario_Total_capacity_annual_data_file='TotalCapacityAnnual.csv'
results_dir:Path=self.clews_results_dir/f'{timeslices}ts_csvs_{self.solver_name}'
self.clews_total_capacity:pd.DataFrame=pd.read_csv(results_dir/scenario_Total_capacity_annual_data_file)
return self.clews_total_capacity
[docs]
def get_future_techs(self,
techs:list,
no_of_existing_techs:int):
future_techs = {k: v for k, v in techs.items() if v.get('status') == 'future'}
return {k: v for k, v in future_techs.items() if int(k[-2:]) > int(no_of_existing_techs)}
[docs]
def get_existing_techs(self,
techs):
existing_techs:dict = {k: v for k, v in techs.items() if v.get('status') == 'existing'}
return existing_techs
[docs]
def create_mapping(self,
future_techs_resource_options:dict,
resource_options:pd.DataFrame,
output_path:str|Path):
mapping = pd.DataFrame(columns=['clews_id', 'name', 'operational_life', 'start_year', 'potential_capacity_GW', 'clews_investment_year', 'invested_capacity_MW'])
rows = []
for k_techid, v_info in future_techs_resource_options.items():
rows.append({
'clews_id': k_techid,
'name': v_info['name'],
'operational_life': v_info['operational_life'],
'start_year': v_info['start_year'],
'potential_capacity_GW': v_info['potential'],
'invested_capacity_MW': 1E3*self.PWR_capacity_year[self.PWR_capacity_year['TECHNOLOGY'] == k_techid].VALUE.values[0] if k_techid in self.PWR_capacity_year['TECHNOLOGY'].values else 0,
'clews_investment_year': self.year
})
mapping = pd.concat([mapping, pd.DataFrame(rows)])
mapping.set_index('name', inplace=True)
non_overlapping_cols = mapping.loc[:, ~mapping.columns.isin(resource_options.columns)]
resource_options = resource_options.join(non_overlapping_cols, how='left')
resource_options.to_csv(output_path)
[docs]
def create_mapping_nuclear(self,
future_techs_resource_options:dict,
resource_options:pd.DataFrame,
output_path:str|Path):
mapping = pd.DataFrame(columns=['clews_id', 'name', 'operational_life', 'clews_investment_year', 'invested_capacity_MW','capex','fom','vom'])
rows = []
for k_techid, v_info in future_techs_resource_options.items():
rows.append({
'clews_id': k_techid,
'name': v_info['name'],
'operational_life': v_info['operational_life'],
# 'start_year': v_info['start_year'],
# 'potential_capacity_GW': v_info['potential'],
'invested_capacity_MW': 1E3*self.PWR_capacity_year[self.PWR_capacity_year['TECHNOLOGY'] == k_techid].VALUE.values[0] if k_techid in self.PWR_capacity_year['TECHNOLOGY'].values else 0,
'clews_investment_year': self.year,
'capex': v_info['capital_cost'], # USD/kW # NREL ATB 2024, Moderate Scenario, Market price of 2035 , CAPEX 7863 USD/kW, FOM 136 USD/kW, VOM 2.6 USD/kWh
'fom': v_info['fom'], #USD/kWh
'vom': v_info['vom'] #USD/kWh
})
mapping = pd.concat([mapping, pd.DataFrame(rows)])
mapping.set_index('name', inplace=True)
if resource_options is None or resource_options.empty:
mapping.to_csv(output_path)
else:
non_overlapping_cols = mapping.loc[:, ~mapping.columns.isin(resource_options.columns)]
resource_options = resource_options.join(non_overlapping_cols, how='left')
resource_options.to_csv(output_path)
[docs]
def get_investments(self):
# Solar Resources
self.existing_tech_sol:dict=self.get_existing_techs(self.techs_sol)
self.future_techs_sol_resource_options = self.get_future_techs(self.techs_sol['PWRSOL'],
len(self.existing_tech_sol.keys()))
print(len(self.existing_tech_sol.keys()))
print(self.future_techs_sol_resource_options)
resource_options_solar = pd.read_csv(self.resource_options_solar, index_col='cluster_id')
self.create_mapping(self.future_techs_sol_resource_options,
resource_options_solar,
self.pypsa_resources_solar_year/f'resource_options_investments_solar_{self.year}.csv')
# Wind Resources
self.existing_tech_wnd:dict=self.get_existing_techs(self.techs_wnd)
self.future_techs_wnd_resource_options = self.get_future_techs(self.techs_wnd['PWRWND'],
len(self.existing_tech_wnd.keys()))
print(len(self.existing_tech_wnd.keys()))
resource_options_wind = pd.read_csv(self.resource_options_wind, index_col='cluster_id')
self.create_mapping(self.future_techs_wnd_resource_options,
resource_options_wind,
self.pypsa_resources_wind_year/f'resource_options_investments_wind_{self.year}.csv')
# Nuclear Resources
if self.techs_nuc:
self.existing_tech_nuc:dict=self.get_existing_techs(self.techs_nuc)
self.future_techs_nuc_resource_options = self.get_future_techs(self.techs_nuc['PWRURN'],
len(self.existing_tech_nuc.keys()))
print(len(self.existing_tech_nuc.keys()))
self.create_mapping_nuclear(self.future_techs_nuc_resource_options, None,
self.pypsa_resources_nuc_year/f'resource_options_investments_nuc_{self.year}.csv')
