"""SupraDashboard Gradio UI."""
from __future__ import annotations
import datetime
import html
import os
try:
from dotenv import load_dotenv
load_dotenv()
except ImportError:
pass
import gradio as gr
from data import FEATURES, get_record, guest_choices, host_record, load_features
from feedback import RATINGS
from inference import available_models, backend_label, default_model, healthcheck
from prompts import ALL_FEATURE_COLS, FEATURE_PRESET_NAMES, preset_text, user_prompt_html
from store import (
db_status,
export_reviews,
fmt_ts,
get_sample,
list_samples,
norm_model,
norm_prompt,
pull_remote,
review_map,
save_review,
)
from ui.cache import _card_from_results, _pred_table, _reason_from_results, predict
from viz.renderers import _reason_panel, _reason_placeholder, _status_line, _tldr_card
from viz.theme import _CSS, _FONT_HEAD, _theme
# --------------------------------------------------------------------------- #
# Feedback #
# --------------------------------------------------------------------------- #
def submit_feedback(
inchikey, guest_name, model, prompt_version, rating, comment, request: gr.Request = None
):
if not rating:
return "
⚠ Pick a rating before submitting.
"
if not guest_name:
return "
⚠ Run a prediction first, then review it.
"
# With Gradio auth= on, the logged-in username is on the request; otherwise "".
reviewer = (getattr(request, "username", None) or "") if request is not None else ""
ts = datetime.datetime.now(datetime.timezone.utc).isoformat()
rid = save_review(
ts=ts,
inchikey=inchikey or "",
guest_name=guest_name or "",
model=model or "",
prompt_version=prompt_version or "",
rating=rating,
comment=comment or "",
reviewer=reviewer,
)
return f"
✓ Saved review #{rid}. Thank you — pick the next guest.
"
_FEEDBACK_COLS = [
"Timestamp", "Guest", "InChIKey", "Model", "Prompt",
"Rating", "Comment", "Reviewer",
]
def _load_export():
# Newest-first; collapse to one row per (guest, model, prompt, reviewer) so a
# reviewer's edited review shows once. `keys` mirrors the visible rows so a
# row click can reopen that exact record in Review.
rows = export_reviews()
if not rows:
return [], "No feedback collected yet.", []
data, keys, seen = [], [], set()
for r in rows:
ik = r.get("inchikey", "")
model = norm_model(r.get("model"))
pv = norm_prompt(r.get("prompt_version"))
reviewer = r.get("reviewer", "") or ""
k = (ik, model, pv, reviewer)
if k in seen:
continue
seen.add(k)
data.append([
fmt_ts(r.get("ts")), r.get("guest_name", ""), ik, model, pv,
r.get("rating", ""), r.get("comment", ""), reviewer,
])
keys.append({
"guest_name": r.get("guest_name", ""), "inchikey": ik,
"model": model, "prompt_version": pv,
"rating": r.get("rating", ""), "comment": r.get("comment", ""),
})
return data, f"{len(data)} review(s) collected.", keys
def _open_feedback(keys, evt: gr.SelectData):
"""Open the clicked Feedback row's guest in Review, prefilling its rating/comment
so the reviewer can edit and re-submit (save_review upserts the same row)."""
idx = evt.index[0] if isinstance(evt.index, (list, tuple)) else evt.index
if not keys or idx is None or idx < 0 or idx >= len(keys):
return (gr.update(),) * 14
k = keys[idx]
inchikey, model, prompt_version = k["inchikey"], k["model"], k["prompt_version"]
guest_name = k["guest_name"]
sample = get_sample(inchikey, model, prompt_version)
results = (sample.get("results") or {}) if sample else {}
return (
_pred_table(results),
_reason_from_results(results),
_card_from_results(results, "physics", "physics"),
_card_from_results(results, "chemistry", "chemistry"),
_status_line(f"Editing your review of {guest_name} — update and re-submit below.", kind="done"),
inchikey,
guest_name,
model,
prompt_version,
guest_name,
model,
gr.update(value=k["rating"] or None),
gr.update(value=k["comment"] or ""),
gr.update(selected="review"),
)
# --------------------------------------------------------------------------- #
# Merged Data board — ranked discovery + Label-Studio-style column chooser #
# --------------------------------------------------------------------------- #
# Core (non-feature) columns of the per-guest record, in display order. "#" and
# "Guest" are pinned (always shown); the rest are toggled by the column chooser.
_PINNED_COLS = ["#", "Guest"]
_CORE_COLS = [
"Host", "Pred logKa", "BatchDate", "True logKa", "Novelty", "Scaffold", "Tmax known",
"SMILES", "Why", "Model", "Source", "Batch", "Prompt", "InChIKey", "Reviews",
]
# Feature display labels (the 22 from data.FEATURES), appended after the core set.
_FEATURE_COLS = [label for label, _col in FEATURES]
# Full ordered column universe and the default (current Candidate set).
_ALL_COLS = _PINNED_COLS + _CORE_COLS + _FEATURE_COLS
_CHOOSABLE_COLS = _CORE_COLS + _FEATURE_COLS # everything except the pinned pair
_DEFAULT_COLS = ["#", "Guest", "Host", "Pred logKa", "BatchDate", "SMILES"]
# Synthetic processing date per docking batch: batch1 = 2026-06-15, +1 day each.
_BATCH_EPOCH = datetime.date(2026, 6, 15)
def _batch_date(batch_tag) -> str:
"""Map a batch tag to its processing date (MM/DD/YYYY): batch1 -> 06/15/2026,
then one day per batch. Accepts plain "batch5" and the legacy "v5·batch1"
prompt-label form. Blank for batch0/Default/unknown."""
s = str(batch_tag or "")
if "·" in s: # legacy "·" label -> take the batch part
s = s.split("·")[-1]
if not s.startswith("batch"):
return ""
try:
n = int(s[len("batch"):])
except ValueError:
return ""
if n < 1:
return ""
return (_BATCH_EPOCH + datetime.timedelta(days=n - 1)).strftime("%m/%d/%Y")
def _batch_date_choices():
"""Dropdown choices for the batch (date) selector: 'All batches' + every
distinct batch processing-date present in the store, newest first."""
dates = set()
try:
for r in list_samples():
d = _batch_date(r.get("batch") or "")
if d:
dates.add(d)
except Exception: # noqa: BLE001 — never block UI build on a store hiccup
return ["All batches"]
ordered = sorted(dates, key=lambda s: datetime.datetime.strptime(s, "%m/%d/%Y"),
reverse=True)
return ["All batches"] + ordered
# Columns offered in the server-side "Sort by" control (the per-column header
# menu only reorders the current page; this sorts the whole filtered set).
_SORTABLE_COLS = ["Pred logKa", "True logKa", "Tmax known", "BatchDate",
"Guest", "Novelty", "Batch", "Model"]
# Numeric columns the "Filter by feature" control can range-filter on: the three
# headline scores plus the 22 docking/chemistry features (label -> source column).
_FEATURE_LABEL_TO_COL = {label: col for label, col in FEATURES}
_FILTERABLE_COLS = ["(none)", "Pred logKa", "True logKa", "Tmax known"] + _FEATURE_COLS
def _feature_value(col, pred, true_logka, tmax_known, frow):
"""Raw numeric value for a filterable column, or None if absent/non-numeric."""
if col == "Pred logKa":
return pred
if col == "True logKa":
try:
return float(true_logka)
except (TypeError, ValueError):
return None
if col == "Tmax known":
return tmax_known
src = _FEATURE_LABEL_TO_COL.get(col)
if src is None:
return None
try:
v = float(frow.get(src))
import math
return None if math.isnan(v) else v
except (TypeError, ValueError):
return None
def _sort_records(records, col, ascending):
"""Sort the full record set by one column. Numeric columns sort numerically;
blanks / NaN always sink to the bottom regardless of direction."""
if not records or not col:
return records
numeric = col in _NUM_COLS or col == "BatchDate"
def val(r):
v = r.get(col)
if col == "BatchDate": # MM/DD/YYYY -> sortable key
try:
return datetime.datetime.strptime(str(v), "%m/%d/%Y").timestamp()
except (TypeError, ValueError):
return None
if numeric:
try:
f = float(v)
import math
return None if math.isnan(f) else f
except (TypeError, ValueError):
return None
s = str(v if v is not None else "").strip().lower()
return s or None
present = [r for r in records if val(r) is not None]
blanks = [r for r in records if val(r) is None]
present.sort(key=val, reverse=not ascending)
return present + blanks
# Per-column datatype hint so gr.Dataframe right-aligns / mono-styles numerics.
_NUM_COLS = {"#", "Pred logKa", "True logKa", "Tmax known"} | set(_FEATURE_COLS)
def _coltype(col: str) -> str:
return "number" if col in _NUM_COLS else "str"
def _fmt_feat(value) -> str:
"""Render a feature value compactly; blank for missing/NaN, never crash."""
if value is None:
return ""
try:
import math
f = float(value)
if math.isnan(f):
return ""
return f"{f:.3g}"
except (TypeError, ValueError):
s = str(value).strip()
return "" if s.lower() in ("nan", "none", "") else s
def _novelty_label(frow: dict):
"""Map the offline novelty annotation (carried on each feature row) to a
display tuple (label, scaffold_family, tmax). Blank when a guest carries no
annotation (e.g. not present in the GEOM discovery pool)."""
fam = frow.get("scaffold_family")
fam = "" if fam is None or str(fam).strip().lower() in ("", "nan", "none", "") else str(fam)
try:
tmax = float(frow.get("tmax_known"))
if tmax != tmax: # NaN
tmax = None
except (TypeError, ValueError):
tmax = None
def _truthy(v):
if isinstance(v, bool):
return v
s = str(v).strip().lower()
return s in ("true", "1", "1.0")
if fam: # known_scaffold <=> non-empty family
label = "Known scaffold"
elif _truthy(frow.get("is_novel")):
label = "Novel"
elif tmax is not None: # annotated, not known, not novel -> near a known binder
label = "Similar to known"
else:
label = "" # unannotated guest (no GEOM row)
return label, fam, tmax
def _build_records(min_logka: float, hide_known: bool, host_filter: str = "All",
refresh: bool = False, hide_known_scaffold: bool = False,
feat_filter: str = "(none)", feat_min=None, feat_max=None,
batch_date: str = "All batches"):
"""Build the full per-guest record set, ranked by predicted logKa (desc).
Returns (records, status_md, names) where each record is a dict keyed by the
full column universe (_ALL_COLS), `status_md` summarizes the result, and
`names` is the ordered guest-name click map (so a row click opens the right
guest regardless of which columns are visible or how the table is filtered).
"""
if refresh:
try:
pull_remote()
except Exception: # noqa: BLE001 — a refresh must never crash the tab
pass
# Feature table: indexed by inchikey, carries smiles + the 22 feature columns.
try:
feat = load_features()
smiles_map = feat["smiles"].astype(str).to_dict()
feat_rows = feat.to_dict("index") # {inchikey: {col: value}}
except Exception: # noqa: BLE001 — board must render even if features fail
smiles_map, feat_rows = {}, {}
rmap = review_map() # {(inchikey, model, prompt_version): {reviewer: rating}}
items = []
for r in list_samples():
host = r.get("host") or "CB[7]"
if host_filter and host_filter != "All" and host != host_filter:
continue
pred = r.get("combined_pred")
if pred is None or pred == "":
continue
try:
pred = float(pred)
except (TypeError, ValueError):
continue
true = r.get("true_logka")
has_true = true is not None and true != ""
if hide_known and has_true:
continue
if pred < float(min_logka or 0):
continue
inchikey = r.get("inchikey", "")
model = r.get("model", "")
prompt_version = r.get("prompt_version", "")
reviews = rmap.get((inchikey, model, prompt_version), {})
reviews_str = " · ".join(f"{rv}:{rt}" for rv, rt in reviews.items())
frow = feat_rows.get(inchikey, {})
novelty, scaffold_fam, tmax_known = _novelty_label(frow)
if hide_known_scaffold and scaffold_fam:
continue
if feat_filter and feat_filter != "(none)":
fv = _feature_value(feat_filter, pred, true, tmax_known, frow)
if fv is None:
continue
if feat_min is not None and feat_min != "" and fv < float(feat_min):
continue
if feat_max is not None and feat_max != "" and fv > float(feat_max):
continue
# Atomic facets straight from the store columns. Legacy rows written
# before `batch`/`prompt` were their own columns are recovered by
# splitting the old concatenated prompt_label ("·").
_batch_tag = r.get("batch") or ""
_prompt_ver = r.get("prompt") or ""
if not _batch_tag or not _prompt_ver:
_pver_head, _sep, _batch_tail = (r.get("prompt_label") or "").partition("·")
if _sep:
_prompt_ver = _prompt_ver or _pver_head
_batch_tag = _batch_tag or _batch_tail
else:
_batch_tag = _batch_tag or (r.get("prompt_label") or prompt_version)
_prompt_ver = _prompt_ver or prompt_version.split(":", 1)[0]
_batch_dt = _batch_date(_batch_tag)
if batch_date and batch_date != "All batches" and _batch_dt != batch_date:
continue
rec = {
"guest": r.get("guest_name", ""),
"pred": pred,
"Host": host,
"Pred logKa": round(pred, 2),
"True logKa": round(float(true), 2) if has_true else None,
"Novelty": novelty,
"Scaffold": scaffold_fam,
"Tmax known": round(tmax_known, 3) if tmax_known is not None else None,
"SMILES": smiles_map.get(inchikey, ""),
"Why": (r.get("combined_tldr") or "")[:120],
"Model": model,
# No dedicated provenance column exists in the feature data; the only
# honest "source"/"batch" tags are the LLM model + the prompt label.
"Source": model,
# prompt_label is "·" (e.g. "v5·batch4").
# Show the two facets separately: Batch = just "batch4", Prompt = "v5".
"Batch": _batch_tag,
"BatchDate": _batch_date(_batch_tag),
"Prompt": _prompt_ver,
"InChIKey": inchikey,
"Reviews": reviews_str,
}
for label, col in FEATURES:
rec[label] = _fmt_feat(frow.get(col))
items.append(rec)
items.sort(key=lambda d: d["pred"], reverse=True)
for i, d in enumerate(items, 1):
d["#"] = i
d["Guest"] = d["guest"]
names = [d["guest"] for d in items]
strong = sum(1 for d in items if d["pred"] >= 10)
status = (
f"**{len(items)}** guests · **{strong}** predicted logKa ≥ 10 · "
"ranked by predicted logKa"
)
return items, status, names
def _project(records, selected_cols):
"""Project records onto the visible column set, preserving _ALL_COLS order.
Returns (rows, headers, datatypes) ready for gr.Dataframe. The two pinned
columns are always present; the chooser only governs the rest.
"""
cols = [c for c in _ALL_COLS if c in set(selected_cols) | set(_PINNED_COLS)]
if not cols:
cols = list(_PINNED_COLS)
headers = cols
datatypes = [_coltype(c) for c in cols]
rows = [[rec.get(c, "") for c in cols] for rec in records]
return rows, headers, datatypes
def _page_slice(records, page, size):
"""One page of the ranked records. Gradio renders only the rows that fit the
grid (no scroll past them), so we page through the full set. Returns
(page_records, clamped_page, total_pages, total_count)."""
try:
size = max(1, int(size))
except (TypeError, ValueError):
size = 50
total = len(records)
pages = max(1, (total + size - 1) // size)
try:
page = int(page)
except (TypeError, ValueError):
page = 1
page = max(1, min(page, pages))
start = (page - 1) * size
return records[start:start + size], page, pages, total
def _render_page(records, selected_cols, page, size):
"""Project one page to the Dataframe. Returns (df_update, names, page, label)."""
disp, page, pages, total = _page_slice(records, page, size)
rows, headers, datatypes = _project(disp, selected_cols or _DEFAULT_COLS)
label = f"of **{pages}** · {total} records"
return (
gr.update(value=rows, headers=headers, datatype=datatypes),
[d["guest"] for d in disp],
page,
label,
)
def _is_ascending(sort_dir) -> bool:
return str(sort_dir or "").strip().startswith("↑")
def _load_data_board(min_logka: float, host_filter: str, selected_cols, size="50",
batch_date: str = "All batches",
sort_col: str = "Pred logKa", sort_dir: str = "↓ High→Low",
feat_filter: str = "(none)", feat_min=None, feat_max=None,
refresh: bool = False):
"""Build the full ranked record set (filters + sort applied) and show page 1.
`cand_records` holds the FULL sorted set so the pager can slice without a
rebuild; `names` is the visible page so clicks/column re-projection stay aligned.
Returns (df_update, status, names, full_records, page, page_label).
"""
records, status, _names = _build_records(
min_logka, False, host_filter, refresh, False,
feat_filter, feat_min, feat_max, batch_date,
)
records = _sort_records(records, sort_col, _is_ascending(sort_dir))
df_u, names, page, label = _render_page(records, selected_cols, 1, size)
return df_u, status, names, records, page, label
def _resort_board(records, sort_col, sort_dir, selected_cols, size):
"""Re-sort the cached full record set in place (no store rebuild) and reset to
page 1. Returns (df_update, sorted_records, names, page, page_label)."""
s = _sort_records(records or [], sort_col, _is_ascending(sort_dir))
df_u, names, page, label = _render_page(s, selected_cols, 1, size)
return df_u, s, names, page, label
def _open_candidate(names, evt: gr.SelectData):
"""Open the clicked board row in the Review tab (robust to ranking/filtering)."""
idx = evt.index[0] if isinstance(evt.index, (list, tuple)) else evt.index
if not names or idx is None or idx < 0 or idx >= len(names):
return (gr.update(),) * 12
guest_name = names[idx]
match = next((r for r in list_samples() if r.get("guest_name") == guest_name), None)
if match is None:
return (gr.update(),) * 12
inchikey, model, prompt_version = match.get("inchikey", ""), match.get("model", ""), match.get("prompt_version", "")
sample = get_sample(inchikey, model, prompt_version)
results = sample.get("results") or {} if sample else {}
return (
_pred_table(results),
_reason_from_results(results),
_card_from_results(results, "physics", "physics"),
_card_from_results(results, "chemistry", "chemistry"),
_status_line(f"Opened {guest_name} from the Data board — review below.", kind="done"),
inchikey,
guest_name,
model,
prompt_version,
guest_name,
model,
gr.update(selected="review"),
)
def _on_prompt_mode(mode, custom):
if mode == "Customize":
return gr.update(value=custom, interactive=True)
if mode == "Default":
return gr.update(value=preset_text("v5"), interactive=False)
if mode == "Simple":
return gr.update(value=preset_text("Default"), interactive=False)
return gr.update(value=preset_text("v5"), interactive=False)
def _persist_custom(mode, text):
if mode == "Customize":
return text
return gr.update()
def _show_custom_features(feature_preset):
return gr.update(visible=(feature_preset == "Custom"))
# --------------------------------------------------------------------------- #
# UI #
# --------------------------------------------------------------------------- #
def _health_html() -> tuple[str, list]:
from viz.theme import ERR, INK, MUTED, OK, WARN
ok, msg = healthcheck()
proxy_dot = OK if ok else ERR
proxy_label = backend_label() if ok else msg
try:
# The Review dropdown only offers guests that have actually been scored
# (the discovery candidates the tab reviews — clicking a Data-board row
# sets the dropdown to one of these). The full feature pool is ~427k
# GEOM+QM9 guests; embedding that as the dropdown's `choices` balloons
# Gradio's api_info enum (replicated per endpoint) to >100 MB and is sent
# even on the login page, so the browser "loads forever". A bounded
# guest_choices() fallback covers a cold/empty store without regressing.
scored = sorted({(r.get("guest_name") or r.get("inchikey") or "")
for r in list_samples()} - {""})
choices = scored or guest_choices()[:2000]
ds_ok = True
except Exception: # noqa: BLE001 — readable status, no crash (FR11/NFR5)
choices, ds_ok = [], False
ds_dot = OK if ds_ok else ERR
ds_msg = "Dataset"
# Store durability is reflected by the dot color + OK/WARN keyword; the
# parenthetical label is kept to a clean "Database". Durable when an HF store
# dataset + token are configured (local SQLite synced by store/hf_sync);
# otherwise the local disk check decides.
hf_store = os.environ.get("HF_STORE_DATASET")
hf_token = os.environ.get("HF_STORE_TOKEN") or os.environ.get("HF_TOKEN")
if hf_store and hf_token:
db_ok = True
else:
_db_path, db_persistent = db_status()
db_ok = db_persistent
db_dot = OK if db_ok else WARN
db_msg = "Database"
def _status_item(color: str, glyph: str, keyword: str, label: str) -> str:
return (
f""
f"{glyph} {html.escape(keyword)} "
f"({html.escape(label)})"
)
# Molecular macrocycle header: heptagon logo mark + Newsreader brand name,
# three LED-style status dots, gradient-tinted header band.
head = (
"
"
f"{_status_item(proxy_dot, '●' if ok else '✕', 'OK' if ok else 'ERROR', proxy_label)}"
f"{_status_item(ds_dot, '●' if ds_ok else '✕', 'OK' if ds_ok else 'ERROR', ds_msg)}"
f"{_status_item(db_dot, '●' if db_ok else '▲', 'OK' if db_ok else 'WARN', db_msg)}"
"
"
"
"
)
return head, choices
def build_ui() -> gr.Blocks:
# Pre-warm the feature table (5.9 MB CSV from HuggingFace, ~14.6k rows) at
# app-build time so its lru_cache is hot before the first request. Otherwise
# the download+parse runs inside the first `demo.load` AFTER login, freezing
# the page right when the user lands. load_features() is lru_cache'd, so this
# one call serves every later page load for the life of the container.
# Non-fatal: _build_records already tolerates a feature-load failure.
try:
load_features()
except Exception: # noqa: BLE001 — pre-warm must never block startup
pass
head_html, choices = _health_html()
with gr.Blocks(
title="SupraDashboard — CB[7] logKa",
theme=_theme(),
css=_CSS,
head=_FONT_HEAD,
) as demo:
inchikey_st = gr.State("")
guest_st = gr.State("")
model_st = gr.State("")
promptver_st = gr.State("")
force_rerun_st = gr.State(False)
menu_open_st = gr.State(False)
custom_text_st = gr.State(preset_text("v5"))
# Land on the Discovery Board (real ranked results) rather than an empty
# "awaiting run" Review tab; a row click still jumps into Review.
with gr.Tabs(selected="data") as tabs:
with gr.Tab("Review", id="review"):
gr.HTML(head_html)
with gr.Row():
host_dd = gr.Dropdown(
choices=["CB[7]", "Other host (unavailable)"],
value="CB[7]",
label="Host Molecule", scale=2,
)
guest_dd = gr.Dropdown(
choices=choices, value=None, label="Guest Molecule",
scale=2, filterable=True, allow_custom_value=True,
)
model_dd = gr.Dropdown(
choices=available_models(), value=default_model(),
label="LLM Model", scale=1,
)
with gr.Column(scale=1, min_width=210, elem_classes=["run-col"]):
# Real split button: main "Run" pill + caret toggle fused as
# a flex btn-group (see .run-group CSS). The caret opens a
# dropdown with the cache / force-re-run choice.
# Run is READ-ONLY: it shows the stored batch prediction for
# the selected guest and never recomputes or writes. The
# force-re-run caret is hidden since there is no live run.
with gr.Row(elem_classes=["run-group"]):
run_btn = gr.Button(
"Run", variant="primary", elem_classes=["run-main"],
)
run_caret = gr.Button(
"▾", variant="primary", elem_classes=["run-caret"],
visible=False,
)
with gr.Column(visible=False, elem_classes=["run-menu"]) as run_menu:
mode_cache_btn = gr.Button(
"✓ Use cache", elem_classes=["run-menu-item"]
)
mode_fresh_btn = gr.Button(
"↻ Force re-run", elem_classes=["run-menu-item"]
)
with gr.Accordion("Prompt", open=False):
with gr.Tabs():
with gr.Tab("System prompt"):
with gr.Row():
prompt_mode = gr.Radio(
["Default", "Simple", "Customize"],
value="Default",
label="Prompt",
elem_classes=["prompt-modes"],
scale=1
)
with gr.Column(scale=3):
system_prompt_ta = gr.Textbox(
value=preset_text("v5"),
label="System prompt",
lines=8,
interactive=False
)
with gr.Tab("User prompt"):
gr.HTML(user_prompt_html())
with gr.Tab("Features"):
feature_preset = gr.Radio(
choices=FEATURE_PRESET_NAMES,
value="Recommended",
label="Feature set",
)
custom_features = gr.Dropdown(
choices=ALL_FEATURE_COLS,
value=[],
multiselect=True,
visible=False,
label="Custom features",
)
host_note = gr.Markdown("", visible=False)
status_html = gr.HTML(_status_line(""))
with gr.Row(equal_height=False):
with gr.Column(scale=2):
with gr.Row(elem_classes=["viz-controls"]):
species_toggle = gr.Radio(
["Guest", "Host"], value="Guest",
show_label=False, container=False,
elem_classes=["pill-toggle", "species-toggle"],
)
view_toggle = gr.Radio(
["3D", "2D"], value="3D",
show_label=False, container=False,
elem_classes=["pill-toggle", "view-toggle"],
)
structure_2d = gr.HTML(
"
"
)
reason_html = gr.HTML(
_reason_panel(
_reason_placeholder(
"The combined reasoning trace will appear here after you Run prediction."
)
)
)
with gr.Column(elem_classes="review-band"):
gr.HTML(
"
"
"Does the reasoning make chemical sense?"
"
"
)
rating = gr.Radio(
choices=RATINGS, label="Reasoning quality", value=None,
elem_classes="rating-radio",
)
comment = gr.Textbox(
label="Expert comment", lines=4,
placeholder="Chemical plausibility, misleading feature "
"interpretations, or where the reasoning diverges from CB[7] "
"host–guest chemistry (cite the rule number).",
)
submit_btn = gr.Button("Submit review", elem_classes=["copper-btn"])
feedback_out = gr.HTML()
with gr.Tab("Data", id="data"):
gr.Markdown("### Discovery Board")
# One consolidated control panel (filters · feature filter · sort ·
# view) instead of three loose rows. Page size feeds pagination
# (Gradio renders only the rows that fit, so we page through all).
with gr.Group(elem_classes=["board-controls"]):
with gr.Row(elem_classes=["viz-controls"]):
cand_min = gr.Slider(0, 12, value=0, step=0.5,
label="Min predicted logKa", scale=3)
host_dd_cand = gr.Dropdown(
choices=["All", "CB[7]"], value="All", label="Host", scale=1,
)
cand_batch = gr.Dropdown(
choices=_batch_date_choices(), value="All batches",
label="Batch (date)", scale=2,
)
with gr.Row(elem_classes=["viz-controls"]):
feat_filter = gr.Dropdown(
choices=_FILTERABLE_COLS, value="(none)",
label="Filter by feature", scale=3,
)
feat_min = gr.Number(label="min", value=None, scale=1)
feat_max = gr.Number(label="max", value=None, scale=1)
cand_sort = gr.Dropdown(
choices=_SORTABLE_COLS, value="Pred logKa",
label="Sort by", scale=2,
)
cand_sort_dir = gr.Radio(
choices=["↓ High→Low", "↑ Low→High"], value="↓ High→Low",
label="Order", scale=2, elem_classes=["pill-toggle"],
)
with gr.Row(elem_classes=["viz-controls"]):
col_chooser = gr.Dropdown(
choices=_CHOOSABLE_COLS,
value=[c for c in _DEFAULT_COLS if c not in _PINNED_COLS],
multiselect=True,
label="Columns · # and Guest are always shown",
elem_classes=["col-chooser"], scale=4,
)
cand_limit = gr.Dropdown(
choices=["5", "10", "50", "100"], value="50",
label="Rows / page", scale=1,
)
cand_refresh = gr.Button("Refresh", elem_classes=["refresh-btn"], scale=1)
cand_status = gr.Markdown("")
cand_names = gr.State([])
cand_records = gr.State([])
cand_df = gr.Dataframe(
headers=_DEFAULT_COLS,
elem_id="candidates-grid",
interactive=False,
max_height=2000, # tall enough to fully render up to 100 rows
# (~18px/row; gradio renders only what fits
# and won't scroll past it). Box auto-shrinks
# for smaller Row counts.
wrap=False, # one line per cell; columns auto-fit to content
# NOTE: pinned_columns is intentionally NOT set. In gradio 5.49 it
# renders a SECOND overlay sub-table; combined with the grid CSS it
# blew the scroll table out to ~1e6px wide and hid every cell. The
# table-wrap below gives a clean horizontal scroll instead.
datatype=[_coltype(c) for c in _DEFAULT_COLS],
)
# Pager (below the grid): walk through every record page by page.
with gr.Row(elem_classes=["pager-row"]):
prev_btn = gr.Button("← Prev", scale=0, min_width=90)
page_num = gr.Number(
value=1, precision=0, minimum=1, step=1,
label="Page", show_label=False, scale=0, min_width=64,
elem_id="pager-input",
)
page_total = gr.Markdown("of **1**", elem_classes=["pager-total"])
next_btn = gr.Button("Next →", scale=0, min_width=90)
with gr.Tab("Feedback"):
gr.Markdown(
"Collected expert reviews. Gated behind the same `APP_AUTH` login. "
"Click a row to open it in **Review** and edit the rating/comment."
)
fb_keys = gr.State([])
export_status = gr.Markdown()
export_refresh_btn = gr.Button("Refresh", elem_classes=["refresh-btn"])
export_df = gr.Dataframe(
headers=_FEEDBACK_COLS,
wrap=True, interactive=False,
)
export_refresh_btn.click(
_load_export, outputs=[export_df, export_status, fb_keys]
)
# predict() is an async generator yielding already-wrapped reasoning HTML
# (every yield applies _reason_panel so the .reason-wrap card chrome persists).
def _on_host_change(host):
if host != "CB[7]":
return (
gr.update(
value="Other host unavailable — predictions are CB[7]-only for now.",
visible=True,
),
gr.update(value="CB[7] only", interactive=False),
)
return (
gr.update(value="", visible=False),
gr.update(value="Run", interactive=True),
)
_PLACEHOLDER_2D = (
"
"
"Pick a guest to view its 2D structure
"
)
_PLACEHOLDER_3D = (
"
"
"Pick a guest to view its 3D structure (drag to rotate)
"
)
async def _render_structure(species, view, guest):
"""Render Host (CB[7]) or Guest structure as a 2D card or 3D viewer.
Returns updates for (structure_2d, structure_3d). Both views carry an
identical in-view caption bar showing "{name} · {source}". PubChem
fetches are module-cached, so toggling species/view after the first
fetch is instant.
"""
import asyncio
from viz.pubchem import fetch_2d, fetch_3d
from viz.viewer import build_2d_html, build_3dmol_html
if species == "Host":
rec = host_record()
name = "CB[7] (host)"
else:
try:
rec = get_record(guest) if guest else None
except KeyError:
rec = None
if not rec:
return (
gr.update(value=_PLACEHOLDER_2D, visible=(view == "2D")),
gr.update(value=_PLACEHOLDER_3D, visible=(view == "3D")),
)
name = f"{rec.get('guest_name', guest)} (guest)"
inchikey = rec.get("inchikey", "")
smiles = rec.get("smiles", "")
def _clean_src(src: str) -> str:
# "RDKit (generated)" → "RDKit"; drop any parenthetical qualifier.
return (src or "").split("(", 1)[0].strip()
if view == "2D":
img, src = await asyncio.to_thread(fetch_2d, inchikey, smiles)
caption = f"{name} · {_clean_src(src)}"
html2d = build_2d_html(img, title=caption) or _PLACEHOLDER_2D
return (
gr.update(value=html2d, visible=True),
gr.update(visible=False),
)
molblock, src = await asyncio.to_thread(fetch_3d, inchikey, smiles)
caption = f"{name} · {_clean_src(src)}"
html3d = build_3dmol_html(molblock, title=caption) or _PLACEHOLDER_3D
return (
gr.update(visible=False),
gr.update(value=html3d, visible=True),
)
_struct_inputs = [species_toggle, view_toggle, guest_dd]
_struct_outputs = [structure_2d, structure_3d]
host_dd.change(_on_host_change, inputs=[host_dd], outputs=[host_note, run_btn])
guest_dd.change(
_render_structure, inputs=_struct_inputs, outputs=_struct_outputs
)
species_toggle.change(
_render_structure, inputs=_struct_inputs, outputs=_struct_outputs
)
view_toggle.change(
_render_structure, inputs=_struct_inputs, outputs=_struct_outputs
)
prompt_mode.change(
_on_prompt_mode,
inputs=[prompt_mode, custom_text_st],
outputs=[system_prompt_ta],
)
system_prompt_ta.input(
_persist_custom,
inputs=[prompt_mode, system_prompt_ta],
outputs=[custom_text_st],
)
feature_preset.change(
_show_custom_features,
inputs=[feature_preset],
outputs=[custom_features],
)
run_caret.click(
lambda o: (not o, gr.update(visible=not o)),
inputs=[menu_open_st],
outputs=[menu_open_st, run_menu],
)
mode_cache_btn.click(
lambda: (False, False, gr.update(visible=False), gr.update(value="Run")),
outputs=[force_rerun_st, menu_open_st, run_menu, run_btn],
)
mode_fresh_btn.click(
lambda: (True, False, gr.update(visible=False), gr.update(value="Run · fresh")),
outputs=[force_rerun_st, menu_open_st, run_menu, run_btn],
)
demo.load(_render_structure, inputs=_struct_inputs, outputs=_struct_outputs)
demo.load(_load_export, outputs=[export_df, export_status, fb_keys])
# ---- Feedback tab: click a review row to edit it in Review ----
export_df.select(
_open_feedback,
inputs=[fb_keys],
outputs=[
pred_html, reason_html, physics_card, chem_card, status_html,
inchikey_st, guest_st, model_st, promptver_st, guest_dd, model_dd,
rating, comment, tabs,
],
).then(_render_structure, inputs=_struct_inputs, outputs=_struct_outputs)
# ---- Data tab: merged ranked discovery board + column chooser ----
# Row click opens the guest in Review (names map keeps clicks correct
# regardless of which columns are visible).
cand_df.select(
_open_candidate,
inputs=[cand_names],
outputs=[
pred_html, reason_html, physics_card, chem_card, status_html,
inchikey_st, guest_st, model_st, promptver_st, guest_dd, model_dd, tabs,
],
).then(_render_structure, inputs=_struct_inputs, outputs=_struct_outputs)
# Data reload (filters / page-size / refresh): rebuild the full record set
# and reset to page 1. cand_records holds the FULL filtered set; the pager
# slices it without a rebuild.
_board_inputs = [cand_min, host_dd_cand, col_chooser, cand_limit, cand_batch,
cand_sort, cand_sort_dir, feat_filter, feat_min, feat_max]
_board_outputs = [cand_df, cand_status, cand_names, cand_records, page_num, page_total]
cand_refresh.click(
lambda mn, hf, cols, lim, bd, sc, sd, ff, fmn, fmx: _load_data_board(
mn, hf, cols, lim, bd, sc, sd, ff, fmn, fmx, refresh=True),
inputs=_board_inputs, outputs=_board_outputs,
)
for _f in (cand_min, host_dd_cand, cand_limit, cand_batch,
feat_filter, feat_min, feat_max):
_f.change(_load_data_board, inputs=_board_inputs, outputs=_board_outputs)
demo.load(_load_data_board, inputs=_board_inputs, outputs=_board_outputs)
# Server-side sort over the WHOLE filtered set (the per-column header menu
# only reorders the current page). Re-sorts cand_records in place, page 1.
_sort_outputs = [cand_df, cand_records, cand_names, page_num, page_total]
for _ctrl in (cand_sort, cand_sort_dir):
_ctrl.change(
_resort_board,
inputs=[cand_records, cand_sort, cand_sort_dir, col_chooser, cand_limit],
outputs=_sort_outputs,
)
# Pager + column chooser: re-slice/re-project the cached full record set
# (no data reload). All emit the same (df, names, page, label) shape.
_page_outputs = [cand_df, cand_names, page_num, page_total]
prev_btn.click(
lambda recs, cols, pg, sz: _render_page(recs, cols, (int(pg or 1) - 1), sz),
inputs=[cand_records, col_chooser, page_num, cand_limit], outputs=_page_outputs,
)
next_btn.click(
lambda recs, cols, pg, sz: _render_page(recs, cols, (int(pg or 1) + 1), sz),
inputs=[cand_records, col_chooser, page_num, cand_limit], outputs=_page_outputs,
)
page_num.submit(
lambda recs, cols, pg, sz: _render_page(recs, cols, pg, sz),
inputs=[cand_records, col_chooser, page_num, cand_limit], outputs=_page_outputs,
)
col_chooser.change(
lambda recs, cols, pg, sz: _render_page(recs, cols, pg, sz),
inputs=[cand_records, col_chooser, page_num, cand_limit], outputs=_page_outputs,
)
run_btn.click(lambda: gr.update(interactive=False), outputs=[run_btn]).then(
predict,
inputs=[
guest_dd,
model_dd,
prompt_mode,
custom_text_st,
force_rerun_st,
feature_preset,
custom_features,
],
outputs=[
pred_html,
reason_html,
physics_card,
chem_card,
status_html,
inchikey_st,
guest_st,
model_st,
promptver_st,
],
show_progress="full",
concurrency_limit=1,
).then(lambda: gr.update(interactive=True), outputs=[run_btn])
submit_btn.click(
submit_feedback,
inputs=[inchikey_st, guest_st, model_st, promptver_st, rating, comment],
outputs=[feedback_out],
)
return demo
def _auth():
"""Parse APP_AUTH='user:pass,user2:pass2' into a Gradio auth list (or None)."""
raw = os.environ.get("APP_AUTH", "").strip()
if not raw:
return None
pairs = [tuple(p.split(":", 1)) for p in raw.split(",") if ":" in p]
return pairs or None