import os import pickle import random import glob import dice_ml import gradio as gr import pandas as pd from sklearn.model_selection import train_test_split def read_csv(dataset): path = './data/' + dataset + '.csv' data = pd.read_csv(path) return data.head(10) def train_model(input_df, target, test_size, model_name, features_to_drop): from sklearn.compose import ColumnTransformer from sklearn.ensemble import RandomForestClassifier, RandomForestRegressor from sklearn.pipeline import Pipeline from sklearn.preprocessing import OneHotEncoder, StandardScaler path = './data/' + input_df + '.csv' data = pd.read_csv(path, index_col=0) data = data.dropna() data = data.drop(features_to_drop, axis=1) metrics = ['TSV','TPV','TCV','TSL'] metrics.remove(target[0]) data = data.drop(metrics, axis=1) features = data.drop(target, axis=1).columns.to_list() target_f = data[target[0]] datasetX = data.drop(target, axis=1) X_train, X_test, y_train, y_test = train_test_split(datasetX, target_f, test_size=test_size, random_state=42) categorical_features = X_train.columns.difference(features) # We create the preprocessing pipelines for both numeric and categorical data. numeric_transformer = Pipeline(steps=[ ('scaler', StandardScaler())]) categorical_transformer = Pipeline(steps=[ ('onehot', OneHotEncoder(handle_unknown='ignore'))]) transformations = ColumnTransformer( transformers=[ ('num', numeric_transformer, features), ('cat', categorical_transformer, categorical_features)]) regr = Pipeline(steps=[('preprocessor', transformations), ('regressor', RandomForestRegressor())]) model = regr.fit(X_train, y_train) pickle.dump(model, open('./' + model_name + '.pkl', 'wb')) return 'Model Saved' def generate_cfs_total(input_df_T, target_T, radio_T, predefined_T, custom_T, dropped_features_T, freeze_features_T, model_T): path = './data/' + input_df_T + '.csv' data = pd.read_csv(path) data = data.dropna() model = pickle.load(open('./' + model_T + '.pkl', 'rb')) data = data.drop(dropped_features_T, axis=1) metrics = ['TSV','TPV','TCV','TSL'] metrics.remove(target_T[0]) data = data.drop(metrics, axis=1) features = data.drop(target_T[0], axis=1).columns.tolist() target = data[target_T[0]] datasetX = data.drop('TSV', axis=1) x_train, x_test, y_train, y_test = train_test_split(datasetX, target, test_size=0.2, random_state=0) always_immutable = ['AvgMaxDailyTemp','AvgMinDailyTemp','School','DAY','StartTime'] freezed = always_immutable + freeze_features_T + [target_T[0]] features_to_vary = data.columns.difference(freezed).to_list() d = dice_ml.Data(dataframe=data, continuous_features=features, outcome_name=target_T[0]) m = dice_ml.Model(model=model, backend='sklearn', model_type='regressor') exp = dice_ml.Dice(d, m, method='random') if radio_T == 'Predefined': random_index = random.randint(0, len(x_train-2)) print(int(predefined_T)) query_instances = x_test[random_index:random_index+int(predefined_T)] elif radio_T == 'Custom': query_instances = custom_T dice_exp = exp.generate_counterfactuals(query_instances, total_CFs=4, desired_range=[0.0, 2.0], features_to_vary=features_to_vary) for i in range(len(dice_exp.cf_examples_list)): if dice_exp.cf_examples_list[i].final_cfs_df is not None: dice_exp.cf_examples_list[i].final_cfs_df.to_csv(path_or_buf=f'c{i}.csv', index=False) df = pd.concat( [ pd.read_csv(filename).assign(source=filename) for filename in glob.glob('*.csv') ], ignore_index=True ) return [query_instances, df] def generate_cfs_individual(input_df_I, target_I, radio_I, predefined_I, custom_I, dropped_features_I, freeze_features_I, model_I): pass with gr.Blocks() as demo: with gr.Tab('Dataset'): gr.Markdown('Visualize the dataset to apply CFML') df_vis = gr.Dropdown(['IndividualClothingBinary','IndividualClothingBinary+3Binary', 'IndividualClothingValue','IndividualClothingValue+3Binary','Multi_TotalCLO_w_Chair', 'Summer','TotalClothingValue','TotalClothingValue+3Binary'], label='Dataset') output_vis = gr.DataFrame() button_vis = gr.Button(label="Run") with gr.Tab('Model'): gr.Markdown('Choose the features to apply CFML') input_df = gr.Dropdown(['IndividualClothingBinary','IndividualClothingBinary+3Binary', 'IndividualClothingValue','IndividualClothingValue+3Binary','Multi_TotalCLO_w_Chair', 'Summer','TotalClothingValue','TotalClothingValue+3Binary'], label='Dataset') target = gr.CheckboxGroup(['TSV','TPV','TCV','TSL'], label='Target Metric', info='Please select only one') test_size = gr.Slider(minimum=0.1, maximum=0.5, step=0.05, value=0.2, label='Test Size', interactive=True) model_name = gr.Textbox(label='Model Name', placeholder='Enter the model name') features_to_drop = gr.CheckboxGroup(['SwC', 'MC', 'Grade', 'Age', 'Gender'], label='Features to Drop', info='Select the features to drop') model_output = gr.Textbox(label='Status') button_model = gr.Button(label="Train Model") #list add .pkl files from models folder models = [] for file in os.listdir('./'): if file.endswith('.pkl'): models.append(file.split('.')[0]) with gr.Tab('Counterfactuals-Total'): gr.Markdown('Generate Counterfactuals for Total CLO Dataset') input_df_T = gr.Dropdown(['Multi_TotalCLO_w_Chair','Summer', 'TotalClothingValue','TotalClothingValue+3Binary'], label='Dataset') target_T = gr.CheckboxGroup(['TSV','TPV','TCV','TSL'], label='Target Metric', info='Please select only one') #target_T_range = gr.Textbox(label='Target Range', placeholder='Enter the target range [start,end]') radio_T = gr.Radio(['Predefined', 'Custom'], label='Type of Input') predefined_T = gr.Number(default=0, label='Number of inputs to provide') custom_T = gr.Dataframe( headers=['DAY','School','SchoolType','StartTime','AvgMaxDailyTemp','AvgMinDailyTemp','AvgIndoorRelativeHumidity', 'IndoorTempDuringSurvey','Grade','Age','Gender','FormalClothing','TotalCLOwithChair'], row_count=(2, 'dynamic') ) dropped_features_T = gr.CheckboxGroup(['SwC', 'MC', 'Grade', 'Age', 'Gender'], label='Features to Drop', info='Select the features that are dropped from feature set') freeze_features_T = gr.CheckboxGroup(['AvgIndoorRelativeHumidity', 'IndoorTempDuringSurvey','Grade','Age','Gender','FormalClothing','TotalCLOwithChair'], info = 'Select the features to be freezed to generate CFs') model_T = gr.Dropdown(models, label='Model', info='Select the model to generate CFs') output_T = gr.DataFrame() output_T1 = gr.DataFrame() button_cf_T = gr.Button(label="Generate CFs") with gr.Tab('Counterfactuals-Individual'): gr.Markdown('Generate Counterfactuals for Individual Clothing Dataset') input_df_I = gr.Dropdown(['IndividualClothingBinary','IndividualClothingBinary+3Binary', 'IndividualClothingValue','IndividualClothingValue+3Binary'], label='Dataset') target_I = gr.CheckboxGroup(['TSV','TPV','TCV','TSL'], label='Target Metric', info='Please select only one') radio_I = gr.Radio(['Predefined', 'Custom'], label='Type of Input') predefined_I = gr.Number(default=0, label='Number of inputs to provide') custom_I = gr.Dataframe( headers=['DAY','School','SchoolType','StartTime','AvgMaxDailyTemp','AvgMinDailyTemp','AvgIndoorRelativeHumidity','IndoorTempDuringSurvey', 'Grade','Age','Gender','FormalClothing','Pant','Trackpant','Halfshirt','Blazer','Jacket','Skirt', 'FullShirt','HalfSweater','Tshirt','Socks','Thermal','Vest','FullSweater','SwC','MC'], row_count=(2, 'dynamic') ) dropped_features_I = gr.CheckboxGroup(['SwC', 'MC', 'Grade', 'Age', 'Gender'], label='Features to Drop', info='Select the features that are dropped from feature set') freeze_features_I = gr.CheckboxGroup(['AvgIndoorRelativeHumidity','IndoorTempDuringSurvey', 'Grade','Age','Gender', 'FormalClothing','Pant','Trackpant','Halfshirt','Blazer','Jacket','Skirt', 'FullShirt','HalfSweater','Tshirt','Socks','Thermal','Vest','FullSweater','SwC','MC'], info='Select the features to be freezed to generate CFs') model_I = gr.Dropdown(models, label='Model', info='Select the model to generate CFs') button_cf_I = gr.Button(label="Generate CFs") button_vis.click(read_csv, df_vis, outputs=output_vis) button_model.click(train_model, [input_df, target, test_size, model_name, features_to_drop], outputs=model_output) button_cf_T.click(generate_cfs_total, [input_df_T, target_T, radio_T, predefined_T, custom_T, dropped_features_T, freeze_features_T, model_T], outputs=[output_T, output_T1]) demo.launch()