import gradio as gr import matplotlib.pyplot as plt import numpy as np import os import pandas as pd import re import selfies as sf import torch import xgboost as xgb from PIL import Image from rdkit import Chem, RDLogger from rdkit.Chem import DataStructs, AllChem, Descriptors, QED, Draw from rdkit.Chem.Crippen import MolLogP from rdkit.Contrib.SA_Score import sascorer from sklearn.kernel_ridge import KernelRidge from sklearn.linear_model import LinearRegression from sklearn.svm import SVR from transformers import BartForConditionalGeneration, AutoTokenizer from transformers.modeling_outputs import BaseModelOutput os.environ["OMP_MAX_ACTIVE_LEVELS"] = "1" import models.fm4m as fm4m RDLogger.logger().setLevel(RDLogger.ERROR) # Function to display molecule image from SMILES def smiles_to_image(smiles): mol = Chem.MolFromSmiles(smiles) return Draw.MolToImage(mol) if mol else None # Dictionary for SMILES strings and corresponding images (you can replace with your actual image paths) smiles_image_mapping = { "Mol 1": { "smiles": "C=C(C)CC(=O)NC[C@H](CO)NC(=O)C=Cc1ccc(C)c(Cl)c1", "image": "img/img1.png", }, # Example SMILES for ethanol "Mol 2": { "smiles": "C=CC1(CC(=O)NC[C@@H](CCCC)NC(=O)c2cc(Cl)cc(Br)c2)CC1", "image": "img/img2.png", }, # Example SMILES for butane "Mol 3": { "smiles": "C=C(C)C[C@H](NC(C)=O)C(=O)N1CC[C@H](NC(=O)[C@H]2C[C@@]2(C)Br)C(C)(C)C1", "image": "img/img3.png", }, # Example SMILES for ethylamine "Mol 4": { "smiles": "C=C1CC(CC(=O)N[C@H]2CCN(C(=O)c3ncccc3SC)C23CC3)C1", "image": "img/img4.png", }, # Example SMILES for diethyl ether "Mol 5": { "smiles": "C=CCS[C@@H](C)CC(=O)OCC", "image": "img/img5.png", }, # Example SMILES for chloroethane } datasets = [" ", "BACE", "ESOL", "Load Custom Dataset"] models_enabled = [ "SELFIES-TED", "MHG-GED", "MolFormer", "SMI-TED", "Mordred", "MorganFingerprint", ] fusion_available = ["Concat"] # Function to handle evaluation and logging def evaluate_and_log(models, dataset, task_type, eval_output, state): task_dic = {'Classification': 'CLS', 'Regression': 'RGR'} result = f"{eval_output}" result = result.replace(" Score", "") new_entry = { "Selected Models": str(models), "Dataset": dataset, "Task": task_dic[task_type], "Result": result, } new_entry_df = pd.DataFrame([new_entry]) state["log_df"] = pd.concat([new_entry_df, state["log_df"]]) return state["log_df"] # Load images for selection def load_image(path): try: return Image.open(smiles_image_mapping[path]["image"]) except: pass # Function to handle image selection def handle_image_selection(image_key): smiles = smiles_image_mapping[image_key]["smiles"] mol_image = smiles_to_image(smiles) return smiles, mol_image def calculate_properties(smiles): mol = Chem.MolFromSmiles(smiles) if mol: qed = QED.qed(mol) logp = MolLogP(mol) sa = sascorer.calculateScore(mol) wt = Descriptors.MolWt(mol) return qed, sa, logp, wt return None, None, None, None # Function to calculate Tanimoto similarity def calculate_tanimoto(smiles1, smiles2): mol1 = Chem.MolFromSmiles(smiles1) mol2 = Chem.MolFromSmiles(smiles2) if mol1 and mol2: fp1 = AllChem.GetMorganFingerprintAsBitVect(mol1, 2) fp2 = AllChem.GetMorganFingerprintAsBitVect(mol2, 2) return round(DataStructs.FingerprintSimilarity(fp1, fp2), 2) return None gen_tokenizer = AutoTokenizer.from_pretrained("ibm/materials.selfies-ted") gen_model = BartForConditionalGeneration.from_pretrained("ibm/materials.selfies-ted") def generate(latent_vector, mask): encoder_outputs = BaseModelOutput(latent_vector) decoder_output = gen_model.generate( encoder_outputs=encoder_outputs, attention_mask=mask, max_new_tokens=64, do_sample=True, top_k=5, top_p=0.95, num_return_sequences=1, ) selfies = gen_tokenizer.batch_decode(decoder_output, skip_special_tokens=True) return [sf.decoder(re.sub(r'\]\s*(.*?)\s*\[', r']\1[', i)) for i in selfies] def perturb_latent(latent_vecs, noise_scale=0.5): return ( torch.tensor( np.random.uniform(0, 1, latent_vecs.shape) * noise_scale, dtype=torch.float32, ) + latent_vecs ) def encode(selfies): encoding = gen_tokenizer( selfies, return_tensors='pt', max_length=128, truncation=True, padding='max_length', ) input_ids = encoding['input_ids'] attention_mask = encoding['attention_mask'] outputs = gen_model.model.encoder( input_ids=input_ids, attention_mask=attention_mask ) model_output = outputs.last_hidden_state return model_output, attention_mask # Function to generate canonical SMILES and molecule image def generate_canonical(smiles): s = sf.encoder(smiles) selfie = s.replace("][", "] [") latent_vec, mask = encode([selfie]) gen_mol = None for i in range(5, 51): print("Searching Latent space") noise = i / 10 perturbed_latent = perturb_latent(latent_vec, noise_scale=noise) gen = generate(perturbed_latent, mask) mol = Chem.MolFromSmiles(gen[0]) if mol: gen_mol = Chem.MolToSmiles(mol) if gen_mol != Chem.MolToSmiles(Chem.MolFromSmiles(smiles)): break else: print('Abnormal molecule:', gen[0]) if gen_mol: # Calculate properties for ref and gen molecules print("calculating properties") ref_properties = calculate_properties(smiles) gen_properties = calculate_properties(gen_mol) tanimoto_similarity = calculate_tanimoto(smiles, gen_mol) # Prepare the table with ref mol and gen mol data = { "Property": ["QED", "SA", "LogP", "Mol Wt", "Tanimoto Similarity"], "Reference Mol": [ ref_properties[0], ref_properties[1], ref_properties[2], ref_properties[3], tanimoto_similarity, ], "Generated Mol": [ gen_properties[0], gen_properties[1], gen_properties[2], gen_properties[3], "", ], } df = pd.DataFrame(data) # Display molecule image of canonical smiles print("Getting image") mol_image = smiles_to_image(gen_mol) return df, gen_mol, mol_image return "Invalid SMILES", None, None # Function to display evaluation score def display_eval(selected_models, dataset, task_type, downstream, fusion_type, state): result = None try: downstream_model = downstream.split("*")[0].lstrip() downstream_model = downstream_model.rstrip() hyp_param = downstream.split("*")[-1].lstrip() hyp_param = hyp_param.rstrip() hyp_param = hyp_param.replace("nan", "float('nan')") params = eval(hyp_param) except: downstream_model = downstream.split("*")[0].lstrip() downstream_model = downstream_model.rstrip() params = None try: if not selected_models: return "Please select at least one enabled model." if len(selected_models) > 1: if task_type == "Classification": if downstream_model == "Default Settings": downstream_model = "DefaultClassifier" params = None ( result, state["roc_auc"], state["fpr"], state["tpr"], state["x_batch"], state["y_batch"], ) = fm4m.multi_modal( model_list=selected_models, downstream_model=downstream_model, params=params, dataset=dataset, ) elif task_type == "Regression": if downstream_model == "Default Settings": downstream_model = "DefaultRegressor" params = None ( result, state["RMSE"], state["y_batch_test"], state["y_prob"], state["x_batch"], state["y_batch"], ) = fm4m.multi_modal( model_list=selected_models, downstream_model=downstream_model, params=params, dataset=dataset, ) else: if task_type == "Classification": if downstream_model == "Default Settings": downstream_model = "DefaultClassifier" params = None ( result, state["roc_auc"], state["fpr"], state["tpr"], state["x_batch"], state["y_batch"], ) = fm4m.single_modal( model=selected_models[0], downstream_model=downstream_model, params=params, dataset=dataset, ) elif task_type == "Regression": if downstream_model == "Default Settings": downstream_model = "DefaultRegressor" params = None ( result, state["RMSE"], state["y_batch_test"], state["y_prob"], state["x_batch"], state["y_batch"], ) = fm4m.single_modal( model=selected_models[0], downstream_model=downstream_model, params=params, dataset=dataset, ) if result == None: result = "Data & Model Setting is incorrect" except Exception as e: return f"An error occurred: {e}" return f"{result}" # Function to handle plot display def display_plot(plot_type, state): fig, ax = plt.subplots() if plot_type == "Latent Space": x_batch, y_batch = state.get("x_batch"), state.get("y_batch") ax.set_title("T-SNE Plot") class_0 = x_batch class_1 = y_batch plt.scatter(class_1[:, 0], class_1[:, 1], c='red', label='Class 1') plt.scatter(class_0[:, 0], class_0[:, 1], c='blue', label='Class 0') ax.set_xlabel('Feature 1') ax.set_ylabel('Feature 2') ax.set_title('Dataset Distribution') elif plot_type == "ROC-AUC": roc_auc, fpr, tpr = state.get("roc_auc"), state.get("fpr"), state.get("tpr") ax.set_title("ROC-AUC Curve") try: ax.plot( fpr, tpr, color='darkorange', lw=2, label=f'ROC curve (area = {roc_auc:.4f})', ) ax.plot([0, 1], [0, 1], color='navy', lw=2, linestyle='--') ax.set_xlim([0.0, 1.0]) ax.set_ylim([0.0, 1.05]) except: pass ax.set_xlabel('False Positive Rate') ax.set_ylabel('True Positive Rate') ax.set_title('Receiver Operating Characteristic') ax.legend(loc='lower right') elif plot_type == "Parity Plot": RMSE, y_batch_test, y_prob = ( state.get("RMSE"), state.get("y_batch_test"), state.get("y_prob"), ) ax.set_title("Parity plot") # change format try: print(y_batch_test) print(y_prob) y_batch_test = np.array(y_batch_test, dtype=float) y_prob = np.array(y_prob, dtype=float) ax.scatter( y_batch_test, y_prob, color="blue", label=f"Predicted vs Actual (RMSE: {RMSE:.4f})", ) min_val = min(min(y_batch_test), min(y_prob)) max_val = max(max(y_batch_test), max(y_prob)) ax.plot([min_val, max_val], [min_val, max_val], 'r-') except: y_batch_test = [] y_prob = [] RMSE = None print(y_batch_test) print(y_prob) ax.set_xlabel('Actual Values') ax.set_ylabel('Predicted Values') ax.legend(loc='lower right') return fig # Predefined dataset paths (these should be adjusted to your file paths) predefined_datasets = { " ": " ", "BACE": f"./data/bace/train.csv, ./data/bace/test.csv, smiles, Class", "ESOL": f"./data/esol/train.csv, ./data/esol/test.csv, smiles, prop", } # Function to load a predefined dataset from the local path def load_predefined_dataset(dataset_name): val = predefined_datasets.get(dataset_name) try: file_path = val.split(",")[0] except: file_path = False if file_path: df = pd.read_csv(file_path) return ( df.head(), gr.update(choices=list(df.columns)), gr.update(choices=list(df.columns)), f"{dataset_name.lower()}", ) return ( pd.DataFrame(), gr.update(choices=[]), gr.update(choices=[]), f"Dataset not found", ) # Function to display the head of the uploaded CSV file def display_csv_head(file): if file is not None: # Load the CSV file into a DataFrame df = pd.read_csv(file.name) return ( df.head(), gr.update(choices=list(df.columns)), gr.update(choices=list(df.columns)), ) return pd.DataFrame(), gr.update(choices=[]), gr.update(choices=[]) # Function to handle dataset selection (predefined or custom) def handle_dataset_selection(selected_dataset): if selected_dataset == "Custom Dataset": # Show file upload fields for train and test datasets if "Custom Dataset" is selected return ( gr.update(visible=True), gr.update(visible=True), gr.update(visible=True), gr.update(visible=True), gr.update(visible=True), gr.update(visible=False), gr.update(visible=True), gr.update(visible=True), ) else: return ( gr.update(visible=True), gr.update(visible=False), gr.update(visible=False), gr.update(visible=False), gr.update(visible=False), gr.update(visible=False), gr.update(visible=False), gr.update(visible=False), ) # Function to select input and output columns and display a message def select_columns(input_column, output_column, train_data, test_data, dataset_name): if input_column and output_column: return f"{train_data.name},{test_data.name},{input_column},{output_column},{dataset_name}" return "Please select both input and output columns." def set_dataname(dataset_name, dataset_selector): if dataset_selector == "Custom Dataset": return f"{dataset_name}" return f"{dataset_selector}" # Function to create model based on user input def create_model( model_name, max_depth=None, n_estimators=None, alpha=None, degree=None, kernel=None ): if model_name == "XGBClassifier": model = xgb.XGBClassifier( objective='binary:logistic', eval_metric='auc', max_depth=max_depth, n_estimators=n_estimators, alpha=alpha, ) elif model_name == "SVR": model = SVR(degree=degree, kernel=kernel) elif model_name == "Kernel Ridge": model = KernelRidge(alpha=alpha, degree=degree, kernel=kernel) elif model_name == "Linear Regression": model = LinearRegression() elif model_name == "Default - Auto": model = "Default Settings" return f"{model}" else: return "Model not supported." return f"{model_name} * {model.get_params()}" # Define the Gradio layout with gr.Blocks() as demo: log_df = pd.DataFrame( {"": [], 'Selected Models': [], 'Dataset': [], 'Task': [], 'Result': []} ) state = gr.State({"log_df": log_df}) with gr.Row(): # Left Column with gr.Column(): gr.HTML( '''

Data & Model Setting

''' ) # Dropdown menu for predefined datasets including "Custom Dataset" option dataset_selector = gr.Dropdown( label="Select Dataset", choices=list(predefined_datasets.keys()) + ["Custom Dataset"], ) # Display the message for selected columns selected_columns_message = gr.Textbox( label="Selected Columns Info", visible=False ) with gr.Accordion("Dataset Settings", open=True): # File upload options for custom dataset (train and test) dataset_name = gr.Textbox(label="Dataset Name", visible=False) train_file = gr.File( label="Upload Custom Train Dataset", file_types=[".csv"], visible=False, ) train_display = gr.Dataframe( label="Train Dataset Preview (First 5 Rows)", visible=False, interactive=False, ) test_file = gr.File( label="Upload Custom Test Dataset", file_types=[".csv"], visible=False, ) test_display = gr.Dataframe( label="Test Dataset Preview (First 5 Rows)", visible=False, interactive=False, ) # Predefined dataset displays predefined_display = gr.Dataframe( label="Predefined Dataset Preview (First 5 Rows)", visible=False, interactive=False, ) # Dropdowns for selecting input and output columns for the custom dataset input_column_selector = gr.Dropdown( label="Select Input Column", choices=[], visible=False ) output_column_selector = gr.Dropdown( label="Select Output Column", choices=[], visible=False ) # When a dataset is selected, show either file upload fields (for custom) or load predefined datasets dataset_selector.change( handle_dataset_selection, inputs=dataset_selector, outputs=[ dataset_name, train_file, train_display, test_file, test_display, predefined_display, input_column_selector, output_column_selector, ], ) # When a predefined dataset is selected, load its head and update column selectors dataset_selector.change( load_predefined_dataset, inputs=dataset_selector, outputs=[ predefined_display, input_column_selector, output_column_selector, selected_columns_message, ], ) # When a custom train file is uploaded, display its head and update column selectors train_file.change( display_csv_head, inputs=train_file, outputs=[ train_display, input_column_selector, output_column_selector, ], ) # When a custom test file is uploaded, display its head test_file.change( display_csv_head, inputs=test_file, outputs=[ test_display, input_column_selector, output_column_selector, ], ) dataset_selector.change( set_dataname, inputs=[dataset_name, dataset_selector], outputs=dataset_name, ) # Update the selected columns information when dropdown values are changed input_column_selector.change( select_columns, inputs=[ input_column_selector, output_column_selector, train_file, test_file, dataset_name, ], outputs=selected_columns_message, ) output_column_selector.change( select_columns, inputs=[ input_column_selector, output_column_selector, train_file, test_file, dataset_name, ], outputs=selected_columns_message, ) model_checkbox = gr.CheckboxGroup( choices=models_enabled, label="Select Model" ) task_radiobutton = gr.Radio( choices=["Classification", "Regression"], label="Task Type" ) ####### adding hyper parameter tuning ########### model_name = gr.Dropdown( [ "Default - Auto", "XGBClassifier", "SVR", "Kernel Ridge", "Linear Regression", ], label="Select Downstream Model", ) with gr.Accordion("Downstream Hyperparameter Settings", open=True): # Create placeholders for hyperparameter components max_depth = gr.Slider(1, 20, step=1, visible=False, label="max_depth") n_estimators = gr.Slider( 100, 5000, step=100, visible=False, label="n_estimators" ) alpha = gr.Slider(0.1, 10.0, step=0.1, visible=False, label="alpha") degree = gr.Slider(1, 20, step=1, visible=False, label="degree") kernel = gr.Dropdown( choices=["rbf", "poly", "linear"], visible=False, label="kernel" ) # Output textbox output = gr.Textbox(label="Loaded Parameters") # Dynamically show relevant hyperparameters based on selected model def update_hyperparameters(model_name): if model_name == "XGBClassifier": return ( gr.update(visible=True), gr.update(visible=True), gr.update(visible=True), gr.update(visible=False), gr.update(visible=False), ) elif model_name == "SVR": return ( gr.update(visible=False), gr.update(visible=False), gr.update(visible=False), gr.update(visible=True), gr.update(visible=True), ) elif model_name == "Kernel Ridge": return ( gr.update(visible=False), gr.update(visible=False), gr.update(visible=True), gr.update(visible=True), gr.update(visible=True), ) elif model_name == "Linear Regression": return ( gr.update(visible=False), gr.update(visible=False), gr.update(visible=False), gr.update(visible=False), gr.update(visible=False), ) elif model_name == "Default - Auto": return ( gr.update(visible=False), gr.update(visible=False), gr.update(visible=False), gr.update(visible=False), gr.update(visible=False), ) # When model is selected, update which hyperparameters are visible model_name.change( update_hyperparameters, inputs=[model_name], outputs=[max_depth, n_estimators, alpha, degree, kernel], ) # Submit button to create the model with selected hyperparameters submit_button = gr.Button("Create Downstream Model") # Function to handle model creation based on input parameters def on_submit(model_name, max_depth, n_estimators, alpha, degree, kernel): if model_name == "XGBClassifier": return create_model( model_name, max_depth=max_depth, n_estimators=n_estimators, alpha=alpha, ) elif model_name == "SVR": return create_model(model_name, degree=degree, kernel=kernel) elif model_name == "Kernel Ridge": return create_model( model_name, alpha=alpha, degree=degree, kernel=kernel ) elif model_name == "Linear Regression": return create_model(model_name) elif model_name == "Default - Auto": return create_model(model_name) # When the submit button is clicked, run the on_submit function submit_button.click( on_submit, inputs=[model_name, max_depth, n_estimators, alpha, degree, kernel], outputs=output, ) ###### End of hyper param tuning ######### fusion_radiobutton = gr.Radio(choices=fusion_available, label="Fusion Type") eval_button = gr.Button("Train downstream model") # Middle Column with gr.Column(): gr.HTML( '''

Downstream Task 1: Property Prediction

''' ) eval_output = gr.Textbox(label="Train downstream model") plot_radio = gr.Radio( choices=["ROC-AUC", "Parity Plot", "Latent Space"], label="Select Plot Type", ) plot_output = gr.Plot(label="Visualization") create_log = gr.Button("Store log") log_table = gr.Dataframe( value=log_df, label="Log of Selections and Results", interactive=False ) eval_button.click( display_eval, inputs=[ model_checkbox, selected_columns_message, task_radiobutton, output, fusion_radiobutton, state, ], outputs=eval_output, ) plot_radio.change( display_plot, inputs=[plot_radio, state], outputs=plot_output ) # Function to gather selected models def gather_selected_models(*models): selected = [model for model in models if model] return selected create_log.click( evaluate_and_log, inputs=[ model_checkbox, dataset_name, task_radiobutton, eval_output, state, ], outputs=log_table, ) # Right Column with gr.Column(): gr.HTML( '''

Downstream Task 2: Molecule Generation

''' ) smiles_input = gr.Textbox(label="Input SMILES String") image_display = gr.Image(label="Molecule Image", height=250, width=250) # Show images for selection with gr.Accordion("Select from sample molecules", open=False): image_selector = gr.Radio( choices=list(smiles_image_mapping.keys()), label="Select from sample molecules", value=None, ) image_selector.change(load_image, image_selector, image_display) generate_button = gr.Button("Generate") gen_image_display = gr.Image( label="Generated Molecule Image", height=250, width=250 ) generated_output = gr.Textbox(label="Generated Output") property_table = gr.Dataframe(label="Molecular Properties Comparison") # Handle image selection image_selector.change( handle_image_selection, inputs=image_selector, outputs=[smiles_input, image_display], ) smiles_input.change( smiles_to_image, inputs=smiles_input, outputs=image_display ) # Generate button to display canonical SMILES and molecule image generate_button.click( generate_canonical, inputs=smiles_input, outputs=[property_table, generated_output, gen_image_display], ) if __name__ == "__main__": demo.launch(server_name="0.0.0.0")