ecran_parametrage_etude.py

# -*- coding: utf-8 -*-
"""
Created on Fri Mar  7 14:42:15 2025

@author: Despouysoli
"""

import streamlit as st
import pandas as pd
import yaml

from battman_communs import DEFAULT_PRICE, DEFAULT_GENERATION, LST_MODES_OPTIM
from battman_communs import validation_parametrage

# =============================================================================
# Lecture des chroniques et paramétrage des pas de temps
# =============================================================================

# Pour avoir une meilleure lisibilité du tableau de configuration du site 
# issu du fichier YAML
class IndentDumper(yaml.Dumper):
    def increase_indent(self, flow=False, indentless=False):
        return super(IndentDumper, self).increase_indent(flow, False)





def display_parametrage_etude():
    
    st.header("Paramétrage de l'étude")
    msg_cfg = '''
    Cet onglet permet de de sélectionner les éléments constitutifs de l'étude :
        
    - configuration d'un site (cf. onglet précédent),
    - choix du pas de temps,
    - sélection de données temporelles (au moins un chronique parmi les suivantes) :
        - chronique de prix de l'électricité [optionnelle]
        - chronique de limitations imposées par RTE [optionnelle]
        - chronique d'injections du site [optionnelle]
    - mode d'optimisation.
    '''
    st.markdown(msg_cfg)
    st.markdown("<br>", unsafe_allow_html=True)


    # =========================================================================
    # Configuration de site
    # =========================================================================
    with st.container(height="content", border=True):
        st.subheader("Configuration du site")
        
        # Création d'une colonne de chargement de la config de site, 
        # et d'une autre de visualisation
        col_upload_cfg_site, col_visu_cfg_site = st.columns([1, 2])  
        
        # Colonne de chargement de la condig du site
        with col_upload_cfg_site:
            label_cfg_site_loader = "Chargez un fichier YAML de configuration de site."
            cfg_site_file = st.file_uploader(label_cfg_site_loader, 
                                             type=["yml"], 
                                             accept_multiple_files=False,
                                             key='yml_uploader',
                                             help='''
                                             Fichier YAML (.yml) généré par l'onglet 
                                             "Configurateur de site".
                                             ''')
            # Traitement du fichier YAML
            if cfg_site_file is not None:
                try:
                     st.session_state.dico_cfg_site = yaml.safe_load(cfg_site_file)
                     if len(st.session_state.dico_cfg_site) > 0:
                        st.success("Fichier chargé avec succès !")
                except Exception as e:
                    st.error(f"Erreur lors du chargement du fichier : {e}")     
            else:
                # On est dans ce cas notamment lorsqu'on supprime un 
                # fichier précédemment chargé ==> il faut RAZ le session_state
                if 'dico_cfg_site' in st.session_state:
                    del st.session_state['dico_cfg_site']

        # Colonne de visualisation de la config du site
        with col_visu_cfg_site:
            st.markdown("<br>", unsafe_allow_html=True)
            if 'dico_cfg_site' in st.session_state:
                md_cfg_site = pd.Series(st.session_state.dico_cfg_site, 
                                        name="Configuration du site").to_markdown()
                st.markdown(md_cfg_site)
            else:
                st.info("Pas de configurationde site... Merci d'en charger une.")



    # =========================================================================
    # Configuration du timestep
    # =========================================================================
    with st.container(height="content", border=True):
        st.subheader("Choix du pas de temps des chroniques")
        
        msg_pas_de_temps = '''
        🚨 **ATTENTION** 🚨 

        Même si les chroniques utilisées pour l'étude sont horodatées, ces 
        indications ne seront pas prise en compte! **Seule la durée entre deux
        pas de temps définie ci-dessous est utilisée.**
        Généralement, les chroniques de données considérés sont horaires.
        '''
        st.markdown(msg_pas_de_temps)
    
        col_timestep, _ = st.columns([1, 3])
        st.session_state['delta_t'] =  col_timestep.number_input(
            "Durée entre deux pas de temps consécutifs (en minutes) :", 
            min_value=1, step=1, value=60,
            help="Durée entre deux pas de temps consécutifs (minutes).")



    # =========================================================================
    # Configuration des chroniques
    # =========================================================================
    with st.container(height="content", border=True):
        st.subheader("Chargement des données temporelles")

    
        col_prix, col_limitations, col_injections = st.columns([1, 1, 1])

        # Chargement d'un fichier CSV de prix
        with col_prix:       
            label_loader_prix = "Chargez un fichier CSV de chronique de prix de l'électricité (€/MWh)"
            prix_file = st.file_uploader(label_loader_prix, 
                                         type=["csv"], 
                                         accept_multiple_files=False,
                                         key='prix_uploader',
                                         help='''
                                              Fichier CSV respectant les conventions suivantes :
                                                 separateur : ';'.
                                                 Colonne 1 : prix de l'électricité (€/MWh)
                                                 Une ligne par pas de temps.
                                                 Pas de label de colonne.
                                              ''')
            # Traitement du fichier CSV
            if prix_file is not None:
                try:
                    # Lecture du fichier CSV
                    df_prix = pd.read_csv(prix_file, sep=';', decimal=",", header=None)
                    if not df_prix.empty:
                        st.success("Fichier chargé avec succès !")
                        st.session_state['df_prix'] = df_prix
                except Exception as e:
                    st.error(f"Erreur lors du chargement du fichier : {e}")
            else:
                # On est dans ce cas notamment lorsqu'on supprime un 
                # fichier précédemment chargé ==> il faut RAZ le session_state
                if 'df_prix' in st.session_state:
                    del st.session_state['df_prix']
    

        # Chargement d'un fichier CSV de limitations
        with col_limitations:       
            label_loader_limitations = "Chargez un fichier CSV de chroniques de limitations en soutirage et en injection (MW)"
            limitations_file = st.file_uploader(label_loader_limitations, 
                                                type=["csv"], 
                                                accept_multiple_files=False,
                                                key='limitations_uploader',
                                                help='''
                                                     Fichier CSV respectant les conventions suivantes :
                                                     separateur : ';'.
                                                     Colonne 1 : limitations en soutirage en MW (valeurs négatives en convention générateur).
                                                     Colonne 2 : limitations en injection en MW (valeurs positives en convention générateur).
                                                     Une ligne par pas de temps.
                                                     Pas de label de colonne.
                                                     ''')
            # Traitement du fichier CSV
            if limitations_file is not None:
                 try:
                     # Lecture du fichier CSV
                     df_limitations = pd.read_csv(limitations_file, sep=';', decimal=",", header=None)
                     if not df_limitations.empty:
                         st.success("Fichier chargé avec succès !")
                         st.session_state['df_limitations'] = df_limitations
                 except Exception as e:
                     st.error(f"Erreur lors du chargement du fichier : {e}")
            else:
                 # On est dans ce cas notamment lorsqu'on supprime un 
                 # fichier précédemment chargé ==> il faut RAZ le session_state
                 if 'df_limitations' in st.session_state:
                     del st.session_state['df_limitations']
     
   

        # Chargement d'un fichier CSV des injections
        with col_injections:       
            label_loader_injections = "Chargez un fichier CSV de chronique d'injections nettes pour le site (MW)"
            injections_file = st.file_uploader(label_loader_injections, 
                                                type=["csv"], 
                                                accept_multiple_files=False,
                                                key='injections_uploader',
                                                help='''
                                                     Fichier CSV respectant les conventions suivantes :
                                                     separateur : ';'.
                                                     Colonne 1 : injections nettes en MW (convention générateur).
                                                     Une ligne par pas de temps.
                                                     Pas de label de colonne.
                                                     ''')
            # Traitement du fichier CSV
            if injections_file is not None:
                 try:
                     # Lecture du fichier CSV
                     df_injections = pd.read_csv(injections_file, sep=';', decimal=",", header=None)
                     if not df_injections.empty:
                         st.success("Fichier chargé avec succès !")
                         st.session_state['df_injections'] = df_injections
                 except Exception as e:
                     st.error(f"Erreur lors du chargement du fichier : {e}")
            else:
                 # On est dans ce cas notamment lorsqu'on supprime un 
                 # fichier précédemment chargé ==> il faut RAZ le session_state
                 if 'df_injections' in st.session_state:
                     del st.session_state['df_injections']
     




    # =========================================================================
    # Choix du mode d'optimisation
    # =========================================================================
    with st.container(height="content", border=True):
        st.subheader("Mode d'optimisation")

        st.session_state['mode_optim'] = st.radio(
            label="Sélectionner le type d'optimisation à effectuer",
            options=[type_optim["label"] for type_optim in LST_MODES_OPTIM],
            captions=[type_optim["explication"] for type_optim in LST_MODES_OPTIM],
            horizontal=True,
        )
            



    # =========================================================================
    # Validation
    # =========================================================================
    with st.container(height="content", border=True):
        st.subheader("Synthèse et validation du paramétrage")
        
        if st.button("Vérification du paramétrage", type="primary"):
            data = validation_parametrage(display_msgs=True, 
                                          default_prices=DEFAULT_PRICE, 
                                          default_generation=DEFAULT_GENERATION)

            if data == None:
                st.error(''' 
                         **La configuration de l'étude n'est pas valide.**
                         Veuillez recommencer.
                         ''')
            else:
                dico_site, df_chroniques, warning_prix_inutiles = data
                
                # Préparation des listes de messages d'information d'une part,
                # de warning d'autre part.
                lst_msg_info, lst_msg_warning = [], []
                lst_msg_info.append(f"Le mode d'optimisation est {st.session_state['mode_optim']}.")
                lst_msg_info.append(f"La durée entre deux pas de temps est de {st.session_state['delta_t']} minutes.")
                if warning_prix_inutiles:
                    lst_msg_warning.append("Une chronique de prix a bien été fournie, mais elle ne sera pas utilisée pour ce mode d'optimisation!")
                if list(df_chroniques.prix.unique()) == [DEFAULT_PRICE] and not warning_prix_inutiles:
                    # La chronique de prix n'a pas été fournie
                    lst_msg_warning.append("En l'absence de données de prix, on initialise arbitrairement tous les prix à une valeur fixe.")
                    lst_msg_info.append(f"Les prix seront identiques pendant toute la simulation, et fixés à {DEFAULT_PRICE} €/MWh.")
                elif warning_prix_inutiles:
                    lst_msg_info.append("Une chronique de prix a bien été fournie, mais elle ne sera pas utilisée.")
                else:
                    lst_msg_info.append("Une chronique de prix a bien été fournie, et elle sera utilisée.")
                    
                if list(df_chroniques.p_inj.unique()) == [DEFAULT_GENERATION]:
                    # La chronique d'injections n'a pas été fournie
                    lst_msg_warning.append("En l'absence de données d'injections, on initialise celles-ci à une valeur fixe.")
                    lst_msg_info.append(f"Les injections seront identiques pendant toute la simulation, et fixées à {DEFAULT_GENERATION} MW.")
                else:
                    lst_msg_info.append("Une chronique d'injections a bien été fournie.")
                if pd.isna(df_chroniques.p_min).all():
                    lst_msg_info.append("Aucune limitation en puissance minimale n'est définie pour cette étude.")
                else:
                    lst_msg_info.append("Une chronique de limitations en puissance minimale sera prise en compte.")
                if pd.isna(df_chroniques.p_max).all():
                    lst_msg_info.append("Aucune limitation en puissance puissance maximale n'est définie pour cette étude.")
                else:
                    lst_msg_info.append("Une chronique de limitations en puissance maximale sera prise en compte.")
                
                
                # Affichage des informations
                col_warnings, _, col_msgs = st.columns([25, 1, 25])

                # Warnings
                with col_warnings:  
                    with st.container(height="content", border=True):
                        st.markdown("**Points de vigilance**")
                        for warn in lst_msg_warning:
                            st.warning(warn)
                        if len(lst_msg_warning) == 0:
                            st.info("RAS")
                # Infos
                with col_msgs:       
                    with st.container(height="content", border=True):
                        st.markdown("**Synthèse de la configuration**")
                        for info in lst_msg_info:
                            st.info(info)
                # Succes !
                st.success("**Les données sont bien validées!** Vous pouvez passer à l'optimisation.", icon="✅")