{text}
', - ) - - -TASK_MAP = { - 'Compound-Protein Interaction': 'DTI', - 'Compound-Protein Binding Affinity': 'DTA', -} - -TASK_METRIC_MAP = { - 'DTI': 'AUROC', - 'DTA': 'CI', - 'Compound-Protein Interaction': 'AUROC', - 'Compound-Protein Binding Affinity': 'CI', - 'CPI': 'DTI', - 'CPA': 'DTA', -} - -PRESET_MAP = { - 'DeepDTA': 'deep_dta', - 'DeepConvDTI': 'deep_conv_dti', - 'GraphDTA': 'graph_dta', - 'MGraphDTA': 'm_graph_dta', - 'HyperAttentionDTI': 'hyper_attention_dti', - 'MolTrans': 'mol_trans', - 'TransformerCPI': 'transformer_cpi', - 'TransformerCPI2': 'transformer_cpi_2', - 'DrugBAN': 'drug_ban', - 'DrugVQA-Seq': 'drug_vqa' -} - -TARGET_FAMILY_MAP = { - 'General': 'general', - 'Kinase': 'kinase', - 'Non-Kinase Enzyme': 'non_kinase_enzyme', - 'Membrane Receptor': 'membrane_receptor', - 'Nuclear Receptor': 'nuclear_receptor', - 'Ion Channel': 'ion_channel', - 'Others': 'others', - # 'general': 'general', - # 'kinase': 'kinase', - # 'non-kinase enzyme': 'non_kinase_enzyme', - # 'membrane receptor': 'membrane_receptor', - # 'nuclear Receptor': 'nuclear_receptor', - # 'ion channel': 'ion_channel', - # 'others': 'others', -} - -TARGET_LIBRARY_MAP = { - 'DrugBank (Human)': 'drugbank_targets.csv', - 'ChEMBL33 (Human)': 'ChEMBL33_human_proteins.csv', -} - -DRUG_LIBRARY_MAP = { - 'DrugBank (Human)': 'drugbank_compounds.csv', - 'Drug Repurposing Hub': 'drug_repurposing_hub.csv', - 'Enamine Discovery Diversity Set (DDS-10)': 'Enamine_Discovery_Diversity_Set_10_10240cmpds_20240130.csv', - 'Enamine Phenotypic Screening Library (PSL-5760)': 'Enamine_Phenotypic_Screening_Library_plated_5760cmds_2020_07_20.csv' -} - -COLUMN_ALIASES = { - 'X1': 'Compound SMILES', - 'X2': 'Target FASTA', - 'ID1': 'Compound ID', - 'ID2': 'Target ID', - 'Y': 'Actual CPI/CPA', - 'Y^': 'Predicted CPI/CPA', -} - -DRUG_SCRENN_CPI_OPTS = [ - 'Calculate Max. Sequence Identity between the Input Target and Targets in the Training Set', - 'Calculate Max. Tanimoto Similarity between the Hit Compound and Known Ligands of the Input Target', - 'Calculate Max. Sequence Identity between the Input Target and Known Targets of Hit Compound', -] - -DRUG_SCRENN_CPA_OPTS = [ - 'Calculate Max. Sequence Identity between the Input Target and Targets in the Training Set', -] - -TARGET_IDENTIFY_CPI_OPTS = [ - 'Calculate Max. Tanimoto Similarity between the Input Compound and Compounds in the Training Set', - 'Calculate Max. Sequence Identity between the Identified Target and Known Targets of the Input Compound', - 'Calculate Max. Tanimoto Similarity between the Input Compound and Known Ligands of the Identified Target', -] - -TARGET_IDENTIFY_CPA_OPTS = [ - 'Calculate Max. Tanimoto Similarity between the Input Compound and Compounds in the Training Set', -] - -pd.set_option('display.float_format', '{:.3f}'.format) -PandasTools.molRepresentation = 'svg' -PandasTools.drawOptions = Draw.rdMolDraw2D.MolDrawOptions() -PandasTools.drawOptions.clearBackground = False -PandasTools.drawOptions.bondLineWidth = 1 -PandasTools.drawOptions.explicitMethyl = True -PandasTools.drawOptions.singleColourWedgeBonds = True -PandasTools.drawOptions.useCDKAtomPalette() -PandasTools.molSize = (100, 64) - - -def remove_job_record(job_id): - # Delete the job from the database - db.remove(Job.id == job_id) - # Delete the corresponding files - files = glob.glob(f"{SERVER_DATA_DIR}/{job_id}*") - for file_path in files: - if os.path.exists(file_path): - os.remove(file_path) - - -def check_expiry(): - Job = Query() - jobs = db.all() - - for job in jobs: - # Check if the job has expired - if job['status'] != 'RUNNING': - expiry_time = job['expiry_time'] if job['expiry_time'] is not None else job['start_time'] + DB_EXPIRY - if expiry_time < time(): - # Delete the job from the database - db.remove(Job.id == job['id']) - # Delete the corresponding file - files = glob.glob(f"{SERVER_DATA_DIR}/{job['id']}*") - for file_path in files: - if os.path.exists(file_path): - os.remove(file_path) - elif job['status'] == 'RUNNING' and time() - job['start_time'] > 4 * 60 * 60: # 4 hours - # Mark the job as failed - db.update({'status': 'FAILED', - 'error': 'Job has timed out by exceeding the maximum running time of 4 hours.'}, - Job.id == job['id']) - if job.get('email'): - send_email(job) - - -def smiles_to_ecfp(smiles): - mol = Chem.MolFromSmiles(smiles) - if mol: - ecfp = AllChem.GetMorganFingerprintAsBitVect(mol, radius=2, nBits=2048) - else: - ecfp = [] - return ecfp - - -def max_tanimoto_similarity(smi, seen_smiles_with_fp): - if smi is None or seen_smiles_with_fp is None or seen_smiles_with_fp.empty: - return {'Max. Tanimoto Similarity': 0, 'Max. Tanimoto Similarity Compound': None} - - if smi in seen_smiles_with_fp['X1'].values: - compound = smi - if 'ID1' in seen_smiles_with_fp.columns: - id1 = seen_smiles_with_fp.loc[seen_smiles_with_fp['X1'] == smi, 'ID1'].values[0] - if pd.notnull(id1) and id1 != '': - compound = id1 - return {'Max. Tanimoto Similarity': 1, 'Max. Tanimoto Similarity Compound': compound} - - mol = Chem.MolFromSmiles(smi) - if mol is None: - return {'Max. Tanimoto Similarity': 0, 'Max. Tanimoto Similarity Compound': None} - - mol_ecfp = AllChem.GetMorganFingerprintAsBitVect(mol, radius=2, nBits=2048) - sims = pd.Series(BulkTanimotoSimilarity(mol_ecfp, seen_smiles_with_fp['FP'].values)).to_numpy() - idx = sims.argmax() - compound = seen_smiles_with_fp.iloc[idx]['X1'] - if 'ID1' in seen_smiles_with_fp.columns: - id1 = seen_smiles_with_fp.iloc[idx]['ID1'] - if pd.notnull(id1) and id1 != '': - compound = id1 - - return {'Max. Tanimoto Similarity': sims[idx], 'Max. Tanimoto Similarity Compound': compound} - - -def alignment_score(query, target): - aligner = PairwiseAligner() - aligner.mode = 'local' - alignment = aligner.align(query, target) - return alignment.score / max(len(query), len(target)) - - -def max_sequence_identity(seq, seen_fastas): - if seq is None or seen_fastas is None or seen_fastas.empty: - return {'Max. Sequence Identity': 0, 'Max. Sequence Identity Target': None} - - if seq in seen_fastas['X2'].values: - target = seq - if 'ID2' in seen_fastas.columns: - id2 = seen_fastas.loc[seen_fastas['X2'] == seq, 'ID2'].values[0] - if pd.notnull(id2) and id2 != '': - target = id2 - return {'Max. Sequence Identity': 1, 'Max. Sequence Identity Target': target} - - cached_alignment_score = cache(alignment_score) - max_iden = 0 - target = None - for fasta in seen_fastas['X2'].values: - identity = cached_alignment_score(seq, fasta) - - if identity > max_iden: - max_iden = identity - target = fasta - if 'ID2' in seen_fastas.columns: - id2 = seen_fastas.loc[seen_fastas['X2'] == fasta, 'ID2'].values[0] - if pd.notnull(id2) and id2 != '': - target = id2 - if max_iden == 1: - break - - cached_alignment_score.cache_clear() - return {'Max. Sequence Identity': max_iden, 'Max. Sequence Identity Target': target} - - -def get_seen_smiles(family, task): - if family == 'General': - family = 'all_families_full' - else: - family = TARGET_FAMILY_MAP[family.title()] - seen_smiles = pd.read_csv( - f'data/benchmarks/seen_compounds/{family}_{task.lower()}_random_split.csv') - return seen_smiles - - -def get_seen_fastas(family, task): - if family == 'General': - family = 'all_families_full' - else: - family = TARGET_FAMILY_MAP[family.title()] - seen_fastas = pd.read_csv( - f'data/benchmarks/seen_targets/{family}_{task.lower()}_random_split.csv') - return seen_fastas - - -@cache -def get_fasta_family_map(): - usecols = ['X2', 'ID2', 'Target Family'] - fasta_family_map = pd.concat([ - pd.read_csv('data/target_libraries/ChEMBL33_all_spe_single_prot_info.csv', usecols=usecols), - pd.read_csv('data/target_libraries/idmapping_not_in_chembl.csv', usecols=usecols) - ]).drop_duplicates(subset=['X2'], keep='first') - return fasta_family_map - - -def lipinski(mol): - """ - Lipinski's rules: - Hydrogen bond donors <= 5 - Hydrogen bond acceptors <= 10 - Molecular weight <= 500 daltons - logP <= 5 - """ - return ( - Lipinski.NumHDonors(mol) <= 5 and - Lipinski.NumHAcceptors(mol) <= 10 and - Descriptors.MolWt(mol) <= 500 and - Crippen.MolLogP(mol) <= 5 - ) - - -def reos(mol): - """ - Rapid Elimination Of Swill filter: - Molecular weight between 200 and 500 - LogP between -5.0 and +5.0 - H-bond donor count between 0 and 5 - H-bond acceptor count between 0 and 10 - Formal charge between -2 and +2 - Rotatable bond count between 0 and 8 - Heavy atom count between 15 and 50 - """ - return ( - 200 <= Descriptors.MolWt(mol) <= 500 and - -5.0 <= Crippen.MolLogP(mol) <= 5.0 and - 0 <= Lipinski.NumHDonors(mol) <= 5 and - 0 <= Lipinski.NumHAcceptors(mol) <= 10 and - -2 <= rdmolops.GetFormalCharge(mol) <= 2 and - 0 <= rdMolDescriptors.CalcNumRotatableBonds(mol) <= 8 and - 15 <= rdMolDescriptors.CalcNumHeavyAtoms(mol) <= 50 - ) - - -def ghose(mol): - """ - Ghose drug like filter: - Molecular weight between 160 and 480 - LogP between -0.4 and +5.6 - Atom count between 20 and 70 - Molar refractivity between 40 and 130 - """ - return ( - 160 <= Descriptors.MolWt(mol) <= 480 and - -0.4 <= Crippen.MolLogP(mol) <= 5.6 and - 20 <= rdMolDescriptors.CalcNumAtoms(mol) <= 70 and - 40 <= Crippen.MolMR(mol) <= 130 - ) - - -def veber(mol): - """ - The Veber filter is a rule of thumb filter for orally active drugs described in - Veber et al., J Med Chem. 2002; 45(12): 2615-23.: - Rotatable bonds <= 10 - Topological polar surface area <= 140 - """ - return ( - rdMolDescriptors.CalcNumRotatableBonds(mol) <= 10 and - rdMolDescriptors.CalcTPSA(mol) <= 140 - ) - - -def rule_of_three(mol): - """ - Rule of Three filter (Congreve et al., Drug Discov. Today. 8 (19): 876–7, (2003).): - Molecular weight <= 300 - LogP <= 3 - H-bond donor <= 3 - H-bond acceptor count <= 3 - Rotatable bond count <= 3 - """ - return ( - Descriptors.MolWt(mol) <= 300 and - Crippen.MolLogP(mol) <= 3 and - Lipinski.NumHDonors(mol) <= 3 and - Lipinski.NumHAcceptors(mol) <= 3 and - rdMolDescriptors.CalcNumRotatableBonds(mol) <= 3 - ) - - -@cache -def load_smarts_patterns(smarts_path): - # Load the CSV file containing SMARTS patterns - smarts_df = pd.read_csv(Path(smarts_path)) - # Convert all SMARTS patterns to molecules - smarts_mols = [Chem.MolFromSmarts(smarts) for smarts in smarts_df['smarts']] - return smarts_mols - - -def smarts_filter(mol, smarts_mols): - for smarts_mol in smarts_mols: - if smarts_mol is not None and mol.HasSubstructMatch(smarts_mol): - return False - return True - - -def pains(mol): - smarts_mols = load_smarts_patterns("data/filters/pains.csv") - return smarts_filter(mol, smarts_mols) - - -def mlsmr(mol): - smarts_mols = load_smarts_patterns("data/filters/mlsmr.csv") - return smarts_filter(mol, smarts_mols) - - -def dundee(mol): - smarts_mols = load_smarts_patterns("data/filters/dundee.csv") - return smarts_filter(mol, smarts_mols) - - -def glaxo(mol): - smarts_mols = load_smarts_patterns("data/filters/glaxo.csv") - return smarts_filter(mol, smarts_mols) - - -def bms(mol): - smarts_mols = load_smarts_patterns("data/filters/bms.csv") - return smarts_filter(mol, smarts_mols) - - -SCORE_MAP = { - 'SAscore': sascorer.calculateScore, - 'LogP': Crippen.MolLogP, - 'Molecular Weight': Descriptors.MolWt, - 'Number of Atoms': rdMolDescriptors.CalcNumAtoms, - 'Number of Heavy Atoms': rdMolDescriptors.CalcNumHeavyAtoms, - 'Molar Refractivity': Crippen.MolMR, - 'H-Bond Donor Count': Lipinski.NumHDonors, - 'H-Bond Acceptor Count': Lipinski.NumHAcceptors, - 'Rotatable Bond Count': rdMolDescriptors.CalcNumRotatableBonds, - 'Topological Polar Surface Area': rdMolDescriptors.CalcTPSA, -} - -FILTER_MAP = { - # TODO support number_of_violations - 'REOS': reos, - "Lipinski's Rule of Five": lipinski, - 'Ghose': ghose, - 'Rule of Three': rule_of_three, - 'Veber': veber, - 'PAINS': pains, - 'MLSMR': mlsmr, - 'Dundee': dundee, - 'Glaxo': glaxo, - 'BMS': bms, -} - - -def validate_columns(df, mandatory_cols): - missing_cols = [col for col in mandatory_cols if col not in df.columns] - if missing_cols: - error_message = (f"The following mandatory columns are missing " - f"in the uploaded dataset: {str(mandatory_cols).strip('[]')}.") - raise ValueError(error_message) - else: - return - - -def process_target_fasta(sequence): - try: - if sequence: - lines = sequence.strip().split("\n") - if lines[0].startswith(">"): - lines = lines[1:] - return ''.join(lines).split(">")[0].strip() - # record = list(SeqIO.parse(io.StringIO(sequence), "fasta"))[0] - # return str(record.seq) - else: - raise ValueError('Empty FASTA sequence.') - except Exception as e: - raise gr.Error(f'Failed to process FASTA due to error: {str(e)}') - - -def send_email(job_info): - if job_info.get('email'): - try: - email_info = job_info.copy() - email_serv = os.getenv('EMAIL_SERV') - email_port = os.getenv('EMAIL_PORT') - email_addr = os.getenv('EMAIL_ADDR') - email_pass = os.getenv('EMAIL_PASS') - email_form = os.getenv('EMAIL_FORM') - email_subj = os.getenv('EMAIL_SUBJ') - - for key, value in email_info.items(): - if key.endswith("time") and value: - email_info[key] = ts_to_str(value, get_timezone_by_ip(email_info['ip'])) - - server = smtplib.SMTP(email_serv, int(email_port)) - # server.starttls() - - server.login(email_addr, email_pass) - msg = MIMEMultipart("alternative") - msg["From"] = email_addr - msg["To"] = email_info['email'] - msg["Subject"] = email_subj.format(**email_info) - msg["Date"] = formatdate(localtime=True) - msg["Message-ID"] = make_msgid() - - msg.attach(MIMEText(markdown(email_form.format(**email_info)), 'html')) - msg.attach(MIMEText(email_form.format(**email_info), 'plain')) - - server.sendmail(email_addr, email_info['email'], msg.as_string()) - server.quit() - gr.Info('Email notification sent.') - except Exception as e: - gr.Warning('Failed to send email notification due to error: ' + str(e)) - - -def check_user_running_job(email, request): - message = ("You already have a running prediction job (ID: {id}) under this {reason}. " - "Please wait for it to complete before submitting another job.") - try: - # with open('jobs.json', 'r') as f: # /data/ - # # Load the JSON data from the file - # jobs = json.load(f) - # - # for job_id, job_info in jobs.items(): - # # check if a job is running for the email - # if email: - # if job_info["email"] == email and job_info["status"] == "running": - # return message.format(id=job_id, reason="email") - # # check if a job is running for the session - # elif request.cookies: - # for key, value in job_info["cookies"].items() and job_info["status"] == "running": - # if key in request.cookies and request.cookies[key] == value: - # return message.format(id=job_id, reason="session") - # # check if a job is running for the IP - # else: - # if job_info["IP"] == request.client.host and job_info["status"] == "running": - # return message.format(id=job_id, reason="IP") - # check if a job is running for the email - Job = Query() - if email: - job = db.search((Job.email == email) & (Job.status == "RUNNING")) - if job: - return message.format(id=job[0]['id'], reason="email") - # check if a job is running for the session - elif request.cookies: - for key, value in request.cookies.items(): - job = db.search((Job.cookies[key] == value) & (Job.status == "RUNNING")) - if job: - return message.format(id=job[0]['id'], reason="session") - # check if a job is running for the IP - else: - job = db.search((Job.IP == request.client.host) & (Job.status == "RUNNING")) - if job: - return message.format(id=job[0]['id'], reason="IP") - - return False - except Exception as e: - raise gr.Error(f'Failed to validate user running jobs due to error: {str(e)}') - - -def get_timezone_by_ip(ip): - try: - data = session.get(f'https://worldtimeapi.org/api/ip/{ip}').json() - return data['timezone'] - except Exception: - return 'UTC' - - -def ts_to_str(timestamp, timezone): - # Create a timezone-aware datetime object from the UNIX timestamp - dt = datetime.fromtimestamp(timestamp, pytz.utc) - - # Convert the timezone-aware datetime object to the target timezone - target_timezone = pytz.timezone(timezone) - localized_dt = dt.astimezone(target_timezone) - - # Format the datetime object to the specified string format - return localized_dt.strftime('%Y-%m-%d %H:%M:%S (%Z%z)') - - -def lookup_job(job_id): - gr.Info('Start querying the job database...') - stop = False - retry = 0 - while not stop: - try: - sleep(5) - Job = Query() - jobs = db.search((Job.id == job_id)) - if jobs: - job = jobs[0] - job_status = job['status'] - job_type = job['type'] - error = job['error'] - start_time = ts_to_str(job['start_time'], get_timezone_by_ip(job['ip'])) - if job.get('end_time'): - end_time = ts_to_str(job['end_time'], get_timezone_by_ip(job['ip'])) - if job.get('expiry_time'): - expiry_time = ts_to_str(job['expiry_time'], get_timezone_by_ip(job['ip'])) - if job_status == "RUNNING": - yield { - pred_lookup_status: f''' -Your **{job_type}** job (ID: **{job_id}**) started at -**{start_time}** and is **RUNNING...** - -It might take a few minutes up to a few hours depending on the prediction dataset, the model, and the queue status. -You may keep the page open and wait for job completion, or close the page and revisit later to look up the job status -using the job id. You will also receive an email notification once the job is done. -''', - pred_lookup_btn: gr.Button(visible=False), - pred_lookup_stop_btn: gr.Button(visible=True) - } - if job_status == "COMPLETED": - stop = True - msg = f"Your {job_type} job (ID: {job_id}) has been **COMPLETED**" - msg += f" at {end_time}" if job.get('end_time') else "" - msg += f" and the results will expire by {expiry_time}." if job.get('expiry_time') else "." - msg += f' Redirecting to the report page...' - - gr.Info(msg) - yield { - pred_lookup_status: msg, - pred_lookup_btn: gr.Button(visible=True), - pred_lookup_stop_btn: gr.Button(visible=False), - tabs: gr.Tabs(selected='Chemical Property Report'), - file_for_report: job['output_file'] - } - if job_status == "FAILED": - stop = True - msg = f'Your {job_type} job (ID: {job_id}) has **FAILED**' - msg += f' at {end_time}' if job.get('end_time') else '' - msg += f' due to error: {error}.' if job.get('expiry_time') else '.' - gr.Info(msg) - yield { - pred_lookup_status: msg, - pred_lookup_btn: gr.Button(visible=True), - pred_lookup_stop_btn: gr.Button(visible=False), - tabs: gr.Tabs(selected='Prediction Status Lookup'), - } - else: - stop = (retry > 3) - if not stop: - msg = f'Job ID {job_id} not found. Retrying... ({retry})' - else: - msg = f'Job ID {job_id} not found after {retry} retries. Please check the job ID and try again.' - gr.Info(msg) - retry += 1 - yield { - pred_lookup_status: msg, - pred_lookup_btn: gr.Button(visible=True), - pred_lookup_stop_btn: gr.Button(visible=False), - tabs: gr.Tabs(selected='Prediction Status Lookup'), - } - - except Exception as e: - raise gr.Error(f'Failed to retrieve job status due to error: {str(e)}') - - -def apply_advanced_opts(prediction_df, opts, df_training): - # Advanced options for Drug Hit Screening - if "Calculate Max. Sequence Identity between the Input Target and Targets in the Training Set" in opts: - x2 = prediction_df['X2'].iloc[0] - - prediction_df[[ - 'Max. Sequence Identity to Training Targets', - 'Max. Id. Training Target' - ]] = pd.Series(max_sequence_identity(x2, df_training)) - - if "Calculate Max. Tanimoto Similarity between the Hit Compound and Known Ligands of the Input Target" in opts: - x2 = prediction_df['X2'].iloc[0] - pos_compounds_df = df_training.loc[(df_training['X2'] == x2) & (df_training['Y'] == 1)].copy() - pos_compounds_df['FP'] = pos_compounds_df['X1'].parallel_apply(smiles_to_ecfp) - - @cache - def max_sim(smiles): - return max_tanimoto_similarity(smiles, pos_compounds_df) - - prediction_df[[ - 'Max. Tanimoto Similarity to Known Ligands', - 'Max. Sim. Ligand' - ]] = prediction_df['X1'].parallel_apply(max_sim).apply(pd.Series) - - max_sim.cache_clear() - - if "Calculate Max. Sequence Identity between the Input Target and Known Targets of Hit Compound" in opts: - x2 = prediction_df['X2'].iloc[0] - prediction_df['X1^'] = prediction_df['X1'].parallel_apply(rdkit_canonicalize) - - @cache - def max_id(compound): - pos_targets_df = df_training.loc[df_training['X1'] == compound] - return max_sequence_identity(x2, pos_targets_df) - - prediction_df[['Max. Sequence Identity to Known Targets of Hit Compound', - 'Max. Id. Target']] = ( - prediction_df['X1^'].parallel_apply(max_id).apply(pd.Series) - ) - prediction_df.drop(['X1^'], axis=1, inplace=True) - - max_id.cache_clear() - - # Advanced options for Target Protein Identification - if "Calculate Max. Tanimoto Similarity between the Input Compound and Compounds in the Training Set" in opts: - x1 = rdkit_canonicalize(prediction_df['X1'].iloc[0]) - if x1 not in df_training['X1'].values: - df_training['FP'] = df_training['X1'].parallel_apply(smiles_to_ecfp) - - prediction_df[[ - 'Max. Tanimoto Similarity to Training Compounds', - 'Max. Sim. Training Compound' - ]] = pd.Series(max_tanimoto_similarity(x1, df_training)) - - if "Calculate Max. Sequence Identity between the Identified Target and Known Targets of the Input Compound" in opts: - x1 = rdkit_canonicalize(prediction_df['X1'].iloc[0]) - pos_targets_df = df_training.loc[(df_training['X1'] == x1) & (df_training['Y'] == 1)].copy() - - @cache - def max_id(fasta): - return max_sequence_identity(fasta, pos_targets_df) - - prediction_df[[ - 'Max. Sequence Identity to Known Targets of Input Compound', - 'Max. Id. Target' - ]] = prediction_df['X2'].parallel_apply(max_id).apply(pd.Series) - - max_id.cache_clear() - - if "Calculate Max. Tanimoto Similarity between the Input Compound and Known Ligands of the Identified Target" in opts: - x1 = rdkit_canonicalize(prediction_df['X1'].iloc[0]) - - @cache - def max_sim(fasta): - pos_targets_df = df_training.loc[(df_training['X2'] == fasta) & (df_training['Y'] == 1)].copy() - if x1 not in pos_targets_df['X1'].values: - pos_targets_df['FP'] = pos_targets_df['X1'].apply(smiles_to_ecfp) - return max_tanimoto_similarity(x1, pos_targets_df) - - prediction_df[[ - 'Max. Tanimoto Similarity to Known Ligands of Identified Target', - 'Max. Sim. Ligand' - ]] = prediction_df['X2'].parallel_apply(max_sim).apply(pd.Series) - - max_sim.cache_clear() - - return prediction_df - - -def submit_predict(predict_filepath, task, preset, target_family, opts, job_info): - job_id = job_info['id'] - status = job_info['status'] - send_email(job_info) - db.insert(job_info) - error = None - task_file_abbr = {'Compound-Protein Interaction': 'CPI', 'Compound-Protein Binding Affinity': 'CPA'} - predictions_file = None - df_training = pd.read_csv(f'data/complete_{TASK_MAP[task].lower()}_dataset.csv') - df_training['X1^'] = df_training['X1'] - orig_df = pd.read_csv(predict_filepath) - alignment_df = get_fasta_family_map() - prediction_df = pd.DataFrame() - - @cache - def detect_family(query): - # Check for an exact match first - exact_match = alignment_df[alignment_df['X2'] == query] - if not exact_match.empty: - row = exact_match.iloc[0] - return row['Target Family'] - # If no exact match, then calculate alignment score - else: - aligner = PairwiseAligner() - aligner.mode = 'local' - - def align_score(target): - alignment = aligner.align(query, target) - return alignment.score / max(len(query), len(target)) - - alignment_df['score'] = alignment_df['X2'].apply(align_score) - row = alignment_df.loc[alignment_df['score'].idxmax()] - return row['Target Family'] - - if 'Target Family' not in orig_df.columns: - orig_df['Target Family'] = None - if orig_df['Target Family'].isna().any(): - orig_df.loc[orig_df['Target Family'].isna(), 'Target Family'] = ( - orig_df.loc[orig_df['Target Family'].isna(), 'X2'].parallel_apply(detect_family) - ) - orig_df['Target Family'] = orig_df['Target Family'].str.capitalize() - detect_family.cache_clear() - - orig_df['X1^'] = orig_df['X1'].parallel_apply(rdkit_canonicalize) - - orig_df = orig_df.merge(df_training[['X1^', 'X2', 'Y']], on=['X1^', 'X2'], how='left', indicator=False) - annotated_df = orig_df[~orig_df['Y'].isna()].copy() - annotated_df.rename(columns={'Y': 'Y^'}, inplace=True) - annotated_df['Source'] = 'Database' - columns_to_drop = ['X1^', 'Compound', 'Scaffold', 'Scaffold SMILES'] - columns_to_drop = [col for col in columns_to_drop if col in annotated_df.columns] - annotated_df.drop(columns_to_drop, axis=1, inplace=True) - - # Save the unannotated data - unannotated_df = orig_df[orig_df['Y'].isna()].drop(['Y'], axis=1) - if not unannotated_df.empty: - unannotated_df.to_csv(predict_filepath, index=False, na_rep='') - else: - annotated_df.to_csv(predictions_file, index=False, na_rep='') - status = "COMPLETED" - return {run_state: False} - - columns_to_drop = ['ID1', 'X1^', 'Compound', 'Scaffold', 'Scaffold SMILES', 'ID2', 'Y', 'Y^'] - columns_to_drop = [col for col in columns_to_drop if col in orig_df.columns] - orig_df.drop(columns_to_drop, axis=1, inplace=True) - - try: - if target_family != 'Family-Specific Auto-Recommendation': - target_family_value = TARGET_FAMILY_MAP[target_family.title()] - task_value = TASK_MAP[task] - preset_value = PRESET_MAP[preset] - predictions_file = (f'{SERVER_DATA_DIR}/' - f'{job_id}_{task_file_abbr[task]}_{preset}_{target_family_value}_predictions.csv') - - cfg = hydra.compose( - config_name="webserver_inference", - overrides=[f"task={task_value}", - f"preset={preset_value}", - f"ckpt_path=resources/checkpoints/{preset_value}-{task_value}-{target_family_value}.ckpt", - f"data.data_file='{str(predict_filepath)}'"]) - - predictions, _ = predict(cfg) - predictions = pd.concat([pd.DataFrame(prediction) for prediction in predictions], ignore_index=True) - predictions['Source'] = f'Predicted ({preset} {target_family})' - df_list = [prediction_df, predictions] - prediction_df = pd.concat([df for df in df_list if not df.empty]) - - else: - predictions_file = f'{SERVER_DATA_DIR}/{job_id}_{task_file_abbr[task]}_family-recommended_predictions.csv' - task_value = TASK_MAP[task] - score = TASK_METRIC_MAP[task] - benchmark_df = pd.read_csv(f'data/benchmarks/{task_value}_test_metrics.csv') - predict_df = pd.read_csv(predict_filepath) - - for family, subset in predict_df.groupby('Target Family'): - predict_subset_filepath = os.path.join( - os.path.dirname(predict_filepath), f'{job_id}_{family}_input.csv' - ) - subset.to_csv(predict_subset_filepath, index=False, na_rep='') - - seen_compounds = get_seen_smiles(family, task_value)['X1'].values - if subset['X1^'].iloc[0] in seen_compounds: - scenario = "Seen Compound" - else: - scenario = "Unseen Compound" - - filtered_df = benchmark_df[(benchmark_df['Family'] == family.title()) - & (benchmark_df['Scenario'] == scenario) - & (benchmark_df['Type'] == 'Family')] - - seen_compounds = get_seen_smiles('General', task_value)['X1'].values - if subset['X1^'].iloc[0] in seen_compounds: - scenario = "Seen Compound" - else: - scenario = "Unseen Compound" - - filtered_df = pd.concat([ - filtered_df, - benchmark_df[(benchmark_df['Family'] == family.title()) - & (benchmark_df['Scenario'] == scenario) - & (benchmark_df['Type'] == 'General')] - ]) - - row = filtered_df.loc[filtered_df[score].idxmax()] - preset_value = PRESET_MAP[row['Model']] - target_family = TARGET_FAMILY_MAP[family.title()] if row['Type'] == 'Family' else 'general' - cfg = hydra.compose( - config_name="webserver_inference", - overrides=[f"task={task_value}", - f"preset={preset_value}", - f"ckpt_path=resources/checkpoints/{preset_value}-{task_value}-{target_family}.ckpt", - f"data.data_file='{str(predict_subset_filepath)}'"]) - - predictions, _ = predict(cfg) - predictions = pd.concat([pd.DataFrame(prediction) for prediction in predictions], ignore_index=True) - predictions['Source'] = (f'Predicted ({row["Model"]} ' - f'{family.title() if row["Type"] == "Family" else "General"})') - df_list = [prediction_df, predictions] - prediction_df = pd.concat([df for df in df_list if not df.empty]) - - prediction_df = prediction_df.merge(orig_df, on=['X1', 'X2'], how='left', indicator=False) - df_list = [prediction_df, annotated_df] - prediction_df = pd.concat([df for df in df_list if not df.empty], ignore_index=True) - - prediction_df = apply_advanced_opts(prediction_df, opts, df_training) - - prediction_df.drop(['N', 'FP'], axis=1, errors='ignore').to_csv(predictions_file, index=False, na_rep='') - status = 'COMPLETED' - - return {run_state: False} - - except Exception as e: - gr.Warning(f"Prediction job failed due to error: {str(e)}") - status = "FAILED" - predictions_file = None - error = str(e) - return {run_state: False} - - finally: - Job = Query() - job_query = (Job.id == job_id) - - end_time = time() - expiry_time = end_time + DB_EXPIRY - - db.update({'end_time': end_time, - 'expiry_time': expiry_time, - 'status': status, - 'error': error, - 'input_file': predict_filepath, - 'output_file': predictions_file}, - job_query) - if job_info := db.search(job_query)[0]: - if job_info.get('email'): - send_email(job_info) - - -def update_df(file, progress=gr.Progress(track_tqdm=True)): - if file and Path(file).is_file(): - task = None - job = None - - if "_CPI_" in str(file): - task = 'Compound-Protein Interaction' - elif "_CPA_" in str(file): - task = 'Compound-Protein Binding Affinity' - - df = pd.read_csv(file) - - if 'N' in df.columns: - df.set_index('N', inplace=True) - - if not any(col in ['X1', 'X2'] for col in df.columns): - gr.Warning("At least one of columns `X1` and `X2` must be in the uploaded dataset.") - return {analyze_btn: gr.Button(interactive=False)} - - if 'X1' in df.columns: - if 'Compound' not in df.columns or df['Compound'].dtype != 'object': - df['Compound'] = df['X1'].parallel_apply( - lambda smiles: PandasTools._MolPlusFingerprint(Chem.MolFromSmiles(smiles))) - df['Scaffold'] = df['Compound'].parallel_apply(MurckoScaffold.GetScaffoldForMol) - df['Scaffold SMILES'] = df['Scaffold'].parallel_apply(lambda x: Chem.MolToSmiles(x)) - df['Pharmacophore'] = None - if task == 'Compound-Protein Binding Affinity': - # Convert Y^ from pIC50 (nM) to IC50 (nM) - if 'Y^' in df.columns: - df['Y^'] = 10 ** (-df['Y^']) * 1e9 - - n_compound = df['X1'].nunique() - n_protein = df['X2'].nunique() - - if n_compound == 1 and n_protein >= 2: - job = 'Target Protein Identification' - if task == 'Compound-Protein Interaction': - opts = TARGET_IDENTIFY_CPI_OPTS - elif task == 'Compound-Protein Binding Affinity': - opts = TARGET_IDENTIFY_CPA_OPTS - if n_compound >= 2 and n_protein == 1: - job = 'Drug Hit Screening' - if task == 'Compound-Protein Interaction': - opts = DRUG_SCRENN_CPI_OPTS - elif task == 'Compound-Protein Binding Affinity': - opts = DRUG_SCRENN_CPA_OPTS - - return { - html_report: create_html_report(df, file=None, task=task), - raw_df: df, - report_df: df.copy(), - analyze_btn: gr.Button(interactive=True), - report_task: task, - job_opts: gr.CheckboxGroup( - label=f'{job} Advanced Options', - choices=opts, visible=True - ) if job else gr.CheckboxGroup(visible=False), - } - else: - return {analyze_btn: gr.Button(interactive=False)} - - -def create_html_report(df, file=None, task=None, opts=(), progress=gr.Progress(track_tqdm=True)): - df_html = df.copy(deep=True) - column_aliases = COLUMN_ALIASES.copy() - cols_left = list(pd.Index([ - 'ID1', 'ID2', 'Compound', 'Scaffold', 'Pharmacophore', 'X1', 'Scaffold SMILES', 'X2', 'Y^' - ]).intersection(df_html.columns)) - # cols_right = list(pd.Index(['X1', 'X2']).intersection(df_html.columns)) - # df_html = df_html[cols_left + (df_html.columns.drop(cols_left + cols_right).tolist()) + cols_right] - df_html = df_html[cols_left + df_html.columns.drop(cols_left).tolist()] - - if isinstance(task, str): - column_aliases.update({ - 'Y^': 'Interaction Probability' if task == 'Compound-Protein Interaction' - else 'Binding Affinity (IC50 [nM])' - }) - - ascending = True if column_aliases['Y^'] == 'Binding Affinity (IC50 [nM])' else False - df_html = df_html.sort_values( - [col for col in ['Y^'] if col in df_html.columns], ascending=ascending - ) - - if not file: - df_html = df_html.iloc[:31] - - # Remove repeated info for one-against-N tasks to save visual and physical space - job = 'Chemical Property' - unique_entity = 'Unique Entity' - unique_df = None - category = None - columns_unique = None - - if 'Exclude Pharmacophore 3D' not in opts: - df_html['Pharmacophore'] = df_html['Compound'].parallel_apply( - lambda x: mol_to_pharm3d(x) if not pd.isna(x) else x) - - if 'Compound' in df_html.columns and 'Exclude Molecular Graph' not in opts: - df_html['Compound'] = df_html['Compound'].parallel_apply( - lambda x: PandasTools.PrintAsImageString(x) if not pd.isna(x) else x) - else: - df_html.drop(['Compound'], axis=1, inplace=True) - - if 'Scaffold' in df_html.columns and 'Exclude Scaffold Graph' not in opts: - df_html['Scaffold'] = df_html['Scaffold'].parallel_apply( - lambda x: PandasTools.PrintAsImageString(x) if not pd.isna(x) else x) - else: - df_html.drop(['Scaffold'], axis=1, inplace=True) - - if 'X1' in df_html.columns and 'X2' in df_html.columns: - n_compound = df_html['X1'].nunique() - n_protein = df_html['X2'].nunique() - - if n_compound == 1 and n_protein >= 2: - unique_entity = 'Compound of Interest' - if any(col in df_html.columns for col in ['Y^', 'Y']): - job = 'Target Protein Identification' - category = 'Target Family' - columns_unique = df_html.columns.isin( - ['ID1', 'Compound', 'Scaffold', 'X1', 'Scaffold SMILES', 'Pharmacophore', - 'Max. Tanimoto Similarity to Training Compounds', 'Max. Sim. Training Compound'] - + list(FILTER_MAP.keys()) + list(SCORE_MAP.keys()) - ) - - elif n_compound >= 2 and n_protein == 1: - unique_entity = 'Target of Interest' - if any(col in df_html.columns for col in ['Y^', 'Y']): - job = 'Drug Hit Screening' - category = 'Scaffold SMILES' - columns_unique = df_html.columns.isin( - ['X2', 'ID2', 'Max. Sequence Identity to Training Targets', 'Max. Id. Training Target'] - ) - - elif 'Y^' in df_html.columns: - job = 'Interaction Pair Inference' - - df_html.rename(columns=column_aliases, inplace=True) - df_html.index.name = 'Index' - if 'Target FASTA' in df_html.columns: - df_html['Target FASTA'] = df_html['Target FASTA'].parallel_apply( - lambda x: wrap_text(x) if not pd.isna(x) else x) - - num_cols = df_html.select_dtypes('number').columns - num_col_colors = sns.color_palette('husl', len(num_cols)) - bool_cols = df_html.select_dtypes(bool).columns - bool_col_colors = {True: 'lightgreen', False: 'lightpink'} - - if columns_unique is not None: - unique_df = df_html.loc[:, columns_unique].iloc[[0]].copy() - df_html = df_html.loc[:, ~columns_unique] - df_html.dropna(how='all', axis=1, inplace=True) - unique_df.dropna(how='all', axis=1, inplace=True) - - if not file: - if 'Compound ID' in df_html.columns: - df_html.drop(['Compound SMILES'], axis=1, inplace=True) - if 'Target ID' in df_html.columns: - df_html.drop(['Target FASTA'], axis=1, inplace=True) - if 'Target FASTA' in df_html.columns: - df_html['Target FASTA'] = df_html['Target FASTA'].parallel_apply( - lambda x: wrap_text(x) if not pd.isna(x) else x) - if 'Scaffold SMILES' in df_html.columns: - df_html.drop(['Scaffold SMILES'], axis=1, inplace=True) - - # FIXME: Temporarily drop pharmacophore column before an image solution is found - if 'Pharmacophore' in df_html.columns: - df_html.drop(['Pharmacophore'], axis=1, inplace=True) - if unique_df is not None and 'Pharmacophore' in unique_df.columns: - unique_df.drop(['Pharmacophore'], axis=1, inplace=True) - - styled_df = df_html.fillna('').style.format(precision=3) - - for i, col in enumerate(num_cols): - cmap = sns.light_palette(num_col_colors[i], as_cmap=True) - if col in df_html.columns: - if col not in ['Binding Affinity (IC50 [nM])']: - cmap.set_bad('white') - styled_df = styled_df.background_gradient( - subset=[col], cmap=cmap) - else: - cmap = cmap.reversed() - cmap.set_bad('white') - styled_df = styled_df.background_gradient( - subset=[col], cmap=cmap) - - if any(df_html.columns.isin(bool_cols)): - styled_df.applymap(lambda val: f'background-color: {bool_col_colors[val]}', subset=bool_cols) - - table_html = styled_df.to_html() - unique_html = '' - if unique_df is not None: - if 'Target FASTA' in unique_df.columns: - unique_df['Target FASTA'] = unique_df['Target FASTA'].str.replace('\n', '{text}
', + ) + + +TASK_MAP = { + 'Compound-Protein Interaction': 'DTI', + 'Compound-Protein Binding Affinity': 'DTA', +} + +TASK_METRIC_MAP = { + 'DTI': 'AUROC', + 'DTA': 'CI', + 'Compound-Protein Interaction': 'AUROC', + 'Compound-Protein Binding Affinity': 'CI', + 'CPI': 'DTI', + 'CPA': 'DTA', +} + +PRESET_MAP = { + 'DeepDTA': 'deep_dta', + 'DeepConvDTI': 'deep_conv_dti', + 'GraphDTA': 'graph_dta', + 'MGraphDTA': 'm_graph_dta', + 'HyperAttentionDTI': 'hyper_attention_dti', + 'MolTrans': 'mol_trans', + 'TransformerCPI': 'transformer_cpi', + 'TransformerCPI2': 'transformer_cpi_2', + 'DrugBAN': 'drug_ban', + 'DrugVQA-Seq': 'drug_vqa' +} + +TARGET_FAMILY_MAP = { + 'General': 'general', + 'Kinase': 'kinase', + 'Non-Kinase Enzyme': 'non_kinase_enzyme', + 'Membrane Receptor': 'membrane_receptor', + 'Nuclear Receptor': 'nuclear_receptor', + 'Ion Channel': 'ion_channel', + 'Others': 'others', + # 'general': 'general', + # 'kinase': 'kinase', + # 'non-kinase enzyme': 'non_kinase_enzyme', + # 'membrane receptor': 'membrane_receptor', + # 'nuclear Receptor': 'nuclear_receptor', + # 'ion channel': 'ion_channel', + # 'others': 'others', +} + +TARGET_LIBRARY_MAP = { + 'DrugBank (Human)': 'drugbank_targets.csv', + 'ChEMBL33 (Human)': 'ChEMBL33_human_proteins.csv', +} + +DRUG_LIBRARY_MAP = { + 'DrugBank (Human)': 'drugbank_compounds.csv', + 'Drug Repurposing Hub': 'drug_repurposing_hub.csv', + 'Enamine Discovery Diversity Set (DDS-10)': 'Enamine_Discovery_Diversity_Set_10_10240cmpds_20240130.csv', + 'Enamine Phenotypic Screening Library (PSL-5760)': 'Enamine_Phenotypic_Screening_Library_plated_5760cmds_2020_07_20.csv' +} + +COLUMN_ALIASES = { + 'X1': 'Compound SMILES', + 'X2': 'Target FASTA', + 'ID1': 'Compound ID', + 'ID2': 'Target ID', + 'Y': 'Actual CPI/CPA', + 'Y^': 'Predicted CPI/CPA', +} + +DRUG_SCRENN_CPI_OPTS = [ + 'Calculate Max. Sequence Identity between the Input Target and Targets in the Training Set', + 'Calculate Max. Tanimoto Similarity between the Hit Compound and Known Ligands of the Input Target', + 'Calculate Max. Sequence Identity between the Input Target and Known Targets of Hit Compound', +] + +DRUG_SCRENN_CPA_OPTS = [ + 'Calculate Max. Sequence Identity between the Input Target and Targets in the Training Set', +] + +TARGET_IDENTIFY_CPI_OPTS = [ + 'Calculate Max. Tanimoto Similarity between the Input Compound and Compounds in the Training Set', + 'Calculate Max. Sequence Identity between the Identified Target and Known Targets of the Input Compound', + 'Calculate Max. Tanimoto Similarity between the Input Compound and Known Ligands of the Identified Target', +] + +TARGET_IDENTIFY_CPA_OPTS = [ + 'Calculate Max. Tanimoto Similarity between the Input Compound and Compounds in the Training Set', +] + +pd.set_option('display.float_format', '{:.3f}'.format) +PandasTools.molRepresentation = 'svg' +PandasTools.drawOptions = Draw.rdMolDraw2D.MolDrawOptions() +PandasTools.drawOptions.clearBackground = False +PandasTools.drawOptions.bondLineWidth = 1 +PandasTools.drawOptions.explicitMethyl = True +PandasTools.drawOptions.singleColourWedgeBonds = True +PandasTools.drawOptions.useCDKAtomPalette() +PandasTools.molSize = (100, 64) + + +def remove_job_record(job_id): + # Delete the job from the database + db.remove(Job.id == job_id) + # Delete the corresponding files + files = glob.glob(f"{SERVER_DATA_DIR}/{job_id}*") + for file_path in files: + if os.path.exists(file_path): + os.remove(file_path) + + +def check_expiry(): + Job = Query() + jobs = db.all() + + for job in jobs: + # Check if the job has expired + if job['status'] != 'RUNNING': + expiry_time = job['expiry_time'] if job['expiry_time'] is not None else job['start_time'] + DB_EXPIRY + if expiry_time < time(): + # Delete the job from the database + db.remove(Job.id == job['id']) + # Delete the corresponding file + files = glob.glob(f"{SERVER_DATA_DIR}/{job['id']}*") + for file_path in files: + if os.path.exists(file_path): + os.remove(file_path) + elif job['status'] == 'RUNNING' and time() - job['start_time'] > 4 * 60 * 60: # 4 hours + # Mark the job as failed + db.update({'status': 'FAILED', + 'error': 'Job has timed out by exceeding the maximum running time of 4 hours.'}, + Job.id == job['id']) + if job.get('email'): + send_email(job) + + +def smiles_to_ecfp(smiles): + mol = Chem.MolFromSmiles(smiles) + if mol: + ecfp = AllChem.GetMorganFingerprintAsBitVect(mol, radius=2, nBits=2048) + else: + ecfp = [] + return ecfp + + +def max_tanimoto_similarity(smi, seen_smiles_with_fp): + if smi is None or seen_smiles_with_fp is None or seen_smiles_with_fp.empty: + return {'Max. Tanimoto Similarity': 0, 'Max. Tanimoto Similarity Compound': None} + + if smi in seen_smiles_with_fp['X1'].values: + compound = smi + if 'ID1' in seen_smiles_with_fp.columns: + id1 = seen_smiles_with_fp.loc[seen_smiles_with_fp['X1'] == smi, 'ID1'].values[0] + if pd.notnull(id1) and id1 != '': + compound = id1 + return {'Max. Tanimoto Similarity': 1, 'Max. Tanimoto Similarity Compound': compound} + + mol = Chem.MolFromSmiles(smi) + if mol is None: + return {'Max. Tanimoto Similarity': 0, 'Max. Tanimoto Similarity Compound': None} + + mol_ecfp = AllChem.GetMorganFingerprintAsBitVect(mol, radius=2, nBits=2048) + sims = pd.Series(BulkTanimotoSimilarity(mol_ecfp, seen_smiles_with_fp['FP'].values)).to_numpy() + idx = sims.argmax() + compound = seen_smiles_with_fp.iloc[idx]['X1'] + if 'ID1' in seen_smiles_with_fp.columns: + id1 = seen_smiles_with_fp.iloc[idx]['ID1'] + if pd.notnull(id1) and id1 != '': + compound = id1 + + return {'Max. Tanimoto Similarity': sims[idx], 'Max. Tanimoto Similarity Compound': compound} + + +def alignment_score(query, target): + aligner = PairwiseAligner() + aligner.mode = 'local' + alignment = aligner.align(query, target) + return alignment.score / max(len(query), len(target)) + + +def max_sequence_identity(seq, seen_fastas): + if seq is None or seen_fastas is None or seen_fastas.empty: + return {'Max. Sequence Identity': 0, 'Max. Sequence Identity Target': None} + + if seq in seen_fastas['X2'].values: + target = seq + if 'ID2' in seen_fastas.columns: + id2 = seen_fastas.loc[seen_fastas['X2'] == seq, 'ID2'].values[0] + if pd.notnull(id2) and id2 != '': + target = id2 + return {'Max. Sequence Identity': 1, 'Max. Sequence Identity Target': target} + + cached_alignment_score = cache(alignment_score) + max_iden = 0 + target = None + for fasta in seen_fastas['X2'].values: + identity = cached_alignment_score(seq, fasta) + + if identity > max_iden: + max_iden = identity + target = fasta + if 'ID2' in seen_fastas.columns: + id2 = seen_fastas.loc[seen_fastas['X2'] == fasta, 'ID2'].values[0] + if pd.notnull(id2) and id2 != '': + target = id2 + if max_iden == 1: + break + + cached_alignment_score.cache_clear() + return {'Max. Sequence Identity': max_iden, 'Max. Sequence Identity Target': target} + + +def get_seen_smiles(family, task): + if family == 'General': + family = 'all_families_full' + else: + family = TARGET_FAMILY_MAP[family.title()] + seen_smiles = pd.read_csv( + f'data/benchmarks/seen_compounds/{family}_{task.lower()}_random_split.csv') + return seen_smiles + + +def get_seen_fastas(family, task): + if family == 'General': + family = 'all_families_full' + else: + family = TARGET_FAMILY_MAP[family.title()] + seen_fastas = pd.read_csv( + f'data/benchmarks/seen_targets/{family}_{task.lower()}_random_split.csv') + return seen_fastas + + +@cache +def get_fasta_family_map(): + usecols = ['X2', 'ID2', 'Target Family'] + fasta_family_map = pd.concat([ + pd.read_csv('data/target_libraries/ChEMBL33_all_spe_single_prot_info.csv', usecols=usecols), + pd.read_csv('data/target_libraries/idmapping_not_in_chembl.csv', usecols=usecols) + ]).drop_duplicates(subset=['X2'], keep='first') + return fasta_family_map + + +def lipinski(mol): + """ + Lipinski's rules: + Hydrogen bond donors <= 5 + Hydrogen bond acceptors <= 10 + Molecular weight <= 500 daltons + logP <= 5 + """ + return ( + Lipinski.NumHDonors(mol) <= 5 and + Lipinski.NumHAcceptors(mol) <= 10 and + Descriptors.MolWt(mol) <= 500 and + Crippen.MolLogP(mol) <= 5 + ) + + +def reos(mol): + """ + Rapid Elimination Of Swill filter: + Molecular weight between 200 and 500 + LogP between -5.0 and +5.0 + H-bond donor count between 0 and 5 + H-bond acceptor count between 0 and 10 + Formal charge between -2 and +2 + Rotatable bond count between 0 and 8 + Heavy atom count between 15 and 50 + """ + return ( + 200 <= Descriptors.MolWt(mol) <= 500 and + -5.0 <= Crippen.MolLogP(mol) <= 5.0 and + 0 <= Lipinski.NumHDonors(mol) <= 5 and + 0 <= Lipinski.NumHAcceptors(mol) <= 10 and + -2 <= rdmolops.GetFormalCharge(mol) <= 2 and + 0 <= rdMolDescriptors.CalcNumRotatableBonds(mol) <= 8 and + 15 <= rdMolDescriptors.CalcNumHeavyAtoms(mol) <= 50 + ) + + +def ghose(mol): + """ + Ghose drug like filter: + Molecular weight between 160 and 480 + LogP between -0.4 and +5.6 + Atom count between 20 and 70 + Molar refractivity between 40 and 130 + """ + return ( + 160 <= Descriptors.MolWt(mol) <= 480 and + -0.4 <= Crippen.MolLogP(mol) <= 5.6 and + 20 <= rdMolDescriptors.CalcNumAtoms(mol) <= 70 and + 40 <= Crippen.MolMR(mol) <= 130 + ) + + +def veber(mol): + """ + The Veber filter is a rule of thumb filter for orally active drugs described in + Veber et al., J Med Chem. 2002; 45(12): 2615-23.: + Rotatable bonds <= 10 + Topological polar surface area <= 140 + """ + return ( + rdMolDescriptors.CalcNumRotatableBonds(mol) <= 10 and + rdMolDescriptors.CalcTPSA(mol) <= 140 + ) + + +def rule_of_three(mol): + """ + Rule of Three filter (Congreve et al., Drug Discov. Today. 8 (19): 876–7, (2003).): + Molecular weight <= 300 + LogP <= 3 + H-bond donor <= 3 + H-bond acceptor count <= 3 + Rotatable bond count <= 3 + """ + return ( + Descriptors.MolWt(mol) <= 300 and + Crippen.MolLogP(mol) <= 3 and + Lipinski.NumHDonors(mol) <= 3 and + Lipinski.NumHAcceptors(mol) <= 3 and + rdMolDescriptors.CalcNumRotatableBonds(mol) <= 3 + ) + + +@cache +def load_smarts_patterns(smarts_path): + # Load the CSV file containing SMARTS patterns + smarts_df = pd.read_csv(Path(smarts_path)) + # Convert all SMARTS patterns to molecules + smarts_mols = [Chem.MolFromSmarts(smarts) for smarts in smarts_df['smarts']] + return smarts_mols + + +def smarts_filter(mol, smarts_mols): + for smarts_mol in smarts_mols: + if smarts_mol is not None and mol.HasSubstructMatch(smarts_mol): + return False + return True + + +def pains(mol): + smarts_mols = load_smarts_patterns("data/filters/pains.csv") + return smarts_filter(mol, smarts_mols) + + +def mlsmr(mol): + smarts_mols = load_smarts_patterns("data/filters/mlsmr.csv") + return smarts_filter(mol, smarts_mols) + + +def dundee(mol): + smarts_mols = load_smarts_patterns("data/filters/dundee.csv") + return smarts_filter(mol, smarts_mols) + + +def glaxo(mol): + smarts_mols = load_smarts_patterns("data/filters/glaxo.csv") + return smarts_filter(mol, smarts_mols) + + +def bms(mol): + smarts_mols = load_smarts_patterns("data/filters/bms.csv") + return smarts_filter(mol, smarts_mols) + + +SCORE_MAP = { + 'SAscore': sascorer.calculateScore, + 'LogP': Crippen.MolLogP, + 'Molecular Weight': Descriptors.MolWt, + 'Number of Atoms': rdMolDescriptors.CalcNumAtoms, + 'Number of Heavy Atoms': rdMolDescriptors.CalcNumHeavyAtoms, + 'Molar Refractivity': Crippen.MolMR, + 'H-Bond Donor Count': Lipinski.NumHDonors, + 'H-Bond Acceptor Count': Lipinski.NumHAcceptors, + 'Rotatable Bond Count': rdMolDescriptors.CalcNumRotatableBonds, + 'Topological Polar Surface Area': rdMolDescriptors.CalcTPSA, +} + +FILTER_MAP = { + # TODO support number_of_violations + 'REOS': reos, + "Lipinski's Rule of Five": lipinski, + 'Ghose': ghose, + 'Rule of Three': rule_of_three, + 'Veber': veber, + 'PAINS': pains, + 'MLSMR': mlsmr, + 'Dundee': dundee, + 'Glaxo': glaxo, + 'BMS': bms, +} + + +def validate_columns(df, mandatory_cols): + missing_cols = [col for col in mandatory_cols if col not in df.columns] + if missing_cols: + error_message = (f"The following mandatory columns are missing " + f"in the uploaded dataset: {str(mandatory_cols).strip('[]')}.") + raise ValueError(error_message) + else: + return + + +def process_target_fasta(sequence): + try: + if sequence: + lines = sequence.strip().split("\n") + if lines[0].startswith(">"): + lines = lines[1:] + return ''.join(lines).split(">")[0].strip() + # record = list(SeqIO.parse(io.StringIO(sequence), "fasta"))[0] + # return str(record.seq) + else: + raise ValueError('Empty FASTA sequence.') + except Exception as e: + raise gr.Error(f'Failed to process FASTA due to error: {str(e)}') + + +def send_email(job_info): + if job_info.get('email'): + try: + email_info = job_info.copy() + email_serv = os.getenv('EMAIL_SERV') + email_port = os.getenv('EMAIL_PORT') + email_addr = os.getenv('EMAIL_ADDR') + email_pass = os.getenv('EMAIL_PASS') + email_form = os.getenv('EMAIL_FORM') + email_subj = os.getenv('EMAIL_SUBJ') + + for key, value in email_info.items(): + if key.endswith("time") and value: + email_info[key] = ts_to_str(value, get_timezone_by_ip(email_info['ip'])) + + server = smtplib.SMTP(email_serv, int(email_port)) + # server.starttls() + + server.login(email_addr, email_pass) + msg = MIMEMultipart("alternative") + msg["From"] = email_addr + msg["To"] = email_info['email'] + msg["Subject"] = email_subj.format(**email_info) + msg["Date"] = formatdate(localtime=True) + msg["Message-ID"] = make_msgid() + + msg.attach(MIMEText(markdown(email_form.format(**email_info)), 'html')) + msg.attach(MIMEText(email_form.format(**email_info), 'plain')) + + server.sendmail(email_addr, email_info['email'], msg.as_string()) + server.quit() + gr.Info('Email notification sent.') + except Exception as e: + gr.Warning('Failed to send email notification due to error: ' + str(e)) + + +def check_user_running_job(email, request): + message = ("You already have a running prediction job (ID: {id}) under this {reason}. " + "Please wait for it to complete before submitting another job.") + try: + # with open('jobs.json', 'r') as f: # /data/ + # # Load the JSON data from the file + # jobs = json.load(f) + # + # for job_id, job_info in jobs.items(): + # # check if a job is running for the email + # if email: + # if job_info["email"] == email and job_info["status"] == "running": + # return message.format(id=job_id, reason="email") + # # check if a job is running for the session + # elif request.cookies: + # for key, value in job_info["cookies"].items() and job_info["status"] == "running": + # if key in request.cookies and request.cookies[key] == value: + # return message.format(id=job_id, reason="session") + # # check if a job is running for the IP + # else: + # if job_info["IP"] == request.client.host and job_info["status"] == "running": + # return message.format(id=job_id, reason="IP") + # check if a job is running for the email + Job = Query() + if email: + job = db.search((Job.email == email) & (Job.status == "RUNNING")) + if job: + return message.format(id=job[0]['id'], reason="email") + # check if a job is running for the session + elif request.cookies: + for key, value in request.cookies.items(): + job = db.search((Job.cookies[key] == value) & (Job.status == "RUNNING")) + if job: + return message.format(id=job[0]['id'], reason="session") + # check if a job is running for the IP + else: + job = db.search((Job.IP == request.client.host) & (Job.status == "RUNNING")) + if job: + return message.format(id=job[0]['id'], reason="IP") + + return False + except Exception as e: + raise gr.Error(f'Failed to validate user running jobs due to error: {str(e)}') + + +def get_timezone_by_ip(ip): + try: + data = session.get(f'https://worldtimeapi.org/api/ip/{ip}').json() + return data['timezone'] + except Exception: + return 'UTC' + + +def ts_to_str(timestamp, timezone): + # Create a timezone-aware datetime object from the UNIX timestamp + dt = datetime.fromtimestamp(timestamp, pytz.utc) + + # Convert the timezone-aware datetime object to the target timezone + target_timezone = pytz.timezone(timezone) + localized_dt = dt.astimezone(target_timezone) + + # Format the datetime object to the specified string format + return localized_dt.strftime('%Y-%m-%d %H:%M:%S (%Z%z)') + + +def lookup_job(job_id): + gr.Info('Start querying the job database...') + stop = False + retry = 0 + while not stop: + try: + sleep(5) + Job = Query() + jobs = db.search((Job.id == job_id)) + if jobs: + job = jobs[0] + job_status = job['status'] + job_type = job['type'] + error = job['error'] + start_time = ts_to_str(job['start_time'], get_timezone_by_ip(job['ip'])) + if job.get('end_time'): + end_time = ts_to_str(job['end_time'], get_timezone_by_ip(job['ip'])) + if job.get('expiry_time'): + expiry_time = ts_to_str(job['expiry_time'], get_timezone_by_ip(job['ip'])) + if job_status == "RUNNING": + yield { + pred_lookup_status: f''' +Your **{job_type}** job (ID: **{job_id}**) started at +**{start_time}** and is **RUNNING...** + +It might take a few minutes up to a few hours depending on the prediction dataset, the model, and the queue status. +You may keep the page open and wait for job completion, or close the page and revisit later to look up the job status +using the job id. You will also receive an email notification once the job is done. +''', + pred_lookup_btn: gr.Button(visible=False), + pred_lookup_stop_btn: gr.Button(visible=True) + } + if job_status == "COMPLETED": + stop = True + msg = f"Your {job_type} job (ID: {job_id}) has been **COMPLETED**" + msg += f" at {end_time}" if job.get('end_time') else "" + msg += f" and the results will expire by {expiry_time}." if job.get('expiry_time') else "." + msg += f' Redirecting to the report page...' + + gr.Info(msg) + yield { + pred_lookup_status: msg, + pred_lookup_btn: gr.Button(visible=True), + pred_lookup_stop_btn: gr.Button(visible=False), + tabs: gr.Tabs(selected='Chemical Property Report'), + file_for_report: job['output_file'] + } + if job_status == "FAILED": + stop = True + msg = f'Your {job_type} job (ID: {job_id}) has **FAILED**' + msg += f' at {end_time}' if job.get('end_time') else '' + msg += f' due to error: {error}.' if job.get('expiry_time') else '.' + gr.Info(msg) + yield { + pred_lookup_status: msg, + pred_lookup_btn: gr.Button(visible=True), + pred_lookup_stop_btn: gr.Button(visible=False), + tabs: gr.Tabs(selected='Prediction Status Lookup'), + } + else: + stop = (retry > 3) + if not stop: + msg = f'Job ID {job_id} not found. Retrying... ({retry})' + else: + msg = f'Job ID {job_id} not found after {retry} retries. Please check the job ID and try again.' + gr.Info(msg) + retry += 1 + yield { + pred_lookup_status: msg, + pred_lookup_btn: gr.Button(visible=True), + pred_lookup_stop_btn: gr.Button(visible=False), + tabs: gr.Tabs(selected='Prediction Status Lookup'), + } + + except Exception as e: + raise gr.Error(f'Failed to retrieve job status due to error: {str(e)}') + + +def apply_advanced_opts(prediction_df, opts, df_training): + # Advanced options for Drug Hit Screening + if "Calculate Max. Sequence Identity between the Input Target and Targets in the Training Set" in opts: + x2 = prediction_df['X2'].iloc[0] + + prediction_df[[ + 'Max. Sequence Identity to Training Targets', + 'Max. Id. Training Target' + ]] = pd.Series(max_sequence_identity(x2, df_training)) + + if "Calculate Max. Tanimoto Similarity between the Hit Compound and Known Ligands of the Input Target" in opts: + x2 = prediction_df['X2'].iloc[0] + pos_compounds_df = df_training.loc[(df_training['X2'] == x2) & (df_training['Y'] == 1)].copy() + pos_compounds_df['FP'] = pos_compounds_df['X1'].parallel_apply(smiles_to_ecfp) + + @cache + def max_sim(smiles): + return max_tanimoto_similarity(smiles, pos_compounds_df) + + prediction_df[[ + 'Max. Tanimoto Similarity to Known Ligands', + 'Max. Sim. Ligand' + ]] = prediction_df['X1'].parallel_apply(max_sim).apply(pd.Series) + + max_sim.cache_clear() + + if "Calculate Max. Sequence Identity between the Input Target and Known Targets of Hit Compound" in opts: + x2 = prediction_df['X2'].iloc[0] + prediction_df['X1^'] = prediction_df['X1'].parallel_apply(rdkit_canonicalize) + + @cache + def max_id(compound): + pos_targets_df = df_training.loc[df_training['X1'] == compound] + return max_sequence_identity(x2, pos_targets_df) + + prediction_df[['Max. Sequence Identity to Known Targets of Hit Compound', + 'Max. Id. Target']] = ( + prediction_df['X1^'].parallel_apply(max_id).apply(pd.Series) + ) + prediction_df.drop(['X1^'], axis=1, inplace=True) + + max_id.cache_clear() + + # Advanced options for Target Protein Identification + if "Calculate Max. Tanimoto Similarity between the Input Compound and Compounds in the Training Set" in opts: + x1 = rdkit_canonicalize(prediction_df['X1'].iloc[0]) + if x1 not in df_training['X1'].values: + df_training['FP'] = df_training['X1'].parallel_apply(smiles_to_ecfp) + + prediction_df[[ + 'Max. Tanimoto Similarity to Training Compounds', + 'Max. Sim. Training Compound' + ]] = pd.Series(max_tanimoto_similarity(x1, df_training)) + + if "Calculate Max. Sequence Identity between the Identified Target and Known Targets of the Input Compound" in opts: + x1 = rdkit_canonicalize(prediction_df['X1'].iloc[0]) + pos_targets_df = df_training.loc[(df_training['X1'] == x1) & (df_training['Y'] == 1)].copy() + + @cache + def max_id(fasta): + return max_sequence_identity(fasta, pos_targets_df) + + prediction_df[[ + 'Max. Sequence Identity to Known Targets of Input Compound', + 'Max. Id. Target' + ]] = prediction_df['X2'].parallel_apply(max_id).apply(pd.Series) + + max_id.cache_clear() + + if "Calculate Max. Tanimoto Similarity between the Input Compound and Known Ligands of the Identified Target" in opts: + x1 = rdkit_canonicalize(prediction_df['X1'].iloc[0]) + + @cache + def max_sim(fasta): + pos_targets_df = df_training.loc[(df_training['X2'] == fasta) & (df_training['Y'] == 1)].copy() + if x1 not in pos_targets_df['X1'].values: + pos_targets_df['FP'] = pos_targets_df['X1'].apply(smiles_to_ecfp) + return max_tanimoto_similarity(x1, pos_targets_df) + + prediction_df[[ + 'Max. Tanimoto Similarity to Known Ligands of Identified Target', + 'Max. Sim. Ligand' + ]] = prediction_df['X2'].parallel_apply(max_sim).apply(pd.Series) + + max_sim.cache_clear() + + return prediction_df + + +def submit_predict(predict_filepath, task, preset, target_family, opts, job_info): + job_id = job_info['id'] + status = job_info['status'] + send_email(job_info) + db.insert(job_info) + error = None + task_file_abbr = {'Compound-Protein Interaction': 'CPI', 'Compound-Protein Binding Affinity': 'CPA'} + predictions_file = None + df_training = pd.read_csv(f'data/complete_{TASK_MAP[task].lower()}_dataset.csv') + df_training['X1^'] = df_training['X1'] + orig_df = pd.read_csv(predict_filepath) + alignment_df = get_fasta_family_map() + prediction_df = pd.DataFrame() + + @cache + def detect_family(query): + # Check for an exact match first + exact_match = alignment_df[alignment_df['X2'] == query] + if not exact_match.empty: + row = exact_match.iloc[0] + return row['Target Family'] + # If no exact match, then calculate alignment score + else: + aligner = PairwiseAligner() + aligner.mode = 'local' + + def align_score(target): + alignment = aligner.align(query, target) + return alignment.score / max(len(query), len(target)) + + alignment_df['score'] = alignment_df['X2'].apply(align_score) + row = alignment_df.loc[alignment_df['score'].idxmax()] + return row['Target Family'] + + if 'Target Family' not in orig_df.columns: + orig_df['Target Family'] = None + if orig_df['Target Family'].isna().any(): + if orig_df['X2'].nunique() > 1: + orig_df = orig_df.reset_index(drop=True) + orig_df.loc[orig_df['Target Family'].isna(), 'Target Family'] = ( + orig_df.loc[orig_df['Target Family'].isna(), 'X2'].parallel_apply(detect_family) + ) + else: + orig_df['Target Family'] = detect_family(orig_df['X2'].iloc[0]) + orig_df['Target Family'] = orig_df['Target Family'].str.capitalize() + detect_family.cache_clear() + + orig_df['X1^'] = orig_df['X1'].parallel_apply(rdkit_canonicalize) + + orig_df = orig_df.merge(df_training[['X1^', 'X2', 'Y']], on=['X1^', 'X2'], how='left', indicator=False) + annotated_df = orig_df[~orig_df['Y'].isna()].copy() + annotated_df.rename(columns={'Y': 'Y^'}, inplace=True) + annotated_df['Source'] = 'Database' + columns_to_drop = ['X1^', 'Compound', 'Scaffold', 'Scaffold SMILES'] + columns_to_drop = [col for col in columns_to_drop if col in annotated_df.columns] + annotated_df.drop(columns_to_drop, axis=1, inplace=True) + + # Save the unannotated data + unannotated_df = orig_df[orig_df['Y'].isna()].drop(['Y'], axis=1) + if not unannotated_df.empty: + unannotated_df.to_csv(predict_filepath, index=False, na_rep='') + else: + annotated_df.to_csv(predictions_file, index=False, na_rep='') + status = "COMPLETED" + return {run_state: False} + + columns_to_drop = ['ID1', 'X1^', 'Compound', 'Scaffold', 'Scaffold SMILES', 'ID2', 'Y', 'Y^'] + columns_to_drop = [col for col in columns_to_drop if col in orig_df.columns] + orig_df.drop(columns_to_drop, axis=1, inplace=True) + + try: + if target_family != 'Family-Specific Auto-Recommendation': + target_family_value = TARGET_FAMILY_MAP[target_family.title()] + task_value = TASK_MAP[task] + preset_value = PRESET_MAP[preset] + predictions_file = (f'{SERVER_DATA_DIR}/' + f'{job_id}_{task_file_abbr[task]}_{preset}_{target_family_value}_predictions.csv') + + cfg = hydra.compose( + config_name="webserver_inference", + overrides=[f"task={task_value}", + f"preset={preset_value}", + f"ckpt_path=resources/checkpoints/{preset_value}-{task_value}-{target_family_value}.ckpt", + f"data.data_file='{str(predict_filepath)}'"]) + + predictions, _ = predict(cfg) + predictions = pd.concat([pd.DataFrame(prediction) for prediction in predictions], ignore_index=True) + predictions['Source'] = f'Predicted ({preset} {target_family})' + df_list = [prediction_df, predictions] + prediction_df = pd.concat([df for df in df_list if not df.empty]) + + else: + predictions_file = f'{SERVER_DATA_DIR}/{job_id}_{task_file_abbr[task]}_family-recommended_predictions.csv' + task_value = TASK_MAP[task] + score = TASK_METRIC_MAP[task] + benchmark_df = pd.read_csv(f'data/benchmarks/{task_value}_test_metrics.csv') + predict_df = pd.read_csv(predict_filepath) + + for family, subset in predict_df.groupby('Target Family'): + predict_subset_filepath = os.path.join( + os.path.dirname(predict_filepath), f'{job_id}_{family}_input.csv' + ) + subset.to_csv(predict_subset_filepath, index=False, na_rep='') + + seen_compounds = get_seen_smiles(family, task_value)['X1'].values + if subset['X1^'].iloc[0] in seen_compounds: + scenario = "Seen Compound" + else: + scenario = "Unseen Compound" + + filtered_df = benchmark_df[(benchmark_df['Family'] == family.title()) + & (benchmark_df['Scenario'] == scenario) + & (benchmark_df['Type'] == 'Family')] + + seen_compounds = get_seen_smiles('General', task_value)['X1'].values + if subset['X1^'].iloc[0] in seen_compounds: + scenario = "Seen Compound" + else: + scenario = "Unseen Compound" + + filtered_df = pd.concat([ + filtered_df, + benchmark_df[(benchmark_df['Family'] == family.title()) + & (benchmark_df['Scenario'] == scenario) + & (benchmark_df['Type'] == 'General')] + ]) + + row = filtered_df.loc[filtered_df[score].idxmax()] + preset_value = PRESET_MAP[row['Model']] + target_family = TARGET_FAMILY_MAP[family.title()] if row['Type'] == 'Family' else 'general' + cfg = hydra.compose( + config_name="webserver_inference", + overrides=[f"task={task_value}", + f"preset={preset_value}", + f"ckpt_path=resources/checkpoints/{preset_value}-{task_value}-{target_family}.ckpt", + f"data.data_file='{str(predict_subset_filepath)}'"]) + + predictions, _ = predict(cfg) + predictions = pd.concat([pd.DataFrame(prediction) for prediction in predictions], ignore_index=True) + predictions['Source'] = (f'Predicted ({row["Model"]} ' + f'{family.title() if row["Type"] == "Family" else "General"})') + df_list = [prediction_df, predictions] + prediction_df = pd.concat([df for df in df_list if not df.empty]) + + prediction_df = prediction_df.merge(orig_df, on=['X1', 'X2'], how='left', indicator=False) + df_list = [prediction_df, annotated_df] + prediction_df = pd.concat([df for df in df_list if not df.empty], ignore_index=True) + + prediction_df = apply_advanced_opts(prediction_df, opts, df_training) + + prediction_df.drop(['N', 'FP'], axis=1, errors='ignore').to_csv(predictions_file, index=False, na_rep='') + status = 'COMPLETED' + + return {run_state: False} + + except Exception as e: + gr.Warning(f"Prediction job failed due to error: {str(e)}") + status = "FAILED" + predictions_file = None + error = str(e) + return {run_state: False} + + finally: + Job = Query() + job_query = (Job.id == job_id) + + end_time = time() + expiry_time = end_time + DB_EXPIRY + + db.update({'end_time': end_time, + 'expiry_time': expiry_time, + 'status': status, + 'error': error, + 'input_file': predict_filepath, + 'output_file': predictions_file}, + job_query) + if job_info := db.search(job_query)[0]: + if job_info.get('email'): + send_email(job_info) + + +def update_df(file, progress=gr.Progress(track_tqdm=True)): + if file and Path(file).is_file(): + task = None + job = None + + if "_CPI_" in str(file): + task = 'Compound-Protein Interaction' + elif "_CPA_" in str(file): + task = 'Compound-Protein Binding Affinity' + + df = pd.read_csv(file) + + if 'N' in df.columns: + df.set_index('N', inplace=True) + + if not any(col in ['X1', 'X2'] for col in df.columns): + gr.Warning("At least one of columns `X1` and `X2` must be in the uploaded dataset.") + return {analyze_btn: gr.Button(interactive=False)} + + if 'X1' in df.columns: + if 'Compound' not in df.columns or df['Compound'].dtype != 'object': + df['Compound'] = df['X1'].parallel_apply( + lambda smiles: PandasTools._MolPlusFingerprint(Chem.MolFromSmiles(smiles))) + df['Scaffold'] = df['Compound'].parallel_apply(MurckoScaffold.GetScaffoldForMol) + df['Scaffold SMILES'] = df['Scaffold'].parallel_apply(lambda x: Chem.MolToSmiles(x)) + df['Pharmacophore'] = None + if task == 'Compound-Protein Binding Affinity': + # Convert Y^ from pIC50 (nM) to IC50 (nM) + if 'Y^' in df.columns: + df['Y^'] = 10 ** (-df['Y^']) * 1e9 + + n_compound = df['X1'].nunique() + n_protein = df['X2'].nunique() + + if n_compound == 1 and n_protein >= 2: + job = 'Target Protein Identification' + if task == 'Compound-Protein Interaction': + opts = TARGET_IDENTIFY_CPI_OPTS + elif task == 'Compound-Protein Binding Affinity': + opts = TARGET_IDENTIFY_CPA_OPTS + if n_compound >= 2 and n_protein == 1: + job = 'Drug Hit Screening' + if task == 'Compound-Protein Interaction': + opts = DRUG_SCRENN_CPI_OPTS + elif task == 'Compound-Protein Binding Affinity': + opts = DRUG_SCRENN_CPA_OPTS + + return { + html_report: create_html_report(df, file=None, task=task), + raw_df: df, + report_df: df.copy(), + analyze_btn: gr.Button(interactive=True), + report_task: task, + job_opts: gr.CheckboxGroup( + label=f'{job} Advanced Options', + choices=opts, visible=True + ) if job else gr.CheckboxGroup(visible=False), + } + else: + return {analyze_btn: gr.Button(interactive=False)} + + +def create_html_report(df, file=None, task=None, opts=(), progress=gr.Progress(track_tqdm=True)): + df_html = df.copy(deep=True) + column_aliases = COLUMN_ALIASES.copy() + cols_left = list(pd.Index([ + 'ID1', 'ID2', 'Compound', 'Scaffold', 'Pharmacophore', 'X1', 'Scaffold SMILES', 'X2', 'Y^' + ]).intersection(df_html.columns)) + # cols_right = list(pd.Index(['X1', 'X2']).intersection(df_html.columns)) + # df_html = df_html[cols_left + (df_html.columns.drop(cols_left + cols_right).tolist()) + cols_right] + df_html = df_html[cols_left + df_html.columns.drop(cols_left).tolist()] + + if isinstance(task, str): + column_aliases.update({ + 'Y^': 'Interaction Probability' if task == 'Compound-Protein Interaction' + else 'Binding Affinity (IC50 [nM])' + }) + + ascending = True if column_aliases['Y^'] == 'Binding Affinity (IC50 [nM])' else False + df_html = df_html.sort_values( + [col for col in ['Y^'] if col in df_html.columns], ascending=ascending + ) + + if not file: + df_html = df_html.iloc[:31] + + # Remove repeated info for one-against-N tasks to save visual and physical space + job = 'Chemical Property' + unique_entity = 'Unique Entity' + unique_df = None + category = None + columns_unique = None + + if 'Exclude Pharmacophore 3D' not in opts: + df_html['Pharmacophore'] = df_html['Compound'].parallel_apply( + lambda x: mol_to_pharm3d(x) if not pd.isna(x) else x) + + if 'Compound' in df_html.columns and 'Exclude Molecular Graph' not in opts: + df_html['Compound'] = df_html['Compound'].parallel_apply( + lambda x: PandasTools.PrintAsImageString(x) if not pd.isna(x) else x) + else: + df_html.drop(['Compound'], axis=1, inplace=True) + + if 'Scaffold' in df_html.columns and 'Exclude Scaffold Graph' not in opts: + df_html['Scaffold'] = df_html['Scaffold'].parallel_apply( + lambda x: PandasTools.PrintAsImageString(x) if not pd.isna(x) else x) + else: + df_html.drop(['Scaffold'], axis=1, inplace=True) + + if 'X1' in df_html.columns and 'X2' in df_html.columns: + n_compound = df_html['X1'].nunique() + n_protein = df_html['X2'].nunique() + + if n_compound == 1 and n_protein >= 2: + unique_entity = 'Compound of Interest' + if any(col in df_html.columns for col in ['Y^', 'Y']): + job = 'Target Protein Identification' + category = 'Target Family' + columns_unique = df_html.columns.isin( + ['ID1', 'Compound', 'Scaffold', 'X1', 'Scaffold SMILES', 'Pharmacophore', + 'Max. Tanimoto Similarity to Training Compounds', 'Max. Sim. Training Compound'] + + list(FILTER_MAP.keys()) + list(SCORE_MAP.keys()) + ) + + elif n_compound >= 2 and n_protein == 1: + unique_entity = 'Target of Interest' + if any(col in df_html.columns for col in ['Y^', 'Y']): + job = 'Drug Hit Screening' + category = 'Scaffold SMILES' + columns_unique = df_html.columns.isin( + ['X2', 'ID2', 'Max. Sequence Identity to Training Targets', 'Max. Id. Training Target'] + ) + + elif 'Y^' in df_html.columns: + job = 'Interaction Pair Inference' + + df_html.rename(columns=column_aliases, inplace=True) + df_html.index.name = 'Index' + if 'Target FASTA' in df_html.columns: + df_html['Target FASTA'] = df_html['Target FASTA'].parallel_apply( + lambda x: wrap_text(x) if not pd.isna(x) else x) + + num_cols = df_html.select_dtypes('number').columns + num_col_colors = sns.color_palette('husl', len(num_cols)) + bool_cols = df_html.select_dtypes(bool).columns + bool_col_colors = {True: 'lightgreen', False: 'lightpink'} + + if columns_unique is not None: + unique_df = df_html.loc[:, columns_unique].iloc[[0]].copy() + df_html = df_html.loc[:, ~columns_unique] + df_html.dropna(how='all', axis=1, inplace=True) + unique_df.dropna(how='all', axis=1, inplace=True) + + if not file: + if 'Compound ID' in df_html.columns: + df_html.drop(['Compound SMILES'], axis=1, inplace=True) + if 'Target ID' in df_html.columns: + df_html.drop(['Target FASTA'], axis=1, inplace=True) + if 'Target FASTA' in df_html.columns: + df_html['Target FASTA'] = df_html['Target FASTA'].parallel_apply( + lambda x: wrap_text(x) if not pd.isna(x) else x) + if 'Scaffold SMILES' in df_html.columns: + df_html.drop(['Scaffold SMILES'], axis=1, inplace=True) + + # FIXME: Temporarily drop pharmacophore column before an image solution is found + if 'Pharmacophore' in df_html.columns: + df_html.drop(['Pharmacophore'], axis=1, inplace=True) + if unique_df is not None and 'Pharmacophore' in unique_df.columns: + unique_df.drop(['Pharmacophore'], axis=1, inplace=True) + + styled_df = df_html.fillna('').style.format(precision=3) + + for i, col in enumerate(num_cols): + cmap = sns.light_palette(num_col_colors[i], as_cmap=True) + if col in df_html.columns: + if col not in ['Binding Affinity (IC50 [nM])']: + cmap.set_bad('white') + styled_df = styled_df.background_gradient( + subset=[col], cmap=cmap) + else: + cmap = cmap.reversed() + cmap.set_bad('white') + styled_df = styled_df.background_gradient( + subset=[col], cmap=cmap) + + if any(df_html.columns.isin(bool_cols)): + styled_df.applymap(lambda val: f'background-color: {bool_col_colors[val]}', subset=bool_cols) + + table_html = styled_df.to_html() + unique_html = '' + if unique_df is not None: + if 'Target FASTA' in unique_df.columns: + unique_df['Target FASTA'] = unique_df['Target FASTA'].str.replace('\n', '