battman_communs.py

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

@author: Despouysoli
"""

import streamlit as st
import pandas as pd
import copy

import optimisation_batterie_EOD as obe


# =============================================================================
# Traces de débug
# =============================================================================
DBG_TRACES = False

# =============================================================================
# Numéro de version
# =============================================================================
VERSION_BATTMAN = "2025-12-03"

# =============================================================================
# Valeurs par défaut 
# =============================================================================
# Prix par défaut (utilisé si aucune chronique de prix n'est fournie)
DEFAULT_PRICE = 50
# Puissance par défaut produite par le site si aucune chronique de production 
# n'est fournie
DEFAULT_GENERATION = 0



# =============================================================================
# Modes d'optimisation disponibles
# =============================================================================

LST_MODES_OPTIM = [
    # 
    {
         "label"          : "**Maximisation des gains économiques, puis minimisation de l'ENE**",
         "explication"    : ("Le site maximise ses bénéfices sans se préoccuper "
                             "d'évacuer au maximum la production hors batterie ;"
                             "si différentes solutions aboutissent aux mêmes "
                             "gains économiques, alors celle qui minimise l'ENE "
                             "est privilégiée."),
        "fonction"        : obe.maximisation_gains_puis_minimisation_ene,
    },
    {
         "label"          : "**Minimisation de l'ENE, puis maximisation des gain économiques**", 
         "explication"    : ("Le site fait en sorte d'évacuer au maximum la "
                             "production hors batterie (minimisation de l'ENE) "
                             "; dans la marge restante, il maximise ses"
                             "revenus."),
        "fonction"        : obe.minimisation_ene_puis_maximisation_gains,
    },
    {
         "label"          : "**Minimisation de l'ENE en décalant la production au plus près des heures de production**",
         "explication"    : ("Le site ne considère pas de critère économique. "
                             "Il fait en sorte d'évacuer autant de production " 
                             "que possible à chaque instant, et pour les cas " 
                             "où l'évacuation est contrainte, la batterie "
                             "décale cette injection aux heures juste après " 
                             "ces contraintes."),
        "fonction"        : obe.minimisation_ene_puis_report_injections,
    },
]


def get_optim_by_label(label_optim):
    for type_optim in LST_MODES_OPTIM:
        if type_optim["label"] == label_optim:
            return type_optim
    # Si on n'a pas trouvé
    return None
                              
                                  
# =============================================================================
# Mise à plat dans un dico du dico structuré issu de la description YAML 
# d'un site
# =============================================================================
def flatten_dict(d):
    d = copy.deepcopy(d)
    if len(d) > 0:
        k = list(d.keys())[0] # première clé
        v = d.pop(k)
        if isinstance(v, dict):
            res = flatten_dict(v)
        elif v == "undef":
            res = {k : None}
        else:
            res = {k : v}
        res.update(flatten_dict(d))
        return res   
    else:
        return {}


# =============================================================================
# Validation du paramétrage d'une étude, et mise en forme de ses éléments 
# constitutifs
# =============================================================================
def validation_parametrage(display_msgs, default_prices, default_generation):
    ''' 
    Valide le paramétrage de l'étude. Si elle est OK, retourne un dataframe 
    avec les chroniques "correctement formatté" et un dico de configuration
    du site.
    Les paramètres servent respectivement à :
        - afficher ou non les messages d'erreur
        - initialiser les prix par défauts s'il n'y a pas de chronique de prix 
          (prix identiques pour tous les pas de temps)
        - initialiser la chronique de production, si elle n'est pas fournie
    '''
    warning_prix_inutiles = False

    # Vérification que la configuration du site est présente
    if 'dico_cfg_site' not in st.session_state:
        if display_msgs:
            st.error("Une configuration de site est requise !")
        return 

    # Vérification qu'on a assez de chroniques pour optimiser qqchose...
    if not (('df_prix' in st.session_state) or ('df_injections' in st.session_state)):
        if display_msgs:
            st.error('''
                     Il n'y a rien à optimiser si une chronique de prix 
                     ou une chronique d'injections ne sont pas fournies
                     (a minima).
                     ''')
        return 

    # Concaténation des données temporelles dans un unique dataframe
    columns = ["prix", "p_min", "p_max", "p_inj"]
    df_chroniques = pd.DataFrame(columns=columns, dtype=float)
    
    # Chronique des prix (prise en compte uniquement si on n'est pas dans le 
    # mode d'optimisation "**Minimisation de l'ENE en décalant la production au plus près des heures de production**")
    if 'df_prix' in st.session_state and st.session_state['mode_optim'] == "**Minimisation de l'ENE en décalant la production au plus près des heures de production**":
        warning_prix_inutiles = True
    elif 'df_prix' in st.session_state:
        df_prix = st.session_state['df_prix']
        if df_prix.shape[1] != 1:
            if display_msgs:
                st.error('''
                         Il doit y avoir 1 colonne dans le fichier des données de prix.
                         ''')
            return
        elif df_prix.shape[0] < 10:
            if display_msgs:
                st.error('''
                         Il y a trop peu de pas de temps dans le fichier des données de prix.
                         Est-ce normal?
                         ''')
            return
        else:
            df_chroniques["prix"] = df_prix.iloc[:, 0].to_list()
            
    # Chroniques des limitations
    if 'df_limitations' in st.session_state:
        df_limitations = st.session_state['df_limitations']
        if df_limitations.shape[1] != 2:
            if display_msgs:
                st.error('''
                         Il doit y avoir 2 colonnes dans le fichier des données des limitations.
                         ''')
            return
        elif df_limitations.shape[0] < 10:
            if display_msgs:
                st.error('''
                         Il y a trop peu de pas de temps dans le fichier des limitations.
                         Est-ce normal?
                         ''')
            return
        elif (df_chroniques.shape[0] > 0) and (df_chroniques.shape[0] != df_limitations.shape[0]):
            if display_msgs:
                st.error('''
                         Il n'y a pas le même nombre de lignes dans le fichier des 
                         limitations que dans les qutres fichiers de données 
                         chronologiques.
                         Vérifier la cohérence des nombres de lignes des fichiers CSV...
                         ''')
            return            
        elif (df_limitations.iloc[1] < df_limitations.iloc[0]).any():
            if display_msgs:
                st.error('''
                         Dans le fichier des données des limitations, chaque valeur 
                         de la première colonne doit  être inférieure ou égale à 
                         la valeur correspondante dans la seconde colonne.
                         ''')
            return
        else:
            df_chroniques["p_min"] = df_limitations.iloc[:, 0].to_list()
            df_chroniques["p_max"] = df_limitations.iloc[:, 1].to_list()
            
    # Chroniques des injections
    if 'df_injections' in st.session_state:
        df_injections = st.session_state['df_injections']
        if df_injections.shape[1] != 1:
            if display_msgs:
                st.error('''
                         Il doit y avoir 1 colonne dans le fichier des données d'injections.
                         ''')
            return
        elif df_injections.shape[0] < 10:
            if display_msgs:
                st.error('''
                         Il y a trop peu de pas de temps dans le fichier des injections.
                         Est-ce normal?
                         ''')
            return
        elif (df_chroniques.shape[0] > 0) and (df_chroniques.shape[0] != df_injections.shape[0]):
            if display_msgs:
                st.error('''
                         Il n'y a pas le même nombre de lignes dans le fichier des 
                         injections que dans les qutres fichiers de données 
                         chronologiques.
                         Vérifier la cohérence des nombres de lignes des fichiers CSV...
                         ''')
            return            
        else:
            df_chroniques["p_inj"] = df_injections.iloc[:, 0].to_list()
    
    # Compléments sur les données du DataFrame et validation de la présence de NaN
    if pd.isna(df_chroniques["prix"]).all():
        df_chroniques["prix"] = default_prices
    if pd.isna(df_chroniques["p_inj"]).all():
        df_chroniques["p_inj"] = default_generation


    # Recherche de NaN (chroniques à trous) 
    if pd.isna(df_chroniques["prix"]).any(): # il ne peut pas y avoir que des NaN dans cette colonne
        if display_msgs:
            st.error('''
                     Il y a des valeurs NaN dans les données de prix.
                     ''')
        return            
    if pd.isna(df_chroniques["p_min"]).any() and not pd.isna(df_chroniques["p_min"]).all():
        if display_msgs:
            st.error('''
                     Il y a des valeurs NaN dans les données de limitation.
                     ''')
        return            
    if pd.isna(df_chroniques["p_max"]).any() and not pd.isna(df_chroniques["p_max"]).all():
        if display_msgs:
            st.error('''
                     Il y a des valeurs NaN dans les données de limitation.
                     ''')
        return            
    if pd.isna(df_chroniques["p_inj"]).any(): # il ne peut pas y avoir que des NaN dans cette colonne
        if display_msgs:
            st.error('''
                     Il y a des valeurs NaN dans les données d'injections.
                     ''')
        return            
    
    # Validation de la cohérence des données de site.
    dico_site = flatten_dict(st.session_state.dico_cfg_site)
    if not isinstance(dico_site['p_max_batt'], float):
        if display_msgs:
            st.error('''
                     Dans la configuration du site, 'p_max_batt' doit être un float.
                     ''')
        return            
    if not isinstance(dico_site['p_min_batt'], float):
        if display_msgs:
            st.error('''
                     Dans la configuration du site, 'p_min_batt' doit être un float.
                     ''')
        return
    if dico_site['p_min_batt'] > dico_site['p_max_batt']:
        if display_msgs:
            st.error('''
                     Dans la configuration du site, 'p_min_batt' doit être inférieur à 'p_max_batt'.
                     ''')
        return
    if not isinstance(dico_site['soc_max'], float):
        if display_msgs:
            st.error('''
                     Dans la configuration du site, 'soc_max' doit être un float.
                     ''')
        return            
    if not isinstance(dico_site['soc_min'], float):
        if display_msgs:
            st.error('''
                     Dans la configuration du site, 'soc_min' doit être un float.
                     ''')
        return
    if dico_site['soc_min'] > dico_site['soc_max'] or dico_site['soc_min'] < 0 or  dico_site['soc_max'] > 100:
        if display_msgs:
            st.error('''
                     Dans la configuration du site, 'soc_min' doit être inférieur à 'soc_max'.
                     Ces deux valeurs doivent être comprises entre 0 et 100.
                     ''')
        return
    if not isinstance(dico_site['Limitations spécifiques'], list):
        if display_msgs:
            st.error('''
                     Dans la configuration du site, 'Limitations spécifiques' doit être une liste.
                     ''')
        return            
    if not isinstance(dico_site['p_racc'], float):
        if display_msgs:
            st.error('''
                     Dans la configuration du site, 'p_racc' doit être un float.
                     ''')
        return            
    if dico_site['p_racc'] <= 0:
        if display_msgs:
            st.error('''
                     Dans la configuration du site, 'p_racc' doit être positif.
                     ''')
        return            
    if not isinstance(dico_site['e_max'], float):
        if display_msgs:
            st.error('''
                     Dans la configuration du site, 'e_max' doit être un float.
                     ''')
        return            
    if dico_site['e_max'] <= 0:
        if display_msgs:
            st.error('''
                     Dans la configuration du site, 'e_max' doit être positif.
                     ''')
        return            
    if not isinstance(dico_site['eta_c'], float):
        if display_msgs:
            st.error('''
                     Dans la configuration du site, 'eta_c' doit être un float.
                     ''')
        return            
    if dico_site['eta_c'] <= 0 or dico_site['eta_c'] > 100:
        if display_msgs:
            st.error('''
                     Dans la configuration du site, 'eta_c' doit être compris entre 0 et 100.
                     ''')
        return            
    if not isinstance(dico_site['eta_d'], float):
        if display_msgs:
            st.error('''
                     Dans la configuration du site, 'eta_d' doit être un float.
                     ''')
        return            
    if dico_site['eta_d'] <= 0 or dico_site['eta_d'] > 100:
        if display_msgs:
            st.error('''
                     Dans la configuration du site, 'eta_d' doit être compris entre 0 et 100.
                     ''')
        return            
    if not isinstance(dico_site['conso_veille'], float):
        if display_msgs:
            st.error('''
                     Dans la configuration du site, 'conso_veille' doit être un float.
                     ''')
        return            
    if dico_site['conso_veille'] < 0 :
        if display_msgs:
            st.error('''
                     Dans la configuration du site, 'conso_veille' doit être positif ou nul.
                     ''')
        return            

    # Envoi des données, qui sont bien valides     
    return dico_site, df_chroniques, warning_prix_inutiles