Spaces:
Running
Running
| # -*- coding: utf-8 -*- | |
| """ | |
| Energy system optimization model | |
| HEMF EWL: Christopher Jahns, Julian Radek, Hendrik Kramer, Cornelia Klüter, Yannik Pflugfelder | |
| """ | |
| import numpy as np | |
| import pandas as pd | |
| import xarray as xr | |
| import plotly.express as px | |
| import streamlit as st | |
| from io import BytesIO | |
| import xlsxwriter | |
| from linopy import Model | |
| import sourced as src | |
| # Main function to run the Streamlit app | |
| def main(): | |
| """ | |
| Main function to set up and solve the energy system optimization model, and handle user inputs and outputs. | |
| """ | |
| setup_page() | |
| settings = load_settings() | |
| # fill session space with variables that are needed on all pages | |
| if 'settings' not in st.session_state: | |
| st.session_state.df = load_settings() | |
| st.session_state.settings = settings | |
| if 'url_excel' not in st.session_state: | |
| st.session_state.url_excel = None | |
| if 'ui_model' not in st.session_state: | |
| st.session_state.url_excel = None | |
| if 'output' not in st.session_state: | |
| st.session_state.output = BytesIO() | |
| setup_sidebar(st.session_state.settings["df"]) | |
| # Navigation | |
| pg = st.navigation([st.Page(page_model, title=st.session_state.settings["df"].loc['menu_modell',st.session_state.lang], icon="📊"), | |
| st.Page(page_documentation, title=st.session_state.settings["df"].loc['menu_doku',st.session_state.lang], icon="📓"), | |
| st.Page(page_about_us, title=st.session_state.settings["df"].loc['menu_impressum',st.session_state.lang], icon="💬")], | |
| expanded=True) | |
| # # Run the app | |
| pg.run() | |
| # Load settings and initial configurations | |
| def load_settings(): | |
| """ | |
| Load settings for the app, including colors and language information. | |
| """ | |
| settings = { | |
| 'write_pickle_from_standard_excel': True, | |
| 'df': pd.read_csv("language.csv", encoding="iso-8859-1", index_col="Label", sep=";"), | |
| 'color_dict': { | |
| 'Biomass': 'lightgreen', | |
| 'Lignite': 'brown', | |
| 'Fossil Gas': 'grey', | |
| 'Fossil Hard coal': 'darkgrey', | |
| 'Fossil Oil': 'maroon', | |
| 'RoR': 'aquamarine', | |
| 'Hydro Water Reservoir': 'azure', | |
| 'Nuclear': 'orange', | |
| 'PV': 'yellow', | |
| 'WindOff': 'darkblue', | |
| 'WindOn': 'green', | |
| 'H2': 'crimson', | |
| 'Pumped Hydro Storage': 'lightblue', | |
| 'Battery storages': 'red', | |
| 'Electrolyzer': 'olive' | |
| }, | |
| 'colors': { | |
| 'hemf_blau_dunkel': "#00386c", | |
| 'hemf_blau_hell': "#00529f", | |
| 'hemf_rot_dunkel': "#8b310d", | |
| 'hemf_rot_hell': "#d04119", | |
| 'hemf_grau': "#dadada" | |
| } | |
| } | |
| return settings | |
| # Initialize Streamlit app | |
| def setup_page(): | |
| """ | |
| Set up the Streamlit page with a specific layout, title, and favicon. | |
| """ | |
| st.set_page_config(layout="wide", page_title="Investment tool", page_icon="media/favicon.ico", initial_sidebar_state="expanded") | |
| # Sidebar for language and links | |
| def setup_sidebar(df): | |
| """ | |
| Set up the sidebar with language options and external links. | |
| """ | |
| st.session_state.lang = st.sidebar.selectbox("Language", ["🇬🇧 EN", "🇩🇪 DE"], key="foo", label_visibility="collapsed")[-2:] | |
| st.sidebar.markdown(""" | |
| <style> | |
| text-align: center; | |
| display: block; | |
| margin-left: auto; | |
| margin-right: auto; | |
| width: 100%; | |
| </style> | |
| """, unsafe_allow_html=True) | |
| with st.sidebar: | |
| left_co, cent_co, last_co = st.columns([0.1, 0.8, 0.1]) | |
| with cent_co: | |
| st.text(" ") # add vertical empty space | |
| ""+df.loc['menu_text', st.session_state.lang] | |
| st.text(" ") # add vertical empty space | |
| if st.session_state.lang == "DE": | |
| st.write("Schaue vorbei beim") | |
| st.markdown(r'[Lehrstuhl für Energiewirtschaft](https://www.ewl.wiwi.uni-due.de)', unsafe_allow_html=True) | |
| elif st.session_state.lang == "EN": | |
| st.write("Get in touch with the") | |
| st.markdown(r'[Chair of Management Science and Energy Economics](https://www.ewl.wiwi.uni-due.de/en)', unsafe_allow_html=True) | |
| st.text(" ") # add vertical empty space | |
| st.image("media/Logo_HEMF.svg", width=200) | |
| st.image("media/Logo_UDE.svg", width=200) | |
| # Load model input data | |
| def load_model_input(df, write_pickle_from_standard_excel): | |
| """ | |
| Load model input data from Excel or Pickle based on user input. | |
| """ | |
| if st.session_state.url_excel is None: | |
| if write_pickle_from_standard_excel: | |
| url_excel = r'Input_Jahr_2021.xlsx' | |
| sets_dict, params_dict = src.load_data_from_excel(url_excel, write_to_pickle_flag=True) | |
| sets_dict, params_dict = src.load_from_pickle() | |
| #st.write(df.loc['model_title1.1', st.session_state.lang]) | |
| # st.write('Running with standard data') | |
| else: | |
| url_excel = st.session_state.url_excel | |
| sets_dict, params_dict = src.load_data_from_excel(url_excel, load_from_pickle_flag=False) | |
| st.write(df.loc['model_title1.2', st.session_state.lang]) | |
| return sets_dict, params_dict | |
| def page_documentation(): | |
| """ | |
| Display documentation and mathematical model details. | |
| """ | |
| df = st.session_state.settings["df"] | |
| st.header(df.loc['constr_header1', st.session_state.lang]) | |
| st.write(df.loc['constr_header2', st.session_state.lang]) | |
| col1, col2 = st.columns([6, 4]) | |
| with col1: | |
| st.header(df.loc['constr_header3', st.session_state.lang]) | |
| with st.container(): | |
| # Objective function | |
| st.subheader(df.loc['constr_subheader_obj_func', st.session_state.lang]) | |
| st.write(df.loc['constr_subheader_obj_func_descr', st.session_state.lang]) | |
| st.latex(r''' \text{min } C^{tot} = C^{op} + C^{inv}''') | |
| # Operational costs minus revenue for produced hydrogen | |
| st.write(df.loc['constr_c_op', st.session_state.lang]) | |
| st.latex(r''' \sum_{i} y_{t,i} \cdot \left( \frac{c^{fuel}_{i}}{\eta_i} + c_{i}^{other} \right) \cdot \Delta t - \sum_{i \in \mathcal{I}^{PtG}} y^{h2}_{t,i} \cdot p^{h2} \cdot \Delta t = C^{op}''') | |
| # Investment costs | |
| st.write(df.loc['constr_c_inv', st.session_state.lang]) | |
| st.latex(r''' \sum_{i} a_{i} \cdot K_{i} \cdot c^{inv}_{i} = C^{inv}''') | |
| # Load-serving constraint | |
| st.write(df.loc['constr_load_serve', st.session_state.lang]) | |
| st.latex(r''' \left( \sum_{i} y_{t,i} - \sum_{i} y_{t,i}^{ch} \right) \cdot \Delta t = D_t \cdot \Delta t, \quad \forall t \in \mathcal{T}''') | |
| # Maximum capacity limit | |
| st.write(df.loc['constr_max_cap', st.session_state.lang]) | |
| st.latex(r''' y_{t,i} - K_{i} \leq K_{0,i}, \quad \forall i \in \mathcal{I}''') | |
| # Capacity limits for investment | |
| st.write(df.loc['constr_inv_cap', st.session_state.lang]) | |
| st.latex(r''' K_{i} \leq 0, \quad \forall i \in \mathcal{I}^{no\_invest}''') | |
| # Prevent power production by PtG | |
| st.write(df.loc['constr_prevent_ptg', st.session_state.lang]) | |
| st.latex(r''' y_{t,i} = 0, \quad \forall i \in \mathcal{I}^{PtG}''') | |
| # Prevent charging for non-storage technologies | |
| st.write(df.loc['constr_prevent_chg', st.session_state.lang]) | |
| st.latex(r''' y_{t,i}^{ch} = 0, \quad \forall i \in \mathcal{I} \setminus \{ \mathcal{I}^{PtG} \cup \mathcal{I}^{Sto} \}''') | |
| # Maximum storage charging and discharging | |
| st.write(df.loc['constr_max_chg', st.session_state.lang]) | |
| st.latex(r''' y_{t,i} + y_{t,i}^{ch} - K_{i} \leq K_{0,i}, \quad \forall i \in \mathcal{I}^{Sto}''') | |
| # Maximum electrolyzer capacity | |
| st.write(df.loc['constr_max_cap_electrolyzer', st.session_state.lang]) | |
| st.latex(r''' y_{t,i}^{ch} - K_{i} \leq K_{0,i}, \quad \forall i \in \mathcal{I}^{PtG}''') | |
| # PtG H2 production | |
| st.write(df.loc['constr_prod_ptg', st.session_state.lang]) | |
| st.latex(r''' y_{t,i}^{ch} \cdot \eta_i = y_{t,i}^{h2}, \quad \forall i \in \mathcal{I}^{PtG}''') | |
| # Infeed of renewables | |
| st.write(df.loc['constr_inf_res', st.session_state.lang]) | |
| st.latex(r''' y_{t,i} + y_{t,i}^{curt} = s_{t,r,i} \cdot (K_{0,i} + K_i), \quad \forall i \in \mathcal{I}^{Res}''') | |
| # Maximum filling level restriction for storage power plants | |
| st.write(df.loc['constr_max_fil_sto', st.session_state.lang]) | |
| # st.latex(r''' l_{t,i} \leq K_{0,i} \cdot e2p_i, \quad \forall i \in \mathcal{I}^{Sto}''') | |
| st.latex(r''' l_{t,i} \leq (K_{0,i} + K_{i}) \cdot \gamma_i^{Sto}, \quad \forall i \in \mathcal{I}^{Sto}''') | |
| # Filling level restriction for hydro reservoir | |
| st.write(df.loc['constr_fil_hyres', st.session_state.lang]) | |
| st.latex(r''' l_{t+1,i} = l_{t,i} + ( h_{t,i} - y_{t,i}) \cdot \Delta t, \quad \forall i \in \mathcal{I}^{HyRes}''') | |
| # Filling level restriction for other storages | |
| st.write(df.loc['constr_fil_sto', st.session_state.lang]) | |
| st.latex(r''' l_{t+1,i} = l_{t,i} - \left(\frac{y_{t,i}}{\eta_i} - y_{t,i}^{ch} \cdot \eta_i \right) \cdot \Delta t, \quad \forall i \in \mathcal{I}^{Sto}''') | |
| # CO2 emission constraint | |
| st.write(df.loc['constr_co2_lim', st.session_state.lang]) | |
| st.latex(r''' \sum_{t} \sum_{i} \frac{y_{t,i}}{\eta_i} \cdot \chi^{CO2}_i \cdot \Delta t \leq L^{CO2}''') | |
| with col2: | |
| symbols_container = st.container() | |
| with symbols_container: | |
| st.header(df.loc['symb_header1', st.session_state.lang]) | |
| st.write(df.loc['symb_header2', st.session_state.lang]) | |
| st.subheader(df.loc['symb_header_sets', st.session_state.lang]) | |
| st.write(f"$\mathcal{{T}}$: {df.loc['symb_time_steps', st.session_state.lang]}") | |
| st.write(f"$\mathcal{{I}}$: {df.loc['symb_tech', st.session_state.lang]}") | |
| st.write(f"$\mathcal{{I}}^{{\\text{{Sto}}}}$: {df.loc['symb_sto_tech', st.session_state.lang]}") | |
| st.write(f"$\mathcal{{I}}^{{\\text{{Conv}}}}$: {df.loc['symb_conv_tech', st.session_state.lang]}") | |
| st.write(f"$\mathcal{{I}}^{{\\text{{PtG}}}}$: {df.loc['symb_ptg', st.session_state.lang]}") | |
| st.write(f"$\mathcal{{I}}^{{\\text{{Res}}}}$: {df.loc['symb_res', st.session_state.lang]}") | |
| st.write(f"$\mathcal{{I}}^{{\\text{{HyRes}}}}$: {df.loc['symb_hyres', st.session_state.lang]}") | |
| st.write(f"$\mathcal{{I}}^{{\\text{{no\_invest}}}}$: {df.loc['symb_no_inv', st.session_state.lang]}") | |
| # Variables section | |
| st.subheader(df.loc['symb_header_variables', st.session_state.lang]) | |
| st.write(f"$C^{{tot}}$: {df.loc['symb_tot_costs', st.session_state.lang]}") | |
| st.write(f"$C^{{op}}$: {df.loc['symb_c_op', st.session_state.lang]}") | |
| st.write(f"$C^{{inv}}$: {df.loc['symb_c_inv', st.session_state.lang]}") | |
| st.write(f"$K_i$: {df.loc['symb_inst_cap', st.session_state.lang]}") | |
| st.write(f"$y_{{t,i}}$: {df.loc['symb_el_prod', st.session_state.lang]}") | |
| st.write(f"$y_{{t, i}}^{{ch}}$: {df.loc['symb_el_ch', st.session_state.lang]}") | |
| st.write(f"$l_{{t,i}}$: {df.loc['symb_sto_fil', st.session_state.lang]}") | |
| st.write(f"$y_{{t, i}}^{{curt}}$: {df.loc['symb_curt', st.session_state.lang]}") | |
| st.write(f"$y_{{t, i}}^{{h2}}$: {df.loc['symb_h2_ptg', st.session_state.lang]}") | |
| # Parameters section | |
| st.subheader(df.loc['symb_header_parameters', st.session_state.lang]) | |
| st.write(f"$D_t$: {df.loc['symb_energy_demand', st.session_state.lang]}") | |
| st.write(f"$p^{{h2}}$: {df.loc['symb_price_h2', st.session_state.lang]}") | |
| st.write(f"$c^{{fuel}}_{{i}}$: {df.loc['symb_fuel_costs', st.session_state.lang]}") | |
| st.write(f"$c_{{i}}^{{other}}$: {df.loc['symb_c_op_other', st.session_state.lang]}") | |
| st.write(f"$c^{{inv}}_{{i}}$: {df.loc['symb_c_inv_tech', st.session_state.lang]}") | |
| st.write(f"$a_{{i}}$: {df.loc['symb_annuity', st.session_state.lang]}") | |
| st.write(f"$\eta_i$: {df.loc['symb_eff_fac', st.session_state.lang]}") | |
| st.write(f"$K_{{0,i}}$: {df.loc['symb_max_cap_tech', st.session_state.lang]}") | |
| st.write(f"$\chi^{{CO2}}_i$: {df.loc['symb_co2_fac', st.session_state.lang]}") | |
| st.write(f"$L^{{CO2}}$: {df.loc['symb_co2_limit', st.session_state.lang]}") | |
| # st.write(f"$e2p_{{\\text{{Sto}}, i}}$: {df.loc['symb_etp', st.session_state.lang]}") | |
| st.write(f"$\gamma^{{\\text{{Sto}}}}_{{i}}$: {df.loc['symb_etp', st.session_state.lang]}") | |
| st.write(f"$s_{{t, r, i}}$: {df.loc['symb_res_supply', st.session_state.lang]}") | |
| st.write(f"$h_{{t, i}}$: {df.loc['symb_hyRes_inflow', st.session_state.lang]}") | |
| # css = float_css_helper(top="50") | |
| # symbols_container.float(css) | |
| def page_about_us(): | |
| """ | |
| Display information about the team and the project. | |
| """ | |
| st.write("About Us/Impressum") | |
| def page_model(): #, write_pickle_from_standard_excel, color_dict): | |
| """ | |
| Display the main model page for energy system optimization. | |
| This function sets up the user interface for the model input parameters, loads data, and configures the | |
| optimization model before solving it and presenting the results. | |
| """ | |
| df = st.session_state.settings["df"] | |
| color_dict = st.session_state.settings["color_dict"] | |
| write_pickle_from_standard_excel = st.session_state.settings["write_pickle_from_standard_excel"] | |
| # Load data from Excel or Pickle | |
| sets_dict, params_dict = load_model_input(df, write_pickle_from_standard_excel) | |
| # Unpack sets_dict into the workspace | |
| t = sets_dict['t'] | |
| t_original = sets_dict['t'] | |
| i = sets_dict['i'] | |
| iSto = sets_dict['iSto'] | |
| iConv = sets_dict['iConv'] | |
| iPtG = sets_dict['iPtG'] | |
| iRes = sets_dict['iRes'] | |
| iHyRes = sets_dict['iHyRes'] | |
| # Unpack params_dict into the workspace | |
| l_co2 = params_dict['l_co2'] | |
| p_co2 = params_dict['p_co2'] | |
| eff_i = params_dict['eff_i'] | |
| life_i = params_dict['life_i'] | |
| c_fuel_i = params_dict['c_fuel_i'] | |
| c_other_i = params_dict['c_other_i'] | |
| c_inv_i = params_dict['c_inv_i'] | |
| co2_factor_i = params_dict['co2_factor_i'] | |
| K_0_i = params_dict['K_0_i'] | |
| e2p_iSto = params_dict['e2p_iSto'] | |
| # Adjust efficiency for storage technologies | |
| eff_i.loc[iSto] = np.sqrt(eff_i.loc[iSto]) # Apply square root to cycle efficiency for storage technologies | |
| # Create columns for UI layout | |
| col1, col2 = st.columns([0.30, 0.70], gap="large") | |
| # Load input data | |
| with col1: | |
| st.title(df.loc['model_title1', st.session_state.lang]) | |
| with open('Input_Jahr_2021.xlsx', 'rb') as f: | |
| st.download_button(df.loc['model_title1.3',st.session_state.lang], f, file_name='Input_Jahr_2021.xlsx') # Download button for Excel template | |
| st.session_state.url_excel = st.file_uploader(label=df.loc['model_title1.4',st.session_state.lang]) # File uploader for user Excel file | |
| st.title(df.loc['model_title2', st.session_state.lang]) | |
| st.download_button(label=df.loc['model_title2.1',st.session_state.lang], disabled=(st.session_state.output.getbuffer().nbytes==0), data=st.session_state.output.getvalue(), file_name="workbook.xlsx", mime="application/vnd.ms-excel") | |
| st.title(df.loc['model_title4', st.session_state.lang]) | |
| if st.session_state.url_excel: | |
| run_model = st.button(df.loc['model_run_info_excel', st.session_state.lang], key="run_model_button", help=df.loc['run_model_button_info',st.session_state.lang]) | |
| else: | |
| run_model = st.button(df.loc['model_run_info_gui', st.session_state.lang], key="run_model_button", help=df.loc['run_model_button_info',st.session_state.lang]) | |
| # Set up user interface for parameters | |
| with col2: | |
| st.title(df.loc['model_title3', st.session_state.lang]) | |
| col1param, col2param = st.columns(2) | |
| with col1param: | |
| l_co2 = st.slider(value=int(params_dict['l_co2']), min_value=0, max_value=750, label=df.loc['model_label_co2',st.session_state.lang], step=50) | |
| price_h2 = st.slider(value=100, min_value=0, max_value=300, label=df.loc['model_label_h2',st.session_state.lang], step=10) | |
| for i_idx in params_dict['c_fuel_i'].get_index('i'): | |
| if i_idx in ['Lignite']: | |
| params_dict['c_fuel_i'].loc[i_idx] = st.slider(value=int(params_dict['c_fuel_i'].loc[i_idx]), | |
| min_value=0, max_value=300, label=df.loc[f'model_label_{i_idx}',st.session_state.lang], step=10) | |
| dt = st.number_input(label=df.loc['model_label_t',st.session_state.lang], min_value=1, max_value=len(t), value=6, | |
| help=df.loc['model_label_t_info',st.session_state.lang]) | |
| with col2param: | |
| for i_idx in params_dict['c_fuel_i'].get_index('i'): | |
| if i_idx in ['Fossil Hard coal', 'Fossil Oil', 'Fossil Gas']: | |
| params_dict['c_fuel_i'].loc[i_idx] = st.slider(value=int(params_dict['c_fuel_i'].loc[i_idx]), | |
| min_value=0, max_value=300, label=df.loc[f'model_label_{i_idx}',st.session_state.lang], step=10) | |
| # Create a dictionary to map German names to English names | |
| tech_mapping_de_to_en = { | |
| df.loc[f'tech_{tech.lower()}', 'DE']: df.loc[f'tech_{tech.lower()}', 'EN'] | |
| for tech in sets_dict['i'] if f'tech_{tech.lower()}' in df.index | |
| } | |
| # Set options and default values based on the selected language | |
| if st.session_state.lang == 'DE': | |
| # German options for the user interface | |
| options = [ | |
| df.loc[f'tech_{tech.lower()}', 'DE'] for tech in sets_dict['i'] if f'tech_{tech.lower()}' in df.index | |
| ] | |
| default = [ | |
| df.loc[f'tech_{tech.lower()}', 'DE'] for tech in ['Lignite', 'Fossil Gas', 'Fossil Hard coal', 'Fossil Oil', 'PV', 'WindOff', 'WindOn', 'H2', 'Pumped Hydro Storage', 'Battery storages', 'Electrolyzer'] | |
| if f'tech_{tech.lower()}' in df.index | |
| ] | |
| else: | |
| # English options for the user interface | |
| options = sets_dict['i'] | |
| default = ['Lignite', 'Fossil Gas', 'Fossil Hard coal', 'Fossil Oil', 'PV', 'WindOff', 'WindOn', 'H2', 'Pumped Hydro Storage', 'Battery storages', 'Electrolyzer'] | |
| # Multiselect for technology options in the user interface | |
| selected_technologies = st.multiselect( | |
| label=df.loc['model_label_tech', st.session_state.lang], | |
| options=options, | |
| default=[tech for tech in default if tech in options] | |
| ) | |
| # If language is German, map selected German names back to their English equivalents | |
| if st.session_state.lang == 'DE': | |
| technologies_invest = [tech_mapping_de_to_en[tech] for tech in selected_technologies] | |
| else: | |
| technologies_invest = selected_technologies | |
| # Technologies that will not be invested in (based on English names) | |
| technologies_no_invest = [tech for tech in sets_dict['i'] if tech not in technologies_invest] | |
| st.markdown("-------") | |
| # Time series aggregation for various parameters | |
| D_t = timstep_aggregate(dt, params_dict['D_t'], t) | |
| s_t_r_iRes = timstep_aggregate(dt, params_dict['s_t_r_iRes'], t) | |
| h_t = timstep_aggregate(dt, params_dict['h_t'], t) | |
| t = D_t.get_index('t') | |
| partial_year_factor = (8760 / len(t)) / dt | |
| if run_model: | |
| # Model setup | |
| m = Model() | |
| # Define Variables | |
| C_tot = m.add_variables(name='C_tot') # Total costs | |
| C_op = m.add_variables(name='C_op', lower=0) # Operational costs | |
| C_inv = m.add_variables(name='C_inv', lower=0) # Investment costs | |
| K = m.add_variables(coords=[i], name='K', lower=0) # Endogenous capacity | |
| y = m.add_variables(coords=[t, i], name='y', lower=0) # Electricity production | |
| y_ch = m.add_variables(coords=[t, i], name='y_ch', lower=0) # Electricity consumption | |
| l = m.add_variables(coords=[t, i], name='l', lower=0) # Storage filling level | |
| y_curt = m.add_variables(coords=[t, i], name='y_curt', lower=0) # RES curtailment | |
| y_h2 = m.add_variables(coords=[t, i], name='y_h2', lower=0) # H2 production | |
| # Define Objective function | |
| C_tot = C_op + C_inv | |
| m.add_objective(C_tot) | |
| # Define Constraints | |
| # Operational costs minus revenue for produced hydrogen | |
| m.add_constraints((y * c_fuel_i / eff_i).sum() * dt - (y_h2.sel(i=iPtG) * price_h2).sum() * dt == C_op, name='C_op_sum') | |
| # Investment costs | |
| m.add_constraints((K * c_inv_i).sum() == C_inv, name='C_inv_sum') | |
| # Load serving | |
| m.add_constraints((((y).sum(dims='i') - y_ch.sum(dims='i')) * dt == D_t.sel(t=t) * dt), name='load') | |
| # Maximum capacity limit | |
| m.add_constraints((y - K <= K_0_i), name='max_cap') | |
| # Capacity limits for investment | |
| m.add_constraints((K.sel(i=technologies_no_invest) <= 0), name='max_cap_invest') | |
| # Prevent power production by PtG | |
| m.add_constraints((y.sel(i=iPtG) <= 0), name='prevent_ptg_prod') | |
| # Prevent charging for non-storage technologies | |
| m.add_constraints((y_ch.sel(i=[x for x in i if x not in iPtG and x not in iSto]) <= 0), name='no_charging') | |
| # Maximum storage charging and discharging | |
| m.add_constraints((y.sel(i=iSto) + y_ch.sel(i=iSto) - K.sel(i=iSto) <= K_0_i.sel(i=iSto)), name='max_cha') | |
| # Maximum electrolyzer capacity | |
| m.add_constraints((y_ch.sel(i=iPtG) - K.sel(i=iPtG) <= K_0_i.sel(i=iPtG)), name='max_cha_ptg') | |
| # PtG H2 production | |
| m.add_constraints(y_ch.sel(i=iPtG) * eff_i.sel(i=iPtG) == y_h2.sel(i=iPtG), name='ptg_h2_prod') | |
| # Infeed of renewables | |
| m.add_constraints((y.sel(i=iRes) - s_t_r_iRes.sel(i=iRes).sel(t=t) * K.sel(i=iRes) + y_curt.sel(i=iRes) == s_t_r_iRes.sel(i=iRes).sel(t=t) * K_0_i.sel(i=iRes)), name='infeed') | |
| # Maximum filling level restriction for storage power plants | |
| m.add_constraints((l.sel(i=iSto) - K.sel(i=iSto) * e2p_iSto.sel(i=iSto) <= K_0_i.sel(i=iSto) * e2p_iSto.sel(i=iSto)), name='max_sto_filling') | |
| # Filling level restriction for hydro reservoir | |
| m.add_constraints(l.sel(i=iHyRes) - l.sel(i=iHyRes).roll(t=-1) + y.sel(i=iHyRes) * dt == h_t.sel(t=t) * dt, name='filling_level_hydro') | |
| # Filling level restriction for other storages | |
| m.add_constraints(l.sel(i=iSto) - (l.sel(i=iSto).roll(t=-1) - (y.sel(i=iSto) / eff_i.sel(i=iSto)) * dt + y_ch.sel(i=iSto) * eff_i.sel(i=iSto) * dt) == 0, name='filling_level') | |
| # CO2 limit | |
| m.add_constraints(((y / eff_i) * co2_factor_i * dt).sum() <= l_co2 * 1_000_000, name='CO2_limit') | |
| # Solve the model | |
| m.solve(solver_name='highs') | |
| # Prepare columns for figures | |
| colb1, colb2 = st.columns(2) | |
| # Generate and display figures | |
| st.markdown("---") | |
| df_total_costs = plot_total_costs(m, colb1, df) | |
| df_CO2_price = plot_co2_price(m, colb2, df) | |
| plot_installed_capacities(m, K_0_i, color_dict) | |
| df_new_capacities = plot_new_capacities(m, color_dict, colb1, df) | |
| # Only plot production for technologies with capacity | |
| i_with_capacity = m.solution['K'].where(m.solution['K'] > 0).dropna(dim='i').get_index('i') | |
| df_production = plot_production(m, i_with_capacity, dt, color_dict, colb2, df) | |
| df_price = plot_electricity_prices(m, dt, colb1, df) | |
| df_contr_marg = plot_contribution_margin(m, dt, color_dict, colb2, df) | |
| df_curtailment = plot_curtailment(m, iRes, color_dict, colb1, df) | |
| df_charging = plot_storage_charging(m, iSto, color_dict, colb2, df) | |
| df_h2_prod = plot_hydrogen_production(m, iPtG, color_dict, colb2, df) | |
| # Export results | |
| st.session_state.output = BytesIO() | |
| with pd.ExcelWriter(st.session_state.output, engine='xlsxwriter') as writer: | |
| disaggregate_df(df_total_costs, t, t_original, dt).to_excel(writer, sheet_name='Total costs', index=False) | |
| disaggregate_df(df_CO2_price, t, t_original, dt).to_excel(writer, sheet_name='CO2 price', index=False) | |
| disaggregate_df(df_price, t, t_original, dt).to_excel(writer, sheet_name='Prices', index=False) | |
| disaggregate_df(df_contr_marg, t, t_original, dt).to_excel(writer, sheet_name='Contribution Margin', index=False) | |
| disaggregate_df(df_new_capacities, t, t_original, dt).to_excel(writer, sheet_name='Capacities', index=False) | |
| disaggregate_df(df_production, t, t_original, dt).to_excel(writer, sheet_name='Production', index=False) | |
| disaggregate_df(df_charging, t, t_original, dt).to_excel(writer, sheet_name='Charging', index=False) | |
| disaggregate_df(D_t.to_dataframe().reset_index(), t, t_original, dt).to_excel(writer, sheet_name='Demand', index=False) | |
| disaggregate_df(df_curtailment, t, t_original, dt).to_excel(writer, sheet_name='Curtailment', index=False) | |
| disaggregate_df(df_h2_prod, t, t_original, dt).to_excel(writer, sheet_name='H2 production', index=False) | |
| st.rerun() | |
| def timstep_aggregate(time_steps_aggregate, xr_data, t): | |
| """ | |
| Aggregates time steps in the data using rolling mean and selects based on step size. | |
| """ | |
| return xr_data.rolling(t=time_steps_aggregate).mean().sel(t=t[0::time_steps_aggregate]) | |
| # Visualization functions | |
| def plot_installed_capacities(m, K_0_i, color_dict): | |
| """ | |
| Plots the total installed capacities. | |
| """ | |
| df_installed_cap = (m.solution['K'] + K_0_i).to_dataframe(name='K').reset_index() | |
| fig = px.bar(df_installed_cap, y='i', x='K', orientation='h', | |
| title='Total Installed Capacities [MW]', color='i', color_discrete_map=color_dict) | |
| return fig | |
| def plot_total_costs(m, col, df): | |
| """ | |
| Displays the total costs. | |
| """ | |
| total_costs = float(m.solution['C_inv'].values) + float(m.solution['C_op'].values) | |
| total_costs_rounded = round(total_costs / 1e9, 2) | |
| with col: | |
| st.write(f"{df.loc['plot_label_total_costs', st.session_state.lang]} {total_costs_rounded}") | |
| # st.write(f'Total costs: {total_costs_rounded} bn. €') | |
| df_total_costs = pd.DataFrame({'Total costs':[total_costs]}) | |
| return df_total_costs | |
| def plot_co2_price(m, col, df): | |
| """ | |
| Displays the CO2 price based on the CO2 constraint dual values. | |
| """ | |
| CO2_price = float(m.constraints['CO2_limit'].dual.values) * (-1) | |
| CO2_price_rounded = round(CO2_price, 2) | |
| df_CO2_price = pd.DataFrame({'CO2 price': [CO2_price]}) | |
| with col: | |
| st.write(f"{df.loc['plot_label_co2_price', st.session_state.lang]} {CO2_price_rounded}") | |
| return df_CO2_price | |
| def plot_new_capacities(m, color_dict, col, df): | |
| """ | |
| Plots the new capacities installed in MW as a bar chart and pie chart. | |
| Includes technologies with 0 MW capacity in the bar chart. | |
| Supports both German and English labels for technologies while ensuring color consistency. | |
| """ | |
| # Convert the solution for new capacities to a DataFrame | |
| df_new_capacities = m.solution['K'].round(0).to_dataframe().reset_index() | |
| # Store the English technology names in a separate column to maintain color consistency | |
| df_new_capacities['i_en'] = df_new_capacities['i'] | |
| # Check if the language is German and map English names to German for display | |
| if st.session_state.lang == 'DE': | |
| tech_mapping_en_to_de = { | |
| df.loc[f'tech_{tech.lower()}', 'EN']: df.loc[f'tech_{tech.lower()}', 'DE'] | |
| for tech in df_new_capacities['i_en'] if f'tech_{tech.lower()}' in df.index | |
| } | |
| # Replace the English technology names with German ones for display | |
| df_new_capacities['i'] = df_new_capacities['i_en'].replace(tech_mapping_en_to_de) | |
| # Bar plot for new capacities (including technologies with 0 MW) | |
| fig_bar = px.bar(df_new_capacities, y='i', x='K', orientation='h', | |
| title=df.loc['plot_label_new_capacities', st.session_state.lang], | |
| color='i_en', # Use the English names for consistent coloring | |
| color_discrete_map=color_dict) | |
| # Hide the legend completely since the labels are already next to the bars | |
| fig_bar.update_layout(showlegend=False) | |
| with col: | |
| st.plotly_chart(fig_bar) | |
| # Pie chart for new capacities (only show technologies with K > 0 in pie chart) | |
| df_new_capacities_filtered = df_new_capacities[df_new_capacities["K"] > 0] | |
| fig_pie = px.pie(df_new_capacities_filtered, names='i', values='K', | |
| title=df.loc['plot_label_new_capacities_pie', st.session_state.lang], | |
| color='i_en', color_discrete_map=color_dict) | |
| # Remove English labels (i_en) from the pie chart legend | |
| fig_pie.update_layout(legend_title_text=df.loc['label_technology', st.session_state.lang]) | |
| fig_pie.for_each_trace(lambda t: t.update(name=df_new_capacities_filtered['i'].iloc[0] if st.session_state.lang == 'DE' else t.name)) | |
| with col: | |
| st.plotly_chart(fig_pie) | |
| return df_new_capacities | |
| def plot_production(m, i_with_capacity, dt, color_dict, col, df): | |
| """ | |
| Plots the energy production for technologies with capacity as an area chart. | |
| Supports both German and English labels for technologies while ensuring color consistency. | |
| """ | |
| # Convert the production data to a DataFrame | |
| df_production = m.solution['y'].sel(i=i_with_capacity).to_dataframe().reset_index() | |
| # Store the English technology names in a separate column to maintain color consistency | |
| df_production['i_en'] = df_production['i'] | |
| # Check if the language is German and map English names to German for display | |
| if st.session_state.lang == 'DE': | |
| tech_mapping_en_to_de = { | |
| df.loc[f'tech_{tech.lower()}', 'EN']: df.loc[f'tech_{tech.lower()}', 'DE'] | |
| for tech in df_production['i_en'] if f'tech_{tech.lower()}' in df.index | |
| } | |
| # Replace the English technology names with German ones for display | |
| df_production['i'] = df_production['i_en'].replace(tech_mapping_en_to_de) | |
| # Area plot for energy production | |
| fig = px.area(df_production, y='y', x='t', | |
| title=df.loc['plot_label_production', st.session_state.lang], | |
| color='i_en', # Use the English names for consistent coloring | |
| color_discrete_map=color_dict) | |
| fig.update_traces(line=dict(width=0)) | |
| fig.for_each_trace(lambda trace: trace.update(fillcolor=trace.line.color)) | |
| with col: | |
| st.plotly_chart(fig) | |
| # Pie chart for total production | |
| df_production_sum = (df_production.groupby(['i', 'i_en'])['y'].sum() * dt / 1000).round(0).reset_index() | |
| # If the language is set to German, display German labels, otherwise use English | |
| pie_column = 'i' if st.session_state.lang == 'DE' else 'i_en' | |
| # Pie chart for total production | |
| fig_pie = px.pie(df_production_sum, names=pie_column, values='y', | |
| title=df.loc['plot_label_total_production_pie', st.session_state.lang], | |
| color='i_en', # Ensure the coloring stays consistent using the 'i_en' column | |
| color_discrete_map=color_dict) | |
| # Update legend title to reflect the correct language | |
| fig_pie.update_layout(legend_title_text=df.loc['label_technology', st.session_state.lang]) | |
| with col: | |
| st.plotly_chart(fig_pie) | |
| return df_production | |
| def plot_electricity_prices(m, dt, col, df): | |
| """ | |
| Plots the electricity price and the price duration curve. | |
| Supports both German and English labels for the plot titles and axis labels. | |
| """ | |
| # Convert the dual constraints to a DataFrame | |
| df_price = m.constraints['load'].dual.to_dataframe().reset_index() | |
| # Line plot for electricity prices | |
| fig_price = px.line(df_price, y='dual', x='t', | |
| title=df.loc['plot_label_electricity_prices', st.session_state.lang], | |
| # range_y=[0, 250], | |
| labels={'dual': df.loc['label_electricity_price', st.session_state.lang], | |
| 't': df.loc['label_time', st.session_state.lang]}) | |
| with col: | |
| st.plotly_chart(fig_price) | |
| # Create the price duration curve | |
| df_sorted_price = df_price["dual"].repeat(dt).sort_values(ascending=False).reset_index(drop=True) / int(dt) | |
| fig_duration = px.line(y=df_sorted_price, x=df_sorted_price.index, | |
| title=df.loc['plot_label_price_duration_curve', st.session_state.lang], | |
| # labels={"x": df.loc['label_hours_of_year', st.session_state.lang]}, | |
| # range_y=[0, 250]) | |
| ) | |
| with col: | |
| st.plotly_chart(fig_duration) | |
| return df_price | |
| def plot_contribution_margin(m, dt, color_dict, col, df): | |
| """ | |
| Plots the contribution margin for each technology. | |
| Supports both German and English labels for titles and axes while ensuring color consistency. | |
| """ | |
| # Convert the dual constraints to a DataFrame | |
| df_contr_marg = m.constraints['max_cap'].dual.to_dataframe().reset_index() | |
| # Adjust the 'dual' values for the contribution margin calculation | |
| df_contr_marg['dual'] = df_contr_marg['dual'] / dt * (-1) | |
| # Store the English technology names in a separate column to maintain color consistency | |
| df_contr_marg['i_en'] = df_contr_marg['i'] | |
| # Check if the language is German and map English names to German for display | |
| if st.session_state.lang == 'DE': | |
| tech_mapping_en_to_de = { | |
| df.loc[f'tech_{tech.lower()}', 'EN']: df.loc[f'tech_{tech.lower()}', 'DE'] | |
| for tech in df_contr_marg['i_en'] if f'tech_{tech.lower()}' in df.index | |
| } | |
| # Replace the English technology names with German ones for display | |
| df_contr_marg['i'] = df_contr_marg['i_en'].replace(tech_mapping_en_to_de) | |
| # Plot contribution margin for each technology | |
| fig = px.line(df_contr_marg, y='dual', x='t', | |
| title=df.loc['plot_label_contribution_margin', st.session_state.lang], | |
| color='i_en', # Use the English names for consistent coloring | |
| range_y=[0, 250], color_discrete_map=color_dict, | |
| # labels={ | |
| # 'dual': df.loc['label_contribution_margin', st.session_state.lang], | |
| # 't': df.loc['label_time', st.session_state.lang], | |
| # 'i_en': df.loc['label_technology', st.session_state.lang] | |
| # } | |
| ) | |
| # Update legend to display the correct language | |
| fig.update_layout(legend_title_text=df.loc['label_technology', st.session_state.lang]) | |
| # For German language, update the legend to show German technology names | |
| if st.session_state.lang == 'DE': | |
| fig.for_each_trace(lambda t: t.update(name=df_contr_marg.loc[df_contr_marg['i_en'] == t.name, 'i'].values[0])) | |
| # Display the plot | |
| with col: | |
| st.plotly_chart(fig) | |
| return df_contr_marg | |
| def plot_curtailment(m, iRes, color_dict, col, df): | |
| """ | |
| Plots the curtailment of renewable energy. | |
| Supports both German and English labels for titles and axes while ensuring color consistency. | |
| """ | |
| # Convert the curtailment solution to a DataFrame | |
| df_curtailment = m.solution['y_curt'].sel(i=iRes).to_dataframe().reset_index() | |
| # Store the English technology names in a separate column to maintain color consistency | |
| df_curtailment['i_en'] = df_curtailment['i'] | |
| # Check if the language is German and map English names to German for display | |
| if st.session_state.lang == 'DE': | |
| tech_mapping_en_to_de = { | |
| df.loc[f'tech_{tech.lower()}', 'EN']: df.loc[f'tech_{tech.lower()}', 'DE'] | |
| for tech in df_curtailment['i_en'] if f'tech_{tech.lower()}' in df.index | |
| } | |
| # Replace the English technology names with German ones for display | |
| df_curtailment['i'] = df_curtailment['i_en'].replace(tech_mapping_en_to_de) | |
| else: | |
| df_curtailment['i'] = df_curtailment['i_en'] # Use English names if not German | |
| # Area plot for curtailment of renewable energy | |
| fig = px.area(df_curtailment, y='y_curt', x='t', | |
| title=df.loc['plot_label_curtailment', st.session_state.lang], | |
| color='i_en', # Use the English names for consistent coloring | |
| color_discrete_map=color_dict) | |
| # Remove line traces and use fill colors for the area plot | |
| fig.update_traces(line=dict(width=0)) | |
| fig.for_each_trace(lambda trace: trace.update(fillcolor=trace.line.color)) | |
| # Update the legend title to reflect the correct language (German or English) | |
| fig.update_layout(legend_title_text=df.loc['label_technology', st.session_state.lang]) | |
| # For German language, update the legend to show German technology names | |
| if st.session_state.lang == 'DE': | |
| fig.for_each_trace(lambda t: t.update(name=df_curtailment.loc[df_curtailment['i_en'] == t.name, 'i'].values[0])) | |
| # Display the plot | |
| with col: | |
| st.plotly_chart(fig) | |
| return df_curtailment | |
| def plot_storage_charging(m, iSto, color_dict, col, df): | |
| """ | |
| Plots the charging of storage technologies. | |
| Supports both German and English labels for titles and axes while ensuring color consistency. | |
| """ | |
| # Convert the storage charging solution to a DataFrame | |
| df_charging = m.solution['y_ch'].sel(i=iSto).to_dataframe().reset_index() | |
| # Store the English technology names in a separate column to maintain color consistency | |
| df_charging['i_en'] = df_charging['i'] | |
| # Check if the language is German and map English names to German for display | |
| if st.session_state.lang == 'DE': | |
| tech_mapping_en_to_de = { | |
| df.loc[f'tech_{tech.lower()}', 'EN']: df.loc[f'tech_{tech.lower()}', 'DE'] | |
| for tech in df_charging['i_en'] if f'tech_{tech.lower()}' in df.index | |
| } | |
| # Replace the English technology names with German ones for display | |
| df_charging['i'] = df_charging['i_en'].replace(tech_mapping_en_to_de) | |
| else: | |
| df_charging['i'] = df_charging['i_en'] # Use English names if not German | |
| # Area plot for storage charging | |
| fig = px.area(df_charging, y='y_ch', x='t', | |
| title=df.loc['plot_label_storage_charging', st.session_state.lang], | |
| color='i_en', # Use the English names for consistent coloring | |
| color_discrete_map=color_dict) | |
| # Remove line traces and use fill colors for the area plot | |
| fig.update_traces(line=dict(width=0)) | |
| fig.for_each_trace(lambda trace: trace.update(fillcolor=trace.line.color)) | |
| # Update the legend title to reflect the correct language (German or English) | |
| fig.update_layout(legend_title_text=df.loc['label_technology', st.session_state.lang]) | |
| # For German language, update the legend to show German technology names | |
| if st.session_state.lang == 'DE': | |
| fig.for_each_trace(lambda t: t.update(name=df_charging.loc[df_charging['i_en'] == t.name, 'i'].values[0])) | |
| # Display the plot | |
| with col: | |
| st.plotly_chart(fig) | |
| return df_charging | |
| def plot_hydrogen_production(m, iPtG, color_dict, col, df): | |
| """ | |
| Plots the hydrogen production. | |
| Supports both German and English labels for titles and axes while ensuring color consistency. | |
| """ | |
| # Convert the hydrogen production data to a DataFrame | |
| df_h2_prod = m.solution['y_h2'].sel(i=iPtG).to_dataframe().reset_index() | |
| # Store the English technology names in a separate column to maintain color consistency | |
| df_h2_prod['i_en'] = df_h2_prod['i'] | |
| # Check if the language is German and map English names to German for display | |
| if st.session_state.lang == 'DE': | |
| tech_mapping_en_to_de = { | |
| df.loc[f'tech_{tech.lower()}', 'EN']: df.loc[f'tech_{tech.lower()}', 'DE'] | |
| for tech in df_h2_prod['i_en'] if f'tech_{tech.lower()}' in df.index | |
| } | |
| # Replace the English technology names with German ones for display | |
| df_h2_prod['i'] = df_h2_prod['i_en'].replace(tech_mapping_en_to_de) | |
| else: | |
| df_h2_prod['i'] = df_h2_prod['i_en'] # Keep English names if not German | |
| # Area plot for hydrogen production | |
| fig = px.area(df_h2_prod, y='y_h2', x='t', | |
| title=df.loc['plot_label_hydrogen_production', st.session_state.lang], | |
| color='i_en', # Use the English names for consistent coloring | |
| color_discrete_map=color_dict) | |
| # Remove line traces and use fill colors for the area plot | |
| fig.update_traces(line=dict(width=0)) | |
| fig.for_each_trace(lambda trace: trace.update(fillcolor=trace.line.color)) | |
| # Update the legend title to reflect the correct language (German or English) | |
| fig.update_layout(legend_title_text=df.loc['label_technology', st.session_state.lang]) | |
| # For German language, update the legend to show German technology names | |
| if st.session_state.lang == 'DE': | |
| fig.for_each_trace(lambda t: t.update(name=df_h2_prod.loc[df_h2_prod['i_en'] == t.name, 'i'].values[0])) | |
| # Display the plot | |
| with col: | |
| st.plotly_chart(fig) | |
| return df_h2_prod | |
| def disaggregate_df(df, t, t_original, dt): | |
| """ | |
| Disaggregates the DataFrame based on the original time steps. | |
| """ | |
| if "t" not in list(df.columns): | |
| return df | |
| df_t_all = pd.DataFrame({"t_all": t_original.to_series(), 't': t.repeat(dt)}).reset_index(drop=True) | |
| df_output = df.merge(df_t_all, on='t').drop('t', axis=1).rename({'t_all': 't'}, axis=1) | |
| df_output = df_output[[df_output.columns[-1]] + list(df_output.columns[:-1])] | |
| return df_output.sort_values('t') | |
| if __name__ == "__main__": | |
| main() |