"""PINO demo Space — blend → predicted odor-pyramid (top/mid/base) trajectory. Physics-Informed Neural Olfaction (PINO). Given a set of ingredients and their concentrations, the model predicts how the perceived odor pyramid evolves over time as the blend dries down (Dortmund-UNIFAC evaporation physics + a Transformer over molecular embeddings). This demo serves the VALIDATED path: multi-ingredient blend trajectories. Single-molecule descriptor lookup is intentionally not exposed (out-of-domain for the trajectory-trained objective head). Checkpoints are pulled from mattbitzesty/pino-pimt-representation-ab. """ from __future__ import annotations import json import os from functools import lru_cache from pathlib import Path import gradio as gr import numpy as np import torch HF_TOKEN = os.environ.get("HF_TOKEN") os.environ.setdefault("HF_HUB_ENABLE_XET", "0") os.environ.setdefault("HF_HUB_DISABLE_XET", "1") AB_REPO = "mattbitzesty/pino-pimt-representation-ab" SOURCE_REPO = "mattbitzesty/pino-source-code" DATASET_REPO = "mattbitzesty/pino-synthetic-dataset-v3" # Arm -> (checkpoint filename, structural_source, objective_dim) # v2 arms (corrected labels) are preferred; fall back to the v1 gpu checkpoint. ARMS = { "OpenPOM (genuine 256-d)": ("pimt_ab_openpom_256_v2.pt", "openpom_256", 138), "Morgan (138-d baseline)": ("pimt_ab_morgan_v2.pt", "morgan", 138), "OpenPOM + all-tags (575-d)": ("pimt_ab_pom_alltags_v2.pt", "pom_alltags", 575), } V1_FALLBACK = { "OpenPOM (genuine 256-d)": "pimt_ab_openpom_256_gpu.pt", "Morgan (138-d baseline)": "pimt_ab_morgan_gpu.pt", } DEFAULT_GENRES = ["citrus_cologne", "fougere", "floral_woody", "amber_oriental", "wildcard"] @lru_cache(maxsize=1) def _bootstrap(): """Download source package + small data, install import path. Returns (src_dir, data_dir).""" from huggingface_hub import snapshot_download src = snapshot_download(SOURCE_REPO, repo_type="model", token=HF_TOKEN) import sys sys.path.insert(0, str(Path(src) / "src")) sys.path.insert(0, str(Path(src))) # for scripts/ imports if needed return src, Path(src) / "data" @lru_cache(maxsize=1) def _dataset_file(name: str) -> str: """Download a file from the canonical dataset repo (large data lives there, not source).""" from huggingface_hub import hf_hub_download return hf_hub_download(DATASET_REPO, name, repo_type="dataset", token=HF_TOKEN) @lru_cache(maxsize=1) def _materials(): """Material records for the picker: list of dicts with name/smiles/cas/aliases/odor/inchikey.""" _, data_dir = _bootstrap() out = [] # Prefer the dataset repo's material_profiles (source repo only carries v11.jsonl). candidates = [] for fn in ["material_profiles_v11_5.jsonl", "material_profiles_v11_4.jsonl"]: try: candidates.append(_dataset_file(fn)) except Exception: pass p = data_dir / "material_profiles_v11.jsonl" if p.exists(): candidates.append(str(p)) for path in candidates: for line in Path(path).read_text().splitlines(): if not line.strip(): continue try: r = json.loads(line) except Exception: continue sm = r.get("smiles") if not sm: continue name = (r.get("name") or (r.get("aliases") or [""])[0] or r.get("cas") or "?").strip() odor = r.get("odor_description") or " ".join(r.get("odor_descriptions") or []) out.append({ "name": name, "smiles": sm, "cas": r.get("cas") or "", "aliases": [a for a in (r.get("aliases") or []) if a], "odor": odor, "inchikey": r.get("inchikey") or "", }) if out: break # de-dup by smiles seen, res = set(), [] for m in out: if m["smiles"] in seen: continue seen.add(m["smiles"]) res.append(m) res.sort(key=lambda m: m["name"].lower()) return res # Curated perfumery trade-name → common-name synonyms. Keys/values lowercase. _PERFUMERY_SYNONYMS = { "otne": "iso e super", "iso e super": "otne", "hedione": "methyl dihydrojasmonate", "mdj": "methyl dihydrojasmonate", "methyl dihydrojasmonate": "hedione", "lyral": "hydroxymethylpentylcyclohexene carboxaldehyde", "lilial": "butylphenyl methylpropional", "galaxolide": "hexamethylindanopyran", "hhb": "galaxolide", "tonalide": "acetyl tetramethyl tetralin", "ahtn": "tonalide", "ambrox": "ambroxide", "ambroxan": "ambroxide", "ambrofix": "ambroxide", "timberol": "trimethylnitrobicycloheptane", "calone": "watermelon ketone", "aldehyde c12": "lauric aldehyde", "aldehyde c11": "undecanal", "aldehyde c10": "decanal", "aldehyde c9": "nonanal", "aldehyde c14": "undecalactone", "aldehyde c16": "ethyl methylphenylglycidate", "aldehyde c18": "nonalactone", "so-called aldehyde c14": "undecalactone", "peach aldehyde": "undecalactone", "strawberry aldehyde": "ethyl methylphenylglycidate", "empg": "ethyl methylphenylglycidate", "vanilla": "vanillin", "musk t": "ethylene brassylate", "ethylene brassylate": "musk t", "exaltolide": "pentadecanolide", "ambrettolide": "hexadecenolide", "civetone": "cycloheptadecenone", "ligustral": "triplal", "triplal": "ligustral", "florhydral": "phenylpropanol dimethyl acetal", "dihydromyrcenol": "dhm", "dhm": "dihydromyrcenol", "orange flower ether": "methyl naphthyl ether", "nerolin": "methyl naphthyl ether", "yara yara": "methyl naphthyl ether", "celestolide": "acetyl tert-butyl dimethyl indane", "phantolide": "acetyl hexamethyl indane", "traseolide": "acetyl tetramethyl indane", "cashmeran": "dpm musk", "dpm musk": "cashmeran", "helvetolide": "muscone analog", "romandolide": "ambrette musk", "nirvanolide": "pentadecenolide", "habanolide": "pentadecanolide", "globalide": "pentadecenolide", "serenolide": "oxacycloheptadecenone", "cosmone": "methyl cyclotetradecenone", "muscopyridine": "musk pyridine", "iso e": "iso e super", "bacdanol": "sandalore", "sandalore": "bacdanol", "ebanol": "methyl branched cyclohexanol", "polysantol": "dimethyl cyclohexenyl pentenol", "javanol": "methyl cyclohexene propanol", "firsantol": "dimethyl norbornene ethanol", "santiff": "sandalore", "myrtenol": "myrtenol", "koavone": "acetyl carene", "florosa": "florol", "florol": "florosa", "lilyflore": "florol", "muguet aldehyde": "cyclamen aldehyde", "cyclal c": "cyclamen aldehyde", "bourgeonal": "methyl propyl phenylpropanal", "adoxal": "tridecene nitrile", "precyclemone b": "methyl octine carbonate", "moc": "methyl octine carbonate", "folione": "methyl heptine carbonate", "mhc": "methyl heptine carbonate", "vert de muguet": "hexenyl salicylate", "cis-3-hexenyl salicylate": "hexenyl salicylate", "leaf alcohol": "cis-3-hexenol", "hexenol": "cis-3-hexenol", "pyrus": "allyl cyclohexylpropionate", "allyl caproate": "allyl hexanoate", "fruity note": "ethyl butyrate", "apple ester": "ethyl butyrate", "pear ester": "ethyl trans-2 cis-4-decadienoate", "cognac oil": "wine lees oil", "iris": "irone", "orris": "irone", "violet leaf": "methyl octine carbonate", "violet": "ionone", "methyl ionone": "ionone", "gamma-methyl ionone": "isoraldeine", "isoraldeine": "methyl ionone", "hedione hc": "methyl dihydrojasmonate", "paradisamide": "methyl dihydrojasmonate", "kharismal": "methyl dihydrojasmonate", "clarycet": "sclareol", "ambrox super": "ambroxide", "cetalox": "ambroxide", "ambermor": "ambroxide", "amberlyn": "ambroxide", "ambermax": "trimethyl cyclohexene butanol", "timberol": "trimethyl cyclohexene butanol", "norlimbanol": "trimethyl cyclohexene butanol", } def _norm(s: str) -> str: return "".join(ch for ch in s.lower() if ch.isalnum() or ch.isspace()).strip() def _tokens(s: str) -> set: return {t for t in _norm(s).split() if len(t) > 1} @lru_cache(maxsize=1) def _search_index(): """Map material -> set of searchable tokens (name+aliases+cas+inchikey+odor+synonyms).""" mats = _materials() idx = [] for i, m in enumerate(mats): toks = set() toks |= _tokens(m["name"]) for a in m["aliases"]: toks |= _tokens(a) if m["cas"]: toks.add(m["cas"].lower()) toks |= _tokens(m["cas"]) if m["inchikey"]: toks.add(m["inchikey"].lower()) toks |= _tokens(m.get("odor", "")) idx.append(toks) return idx def _display_label(m) -> str: cas = m["cas"] or "?" return f"{m['name']} [{cas}]" def search_materials(query: str, limit: int = 25): """Search materials by name/alias/CAS/InChIKey/odor-term, with synonym expansion. Returns list of display labels best-matching the query. Empty query returns the first `limit` materials (catalog browsing). """ mats = _materials() idx = _search_index() q = _norm(query or "") if not q: return [_display_label(m) for m in mats[:limit]] # synonym expansion: map query terms through the curated map q_terms = _tokens(q) expanded = set(q_terms) for t in list(q_terms): syn = _PERFUMERY_SYNONYMS.get(t) if syn: expanded |= _tokens(syn) # also try whole-phrase synonym lookup phrase_syn = _PERFUMERY_SYNONYMS.get(q) if phrase_syn: expanded |= _tokens(phrase_syn) q_lower = q.lower() scored = [] for i, toks in enumerate(idx): m = mats[i] score = 0 # exact CAS / inchikey match = strongest if m["cas"] and q_lower == m["cas"].lower(): score += 100 if m["inchikey"] and q_lower == m["inchikey"].lower(): score += 100 # name substring / prefix name_n = _norm(m["name"]) if q and q in name_n: score += 60 if name_n.startswith(q) else 40 # alias substring if any(q in _norm(a) for a in m["aliases"]): score += 35 # token overlap (with synonym expansion) overlap = len(expanded & toks) score += overlap * 10 if score > 0: scored.append((score, i)) scored.sort(key=lambda x: (-x[0], mats[x[1]]["name"].lower())) return [_display_label(mats[i]) for _, i in scored[:limit]] @lru_cache(maxsize=1) def _label_to_material(): """display label -> (smiles, cas) for prediction.""" return {_display_label(m): (m["smiles"], m["cas"]) for m in _materials()} @lru_cache(maxsize=1) def _vocab(): _, data_dir = _bootstrap() return json.loads((data_dir / "pyrfume_vocabulary.json").read_text())["vocabulary"] @lru_cache(maxsize=8) def _load_arm(arm_label: str): """Load model+heads+engine for an arm; returns (model, heads, engine, resolved_ckpt).""" from huggingface_hub import hf_hub_download src, _ = _bootstrap() from pino.pimt_model import PhysicsInformedMixtureTransformer # noqa from pino.heads import PIMTHeads # noqa from pino.embeddings import OlfactoryEmbeddingEngine # noqa import pino.embeddings as _emb ckpt_name, structural_source, objective_dim = ARMS[arm_label] try: ckpt_path = hf_hub_download(AB_REPO, ckpt_name, repo_type="model", token=HF_TOKEN) resolved = ckpt_name except Exception: fb = V1_FALLBACK.get(arm_label) if not fb: raise ckpt_path = hf_hub_download(AB_REPO, fb, repo_type="model", token=HF_TOKEN) resolved = fb + " (v1)" # Genuine-POM arms need the embedding assets; fetch them from the dataset repo # into the location the engine expects (/artifacts/pom). if structural_source in ("openpom_256", "pom_alltags"): pom_dir = Path(src) / "artifacts" / "pom" pom_dir.mkdir(parents=True, exist_ok=True) for f in ("pom_embeddings.npy", "pom_index.json"): dst = pom_dir / f if not dst.exists(): got = hf_hub_download(DATASET_REPO, f"artifacts/pom/{f}", repo_type="dataset", token=HF_TOKEN) dst.write_bytes(Path(got).read_bytes()) _emb._POM_DIR = pom_dir # ensure engine resolves the fetched assets engine_source = "openpom_256" if structural_source == "pom_alltags" else structural_source engine = OlfactoryEmbeddingEngine(structural_source=engine_source) ckpt = torch.load(ckpt_path, map_location="cpu", weights_only=False) msd = ckpt["model_state_dict"] embedding_dim = int(msd["input_proj.weight"].shape[1]) hidden_dim = int(msd["input_proj.weight"].shape[0]) num_layers = max(int(k.split(".")[2]) for k in msd if k.startswith("encoder.layers.")) + 1 model = PhysicsInformedMixtureTransformer( embedding_dim=embedding_dim, state_dim=2, hidden_dim=hidden_dim, num_heads=4, num_layers=num_layers, ) model.load_state_dict(msd) model.eval() heads = PIMTHeads(hidden_dim=int(hidden_dim), objective_dim=objective_dim) heads.load_state_dict(ckpt["heads_state_dict"], strict=False) heads.eval() return model, heads, engine, resolved, objective_dim def _pyramid_for_ingredient(model, heads, engine, smiles, cas, objective_dim): """Encode one molecule through the trajectory model; return (3, objective_dim) pyramid.""" z = torch.from_numpy(engine.get_embedding(smiles, cas=cas)).float() tokens = z.unsqueeze(0).unsqueeze(0) # (1, S=1, E) physics = torch.zeros(1, 1, 1, 2) # (1, T=1, S=1, 2) with torch.no_grad(): latent = model(tokens, physics) out = heads(latent, physics) pyr = out["objective"].squeeze(0).numpy() # (3, objective_dim) return pyr def _row_pair(r): """Extract (label, conc) from a Dataframe row regardless of shape. Gradio 6 may deliver rows as lists, tuples, numpy arrays, or dicts keyed by header name. Normalise to (str label, float conc); return ("", 0.0) on junk. """ try: if isinstance(r, dict): vals = list(r.values()) label = vals[0] if vals else "" conc = vals[1] if len(vals) > 1 else 0 else: label = r[0] conc = r[1] if len(r) > 1 else 0 label = "" if label is None else str(label) if hasattr(conc, "item"): # numpy scalar conc = conc.item() conc = float(conc) if conc not in (None, "") else 0.0 return label, conc except (TypeError, ValueError, IndexError): return "", 0.0 def predict(arm_label, rows, progress=gr.Progress()): # Gradio 6 delivers a Dataframe input as a pandas DataFrame server-side; # normalise to a list of rows first (never use `rows or []` on a DataFrame). if rows is None: row_list = [] elif hasattr(rows, "values") and hasattr(rows, "columns"): # pandas DataFrame row_list = rows.values.tolist() else: row_list = list(rows) pairs = [_row_pair(r) for r in row_list] pairs = [(l, c) for l, c in pairs if l and c > 0] if not pairs: return None, "Add at least one ingredient (with a concentration > 0)." try: model, heads, engine, resolved, objective_dim = _load_arm(arm_label) except Exception as e: return None, f"Model load failed: {e}" mats = _label_to_material() total_w = sum(c for _, c in pairs) or 1.0 blend = np.zeros(objective_dim, dtype=np.float64) # (objective_dim,) tier-summed tier_blend = np.zeros((3, objective_dim), dtype=np.float64) used = [] skipped = [] for label, conc in progress.tqdm(pairs, desc="Encoding"): if label not in mats: skipped.append(label.split(" [")[0]) continue sm, cas = mats[label] pyr = _pyramid_for_ingredient(model, heads, engine, sm, cas, objective_dim) # (3, D) w = conc / total_w tier_blend += w * pyr used.append(label.split(" [")[0]) if not used: return None, ("None of the selected ingredients matched the catalog. " "Pick ingredients from the Quick-add list." + (f" Unmatched: {', '.join(skipped[:5])}" if skipped else "")) # Tier masses = perceived intensity over dry-down tier_mass = tier_blend.sum(axis=1) # (3,) # Top descriptors overall (blend across tiers) overall = tier_blend.sum(axis=0) # (D,) if objective_dim == 138: vocab = _vocab() top_idx = np.argsort(-overall)[:12] top_desc = [(vocab[i], float(overall[i])) for i in top_idx if overall[i] > 0] else: top_desc = [("(575-dim tag space — descriptor names n/a)", 0.0)] # Plot: pyramid trajectory -> render to PNG bytes for a plain gr.Image (reliable in Gradio 6). import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt import io fig, ax = plt.subplots(figsize=(6, 3.6), dpi=110) x = [0, 1, 2] labels = ["Top (0–15 min)", "Heart (15–60 min)", "Base (1h+)"] m = tier_mass / (tier_mass.max() or 1.0) ax.fill_between(x, 0, m, alpha=0.35, color="#7c5cbf") ax.plot(x, m, "o-", color="#5a3d99", lw=2.5, ms=8) ax.set_xticks(x); ax.set_xticklabels(labels, fontsize=9) ax.set_ylabel("Relative perceived intensity"); ax.set_ylim(0, 1.1) ax.set_title(f"Predicted dry-down pyramid — {resolved}", fontsize=10) ax.grid(alpha=0.25) fig.tight_layout() # Render to base64 data-URL embedded in HTML — no file serving, always renders in Gradio 6. import io as _io, base64 as _b64 buf = _io.BytesIO() fig.savefig(buf, format="png", bbox_inches="tight", dpi=110) plt.close(fig) b64 = _b64.b64encode(buf.getvalue()).decode("ascii") img = (f'
' f'
') lines = [f"**Model:** {resolved}", "", f"**Blend ({len(used)} ingredients):** {', '.join(used)}", "", "**Tier balance (relative intensity):** " f"top {m[0]:.2f} · heart {m[1]:.2f} · base {m[2]:.2f}", ""] if top_desc and objective_dim == 138: lines.append("**Dominant predicted descriptors:**") lines += [f"- {d} ({v:.2f})" for d, v in top_desc[:10]] return img, "\n".join(lines) def _material_labels(): return [_display_label(m) for m in _materials()] with gr.Blocks(title="PINO — Physics-Informed Neural Olfaction") as demo: gr.Markdown( "# PINO — Physics-Informed Neural Olfaction\n" "Predict how a fragrance blend's **odor pyramid (top / heart / base)** evolves as it dries down.\n\n" "Search ingredients (by name, trade name / synonym, CAS, or odor note), set concentrations, " "choose a representation arm, and the model renders the predicted dry-down. " "Validated path: multi-ingredient blend trajectories." ) with gr.Row(): arm = gr.Radio(list(ARMS.keys()), value="OpenPOM (genuine 256-d)", label="Representation arm") ingredient_table = gr.Dataframe( headers=["ingredient", "concentration (parts)"], datatype=["str", "number"], row_count=(4, "dynamic"), col_count=(2, "fixed"), label="Blend formula", interactive=True, ) with gr.Row(): # Live search: type to filter, results populate the picker below. search_box = gr.Textbox( label="Search ingredients", placeholder="e.g. 'iso e super', 'hedione', 'ambroxan', '106-22-9', 'sandalwood'…", interactive=True, scale=3, ) with gr.Row(): results = gr.Dropdown( choices=[], label="Matches (pick one, then Add)", interactive=True, scale=3, ) add_parts = gr.Number(value=10, label="parts", scale=1) add_btn = gr.Button("Add", scale=1) status = gr.Markdown() run = gr.Button("Predict dry-down", variant="primary") out_plot = gr.HTML(label="Odor-pyramid trajectory") out_text = gr.Markdown() def _do_search(query): try: labels = search_materials(query, limit=30) if not labels: return gr.update(choices=[], value=None), "No matches." return gr.update(choices=labels, value=labels[0]), f"{len(labels)} match(es)." except Exception as e: return gr.update(choices=[], value=None), f"Search failed: {e}" def _add_row(current, label, parts): # Normalise current (list-of-lists, DataFrame, or None) then append. rows = [] if current is not None: if hasattr(current, "values") and hasattr(current, "columns"): rows = current.values.tolist() else: rows = [list(r) for r in current] if label: p = parts.item() if hasattr(parts, "item") else parts rows.append([str(label), float(p or 0)]) # Return an explicit DataFrame so the Dataframe component re-renders cleanly. import pandas as pd return pd.DataFrame(rows, columns=["ingredient", "concentration (parts)"]) # Search as the user types (and on submit). Populate initial browse list. search_box.change(_do_search, search_box, [results, status]) search_box.submit(_do_search, search_box, [results, status]) demo.load(lambda: _do_search(""), None, [results, status]) add_btn.click(_add_row, [ingredient_table, results, add_parts], ingredient_table) run.click(predict, [arm, ingredient_table], [out_plot, out_text]) gr.Markdown( "*Model: PhysicsInformedMixtureTransformer over OpenPOM / Morgan embeddings + Dortmund-UNIFAC " "evaporation physics. Trained on 5,708 empirical formulations (molecule-disjoint holdout). " "Single-molecule descriptor lookup intentionally not exposed (out-of-domain).*" ) if __name__ == "__main__": demo.queue().launch(server_name="0.0.0.0", server_port=int(os.environ.get("PORT", 7860)), show_error=True)