Spaces:
Sleeping
Sleeping
| """ | |
| 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 "<div class='imgcell'><div class='imgbox empty'>[no image]</div></div>" | |
| return f""" | |
| <div class='imgcell'> | |
| <div class='imgbox'><img id='{dom_id}' src='{uri}' data-bright='1' data-contrast='1' data-zoom='1'></div> | |
| <div class='strip'> | |
| <button type='button' onclick="rsZoom('{dom_id}',-0.25)" title='zoom out (this image only)'>−</button> | |
| <span class='lbl'>zoom</span> | |
| <button type='button' onclick="rsZoom('{dom_id}',0.25)" title='zoom in (this image only)'>+</button> | |
| <label class='lbl'>bright<input type='range' min='0.3' max='2.5' step='0.05' value='1' | |
| title='brightness (display only)' oninput="rsSet('{dom_id}','bright',this.value)"></label> | |
| <label class='lbl'>contrast<input type='range' min='0.3' max='2.5' step='0.05' value='1' | |
| title='contrast (display only)' oninput="rsSet('{dom_id}','contrast',this.value)"></label> | |
| </div> | |
| </div>""" | |
| 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"<div class='{cls}'>{cells}</div>" | |
| # ----------------------------------------------------------------------------- UI text (self-explanatory, on-screen) | |
| PROVENANCE = ( | |
| "<div class='prov'><b>Where these images come from.</b> This app validates an AI system for " | |
| "<b>Stimulated Raman Histology (SRH)</b> of brain tumors. In <b>Task A</b> a single patch is shown; it is " | |
| "either a <b>real</b> acquired SRH field or an <b>AI-generated</b> one, shown blinded and rendered identically " | |
| "as virtual H&E. In <b>Task B</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.</div>") | |
| RESOLUTION_NOTE = ( | |
| "<div class='legend' style='background:#fff5f5;border-color:#f0b3b3;'>" | |
| "<b style='color:#c0261c;'>About image resolution:</b> these are stimulated-Raman tissue fields shown at " | |
| "their native acquisition resolution, which is <b>uniform across every image</b> (it is not a quality " | |
| "defect and is unrelated to whether an image is real or AI-generated). Please assess <b>tissue morphology " | |
| "and pattern</b> rather than sharpness; use the <b>+ zoom</b> control on any image if helpful.</div>") | |
| 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 + | |
| "<div class='legend'>" | |
| "<b>Real vs AI-generated:</b> Real = a genuine acquired SRH field; AI-generated = synthesized by a model.<br>" | |
| "<b>Confidence 1-5:</b> 1 = pure guess, 2 = low, 3 = moderate, 4 = high, 5 = certain.<br>" | |
| "<b>Looks copied/memorized:</b> Yes if it looks like a near-duplicate of a specific real example.<br>" | |
| "<b>Clinical plausibility:</b> Plausible = could be genuine tissue; Minor artifacts = mostly realistic with " | |
| "small oddities; Implausible = clearly not real tissue.</div>") | |
| DEF_A = ("<span class='reddef'><b>Copied/memorized = a plausible-looking near-duplicate of a specific real field, " | |
| "not a fresh example.</b> Example: nuclei arrangement and background that appear lifted verbatim from one " | |
| "known image rather than a new, independently generated field.</span>") | |
| 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 + | |
| "<div class='legend'>" | |
| "<b>Coherent & meaningful (Yes):</b> the patches share one recognizable morphology that could correspond " | |
| "to a real tumor type/pattern.<br>" | |
| "<b>Partial:</b> a dominant shared pattern plus some outliers.<br>" | |
| "<b>No:</b> an incoherent mix with no shared morphology (a clustering artifact).<br>" | |
| "<b>Confidence 1-5:</b> 1 = pure guess ... 5 = certain.</div>") | |
| DEF_B = ("<span class='reddef'><b>Meaningful = a morphology a pathologist would recognize as one entity/pattern, " | |
| "not merely visually similar noise.</b> 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).</span>") | |
| 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 = """ | |
| <style> | |
| .grid { display:flex; flex-wrap:wrap; gap:8px; } | |
| .grid .imgcell { width:150px; } | |
| .grid.single .imgcell { width:384px; } | |
| .imgcell { display:flex; flex-direction:column; gap:3px; } | |
| .imgbox { overflow:auto; max-height:260px; border:1px solid #d0d5dd; border-radius:6px; background:#0b0b0b; } | |
| .grid.single .imgbox { max-height:384px; } | |
| .imgbox.empty { display:flex; align-items:center; justify-content:center; color:#999; height:120px; background:#f3f4f6; } | |
| .imgbox img { display:block; width:100%; height:auto; } | |
| .strip { display:flex; align-items:center; gap:6px; flex-wrap:wrap; font-size:11px; } | |
| .strip button { width:24px; height:22px; font-weight:700; cursor:pointer; } | |
| .strip .lbl { color:#475467; } | |
| .strip input[type=range] { width:64px; vertical-align:middle; } | |
| .reddef { color:#c0261c; font-size:13px; display:block; margin:2px 0 6px; line-height:1.35; } | |
| .legend { background:#f8fafc; border:1px solid #e4e7ec; border-radius:6px; padding:6px 9px; font-size:12.5px; | |
| color:#1d2939; margin:2px 0 6px; line-height:1.45; } | |
| .prov { background:#eef4ff; border:1px solid #cdddff; border-radius:6px; padding:7px 10px; font-size:12.5px; | |
| color:#1d2939; margin-bottom:6px; line-height:1.45; } | |
| .badge { font-size:15px; color:#101828; } | |
| </style> | |
| <script> | |
| window.rsApply = function(id){ | |
| var img = document.getElementById(id); if(!img) return; | |
| var b = img.dataset.bright||1, c = img.dataset.contrast||1, z = img.dataset.zoom||1; | |
| img.style.filter = 'brightness('+b+') contrast('+c+')'; | |
| img.style.transform = 'scale('+z+')'; img.style.transformOrigin = 'top left'; | |
| }; | |
| window.rsSet = function(id, kind, val){ var i=document.getElementById(id); if(!i) return; i.dataset[kind]=val; window.rsApply(id); }; | |
| window.rsZoom = function(id, d){ var i=document.getElementById(id); if(!i) return; var z=parseFloat(i.dataset.zoom||1)+d; if(z<0.2)z=0.2; if(z>6)z=6; i.dataset.zoom=z; window.rsApply(id); }; | |
| </script> | |
| """ | |
| # ----------------------------------------------------------------------------- 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="<span class='badge'><b>TASK A</b> - realism</span>"), | |
| 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="<span class='badge'><b>TASK B</b> - discovered-category review</span>"), | |
| 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) | |