#!/usr/bin/env python3 """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.*') # The FDA-12 approved gliptins — verbatim from the project generation # scripts (names + SMILES as used for novelty-vs-FDA12 reporting). 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: _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())