# 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