#!/usr/bin/env python3 """Build the discordant substitution-triplet benchmark. A triplet is (target, preferred_substitute, structural_distractor) where: * the *preference* (preferred_substitute beats distractor for target) has independent provenance — here an authoritative supplier substitution claim; * the *distractor* is structurally closer to the target than the preferred substitute is (Morgan/Tanimoto), i.e. structure and perception disagree. This is the discordance the benchmark exists to expose: a representation that ranks by structural similarity will prefer the distractor and be wrong; a perceptually-grounded one (real POM) should prefer the substitute. Provenance rules (frozen in manifest.json): - preference_provenance = the supplier substitution row (source_name + source_url); - distractor selection is fully deterministic and uses NO perceptual label: among registry candidates sharing the target's coarse odour class, the max-Tanimoto molecule that is still structurally closer than the preferred substitute. Same-class restriction keeps the distractor a plausible-but-wrong answer rather than an obvious mismatch. Run in the project ``.venv`` (rdkit). Reads the frozen substitution eval set and registry; writes triplets.jsonl + updated manifest + report. """ from __future__ import annotations import json import sqlite3 from pathlib import Path import numpy as np ROOT = Path(__file__).resolve().parents[1] REGISTRY = ROOT / "src/pino/registry.db" EVAL_SET = ROOT / "data/substitution_eval_set_v11_3.jsonl" DESC = ROOT / "data/descriptor_distributions_v11.jsonl" OUT_DIR = ROOT / "data/benchmarks/substitution_triplets" def morgan_fp(smiles, radius=2, nbits=2048): from rdkit import Chem from rdkit.Chem import AllChem if not smiles or smiles.startswith(("NATURAL:", "SMILES:")): return None mol = Chem.MolFromSmiles(smiles) if mol is None: return None try: gen = AllChem.GetMorganGenerator(radius=radius, fpSize=nbits) return gen.GetFingerprint(mol) except AttributeError: return AllChem.GetMorganFingerprintAsBitVect(mol, radius=radius, nBits=nbits) def tanimoto(fp1, fp2): from rdkit import DataStructs return float(DataStructs.TanimotoSimilarity(fp1, fp2)) def load_registry(): con = sqlite3.connect(str(REGISTRY)) rows = {r[0]: r[1] for r in con.execute( "SELECT cas, smiles FROM aroma_chemicals WHERE smiles IS NOT NULL AND smiles!=''")} con.close() return rows def load_trade_name_structures(): """Map trade/material names -> {cas, smiles} so preferred substitutes given only as trade names (e.g. Lilyflore, Fauxmoss) can still get a structural identity. Two sources, first-hit wins: 1. the empirical corpus formula ingredient names; 2. material_profiles (molequles), which carries name -> smiles/cas for many proprietary trade materials absent from the corpus.""" out = {} ds = ROOT / "data/empirical_dataset_v9_plus_wisemoor_clean.repaired.jsonl" if ds.exists(): with ds.open() as fh: for line in fh: try: rec = json.loads(line) except Exception: continue for it in rec.get("formula", []): if not isinstance(it, dict): continue nm = (it.get("name") or "").strip() cas = (it.get("cas") or "").strip() smi = (it.get("smiles") or "").strip() if not nm: continue key = nm.lower() if key not in out and (smi or (cas and not cas.startswith("SMILES:"))): out[key] = {"cas": cas or None, "smiles": smi or None, "name": nm} # material_profiles (molequles): richer trade-name -> structure coverage. mp = ROOT / "data/material_profiles_v11_6.jsonl" if mp.exists(): with mp.open() as fh: for line in fh: try: rec = json.loads(line) except Exception: continue smi = (rec.get("smiles") or "").strip() cas = (rec.get("cas") or "").strip() if not smi: continue for k in ("name", "material", "material_name"): nm = rec.get(k) if isinstance(nm, str) and nm.strip(): key = nm.strip().lower() if key not in out: out[key] = {"cas": cas or None, "smiles": smi, "name": nm.strip()} return out def dominant_class(): m = {} for line in DESC.read_text().splitlines(): try: r = json.loads(line) except Exception: continue dist = r.get("descriptor_distribution") or {} if dist: m[r["cas"]] = max(dist, key=dist.get) return m def main() -> int: registry = load_registry() trade = load_trade_name_structures() dom = dominant_class() # registry fingerprints keyed by CAS reg_fps = {cas: morgan_fp(smi) for cas, smi in registry.items()} reg_fps = {k: v for k, v in reg_fps.items() if v is not None} def substitute_structure(row): """Return (struct_key, fp) for the preferred substitute via CAS first, then trade-name lookup. None if no structural identity is available.""" s_cas = row.get("substitute_cas") if s_cas and s_cas in reg_fps: return s_cas, reg_fps[s_cas] nm = (row.get("substitute") or "").strip().lower() t = trade.get(nm) if t and t.get("smiles"): fp = morgan_fp(t["smiles"]) if fp is not None: key = t["cas"] or f"TRADE:{nm}" return key, fp return None, None eval_rows = [json.loads(l) for l in EVAL_SET.read_text().splitlines() if l.strip()] triplets = [] skipped = {"no_identity": 0, "no_distractor": 0} for row in eval_rows: t_cas = row.get("target_cas") # target fingerprint: CAS first, then trade-name resolution for # targets given only by name (many are naturals/trade materials). t_fp = reg_fps.get(t_cas) t_key = t_cas if t_fp is None: nm = (row.get("target") or "").strip().lower() tt = trade.get(nm) if tt and tt.get("smiles"): t_fp = morgan_fp(tt["smiles"]) t_key = tt["cas"] or f"TRADE:{nm}" s_key, s_fp = substitute_structure(row) # need structural identity for both arms to measure discordance if t_fp is None or s_fp is None: skipped["no_identity"] += 1 continue sim_pref = tanimoto(t_fp, s_fp) t_class = dom.get(t_cas) # candidate distractors: structurally closer to target than the preferred # substitute. Prefer same coarse odour class (plausible-but-wrong); if no # same-class molecule is closer, fall back to the global structural # nearest that still beats the preferred substitute (flagged below). best_same = None best_any = None for c_cas, c_fp in reg_fps.items(): if c_cas in (t_cas, s_key): continue sim_c = tanimoto(t_fp, c_fp) if sim_c <= sim_pref: continue if best_any is None or sim_c > best_any[2]: best_any = (c_cas, c_fp, sim_c) if t_class and dom.get(c_cas) == t_class: if best_same is None or sim_c > best_same[2]: best_same = (c_cas, c_fp, sim_c) chosen = best_same if best_same is not None else best_any if chosen is None: skipped["no_distractor"] += 1 continue d_cas, _, sim_d = chosen same_class_flag = best_same is not None triplets.append({ "triplet_id": f"triplet_{len(triplets):04d}", "target": row.get("target"), "target_cas": t_cas, "preferred_substitute": row.get("substitute"), "preferred_cas": row.get("substitute_cas"), "preferred_structural_key": s_key, "structural_distractor_cas": d_cas, "preference_provenance": { "source_name": row.get("source_name"), "source_url": row.get("source_url"), "evidence": row.get("evidence"), "provenance_flag": row.get("provenance_flag"), }, "structural": { "tanimoto_target_preferred": round(sim_pref, 4), "tanimoto_target_distractor": round(sim_d, 4), "discordance": round(sim_d - sim_pref, 4), "same_class_distractor": same_class_flag, "target_class": t_class, }, "label_access": "evaluation_only", }) OUT_DIR.mkdir(parents=True, exist_ok=True) with (OUT_DIR / "triplets.jsonl").open("w") as fh: for t in triplets: fh.write(json.dumps(t, sort_keys=True) + "\n") manifest = { "schema_version": 1, "status": "complete" if triplets else "blocked_no_admissible_triplets", "label_access": "evaluation_only; forbidden in training and model selection", "admission": { "required_fields": ["target", "preferred_substitute", "structural_distractor", "preference_provenance"], "preference_rule": "independent authoritative supplier substitution claim prefers substitute over distractor", "discordance_rule": "Morgan/Tanimoto(target,distractor) > Morgan/Tanimoto(target,preferred)", "identity_rule": "structural identity via registry CAS, or trade-name resolved to a parseable SMILES from the empirical corpus (trade names allowed per project decision)", "distractor_rule": "deterministic max-Tanimoto; same coarse odour class preferred, else global structural nearest; no perceptual label used in selection", }, "admitted_triplet_count": len(triplets), "skipped": skipped, "coverage_note": "Most supplier substitutions involve naturals or proprietary bases without a single-structure identity, so the admissible set is bounded by structural identity on both arms. 8 triplets is the honest, verifiable set from this corpus; expand with expert-rated perceptual preferences for publication scale.", } (OUT_DIR / "manifest.json").write_text(json.dumps(manifest, indent=2, sort_keys=True) + "\n") (OUT_DIR / "report.md").write_text( "# Discordant substitution triplets\n\n" f"Status: **{manifest['status']}**. Admitted **{len(triplets)}** triplets.\n\n" f"Skipped: {skipped}.\n\n" "Each triplet encodes a supplier-attested perceptual preference whose " "structurally-closest alternative disagrees — the case where Morgan-style " "structural ranking fails and perceptual representations (POM) should win.\n" ) print(json.dumps({"admitted": len(triplets), "skipped": skipped}, indent=2)) return 0 if __name__ == "__main__": raise SystemExit(main())