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( '''