src/WattFieldsCommon/optim_analytics.py

# Functions to generate dataframes from run results (Runs dict results are part of simu output pkl file)

# PKL STRUCTURE FILE AS INPUT THAT IS THE SAVED OPTIMISATION -> CONTAINS:
# SIMU DICT : {'config':self.config, 'runs':pareto_optimal_runs, 'computation_time':computation_time}
# and ech element in runs :
#             {'general_metrics':general_metrics,
#'df_weekly_blocks':df_weekly_blocks,
#'df_daily_pv_energy':df_daily_pv_energy,
#'list_dssat_outputs_per_row':list_dssat_outputs_per_row}
# and each list_dssat_outputs_per_row: CONTAINS {'index_row', 'x_row', 'dssat_summary', 'df_dssat_daily'}
# and each dssat_summary': """{'RUN': '1','TRT': '1','FLO': '213','MAT': '248','TOPWT': '3876','HARWT': '3239','RAIN': '644','TIRR': '0','CET': '431','PESW': '169','TNUP': '0','TNLF': '-99','TSON': '0','TSOC': '100'},"""

"""And general_metrics :
        general_metrics = {'run_id':self.id,
                    'average_agri_yield':self.get_average_agri_yield(),
                    'average_pv_energy':self.get_average_pv_energy(),
                    'max_dssat_end_time':self.max_dssat_end_time} 
                    'normalized_ac_collected_energy','normalized_agri_yield'}"""


# REMEMBER THAT ALL THOSE METHODS GENERATE DATASETS FOR A SINGLE RUN !


import datetime

import pandas as pd
import numpy as np

import warnings
warnings.simplefilter(action='ignore', category=FutureWarning)

from WattFieldsCommon.constants import DSSAT_OUTPUT_KEYS

THRESHOLD_WFGD_WATER_STRESS = 0.7

def generate_run_df_summary(
    dict_run: dict,
    ref_agri_yield: float,
    ref_pv_energy: float,
    surface_panel: float,
    pv_module_pnom: float,
    planting_date: datetime.datetime,
) -> pd.DataFrame:
    """Based on a dict run, it generates dataframe with key metrics.

    Args:
        dict_run (dict): input dict run with keys (general_metrics, df_weekly_blocks, df_daily_pv_energy, list_dssat_outputs_per_row)
        ref_agri_yield (float): Ref agri yield
        ref_pv_energy (float): Ref pv energy
        surface_panel (float): Surface of the panel
        pv_module_pnom (float): PV module nominal power
        planting_date (datetime.datetime): Date of plantation

    Returns:
        pd.DataFrame: Return dataaframe with key metrics
    """

    df_recap = pd.Series()

    # AGRI SCORES
    df_recap["agri_average_yield"] = dict_run["general_metrics"]["average_agri_yield"]
    df_recap["normalized_agri_average_yield"] = int(
        df_recap["agri_average_yield"] / ref_agri_yield * 100
    )

    # PV SCORES
    df_recap["collected_ac_energy"] = dict_run["general_metrics"]["average_pv_energy"]
    df_recap["normalized_collected_ac_energy"] = int(
        df_recap["collected_ac_energy"] / ref_pv_energy * 100
    )
    df_recap["collected_ac_energy_per_meter2"] = int(
        df_recap["collected_ac_energy"] / surface_panel
    )
    df_recap["collected_ac_energy_per_peak_power"] = int(
        df_recap["collected_ac_energy"] / (pv_module_pnom / 1000)
    )  # in kW

    # OTHERS
    df_recap["run_duration"] = int(
        (dict_run["general_metrics"]["max_dssat_end_time"] - planting_date).days
    )
    weekly_choices = dict_run["df_weekly_blocks"]["weekly_strategy_choice"]
    df_recap["nb_block_anti-tracking"] = weekly_choices[weekly_choices == "anti-tracking"].count()

    return df_recap


def generate_df_temporal_analysis(dict_run: dict, dict_ref_agri_run: dict):
    """Generate a dataframe with temporal evolution of DSSAT variables One
    column for each dual (row, dssat_var)

    Args:
        dict_run (dict): Input run
        dict_ref_agri_run (dict): Ref run agri

    Returns:
        pd.DataFRame: dataframe with temporal evolution (one row per day)
    """
    all_dfs = []
    for row in dict_run["list_dssat_outputs_per_row"]:
        x_row = round(row["x_row"], 2)
        keys_to_select = list(set(DSSAT_OUTPUT_KEYS["ALL"].keys()).intersection(set(row["df_dssat_daily"].columns)))
        df_row = row["df_dssat_daily"][keys_to_select]
        df_row = df_row.rename(columns={col: f"x={x_row}_{col}" for col in df_row.columns})
        all_dfs.append(df_row)
    df_all_rows = pd.concat(all_dfs, axis=1)

    variables = list(set([col.split('_')[1] for col in df_row.columns]))

    # ADD MEAN AS EXTRA DF
    dict_mean_rows = {}
    for var in variables:
        df_selected_var_all_rows = df_all_rows.filter(like=var)
        dict_mean_rows[f'MOYENNE_{var}'] = list(df_selected_var_all_rows.mean(axis=1))
    df_mean_col = pd.DataFrame(dict_mean_rows)
    df_mean_col.loc[:, 'date'] = list(df_all_rows.index)
    df_mean_col = df_mean_col.set_index('date')

    # ADD REF AS EXTRA DF
    df_ref_dssat_ref = dict_ref_agri_run["list_dssat_outputs_per_row"][0]["df_dssat_daily"]
    df_ref_dssat_ref = df_ref_dssat_ref.rename(
        columns={col: f"AGRI_REF_{col}" for col in df_ref_dssat_ref.columns}
    )
    return pd.concat([df_all_rows, df_mean_col, df_ref_dssat_ref], axis=1)



def generate_df_spatial_analysis(dict_run: dict, dict_ref_agri_run: dict):
    """Generate a dataframe with spatial evolution of DSSAT variables (one var
    per dssat row, for example HWAT_row_x=2)

    Args:
        dict_run (dict): input run
        dict_ref_agri_run (dict): ref run agri

    Returns:
        pd.DataFRame: Dataframe where each column accounts for a row
    """
    column_names = [
        f'x_row={round(row["x_row"],2)}' for row in dict_run["list_dssat_outputs_per_row"]
    ]

    dicts_summary_per_row = [row["dssat_summary"] for row in dict_run["list_dssat_outputs_per_row"]]

    # FRACTION IRRAD
    sum_rad_ref_agri = dict_ref_agri_run["list_dssat_outputs_per_row"][0]['df_dssat_daily']['SRAD'].sum()
    for i in range(len(dicts_summary_per_row)):
        dicts_summary_per_row[i]['RATIO_IRRAD_AGRI_REF'] = round(dict_run["list_dssat_outputs_per_row"][i]['df_dssat_daily']['SRAD'].sum() / sum_rad_ref_agri, 3) * 100

    # NB DAYS WHERE HYDRIC STRESS
    for i in range(len(dicts_summary_per_row)):
        water_stress_level = dict_run["list_dssat_outputs_per_row"][i]['df_dssat_daily']['WFGD']
        dicts_summary_per_row[i]['NB_DAYS_WATER_STRESS'] = len(water_stress_level[water_stress_level>THRESHOLD_WFGD_WATER_STRESS])

    # ADD REF AS EXTRA ROW
    column_names.append("AGRI_REF_ROW")
    dicts_summary_per_row.append(dict_ref_agri_run["list_dssat_outputs_per_row"][0]["dssat_summary"])
    dicts_summary_per_row[-1]['RATIO_IRRAD_AGRI_REF'] = 100
    ref_agri_water_stress_level = dict_ref_agri_run["list_dssat_outputs_per_row"][0]['df_dssat_daily']['WFGD']
    dicts_summary_per_row[-1]['NB_DAYS_WATER_STRESS'] = len(ref_agri_water_stress_level[ref_agri_water_stress_level>THRESHOLD_WFGD_WATER_STRESS])

    # Compute total irradience per row
    # total_irrad = int(np.nanmean(row["df_dssat_daily"]["SRAD"]))

    df_all_rows = pd.DataFrame(dicts_summary_per_row).T
    df_all_rows.columns = column_names
    return df_all_rows


def generate_df_block_temporal_analysis(dict_run: dict, ref_pv_run: dict, pv_module_pnom: float):
    """From run to PV energy temporal data.

    Args:
        dict_run (dict): Input run
        ref_pv_run (dict): Ref run PV (full backtracking)
        pv_module_pnom (float): Nominal Power of PV module

    Returns:
        pd.DataFrame: DataFrame with nb_rows = nb_blocks (account generally for one week)
    """
    df_block = dict_run["df_weekly_blocks"]
    nb_blocks = len(df_block)
    df_block["PV_REF_block_pv_power_per_peak_power"] = list(
        ref_pv_run["df_weekly_blocks"]["weekly_collected_ac_energy"] / pv_module_pnom
    )[:nb_blocks]
    return df_block

def generate_df_gstd_block(dict_run, df_temporal_analysis, dict_ref_agri_run):

    # GENERATE GSTD BINS
    selected_columns = [col for col in df_temporal_analysis.columns if 'GSTD' in col and 'x=' in col]
    gstd_all_rows = df_temporal_analysis[selected_columns]
    max_gstd = max(gstd_all_rows.max().values)
    #Range of std can vary a lot so different use cases...
    if max_gstd > 20:
        # Case of wheat for example
        bins = np.arange(0, max_gstd + 10, 10)
    else:
        # Others species
        bins = np.arange(0, max_gstd + 1, 1)
    #bins = [(a,b) for a,b in zip(bins[:-1, 1:])]
    labels = [f'{a} - {b}' for a,b in zip(bins[:-1],bins[1:])]

    x_row_cols = [col for col in df_temporal_analysis.columns if 'x=' in col]
    x_row_cols = [col for col in df_temporal_analysis.columns if (('GSTD' in col) or ('SRAD' in col))]
    df_temporal_analysis = df_temporal_analysis[x_row_cols]

    x_positions = [round(row["x_row"],2) for row in dict_run["list_dssat_outputs_per_row"]]
    average_outputs = []

    df_dssat_ref = dict_ref_agri_run["list_dssat_outputs_per_row"][0]['df_dssat_daily']
    df_dssat_ref.loc[:, 'GSTD_bin'] = pd.cut(df_dssat_ref['GSTD'], bins=bins, labels=labels, right=True)
    ref_bins_average = df_dssat_ref.groupby('GSTD_bin')['SRAD'].mean().reset_index()[['GSTD_bin', f'SRAD']]
    ref_bins_average = ref_bins_average.rename(columns={'SRAD':'REF_SRAD'})

    for x_pos in x_positions:
        df_x_pos = df_temporal_analysis[[f'x={x_pos}_GSTD', f'x={x_pos}_SRAD']]
        df_x_pos.loc[:, 'GSTD_bin'] = pd.cut(df_x_pos[f'x={x_pos}_GSTD'], bins=bins, labels=labels, right=True)
        x_pos_average = df_x_pos.groupby('GSTD_bin')[f'x={x_pos}_SRAD'].mean().reset_index()[['GSTD_bin', f'x={x_pos}_SRAD']]
        x_pos_average = x_pos_average.merge(ref_bins_average, on='GSTD_bin')
        x_pos_average[f'x={x_pos}_SRAD_ref_ratio'] = np.divide(
            100 * np.array(x_pos_average[f'x={x_pos}_SRAD']), #percentage
            np.array(x_pos_average['REF_SRAD']),
            out=np.full_like(np.array(x_pos_average[f'x={x_pos}_SRAD']), np.nan),  # Fill with NaN
            where=np.array(x_pos_average['REF_SRAD']) != 0  # Only divide where REF_SRAD is not zero
        )       
        average_outputs.append(x_pos_average[[f'x={x_pos}_SRAD_ref_ratio', 'GSTD_bin']].set_index('GSTD_bin'))

    df_row_srad_averages = pd.concat(average_outputs, axis=1).round(5)
    df_row_srad_averages['MOYENNE_SRAD_ref_ratio'] = df_row_srad_averages.mean(axis=1)

    return df_row_srad_averages