| |
| """RDKit validation pipeline for generated SMILES (repo script). |
| |
| This is the validation method used throughout the DPP-4 generative |
| project, extracted as a standalone module/CLI: |
| |
| * the canonicalization / validity / uniqueness / novelty core is the |
| ``validate()`` logic of the generation scripts |
| (``generate_molgemma.py`` in this repo; originally |
| kaggle_molgemma_finetune.py and the ChemBERTa/Mistral equivalents); |
| * the second pass is the largest-fragment cleaning applied before LTN |
| scoring of the Mistral and MolGemma pools: multi-fragment |
| generations (salts / mixtures, e.g. ``mol.C(=O)(C(F)(F)F)O``) are |
| reduced to their largest fragment (by heavy-atom count), then |
| re-canonicalized and re-deduplicated. |
| |
| Definitions (project-wide, do not change): |
| |
| * **valid** — RDKit ``MolFromSmiles`` parses the raw string. |
| * **canonical** — ``MolToSmiles`` of the parsed molecule (the |
| identity used for every comparison). |
| * **unique** — distinct canonical SMILES among the valid samples. |
| * **novel** — a unique molecule whose canonical SMILES is absent |
| from the reference set (the fine-tune set, canonicalized the same |
| way); novelty vs the FDA-12 approved gliptins is reported |
| separately, exactly as in the project. |
| * **cleaned** — largest fragment of the canonical molecule; |
| **cleaned-unique** — distinct cleaned molecules among the novel |
| set. |
| |
| CLI |
| --- |
| python validate_rdkit.py generations.csv \ |
| --reference-csv dpp4_finetune_potent_moderate_plus_fda12.csv \ |
| [--fda-csv fda12.csv] [--smiles-col raw_smiles] \ |
| [--out generations_validated.csv] |
| |
| The generations CSV needs a SMILES column (auto-detected: |
| ``raw_smiles``, ``smiles`` or ``canonical_smiles``). The reference |
| CSV needs a ``smiles`` column. The FDA-12 reference defaults to the |
| project's built-in list (the same SMILES hard-coded in the |
| generation scripts); ``--fda-csv`` (columns ``name``, ``smiles``) |
| overrides it. The funnel (n -> valid -> unique -> novel -> |
| cleaned-unique) is printed and the per-row validated CSV is written. |
| """ |
|
|
| import argparse |
| import csv |
| import sys |
|
|
| from rdkit import Chem |
| from rdkit import RDLogger |
|
|
| RDLogger.DisableLog('rdApp.*') |
|
|
| |
| |
| FDA12 = [ |
| ('Alogliptin', 'Cn1c(=O)cc(N2CCC[C@@H](N)C2)n(Cc2ccccc2C#N)c1=O'), |
| ('Saxagliptin', 'N#C[C@@H]1C[C@@H]2C[C@@H]2N1C(=O)[C@@H](N)C12CC3CC(CC(O)(C3)C1)C2'), |
| ('Teneligliptin', 'Cc1cc(N2CCN([C@@H]3CN[C@H](C(=O)N4CCSC4)C3)CC2)n(-c2ccccc2)n1'), |
| ('Vildagliptin', 'N#C[C@@H]1CCCN1C(=O)CNC12CC3CC(CC(O)(C3)C1)C2'), |
| ('Anagliptin', 'Cc1cc2ncc(C(=O)NCC(C)(C)NCC(=O)N3CCC[C@H]3C#N)cn2n1'), |
| ('Evogliptin', 'CC(C)(C)OC[C@@H]1C(=O)NCCN1C(=O)C[C@H](N)Cc1cc(F)c(F)cc1F'), |
| ('Gemigliptin', 'N[C@@H](CC(=O)N1CCc2c(nc(C(F)(F)F)nc2C(F)(F)F)C1)CN1CC(F)(F)CCC1=O'), |
| ('Gosogliptin', 'O=C([C@@H]1C[C@H](N2CCN(c3ncccn3)CC2)CN1)N1CCC(F)(F)C1'), |
| ('Linagliptin', 'CC#CCn1c(N2CCC[C@@H](N)C2)nc2c1c(=O)n(Cc1nc(C)c3ccccc3n1)c(=O)n2C'), |
| ('Omarigliptin', 'CS(=O)(=O)n1cc2c(n1)CN([C@H]1CO[C@H](c3cc(F)ccc3F)[C@@H](N)C1)C2'), |
| ('Sitagliptin', 'N[C@@H](CC(=O)N1CCn2c(nnc2C(F)(F)F)C1)Cc1cc(F)c(F)cc1F'), |
| ('Trelagliptin', 'Cn1c(=O)cc(N2CCC[C@@H](N)C2)n(Cc2cc(F)ccc2C#N)c1=O'), |
| ] |
|
|
|
|
| def canon(smiles): |
| """RDKit canonical SMILES, or None if the string does not parse. |
| (Verbatim from the project generation scripts.)""" |
| m = Chem.MolFromSmiles(str(smiles)) |
| return Chem.MolToSmiles(m) if m is not None else None |
|
|
|
|
| def is_valid(smiles): |
| """True if RDKit can parse the SMILES.""" |
| return Chem.MolFromSmiles(str(smiles)) is not None |
|
|
|
|
| def canonical_set(smiles_iterable): |
| """Canonical reference set: canonicalize every entry, drop the |
| ones that do not parse. (How the fine-tune reference set is |
| built in the project: 9,338 rows -> 6,835 compounds.)""" |
| out = set() |
| for s in smiles_iterable: |
| c = canon(s) |
| if c: |
| out.add(c) |
| return out |
|
|
|
|
| def largest_fragment(canonical_smiles): |
| """Largest fragment (by heavy-atom count) of a canonical SMILES, |
| re-canonicalized. Single-fragment molecules are returned |
| unchanged. This is the cleaning rule applied to the Mistral and |
| MolGemma pools before LTN scoring.""" |
| if '.' not in canonical_smiles: |
| return canonical_smiles |
| best, best_heavy = None, -1 |
| for frag in canonical_smiles.split('.'): |
| m = Chem.MolFromSmiles(frag) |
| if m is None: |
| continue |
| heavy = m.GetNumHeavyAtoms() |
| if heavy > best_heavy: |
| best, best_heavy = Chem.MolToSmiles(m), heavy |
| return best if best is not None else canonical_smiles |
|
|
|
|
| def n_fragments(canonical_smiles): |
| return canonical_smiles.count('.') + 1 |
|
|
|
|
| def validate_pool(raw_smiles_list, reference_canon, fda_canon): |
| """Run the full project validation over raw generated SMILES. |
| |
| Returns (rows, funnel): rows is a list of per-sample dicts with |
| raw_smiles, canonical_smiles, valid, n_fragments, cleaned_smiles, |
| novel_vs_finetune, novel_vs_fda12; funnel is the summary dict |
| (n -> valid -> unique -> novel -> novel+fda -> mixtures -> |
| cleaned-unique), computed exactly as in the generation scripts |
| plus the fragment-cleaning pass.""" |
| rows, valid_canon = [], [] |
| for s in raw_smiles_list: |
| c = canon(s) |
| rows.append({'raw_smiles': s, 'canonical_smiles': c, |
| 'valid': c is not None}) |
| if c is not None: |
| valid_canon.append(c) |
| n = len(raw_smiles_list) |
| nv = len(valid_canon) |
| uniq = set(valid_canon) |
| novel_train = {c for c in uniq if c not in reference_canon} |
| novel_both = {c for c in novel_train if c not in fda_canon} |
| mixtures = {c for c in novel_train if '.' in c} |
| cleaned = {largest_fragment(c) for c in novel_train} |
| funnel = { |
| 'n_generated': n, |
| 'n_valid': nv, |
| 'validity_pct': round(100.0 * nv / n, 2) if n else 0.0, |
| 'n_unique_valid': len(uniq), |
| 'uniqueness_pct_of_valid': round(100.0 * len(uniq) / nv, 2) if nv else 0.0, |
| 'n_novel_vs_finetune': len(novel_train), |
| 'novelty_vs_finetune_pct_of_unique': round(100.0 * len(novel_train) / len(uniq), 2) if uniq else 0.0, |
| 'n_novel_vs_finetune_and_fda12': len(novel_both), |
| 'n_mixtures_among_novel': len(mixtures), |
| 'n_cleaned_unique_novel': len(cleaned), |
| } |
| for row in rows: |
| c = row['canonical_smiles'] |
| if c is None: |
| row.update({'n_fragments': None, 'cleaned_smiles': None, |
| 'novel_vs_finetune': False, 'novel_vs_fda12': False}) |
| else: |
| row.update({ |
| 'n_fragments': n_fragments(c), |
| 'cleaned_smiles': largest_fragment(c), |
| 'novel_vs_finetune': c in novel_train, |
| 'novel_vs_fda12': c in {u for u in uniq if u not in fda_canon}, |
| }) |
| return rows, funnel |
|
|
|
|
| def print_funnel(funnel): |
| f = funnel |
| print(f"n generated: {f['n_generated']}") |
| print(f"valid: {f['n_valid']} ({f['validity_pct']}%)") |
| print(f"unique (canonical): {f['n_unique_valid']} " |
| f"({f['uniqueness_pct_of_valid']}% of valid)") |
| print(f"novel vs fine-tune set: {f['n_novel_vs_finetune']} " |
| f"({f['novelty_vs_finetune_pct_of_unique']}% of unique)") |
| print(f"novel vs set + FDA-12: {f['n_novel_vs_finetune_and_fda12']}") |
| print(f"mixtures among novel: {f['n_mixtures_among_novel']}") |
| print(f"cleaned-unique novel: {f['n_cleaned_unique_novel']}") |
|
|
|
|
| def _read_csv(path): |
| with open(path, newline='') as fh: |
| return list(csv.DictReader(fh)) |
|
|
|
|
| def main(argv=None): |
| ap = argparse.ArgumentParser( |
| description="Validate generated SMILES with the project's " |
| "RDKit pipeline (validity / uniqueness / novelty " |
| "/ fragment cleaning).") |
| ap.add_argument("generations_csv", help="CSV of generated SMILES") |
| ap.add_argument("--reference-csv", required=True, |
| help="fine-tune/reference CSV with a 'smiles' column") |
| ap.add_argument("--fda-csv", default=None, |
| help="optional FDA reference CSV (name, smiles); " |
| "defaults to the built-in FDA-12 list") |
| ap.add_argument("--smiles-col", default=None, |
| help="SMILES column in the generations CSV " |
| "(default: auto-detect raw_smiles/smiles/" |
| "canonical_smiles)") |
| ap.add_argument("--out", default=None, |
| help="output CSV (default: <input>_validated.csv)") |
| args = ap.parse_args(argv) |
|
|
| gen_rows = _read_csv(args.generations_csv) |
| if not gen_rows: |
| print("empty generations CSV", file=sys.stderr) |
| return 1 |
| cols = gen_rows[0].keys() |
| smiles_col = args.smiles_col |
| if smiles_col is None: |
| for cand in ('raw_smiles', 'smiles', 'canonical_smiles'): |
| if cand in cols: |
| smiles_col = cand |
| break |
| if smiles_col is None or smiles_col not in cols: |
| print(f"no SMILES column found among {list(cols)}", file=sys.stderr) |
| return 1 |
| raw = [r[smiles_col] for r in gen_rows] |
|
|
| ref_rows = _read_csv(args.reference_csv) |
| reference_canon = canonical_set(r['smiles'] for r in ref_rows) |
| print(f"reference set: {len(ref_rows)} rows -> " |
| f"{len(reference_canon)} canonical compounds") |
|
|
| if args.fda_csv: |
| fda_rows = _read_csv(args.fda_csv) |
| fda_pairs = [(r.get('name', ''), r['smiles']) for r in fda_rows] |
| else: |
| fda_pairs = FDA12 |
| fda_canon = canonical_set(s for _, s in fda_pairs) |
| print(f"FDA reference: {len(fda_pairs)} drugs -> " |
| f"{len(fda_canon)} canonical compounds") |
|
|
| rows, funnel = validate_pool(raw, reference_canon, fda_canon) |
| print_funnel(funnel) |
|
|
| out_path = args.out or args.generations_csv.replace( |
| '.csv', '_validated.csv') |
| fields = ['raw_smiles', 'canonical_smiles', 'valid', 'n_fragments', |
| 'cleaned_smiles', 'novel_vs_finetune', 'novel_vs_fda12'] |
| with open(out_path, 'w', newline='') as fh: |
| w = csv.DictWriter(fh, fieldnames=fields) |
| w.writeheader() |
| w.writerows(rows) |
| print(f"wrote {out_path} ({len(rows)} rows)") |
| return 0 |
|
|
|
|
| if __name__ == '__main__': |
| sys.exit(main()) |
|
|
