Spaces:
Sleeping
Sleeping
| # -*- 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 plotly.graph_objects as go | |
| import streamlit as st | |
| from io import BytesIO | |
| import xlsxwriter | |
| from linopy import Model | |
| import sourced as src | |
| import time | |
| # 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() | |
| # Create tabs for navigation | |
| tabs = st.tabs([ | |
| st.session_state.settings["df"].loc['menu_modell', st.session_state.lang], | |
| st.session_state.settings["df"].loc['menu_doku', st.session_state.lang], | |
| st.session_state.settings["df"].loc['menu_impressum', st.session_state.lang] | |
| ]) | |
| # Load and display content based on the selected tab | |
| with tabs[0]: # Model page | |
| page_model() | |
| with tabs[1]: # Documentation page | |
| page_documentation() | |
| with tabs[2]: # About Us page | |
| page_about_us() | |
| # 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': 'saddlebrown', | |
| 'Fossil Hard coal': 'chocolate', # Ein Braunton ähnlich Lignite | |
| 'Fossil Oil': 'black', | |
| 'CCGT': 'lightgray', # Hellgrau | |
| 'OCGT': 'darkgray', # Dunkelgrau | |
| 'RoR': 'aquamarine', | |
| 'Hydro Water Reservoir': 'lightsteelblue', | |
| 'Nuclear': 'gold', | |
| 'PV': 'yellow', | |
| 'WindOff': 'darkblue', | |
| 'WindOn': 'green', | |
| 'H2': 'tomato', | |
| 'Pumped Hydro Storage': 'skyblue', | |
| 'Battery storages': 'firebrick', | |
| 'Electrolyzer': 'yellowgreen' | |
| }, | |
| '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_2023.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''' C^{op} = \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''') | |
| # Investment costs | |
| st.write(df.loc['constr_c_inv', st.session_state.lang]) | |
| st.latex(r''' C^{inv} = \sum_{i} a_{i} \cdot K_{i} \cdot c^{inv}_{i}''') | |
| # Constraints | |
| st.subheader(df.loc['subheader_constr', st.session_state.lang]) | |
| # 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_2023.xlsx', 'rb') as f: | |
| st.download_button(df.loc['model_title1.3',st.session_state.lang], f, file_name='Input_Jahr_2023.xlsx') # Download button for Excel template | |
| with st.form("input_file"): | |
| 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_title4', st.session_state.lang]) | |
| run_model_excel = st.form_submit_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]) | |
| with st.form("input_custom"): | |
| col1form, col2form, col3form = st.columns([0.25, 0.25, 0.50]) | |
| # colum 1 form | |
| l_co2 = col1form.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 = col1form.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] = col1form.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) | |
| # colum 1 form | |
| for i_idx in params_dict['c_fuel_i'].get_index('i'): | |
| if i_idx in ['Fossil Hard coal', 'Fossil Oil', 'CCGT']: | |
| params_dict['c_fuel_i'].loc[i_idx] = col2form.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) | |
| params_dict['c_fuel_i'].loc['OCGT'] = params_dict['c_fuel_i'].loc['CCGT'] | |
| # 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', 'CCGT', 'OCGT', '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', 'CCGT', 'OCGT', '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 = col3form.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] | |
| col4form, col5form = st.columns([0.25, 0.75]) | |
| dt = col4form.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]) | |
| run_model_manual = col5form.form_submit_button(df.loc['model_run_info_gui', st.session_state.lang]) | |
| #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]) | |
| st.markdown("-------") | |
| # run_model_manual = True | |
| if run_model_excel or run_model_manual: | |
| # Model setup | |
| info_yellow_build = st.info(df.loc['label_build_model', st.session_state.lang]) | |
| if run_model_excel: # overwrite with excel values | |
| #sets_dict, params_dict = load_model_input(df, write_pickle_from_standard_excel) | |
| sets_dict, params_dict = src.load_data_from_excel(st.session_state.url_excel, write_to_pickle_flag=True) | |
| # 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 | |
| # 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 | |
| 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 | |
| info_yellow_build.empty() | |
| info_green_build = st.success(df.loc['label_build_model', st.session_state.lang]) | |
| info_yellow_solve = st.info(df.loc['label_solve_model', st.session_state.lang]) | |
| m.solve(solver_name='highs') | |
| info_yellow_solve.empty() | |
| info_green_solve = st.success(df.loc['label_solve_model', st.session_state.lang]) | |
| info_yellow_plot = st.info(df.loc['label_generate_plots', st.session_state.lang]) | |
| # 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) | |
| 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) & (m.solution['i'] != 'Electrolyzer')).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, colb2, df) | |
| df_curtailment = plot_curtailment(m, iRes, color_dict, colb1, df) | |
| df_residual_load_duration = plot_residual_load_duration(m, dt, colb1, df, D_t, i_with_capacity, iRes, color_dict, df_curtailment, iConv) | |
| df_price = plot_electricity_prices(m, dt, colb2, df, df_residual_load_duration) | |
| df_contr_marg = plot_contribution_margin(m, dt, i_with_capacity, color_dict, colb1, 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, colb1, df) | |
| # df_stackplot = plot_stackplot(m) | |
| # 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=df.loc['sheet_name_total_costs', st.session_state.lang], index=False) | |
| disaggregate_df(df_CO2_price, t, t_original, dt).to_excel(writer, sheet_name=df.loc['sheet_name_co2_price', st.session_state.lang], index=False) | |
| disaggregate_df(df_price, t, t_original, dt).to_excel(writer, sheet_name=df.loc['sheet_name_prices', st.session_state.lang], index=False) | |
| disaggregate_df(df_contr_marg, t, t_original, dt).to_excel(writer, sheet_name=df.loc['sheet_name_contribution_margin', st.session_state.lang], index=False) | |
| disaggregate_df(df_new_capacities, t, t_original, dt).to_excel(writer, sheet_name=df.loc['sheet_name_capacities', st.session_state.lang], index=False) | |
| disaggregate_df(df_production, t, t_original, dt).to_excel(writer, sheet_name=df.loc['sheet_name_production', st.session_state.lang], index=False) | |
| disaggregate_df(df_charging, t, t_original, dt).to_excel(writer, sheet_name=df.loc['sheet_name_charging', st.session_state.lang], index=False) | |
| disaggregate_df(D_t.to_dataframe().reset_index(), t, t_original, dt).to_excel(writer, sheet_name=df.loc['sheet_name_demand', st.session_state.lang], index=False) | |
| disaggregate_df(df_curtailment, t, t_original, dt).to_excel(writer, sheet_name=df.loc['sheet_name_curtailment', st.session_state.lang], index=False) | |
| disaggregate_df(df_h2_prod, t, t_original, dt).to_excel(writer, sheet_name=df.loc['sheet_name_h2_production', st.session_state.lang], index=False) | |
| with col1: | |
| 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") | |
| info_yellow_plot.empty() | |
| info_green_plot = st.success(df.loc['label_generate_plots', st.session_state.lang]) | |
| time.sleep(1) | |
| info_green_build.empty() | |
| info_green_solve.empty() | |
| info_green_plot.empty() | |
| st.stop() | |
| # 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_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.markdown( | |
| f"<h3><b>{df.loc['plot_label_total_costs', st.session_state.lang]} {total_costs_rounded}</b></h3>", | |
| unsafe_allow_html=True | |
| ) | |
| 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.markdown( | |
| f"<h3><b>{df.loc['plot_label_co2_price', st.session_state.lang]} {CO2_price_rounded}</b></h3>", | |
| unsafe_allow_html=True | |
| ) | |
| 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, | |
| labels={'K': '', 'i': ''} # Delete double labeling | |
| ) | |
| # 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'] | |
| # Convert 't'-column in a datetime format | |
| df_production['t'] = df_production['t'].str.strip("'") | |
| df_production['t'] = pd.to_datetime(df_production['t'], format='%Y-%m-%d %H:%M %z') | |
| # 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, | |
| labels={'y': '', 't': '', 'i_en': df.loc['label_technology', st.session_state.lang]} # Delete double labeling | |
| ) | |
| # Update legend labels to display German names instead of English | |
| if st.session_state.lang == 'DE': | |
| fig.for_each_trace(lambda trace: trace.update(name=tech_mapping_en_to_de[trace.name])) | |
| fig.update_traces(line=dict(width=0)) | |
| fig.for_each_trace(lambda trace: trace.update(fillcolor=trace.line.color)) | |
| # # Customize x-axis for better date formatting | |
| # fig.update_layout( | |
| # xaxis=dict( | |
| # tickformat="%d/%m/%Y", # Display months and years in MM/YYYY format | |
| # title='', # No title for the x-axis | |
| # type="date" # Ensure x-axis is treated as a date axis | |
| # ), | |
| # xaxis_tickangle=-45 # Tilt the ticks for better readability | |
| # ) | |
| 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, df_residual_load_duration): | |
| """ | |
| 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() | |
| # Convert 't'-column in a datetime format | |
| df_price['t'] = df_price['t'].str.strip("'") | |
| df_price['t'] = pd.to_datetime(df_price['t'], format='%Y-%m-%d %H:%M %z') | |
| # 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': '', 't': ''} | |
| ) | |
| 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) | |
| df_residual_load_sorted = df_residual_load_duration.sort_values(by='Residual_Load', ascending=False).reset_index(drop=True) | |
| df_axis2 = df_residual_load_sorted['Residual_Load'] | |
| ax2_max = np.max(df_axis2) | |
| ax2_min = np.min(df_axis2) | |
| fig_duration = go.Figure() | |
| # Add primary y-axis trace (Price duration curve) | |
| fig_duration.add_trace(go.Scatter( | |
| x=df_sorted_price.index, | |
| y=df_sorted_price, | |
| mode='lines', | |
| name=df.loc['plot_label_price_duration_curve', st.session_state.lang], # Price duration label | |
| line=dict(color='blue', width=2) # Blue line for primary y-axis | |
| )) | |
| # Add secondary y-axis trace (Residual load) | |
| fig_duration.add_trace(go.Scatter( | |
| x=df_axis2.index, | |
| y=df_axis2, | |
| mode='lines', | |
| name=df.loc['plot_label_residual_load', st.session_state.lang], # Residual load label | |
| line=dict(color='red', width=2), # Red line for secondary y-axis | |
| yaxis='y2' # Link this trace to the secondary y-axis | |
| )) | |
| # Layout mit separaten Achsen | |
| fig_duration.update_layout( | |
| title=df.loc['plot_label_price_duration_curve', st.session_state.lang], | |
| xaxis=dict( | |
| title=df.loc['label_hours', st.session_state.lang] # Common x-axis | |
| ), | |
| yaxis=dict( | |
| title=df.loc['plot_label_price_duration_curve', st.session_state.lang], # Title for primary y-axis | |
| range=[-(100/(ax2_max/(ax2_max-ax2_min))-100), 100], # Primary y-axis range | |
| titlefont=dict(color='blue'), # Blue color for primary axis title | |
| tickfont=dict(color='blue') # Blue ticks for primary axis | |
| ), | |
| yaxis2=dict( | |
| title=df.loc['plot_label_residual_load', st.session_state.lang], # Title for secondary y-axis | |
| range=[ax2_min, ax2_max], # Secondary y-axis range | |
| titlefont=dict(color='red'), # Red color for secondary axis title | |
| tickfont=dict(color='red'), # Red ticks for secondary axis | |
| overlaying='y', # Overlay secondary axis on primary | |
| side='right' # Place secondary y-axis on the right side | |
| ), | |
| legend=dict( | |
| x=1, # Positioniert die Legende am rechten Rand | |
| y=1, # Positioniert die Legende am oberen Rand | |
| xanchor='right', # Verankert die Legende am rechten Rand | |
| yanchor='top', # Verankert die Legende am oberen Rand | |
| bgcolor='rgba(255, 255, 255, 0.5)', # Weißer Hintergrund mit Transparenz | |
| bordercolor='black', | |
| borderwidth=1 | |
| ) | |
| ) | |
| with col: | |
| st.plotly_chart(fig_duration) | |
| return df_price | |
| def plot_residual_load_duration(m, dt, col, df, D_t, i_with_capacity, iRes, color_dict, df_curtailment, iConv): | |
| """ | |
| Plots the residual load and corresponding production as a stacked area chart. | |
| Supports both German and English labels for the plot titles and axis labels. | |
| Consistent color coding for technologies using a predefined color dictionary. | |
| """ | |
| # Extract load data and repeat each value to match the total number of hours in the year | |
| df_load = D_t.values.flatten() | |
| total_hours = len(df_load) * dt # Calculate the total number of hours dynamically | |
| repeated_load = np.repeat(df_load, dt)[:total_hours] # Repeat values to represent each hour | |
| # Convert production data to DataFrame | |
| df_production = m.solution['y'].sel(i=i_with_capacity).to_dataframe().reset_index() | |
| # Pivot production data to get technologies as columns and time 't' as index | |
| df_production_pivot = df_production.pivot(index='t', columns='i', values='y') | |
| # Repeat the pivoted production data to match the number of hours | |
| repeated_index = np.repeat(df_production_pivot.index, dt)[:total_hours] # Create repeated index | |
| df_production_repeated = df_production_pivot.loc[repeated_index].reset_index(drop=True) | |
| # Create load series with the same index as the repeated production data | |
| df_load_series = pd.Series(repeated_load, index=df_production_repeated.index, name='Load') | |
| # Combine load with repeated production data | |
| df_combined = df_production_repeated.copy() | |
| df_combined['Load'] = df_load_series | |
| # Identify renewable technologies from iRes | |
| iRes_list = iRes.tolist() # Convert the Index to a list | |
| # Calculate renewable generation (only include available technologies in df_combined) | |
| renewable_columns = [col for col in iRes_list if col in df_combined.columns] | |
| df_combined['Renewable_Generation'] = df_combined[renewable_columns].sum(axis=1) if renewable_columns else 0 | |
| # Create pivot table of curtailment | |
| df_curtailment_pivot = df_curtailment.pivot(index='t', columns='i', values='y_curt') | |
| repeated_index = np.repeat(df_curtailment_pivot.index, dt)[:total_hours] # Create repeated index | |
| df_curtailment_repeated = df_curtailment_pivot.loc[repeated_index].reset_index(drop=True) | |
| df_curtailment_repeated['Sum'] = df_curtailment_repeated.sum(axis=1) | |
| df_combined['Sum_curtailment'] = -df_curtailment_repeated['Sum'] | |
| # Calculate residual load as the difference between total load and renewable generation | |
| df_combined['Residual_Load'] = df_combined['Load'] - df_combined['Renewable_Generation'] + df_combined['Sum_curtailment'] | |
| # Sort DataFrame by residual load (descending order) to create the duration curve | |
| df_sorted = df_combined.sort_values(by='Residual_Load', ascending=False).reset_index(drop=True) | |
| # Identify all technology columns except 'Load', 'Residual_Load', 'Renewable_Generation' | |
| technology_columns = [col for col in df_combined.columns if col not in ['Load', 'Residual_Load', 'Renewable_Generation', 'Sum_curtailment']] | |
| # Mapping English technology names to German (if desired) | |
| 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 technology_columns if f'tech_{tech.lower()}' in df.index | |
| } | |
| else: | |
| tech_mapping_en_to_de = {tech: tech for tech in technology_columns} # Use the original names if not in German | |
| # Plotting with Plotly - Creating stacked area chart | |
| fig = go.Figure() | |
| # Sort technology_columns based on the highest index in df_sorted (only for iConv); others are placed at the end | |
| sorted_technology_columns = sorted( | |
| technology_columns, | |
| key=lambda tech: ( | |
| tech not in iConv, # Place non-iConv technologies at the end | |
| -df_sorted[df_sorted[tech] != 0].index.max() if tech in iConv and not df_sorted[df_sorted[tech] != 0].empty else float('inf') | |
| ) | |
| ) | |
| # Add stacked area traces for each production technology with consistent colors and language-specific names | |
| for tech in sorted_technology_columns: | |
| tech_name = tech_mapping_en_to_de.get(tech, tech) # Get the translated name or fallback to the original | |
| fig.add_trace(go.Scatter( | |
| x=df_sorted.index, | |
| y=df_sorted[tech], | |
| mode='lines', | |
| stackgroup='one', # For stacking traces | |
| name=tech_name, | |
| line=dict(width=0.5, color=color_dict.get(tech)) | |
| )) | |
| # Add residual load trace as a red line | |
| fig.add_trace(go.Scatter( | |
| x=df_sorted.index, | |
| y=df_sorted['Residual_Load'], | |
| mode='lines', | |
| name=df.loc['plot_label_residual_load', st.session_state.lang], # Residual load label in current language | |
| line=dict(color='red', width=2) | |
| )) | |
| # Add curtailment trace as a shaded area with a dark yellow tone | |
| fig.add_trace(go.Scatter( | |
| x=df_sorted.index, | |
| y=df_sorted['Sum_curtailment'], | |
| mode='lines', # Line mode for the boundary of the area | |
| name=df.loc['plot_label_sum_curtailment', st.session_state.lang], # Curtailment label in current language | |
| line=dict(color='rgba(204, 153, 0, 1)', width=1.5), # Dark yellow line | |
| fill='tozeroy', # Fill area down to the x-axis | |
| fillcolor='rgba(204, 153, 0, 0.3)' # Semi-transparent dark yellow for the fill | |
| )) | |
| # Layout settings for the plot | |
| fig.update_layout( | |
| title=df.loc['plot_label_residual_load_curve', st.session_state.lang], | |
| xaxis_title=df.loc['label_hours', st.session_state.lang], | |
| template="plotly_white", | |
| ) | |
| # Display the plot in Streamlit | |
| with col: | |
| st.plotly_chart(fig) | |
| return df_combined | |
| def plot_contribution_margin(m, dt, i_with_capacity, 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.sel(i=i_with_capacity).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'] | |
| # Convert 't'-column in a datetime format | |
| df_contr_marg['t'] = pd.to_datetime(df_contr_marg['t'].str.strip("'"), format='%Y-%m-%d %H:%M %z') | |
| # 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':'', 't':'', 'i_en':''} | |
| ) | |
| # 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() | |
| # Convert 't'-column in a datetime format | |
| df_curtailment['t'] = pd.to_datetime(df_curtailment['t'].str.strip("'"), format='%Y-%m-%d %H:%M %z') | |
| # 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, | |
| labels={'y_curt': '', 't': ''} # Delete double labeling | |
| ) | |
| # 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() | |
| # Drop out infinitesimal numbers | |
| df_charging['y_ch'] = df_charging['y_ch'].apply(lambda x: 0 if x < 0.01 else x) | |
| # Convert 't'-column in a datetime format | |
| df_charging['t'] = pd.to_datetime(df_charging['t'].str.strip("'"), format='%Y-%m-%d %H:%M %z') | |
| # 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, | |
| labels={'y_ch': '', 't': ''} # Delete double labeling | |
| ) | |
| # 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() | |
| # Convert 't'-column in a datetime format | |
| df_h2_prod['t'] = pd.to_datetime(df_h2_prod['t'].str.strip("'"), format='%Y-%m-%d %H:%M %z') | |
| # 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, | |
| labels={'y_h2': '', 't': ''} # Delete double labeling | |
| ) | |
| # 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 | |
| # Change format of t back | |
| df['t'] = "'" + pd.to_datetime(df['t'], utc=True).dt.tz_convert('Europe/Berlin').dt.strftime('%Y-%m-%d %H:%M %z') + "'" | |
| 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])] | |
| # Drop the helping column i_en | |
| df_output = df_output.drop(columns=['i_en'], errors='ignore') | |
| return df_output.sort_values('t') | |
| if __name__ == "__main__": | |
| main() |