""" SRH Pathology Validation Study - expert annotation app (Gradio / Hugging Face Space). Design goal (strict): judging one case requires ZERO scrolling and ZERO guessing about what controls mean. Every scale legend and subjective definition is printed on-screen next to the control that uses it, and every image carries its own zoom / brightness / contrast strip directly beneath it. Two arms (one grading unit per screen; order randomized per reader; blinded to source): ARM A - Realism & memorization: ONE SRH patch; judge Real vs AI-generated, confidence, looks-copied, clinical plausibility. ARM B - Discovered-category meaningfulness: a compact grid of patches the model grouped as ONE discovered category; judge whether it is a coherent, clinically meaningful morphology (+ optional description). Credentials: simple first-login logins (e.g. Pathologist_1 / Path1234@) that the reader can CHANGE after first login (username and/or password) from an on-screen Settings panel; changing the username migrates saved work. Robust storage: every answer is keyed by (annotator, case_id, item_id, dimension) in an append-only JSONL in a private dataset, so later UI/wording changes can never invalidate or overwrite prior annotations. Config via env / HF Space secrets: HF_TOKEN : write token for the private response/account dataset. RESPONSE_DATASET : private dataset repo id (default DrSyedFaizan/srh-reader-responses). CASES_DATASET : private dataset holding cases.json + images (snapshot at boot). READER_CREDENTIALS : JSON {"login_name": "login_password", ...} for first-time login only. APP_SECRET : secret string signing resume tokens (defaults derived from HF_TOKEN). """ import os import io import json import time import base64 import random import hashlib import hmac import secrets from pathlib import Path import gradio as gr from PIL import Image # ----------------------------------------------------------------------------- config APP_DIR = Path(__file__).parent DATA_DIR = APP_DIR / "data" LOCAL_BACKUP_DIR = APP_DIR / "local_responses" LOCAL_BACKUP_DIR.mkdir(exist_ok=True) HF_TOKEN = os.environ.get("HF_TOKEN", "").strip() RESPONSE_DATASET = os.environ.get("RESPONSE_DATASET", "DrSyedFaizan/srh-reader-responses").strip() CASES_DATASET = os.environ.get("CASES_DATASET", "").strip() APP_SECRET = os.environ.get("APP_SECRET", "").strip() or ( "srh-" + hashlib.sha256(HF_TOKEN.encode()).hexdigest()[:16] if HF_TOKEN else "srh-dev-secret") DISPLAY_MAX_W = 700 ARMB_MAX_TILES = 9 # compact grid so an Arm-B case stays near a single viewport SCHEMA_VERSION = 2 APP_BUILD = 3 # bumped 2026-07-08: 512px crisp images + resolution note. Analysis counts app_build>=3 # only (silently discards pre-fix, resolution-confounded ratings). ACCOUNTS_PATH = "accounts/accounts.json" try: _invite = json.loads(os.environ.get("READER_CREDENTIALS", "").strip() or "{}") except Exception: _invite = {} if not _invite: _invite = {"Pathologist_1": "Path1234@", "Pathologist_2": "Path1234@", "Pathologist_3": "Path1234@"} print("[WARN] READER_CREDENTIALS not set - using default simple logins. Set the secret before the real study.") INVITES = {str(k): str(v) for k, v in _invite.items()} # ----------------------------------------------------------------------------- HF api try: from huggingface_hub import HfApi _api = HfApi(token=HF_TOKEN) if HF_TOKEN else None except Exception as e: _api = None print(f"[WARN] huggingface_hub unavailable: {e}") def _ensure_response_dataset(): if not _api: return try: _api.create_repo(RESPONSE_DATASET, repo_type="dataset", private=True, exist_ok=True) except Exception as e: print(f"[WARN] could not ensure response dataset: {e}") # ----------------------------------------------------------------------------- cases def _maybe_pull_cases(): if not CASES_DATASET or not _api: return try: from huggingface_hub import snapshot_download snapshot_download(CASES_DATASET, repo_type="dataset", local_dir=str(DATA_DIR), token=HF_TOKEN, local_dir_use_symlinks=False) print(f"[info] pulled cases from {CASES_DATASET}") except Exception as e: print(f"[WARN] could not pull CASES_DATASET: {e}") def load_cases(): _maybe_pull_cases() cj = DATA_DIR / "cases.json" if not cj.exists(): print("[WARN] no data/cases.json found.") return [] with open(cj, "r", encoding="utf-8") as f: return json.load(f).get("cases", []) CASES = load_cases() N_CASES = len(CASES) CASES_BY_ID = {c["case_id"]: c for c in CASES} _IMG_CACHE = {} def img_data_uri(rel_path): if not rel_path: return "" if rel_path in _IMG_CACHE: return _IMG_CACHE[rel_path] p = DATA_DIR / rel_path try: im = Image.open(p).convert("RGB") if im.width > DISPLAY_MAX_W: h = int(im.height * DISPLAY_MAX_W / im.width) im = im.resize((DISPLAY_MAX_W, h)) buf = io.BytesIO() im.save(buf, format="PNG") uri = "data:image/png;base64," + base64.b64encode(buf.getvalue()).decode() except Exception as e: print(f"[WARN] image load failed {p}: {e}") uri = "" _IMG_CACHE[rel_path] = uri return uri # ----------------------------------------------------------------------------- tokens / passwords / accounts def make_token(name): return hmac.new(APP_SECRET.encode(), name.encode(), hashlib.sha256).hexdigest() def valid_token(name, tok): return bool(name) and bool(tok) and hmac.compare_digest(make_token(name), tok) def hash_pw(pw, salt): return hashlib.pbkdf2_hmac("sha256", pw.encode(), bytes.fromhex(salt), 100_000).hex() def verify_pw(pw, rec): try: return hmac.compare_digest(hash_pw(pw, rec["salt"]), rec["hash"]) except Exception: return False def new_pw_record(pw, invite=""): salt = secrets.token_hex(8) return {"salt": salt, "hash": hash_pw(pw, salt), "invite": invite, "created_ts": int(time.time())} def load_accounts(): default = {"accounts": {}, "claimed_invites": []} if _api: try: from huggingface_hub import hf_hub_download fp = hf_hub_download(RESPONSE_DATASET, ACCOUNTS_PATH, repo_type="dataset", token=HF_TOKEN) with open(fp, "r", encoding="utf-8") as f: d = json.load(f) d.setdefault("accounts", {}); d.setdefault("claimed_invites", []) return d except Exception: pass lb = LOCAL_BACKUP_DIR / "accounts.json" if lb.exists(): with open(lb, "r", encoding="utf-8") as f: return json.load(f) return default def save_accounts(acc): payload = json.dumps(acc, ensure_ascii=False, indent=2) with open(LOCAL_BACKUP_DIR / "accounts.json", "w", encoding="utf-8") as f: f.write(payload) if _api: try: _api.upload_file(path_or_fileobj=payload.encode("utf-8"), path_in_repo=ACCOUNTS_PATH, repo_id=RESPONSE_DATASET, repo_type="dataset", commit_message="update accounts") except Exception as e: print(f"[WARN] account upload failed (kept local): {e}") # ----------------------------------------------------------------------------- storage def _resp_path(annotator): return f"responses/{annotator}.jsonl" def load_existing_responses(annotator): if _api: try: from huggingface_hub import hf_hub_download fp = hf_hub_download(RESPONSE_DATASET, _resp_path(annotator), repo_type="dataset", token=HF_TOKEN) with open(fp, "r", encoding="utf-8") as f: return [json.loads(l) for l in f if l.strip()] except Exception: pass lb = LOCAL_BACKUP_DIR / f"{annotator}.jsonl" if lb.exists(): with open(lb, "r", encoding="utf-8") as f: return [json.loads(l) for l in f if l.strip()] return [] def completed_case_ids(records): return {r["case_id"] for r in records if r.get("item_id") == "__case__"} def _write_responses(annotator, records): payload = "\n".join(json.dumps(r, ensure_ascii=False) for r in records) + ("\n" if records else "") with open(LOCAL_BACKUP_DIR / f"{annotator}.jsonl", "w", encoding="utf-8") as f: f.write(payload) if _api: try: _api.upload_file(path_or_fileobj=payload.encode("utf-8"), path_in_repo=_resp_path(annotator), repo_id=RESPONSE_DATASET, repo_type="dataset", commit_message=f"responses {annotator} {int(time.time())}") except Exception as e: print(f"[WARN] HF upload failed (kept local backup): {e}") def save_records(annotator, new_records): existing = load_existing_responses(annotator) existing.extend(new_records) _write_responses(annotator, existing) return existing def migrate_responses(old, new): """Rename an annotator: copy their append-only responses to the new key. Old data is preserved as-is.""" recs = load_existing_responses(old) for r in recs: r["annotator"] = new if recs: _write_responses(new, recs) # ----------------------------------------------------------------------------- ordering def reader_order(annotator): # Arm B (discovered-category coherence) is the paper's primary endpoint and time is short, so we present # every reader's remaining Arm B cases first, then Arm A. Each arm is still shuffled by the reader's own # seed (internal blinding of the hidden pos/neg controls is preserved), and completed cases are skipped by # first_unfinished_idx, so a reader resumes into her next unfinished Arm B case and keeps all prior work. seed = int(hashlib.sha256(annotator.encode()).hexdigest(), 16) % (2**32) b_ids = [c["case_id"] for c in CASES if c.get("arm", "A") == "B"] a_ids = [c["case_id"] for c in CASES if c.get("arm", "A") != "B"] random.Random(seed).shuffle(b_ids) random.Random(seed + 1).shuffle(a_ids) return b_ids + a_ids def first_unfinished_idx(order, records): done = completed_case_ids(records) for i, cid in enumerate(order): if cid not in done: return i return len(order) # ----------------------------------------------------------------------------- UI (images) def image_html(dom_id, uri): if not uri: return "
[no image]
" return f"""
zoom
""" def image_grid_html(prefix, uris, single=False): cls = "grid single" if single else "grid" cells = "".join(image_html(f"{prefix}_{k}", u) for k, u in enumerate(uris)) or "[no images]" return f"
{cells}
" # ----------------------------------------------------------------------------- UI text (self-explanatory, on-screen) PROVENANCE = ( "
Where these images come from. This app validates an AI system for " "Stimulated Raman Histology (SRH) of brain tumors. In Task A a single patch is shown; it is " "either a real acquired SRH field or an AI-generated one, shown blinded and rendered identically " "as virtual H&E. In Task B a group of patches that the model discovered as one category is shown; " "some groups are real known-tumor categories, some are candidate novel/rare categories, and one is a " "deliberately scrambled (random) group used as a hidden control. You are blinded to all sources. " "You can stop and resume anytime; your work saves as you go.
") RESOLUTION_NOTE = ( "
" "About image resolution: these are stimulated-Raman tissue fields shown at " "their native acquisition resolution, which is uniform across every image (it is not a quality " "defect and is unrelated to whether an image is real or AI-generated). Please assess tissue morphology " "and pattern rather than sharpness; use the + zoom control on any image if helpful.
") PROMPT_A = ("### Task A - Is this single SRH patch real or AI-generated?\n" "Judge purely from morphology and texture, not sharpness. You are blinded to the source.") LEGEND_A = ( RESOLUTION_NOTE + "
" "Real vs AI-generated: Real = a genuine acquired SRH field; AI-generated = synthesized by a model.
" "Confidence 1-5: 1 = pure guess, 2 = low, 3 = moderate, 4 = high, 5 = certain.
" "Looks copied/memorized: Yes if it looks like a near-duplicate of a specific real example.
" "Clinical plausibility: Plausible = could be genuine tissue; Minor artifacts = mostly realistic with " "small oddities; Implausible = clearly not real tissue.
") DEF_A = ("Copied/memorized = a plausible-looking near-duplicate of a specific real field, " "not a fresh example. Example: nuclei arrangement and background that appear lifted verbatim from one " "known image rather than a new, independently generated field.") PROMPT_B = ("### Task B - Is this a coherent, clinically meaningful category?\n" "The patches were grouped by the model as ONE discovered category. Decide if they share a coherent, " "clinically meaningful morphology (a real entity/pattern) or are an incoherent mix (an artifact). " "Optionally name the morphology or putative entity.") LEGEND_B = ( RESOLUTION_NOTE + "
" "Coherent & meaningful (Yes): the patches share one recognizable morphology that could correspond " "to a real tumor type/pattern.
" "Partial: a dominant shared pattern plus some outliers.
" "No: an incoherent mix with no shared morphology (a clustering artifact).
" "Confidence 1-5: 1 = pure guess ... 5 = certain.
") DEF_B = ("Meaningful = a morphology a pathologist would recognize as one entity/pattern, " "not merely visually similar noise. Example: uniformly monomorphic cells with salt-and-pepper nuclei " "read as one entity (Yes); a mix of fibrous, cellular and necrotic fields with nothing in common (No).") LOGIN_OVERVIEW = ( "### What this study is\n" "You are validating an AI system that analyzes **Stimulated Raman Histology (SRH)** of brain tumors, in two " "short blinded tasks:\n" "- **Task A (realism):** decide whether a single SRH patch is **real** or **AI-generated**, and whether it " "looks copied.\n" "- **Task B (category review):** decide whether a group of patches the model discovered as one category is a " "**coherent, clinically meaningful** morphology or an artifact.\n\n" "### Signing in\n" "Use the login you were given (for example **Pathologist_1** with the password provided). After you sign in " "you can **change your username and password** from the **Account settings** panel. Closing the tab does " "**not** sign you out; a **progress counter** shows how many items you have finished, and you resume where " "you left off.") HOWTO = ( "**One item per screen; order is randomized; you are blinded to all sources. Every scale is printed on screen " "next to the control, so you never need to scroll to recall a meaning.**\n\n" "**Task A - Real vs AI-generated (single patch):** answer Real/Synthetic, Confidence (1-5), Looks copied? " "(Yes/No), and Clinical plausibility. The full meaning of each option is shown inline.\n\n" "**Task B - Discovered-category review (grid of patches):** answer Coherent & meaningful? (Yes/Partial/No), " "Confidence (1-5), and an optional description.\n\n" "**Viewing each image:** under every image there is **-/+** zoom and **bright** / **contrast** sliders that " "change only your view, never the stored data.\n\n" "**Saving & resuming:** press **Save & Next** to store the item and continue (required fields must be filled). " "Stop anytime and sign back in to resume. Use **Logout** to end the session.") DONE_MSG = ( "### All items complete. Thank you.\n\n" "Your responses are saved securely. You may close this tab. If more items are added later, sign back in and " "you will continue from the new items.") SET_LS_JS = "(u,t)=>{ if(t){ localStorage.setItem('srh_reader_user',u); localStorage.setItem('srh_reader_token',t);} }" GET_LS_JS = "()=>[localStorage.getItem('srh_reader_user')||'', localStorage.getItem('srh_reader_token')||'']" CLR_LS_JS = "()=>{ localStorage.removeItem('srh_reader_user'); localStorage.removeItem('srh_reader_token'); location.reload(); }" HEAD = """ """ # ----------------------------------------------------------------------------- app def build_app(): with gr.Blocks(title="SRH Pathology Validation Study", head=HEAD, theme=gr.themes.Soft()) as demo: annotator_state = gr.State("") index_state = gr.State(0) order_state = gr.State([]) user_ls = gr.Textbox(visible=False) tok_ls = gr.Textbox(visible=False) tok_out = gr.Textbox(visible=False) user_out = gr.Textbox(visible=False) # ------------------------------------------------- LOGIN with gr.Column(visible=True) as login_col: gr.Markdown("## SRH Pathology Validation Study") gr.Markdown(LOGIN_OVERVIEW) name_in = gr.Textbox(label="Username", placeholder="e.g. Pathologist_1") pass_in = gr.Textbox(label="Password", type="password") login_btn = gr.Button("Sign in", variant="primary") login_msg = gr.Markdown("") # ------------------------------------------------- STUDY with gr.Column(visible=False) as study_col: with gr.Row(): gr.Markdown("## SRH Pathology Validation Study") progress_md = gr.Markdown("0 / 0") logout_btn = gr.Button("Logout", scale=0) with gr.Accordion("How to grade (click to expand or collapse)", open=True): gr.Markdown(HOWTO) with gr.Accordion("Account settings (change your username or password)", open=False): gr.Markdown("Change your login. Leave a field blank to keep it. Changing your username keeps all " "your saved work.") cs_user = gr.Textbox(label="New username (optional)") cs_pw = gr.Textbox(label="New password (optional, min 6 chars)", type="password") cs_confirm = gr.Textbox(label="Confirm new password", type="password") cs_btn = gr.Button("Save account changes") cs_msg = gr.Markdown("") gr.HTML(PROVENANCE) arm_badge_md = gr.Markdown("") prompt_md = gr.Markdown("") images_html = gr.HTML("") # Arm A controls (definitions + legend printed inline, right where scored) with gr.Group(visible=False) as a_group: gr.HTML(LEGEND_A) gr.HTML(DEF_A) a_source = gr.Radio(["Real", "Synthetic"], label="Is this patch real or AI-generated?", info="Real = genuine acquired SRH; Synthetic = AI-generated.") a_conf = gr.Radio(["1", "2", "3", "4", "5"], label="Confidence", info="1 = pure guess ... 5 = certain") a_copied = gr.Radio(["No", "Yes"], label="Does it look copied/memorized from a real example?") a_quality = gr.Radio(["Plausible", "Minor artifacts", "Implausible"], label="Clinical plausibility") a_note = gr.Textbox(label="Note (optional)", lines=1) # Arm B controls with gr.Group(visible=False) as b_group: gr.HTML(LEGEND_B) gr.HTML(DEF_B) b_coherent = gr.Radio(["Yes", "Partial", "No"], label="Is this a coherent, clinically meaningful category?", info="Yes = one recognizable morphology; Partial = dominant + outliers; No = incoherent mix.") b_conf = gr.Radio(["1", "2", "3", "4", "5"], label="Confidence", info="1 = pure guess ... 5 = certain") b_desc = gr.Textbox(label="Optional: describe the morphology / putative entity", lines=1) b_note = gr.Textbox(label="Note (optional)", lines=1) status_md = gr.Markdown("") save_btn = gr.Button("Save & Next", variant="primary") done_md = gr.Markdown("", visible=False) CASE_OUTPUTS = [arm_badge_md, prompt_md, images_html, progress_md, done_md, a_group, a_source, a_conf, a_copied, a_quality, a_note, b_group, b_coherent, b_conf, b_desc, b_note] def _done_updates(progress): return [gr.update(value=""), gr.update(value=""), gr.update(value=""), gr.update(value=progress), gr.update(value=DONE_MSG, visible=True), gr.update(visible=False), gr.update(value=None), gr.update(value=None), gr.update(value=None), gr.update(value=None), gr.update(value=""), gr.update(visible=False), gr.update(value=None), gr.update(value=None), gr.update(value=""), gr.update(value="")] def render_case(annotator, order, idx, records=None): if records is None: records = load_existing_responses(annotator) progress = f"**{len(completed_case_ids(records))} / {N_CASES}** items completed" if idx >= len(order): return _done_updates(progress) case = CASES_BY_ID[order[idx]] arm = case.get("arm", "A") if arm == "A": imgs = image_grid_html(f"a{idx}", [img_data_uri(case.get("image", ""))], single=True) return [gr.update(value="TASK A - realism"), gr.update(value=PROMPT_A), gr.update(value=imgs), gr.update(value=progress), gr.update(value="", visible=False), gr.update(visible=True), gr.update(value=None), gr.update(value=None), gr.update(value=None), gr.update(value=None), gr.update(value=""), gr.update(visible=False), gr.update(value=None), gr.update(value=None), gr.update(value=""), gr.update(value="")] imgs = image_grid_html(f"b{idx}", [img_data_uri(p) for p in case.get("images", [])][:ARMB_MAX_TILES]) return [gr.update(value="TASK B - discovered-category review"), gr.update(value=PROMPT_B), gr.update(value=imgs), gr.update(value=progress), gr.update(value="", visible=False), gr.update(visible=False), gr.update(value=None), gr.update(value=None), gr.update(value=None), gr.update(value=None), gr.update(value=""), gr.update(visible=True), gr.update(value=None), gr.update(value=None), gr.update(value=""), gr.update(value="")] def _blank(): return [gr.update() for _ in CASE_OUTPUTS] def nav_login(msg=""): return ([gr.update(value=""), gr.update(value=""), gr.update(visible=True), gr.update(visible=False), "", 0, [], gr.update(value=msg)] + _blank()) def nav_study(username, token, msg=""): records = load_existing_responses(username) order = reader_order(username) idx = first_unfinished_idx(order, records) ups = render_case(username, order, idx, records) return ([gr.update(value=token), gr.update(value=username), gr.update(visible=False), gr.update(visible=True), username, idx, order, gr.update(value=msg)] + ups) ALL_NAV = ([tok_out, user_out, login_col, study_col, annotator_state, index_state, order_state, login_msg] + CASE_OUTPUTS) # ---- auth ---- def do_login(name, pw): name = (name or "").strip() acc = load_accounts() custom = acc.get("accounts", {}); claimed = set(acc.get("claimed_invites", [])) if name in custom and verify_pw(pw, custom[name]): return nav_study(name, make_token(name)) if name in INVITES and name not in claimed and pw == INVITES[name]: # first login: create the account under the login name; the reader can rename later custom = acc.setdefault("accounts", {}); acc.setdefault("claimed_invites", []) custom[name] = new_pw_record(pw, invite=name) acc["claimed_invites"].append(name) save_accounts(acc) return nav_study(name, make_token(name)) if name in INVITES and name in claimed and name not in custom: return nav_login("That login was already used. Sign in with your current password.") return nav_login("Invalid username or password.") def do_auto_login(name, tok): name = (name or "").strip() if not valid_token(name, tok): return nav_login("") if name in load_accounts().get("accounts", {}): return nav_study(name, tok) return nav_login("") login_btn.click(do_login, [name_in, pass_in], ALL_NAV).then(None, [user_out, tok_out], None, js=SET_LS_JS) pass_in.submit(do_login, [name_in, pass_in], ALL_NAV).then(None, [user_out, tok_out], None, js=SET_LS_JS) # ---- change credentials (post-login) ---- def do_change(annotator, new_user, new_pw, confirm): new_user = (new_user or "").strip() acc = load_accounts(); custom = acc.setdefault("accounts", {}) if annotator not in custom: return [gr.update(value="Session error, please sign in again."), annotator, gr.update(value=annotator), gr.update(value=make_token(annotator))] rec = dict(custom[annotator]) if new_pw: if len(new_pw) < 6: return [gr.update(value="New password must be at least 6 characters."), annotator, gr.update(value=annotator), gr.update(value=make_token(annotator))] if new_pw != confirm: return [gr.update(value="New passwords do not match."), annotator, gr.update(value=annotator), gr.update(value=make_token(annotator))] rec = new_pw_record(new_pw, invite=rec.get("invite", "")) target = annotator if new_user and new_user != annotator: if len(new_user) < 3: return [gr.update(value="New username must be at least 3 characters."), annotator, gr.update(value=annotator), gr.update(value=make_token(annotator))] if new_user in custom or new_user in INVITES: return [gr.update(value="That username is taken. Choose another."), annotator, gr.update(value=annotator), gr.update(value=make_token(annotator))] migrate_responses(annotator, new_user) # keep the reader's saved work under the new name custom.pop(annotator, None) target = new_user custom[target] = rec save_accounts(acc) return [gr.update(value=f"Account updated. You are now signed in as **{target}**."), target, gr.update(value=target), gr.update(value=make_token(target))] cs_btn.click(do_change, [annotator_state, cs_user, cs_pw, cs_confirm], [cs_msg, annotator_state, user_out, tok_out]).then( None, [user_out, tok_out], None, js=SET_LS_JS) # ---- save & next ---- def do_save(annotator, order, idx, a_src, a_cf, a_cp, a_q, a_nt, b_co, b_cf, b_ds, b_nt): if idx >= len(order): return [idx, gr.update(value="Nothing to save.")] + _blank() case = CASES_BY_ID[order[idx]] arm = case.get("arm", "A") if arm == "A": missing = [lbl for lbl, v in [("real/synthetic", a_src), ("confidence", a_cf), ("copied?", a_cp), ("plausibility", a_q)] if not v] dims = {"task": "A", "judged_source": a_src, "confidence": a_cf, "looks_copied": a_cp, "plausibility": a_q, "note": a_nt or "", "true_source": case.get("true_source")} else: missing = [lbl for lbl, v in [("coherent?", b_co), ("confidence", b_cf)] if not v] dims = {"task": "B", "coherent": b_co, "confidence": b_cf, "description": b_ds or "", "note": b_nt or "", "cluster_id": case.get("cluster_id"), "is_control": case.get("is_control", "none")} if missing: return [idx, gr.update(value="Please answer: " + ", ".join(missing))] + _blank() rec = {"schema_version": SCHEMA_VERSION, "app_build": APP_BUILD, "annotator": annotator, "case_id": case["case_id"], "arm": arm, "item_id": "__case__", "dims": dims, "shown_position": idx, "ts": int(time.time())} records = save_records(annotator, [rec]) nxt = first_unfinished_idx(order, records) return [nxt, gr.update(value="Saved.")] + render_case(annotator, order, nxt, records) SAVE_INPUTS = [annotator_state, order_state, index_state, a_source, a_conf, a_copied, a_quality, a_note, b_coherent, b_conf, b_desc, b_note] SAVE_OUTPUTS = [index_state, status_md] + CASE_OUTPUTS save_btn.click(do_save, SAVE_INPUTS, SAVE_OUTPUTS) # ---- logout ---- def do_logout(): return (gr.update(visible=True), gr.update(visible=False), "", 0, []) logout_btn.click(do_logout, None, [login_col, study_col, annotator_state, index_state, order_state]).then( None, None, None, js=CLR_LS_JS) # ---- boot / resume ---- demo.load(None, None, [user_ls, tok_ls], js=GET_LS_JS).then(do_auto_login, [user_ls, tok_ls], ALL_NAV) return demo if __name__ == "__main__": _ensure_response_dataset() build_app().queue().launch(server_name="0.0.0.0", server_port=7860)