Spaces:
Sleeping
Sleeping
| """Backend for the partial-prompt human study (Hugging Face Space, Docker SDK). | |
| - Builds counterbalanced lists from the stimuli dataset (one member of each minimal-pair group per participant). | |
| - Assigns each new participant to the least-filled list. | |
| - Receives the final session JSON and writes it to DATA_DIR; a CommitScheduler pushes DATA_DIR | |
| to a private dataset repo every few minutes. | |
| Stimuli are loaded at startup from a Hugging Face dataset (DATASET_REPO), either an imagefolder | |
| (`metadata.jsonl` + image files) or parquet with embedded images. Required columns: `sample_id`, | |
| `prompt`, and the image (`file_name` / `image`). Optional: `human_study` (false = never shown), | |
| `role` + `control_type` (attention checks), `pair_ids`, | |
| `image_id`, plus any label columns (copied into the saved responses). Ranking questions come from | |
| `groups/groups.jsonl` when the dataset has one (one group per line: `group_id`, `prompt`, | |
| `visual_style`, `human_study`, `members` with `image_id` and `file_name` relative to `data/`). | |
| Restart the Space to pick up a new version of the dataset; the exact dataset commit is stored with | |
| every session. | |
| Environment variables (set them as Space secrets / variables): | |
| DATASET_REPO stimuli dataset (default "Enzo-py/partial-prompting-ds") | |
| DATASET_REVISION optional branch / tag / commit to pin (default "main") | |
| DATASET_SPLIT optional split (default "train") | |
| RESPONSES_REPO dataset repo that receives responses, e.g. "Enzo-py/partial-prompting-responses" | |
| HF_TOKEN write token for RESPONSES_REPO (secret) | |
| PROLIFIC_COMPLETION_CODE optional; participants are redirected to Prolific with this code | |
| STUDY_VERSION optional label stored with every session (default "pilot-v1") | |
| ADMIN_KEY optional; protects /api/status and enables /admin | |
| STUDY_CONTACT optional; researcher contact shown on the consent page | |
| PID_SALT optional secret; saves a salted hash of the Prolific ID (never the ID itself) | |
| LIST_COUNT, LIST_SIZE optional; number of lists (default: just enough to show every stimulus) and | |
| study images per list (default 14, at most) | |
| CHECKS_PER_LIST optional; attention checks per list, rotating (default 2) | |
| RANKINGS_PER_LIST optional; ranking questions after the ratings (default 5, 0 = none) | |
| """ | |
| import hashlib | |
| import json | |
| import math | |
| import os | |
| import random | |
| import threading | |
| import time | |
| import uuid | |
| from collections import Counter, defaultdict | |
| from datetime import datetime, timezone | |
| from pathlib import Path | |
| from fastapi import FastAPI, HTTPException, Request | |
| from fastapi.responses import FileResponse, JSONResponse | |
| from fastapi.staticfiles import StaticFiles | |
| ROOT = Path(__file__).resolve().parent | |
| STIM_DIR = Path(os.environ.get("STIM_DIR", ROOT / "stimuli_cache")) # images extracted from the dataset | |
| DATA_DIR = Path(os.environ.get("DATA_DIR", ROOT / "data")) | |
| DATA_DIR.mkdir(parents=True, exist_ok=True) | |
| RESPONSES_REPO = os.environ.get("RESPONSES_REPO") | |
| HF_TOKEN = os.environ.get("HF_TOKEN") | |
| COMPLETION_CODE = os.environ.get("PROLIFIC_COMPLETION_CODE") | |
| STUDY_VERSION = os.environ.get("STUDY_VERSION", "pilot-v1") | |
| ADMIN_KEY = os.environ.get("ADMIN_KEY") | |
| PID_SALT = os.environ.get("PID_SALT") # optional secret; stores a hash of the Prolific ID | |
| STUDY_CONTACT = os.environ.get("STUDY_CONTACT") # optional; who to contact, shown on the consent page | |
| PENDING_TTL = 30 * 60 # a started session counts towards its list for 30 min | |
| # ---------------------------------------------------------------- stimuli & lists | |
| DATASET_REPO = os.environ.get("DATASET_REPO", "Enzo-py/partial-prompting-ds") | |
| DATASET_REVISION = os.environ.get("DATASET_REVISION", "main") | |
| DATASET_SPLIT = os.environ.get("DATASET_SPLIT", "train") | |
| def _unique_paths(api, sha, files): | |
| """Drop files whose content duplicates an earlier one (the same file uploaded at two paths).""" | |
| seen, unique = set(), [] | |
| for info in sorted(api.get_paths_info(DATASET_REPO, files, repo_type="dataset", revision=sha), | |
| key=lambda i: i.path): | |
| oid = info.lfs.sha256 if info.lfs else info.blob_id | |
| if oid in seen: | |
| print(f"Skipping {info.path}: identical to an already selected file") | |
| continue | |
| seen.add(oid) | |
| unique.append(info.path) | |
| return unique | |
| def _parquet_rows(local, files): | |
| """Rows from parquet files with embedded images, as (row, image bytes, original image path).""" | |
| import pyarrow.parquet as pq | |
| for f in files: | |
| for r in pq.read_table(local / f).to_pylist(): | |
| img = r.pop("image", None) | |
| data = img.get("bytes") if isinstance(img, dict) else None | |
| if not data: | |
| raise RuntimeError(f"{r.get('sample_id')}: image bytes are not embedded in {f}") | |
| yield r, data, img.get("path") or "" | |
| def _imagefolder_rows(local, files): | |
| """Rows from imagefolder metadata.jsonl files; `file_name` is relative to the metadata file.""" | |
| for f in files: | |
| folder = (local / f).parent | |
| for n, line in enumerate((local / f).read_text().splitlines(), 1): | |
| if not line.strip(): | |
| continue | |
| r = json.loads(line) | |
| if not r.get("file_name"): | |
| raise RuntimeError(f"{f}, line {n}: no 'file_name'") | |
| path = folder / r.pop("file_name") | |
| if not path.is_file(): | |
| raise RuntimeError(f"{f}, line {n}: image {path.relative_to(local)} not found") | |
| yield r, path.read_bytes(), path.name | |
| WEB_MAX_SIDE = 1024 # images are never shown larger than this | |
| def web_image(data, path): | |
| """Write `data` to STIM_DIR/<path>.webp, resized to WEB_MAX_SIDE and compressed (a 2 MB PNG becomes | |
| ~150 KB), and return that name. Click marks are stored relative to the image, so resizing keeps them valid.""" | |
| import io | |
| from PIL import Image | |
| name = f"{path}.webp" | |
| with Image.open(io.BytesIO(data)) as im: | |
| im = im.convert("RGBA" if im.mode in ("RGBA", "LA", "P") else "RGB") | |
| im.thumbnail((WEB_MAX_SIDE, WEB_MAX_SIDE), Image.LANCZOS) | |
| (STIM_DIR / name).parent.mkdir(parents=True, exist_ok=True) | |
| im.save(STIM_DIR / name, "WEBP", quality=85, method=4) | |
| return name | |
| def load_stimuli(): | |
| """Download the stimuli, write the images to STIM_DIR, return (rows, commit sha). | |
| Reads either an imagefolder (`metadata.jsonl` next to the images) or parquet files with the | |
| images embedded, the two layouts the Hub uses for image datasets. | |
| """ | |
| from huggingface_hub import HfApi, snapshot_download | |
| api = HfApi(token=HF_TOKEN) | |
| sha = api.dataset_info(DATASET_REPO, revision=DATASET_REVISION).sha | |
| repo_files = api.list_repo_files(DATASET_REPO, repo_type="dataset", revision=sha) | |
| meta = [f for f in repo_files if f.split("/")[-1] == "metadata.jsonl"] | |
| if meta: | |
| # An imagefolder split lives in a folder named after it; a single folder is the default split. | |
| in_split = [f for f in meta if DATASET_SPLIT in f.split("/")[:-1]] | |
| files = _unique_paths(api, sha, in_split or meta) | |
| patterns = [f"{f.rsplit('/', 1)[0]}/**" if "/" in f else "**" for f in files] | |
| reader = _imagefolder_rows | |
| else: | |
| files = [f for f in repo_files | |
| if f.endswith(".parquet") and (f"/{DATASET_SPLIT}-" in f or f"/{DATASET_SPLIT}/" in f | |
| or f.split("/")[-1].startswith(DATASET_SPLIT))] | |
| files = _unique_paths(api, sha, files) if files else [] | |
| patterns, reader = files, _parquet_rows | |
| if not files: | |
| raise RuntimeError(f"No metadata.jsonl or parquet file for split '{DATASET_SPLIT}' in {DATASET_REPO}@{sha[:8]}") | |
| local = Path(snapshot_download(DATASET_REPO, repo_type="dataset", revision=sha, | |
| allow_patterns=patterns, token=HF_TOKEN)) | |
| STIM_DIR.mkdir(parents=True, exist_ok=True) | |
| rows = [] | |
| for r, data, src in reader(local, sorted(files)): | |
| for col in ("sample_id", "prompt"): | |
| if r.get(col) is None: | |
| raise RuntimeError(f"Row without '{col}' in {', '.join(files)}") | |
| r["file_name"] = web_image(data, r["sample_id"]) | |
| r["role"] = r.get("role") or "test" | |
| r["pair_ids"] = list(r.get("pair_ids") or []) | |
| r["image_id"] = r.get("image_id") or r["sample_id"] | |
| rows.append(r) | |
| ids = [r["sample_id"] for r in rows] | |
| if len(ids) != len(set(ids)): | |
| raise RuntimeError("sample_id values must be unique") | |
| print(f"Loaded {len(rows)} stimuli from {DATASET_REPO}@{sha[:8]} ({', '.join(files)})") | |
| return rows, sha, repo_files | |
| GROUPS_FILE = "groups/groups.jsonl" | |
| def load_groups(sha, repo_files): | |
| """Ranking groups defined by the dataset: images to put in order under one prompt. Each image is | |
| written to STIM_DIR/groups/<image_id>; groups with fewer than 2 images are skipped.""" | |
| if GROUPS_FILE not in repo_files: | |
| print(f"No {GROUPS_FILE} in the dataset: no ranking questions") | |
| return [] | |
| from huggingface_hub import hf_hub_download, snapshot_download | |
| path = hf_hub_download(DATASET_REPO, GROUPS_FILE, repo_type="dataset", revision=sha, token=HF_TOKEN) | |
| lines = [json.loads(l) for l in Path(path).read_text().splitlines() if l.strip()] | |
| sources = {f"data/{m['file_name']}" for g in lines for m in g.get("members") or []} | |
| local = Path(snapshot_download(DATASET_REPO, repo_type="dataset", revision=sha, | |
| allow_patterns=sorted(sources), token=HF_TOKEN)) | |
| (STIM_DIR / "groups").mkdir(parents=True, exist_ok=True) | |
| groups = [] | |
| for n, g in enumerate(lines, 1): | |
| if g.get("human_study") is False: | |
| continue | |
| images = {} | |
| for m in g.get("members") or []: | |
| src = local / "data" / m["file_name"] | |
| if not src.is_file(): | |
| raise RuntimeError(f"{GROUPS_FILE}, line {n}: image data/{m['file_name']} not found") | |
| name = web_image(src.read_bytes(), f"groups/{m['image_id']}") | |
| images[m["image_id"]] = {"image_id": m["image_id"], "sample_id": m.get("sample_id"), "file_name": name} | |
| if len(images) < 2: | |
| print(f"Skipping group {g.get('group_id')}: {len(images)} image(s), nothing to rank") | |
| continue | |
| groups.append({"set_id": g["group_id"], "prompt": g["prompt"], "visual_style": g.get("visual_style"), | |
| "images": list(images.values())}) | |
| ids = [g["set_id"] for g in groups] | |
| if len(ids) != len(set(ids)): | |
| raise RuntimeError("group_id values must be unique") | |
| return groups | |
| def load_segments(sha, repo_files, stimuli): | |
| """Segmentations from the dataset (`segmentation/<image_id>.png` + `.json`, COCO panoptic). For each | |
| stimulus whose image has one, the map is copied to STIM_DIR/seg/<image_id>.png (`seg_file`) so that | |
| question 2 can let participants pick segments instead of clicking points. The category names stay | |
| server-side; returns {image_id: {segment id: {category, names, isthing}}}. Images without a usable | |
| segmentation (missing, unreadable or empty) keep point marking; a failure never stops the app.""" | |
| wanted = sorted({s["image_id"] for s in stimuli if in_study(s)} | |
| & {f[len("segmentation/"):-5] for f in repo_files | |
| if f.startswith("segmentation/") and f.endswith(".json")}) | |
| wanted = [i for i in wanted if f"segmentation/{i}.png" in repo_files] | |
| if not wanted: | |
| return {} | |
| from huggingface_hub import snapshot_download | |
| from PIL import Image | |
| try: | |
| local = Path(snapshot_download(DATASET_REPO, repo_type="dataset", revision=sha, token=HF_TOKEN, | |
| allow_patterns=[f"segmentation/{i}.{e}" for i in wanted for e in ("png", "json")])) | |
| except Exception as e: | |
| print(f"Segmentations not loaded, point marking everywhere: {e}") | |
| return {} | |
| (STIM_DIR / "seg").mkdir(parents=True, exist_ok=True) | |
| segments = {} | |
| for i in wanted: | |
| try: | |
| info = json.loads((local / "segmentation" / f"{i}.json").read_text()) | |
| known = {seg["id"]: {"category": seg.get("category"), "names": seg.get("names"), | |
| "isthing": seg.get("isthing")} | |
| for seg in info.get("segments_info") or [] if isinstance(seg.get("id"), int) and seg["id"]} | |
| with Image.open(local / "segmentation" / f"{i}.png") as im: | |
| im.verify() | |
| if not known: | |
| raise ValueError("no segments") | |
| (STIM_DIR / "seg" / f"{i}.png").write_bytes((local / "segmentation" / f"{i}.png").read_bytes()) | |
| except Exception as e: | |
| print(f"Segmentation of {i} skipped, point marking instead: {e}") | |
| continue | |
| segments[i] = known | |
| for st in stimuli: | |
| if st["image_id"] in segments: | |
| st["seg_file"] = f"seg/{st['image_id']}.png" | |
| print(f"Segmentations for {len(segments)} images") | |
| return segments | |
| STIMULI, DATASET_SHA, REPO_FILES = load_stimuli() | |
| BY_ID = {s["sample_id"]: s for s in STIMULI} | |
| # Labels for the saved files: every dataset column except the ones already sent or stored elsewhere. | |
| LABEL_FIELDS = list(dict.fromkeys(k for s in STIMULI for k in s if k not in ("sample_id", "prompt", "file_name"))) | |
| # Public fields sent to the browser. Taxonomy labels stay server-side so participants never see them; | |
| # visual_style is shown above the image so the drawing style is not mistaken for a mismatch. | |
| PUBLIC_FIELDS = ("sample_id", "prompt", "file_name", "visual_style", "seg_file") | |
| LIST_COUNT = int(os.environ.get("LIST_COUNT") or 0) # number of lists; 0 = just enough to show every stimulus | |
| LIST_SIZE = int(os.environ.get("LIST_SIZE", 14)) # study images per list (an upper bound), checks not included | |
| CHECKS_PER_LIST = int(os.environ.get("CHECKS_PER_LIST", 2)) # attention checks per list, rotating | |
| RANKINGS_PER_LIST = int(os.environ.get("RANKINGS_PER_LIST", 5)) # ranking questions after the ratings | |
| def in_study(s): | |
| """Stimuli marked human_study = false only serve the pair table and are never shown.""" | |
| return s.get("human_study", True) is not False | |
| def is_attention_check(s): | |
| """A control whose text the image plainly contradicts (`control_type`: absent, relation, ...): its | |
| answer is a rating of 1. Other controls (e.g. "every detail true") are the control condition of a | |
| pair and are counterbalanced like any other stimulus. Older datasets mark the check as case E1.""" | |
| return s.get("role") == "control" and bool( | |
| s.get("control_type") or s.get("subcategory") == "Attention check" or s.get("case_id") == "E1") | |
| def conflict(a, b): | |
| """Two stimuli never go in the same list: same minimal pair, same image, or same text.""" | |
| return bool(set(a["pair_ids"]) & set(b["pair_ids"])) or a["image_id"] == b["image_id"] or a["prompt"] == b["prompt"] | |
| def design_lists(ids, neighbours, n_lists, size, fixed=None, seed=0, restarts=10, steps=20000): | |
| """n_lists conflict-free lists of `size` items from `ids`, each item shown about equally often. | |
| `fixed[l]` are items already in list l (e.g. its attention checks): nothing conflicting with them | |
| goes in that list. Local search, deterministic for a given dataset: swap an item out of a list | |
| and another in, keeping the list conflict-free, whenever exposures get more even. Conflicting | |
| items (the members of a pair) are also pushed towards equal exposure, so cells stay balanced. | |
| """ | |
| fixed = fixed or [set() for _ in range(n_lists)] | |
| pool = set(ids) | |
| pairs = [(a, b) for a in ids for b in neighbours[a] if b in pool and a < b] | |
| target = n_lists * size / len(ids) | |
| def cost(count): | |
| return (sum((count[i] - target) ** 2 + (1000 if count[i] == 0 else 0) for i in ids) | |
| + sum((count[a] - count[b]) ** 2 for a, b in pairs)) | |
| best = None | |
| for r in range(restarts): | |
| rng = random.Random(seed * 1000 + r) | |
| count, lists = Counter(), [] | |
| for l in range(n_lists): # greedy start: least shown items first | |
| cur = set() | |
| for i in sorted(ids, key=lambda i: (count[i], rng.random())): | |
| if len(cur) < size and not neighbours[i] & (cur | fixed[l]): | |
| cur.add(i) | |
| lists.append(cur) | |
| count.update(cur) | |
| if any(len(cur) < size for cur in lists): | |
| continue | |
| c = cost(count) | |
| for _ in range(steps): | |
| l = rng.randrange(n_lists) | |
| cur = lists[l] | |
| out = rng.choice(sorted(cur)) | |
| blocked = (cur - {out}) | fixed[l] | |
| cand = [i for i in ids if i not in cur and not neighbours[i] & blocked] | |
| if not cand: | |
| continue | |
| new = rng.choice(cand) | |
| count[out] -= 1 | |
| count[new] += 1 | |
| c2 = cost(count) | |
| if c2 <= c: | |
| cur.remove(out) | |
| cur.add(new) | |
| c = c2 | |
| else: | |
| count[out] += 1 | |
| count[new] -= 1 | |
| if best is None or c < best[0]: | |
| best = (c, [sorted(cur) for cur in lists]) | |
| return best[1] if best else None | |
| def build_lists(stimuli): | |
| """Counterbalanced lists: CHECKS_PER_LIST attention checks plus up to LIST_SIZE study images each, | |
| never two conflicting stimuli together (see `conflict`), so each comparison is between-subjects. | |
| Returns (lists, attention checks, stimuli not in the study). | |
| """ | |
| study = [s for s in stimuli if in_study(s)] | |
| checks = sorted(s["sample_id"] for s in study if is_attention_check(s)) | |
| items = sorted(s["sample_id"] for s in study if not is_attention_check(s)) | |
| neighbours = {s["sample_id"]: set() for s in study} | |
| for a in study: | |
| for b in study: | |
| if a is not b and conflict(a, b): | |
| neighbours[a["sample_id"]].add(b["sample_id"]) | |
| # Stimuli sharing a text, an image or a pair all conflict with each other, so each needs its own | |
| # list: the largest such group is a lower bound on the number of lists. | |
| groups = Counter() | |
| for s in study: | |
| if not is_attention_check(s): | |
| groups.update([("text", s["prompt"]), ("image", s["image_id"])] + [("pair", p) for p in s["pair_ids"]]) | |
| min_lists = max(groups.values(), default=1) | |
| # Largest list size that still fits: there may be fewer conflict-free items than LIST_SIZE asks. | |
| for size in range(min(LIST_SIZE, len(items)), 0, -1): | |
| n_lists = LIST_COUNT or max(math.ceil(len(items) / size), min_lists) | |
| if not LIST_COUNT: # spread the items evenly over those lists | |
| size = math.ceil(len(items) / n_lists) | |
| k = min(CHECKS_PER_LIST, len(checks)) | |
| check_lists = (design_lists(checks, neighbours, n_lists, k) if k else None) or [[] for _ in range(n_lists)] | |
| lists = design_lists(items, neighbours, n_lists, size, fixed=[set(c) for c in check_lists]) | |
| if lists: | |
| break | |
| else: | |
| check_lists, lists = [[]], [[]] | |
| not_shown = [s["sample_id"] for s in stimuli if not in_study(s)] | |
| return [sorted(c + l) for c, l in zip(check_lists, lists)], checks, not_shown | |
| def assign_rankings(sets, n_lists): | |
| """RANKINGS_PER_LIST groups per list, each group in about as many lists, never two groups with | |
| the same text in one list.""" | |
| if not sets or not RANKINGS_PER_LIST: | |
| return [[] for _ in range(n_lists)] | |
| ids = [r["set_id"] for r in sets] | |
| by_id = {r["set_id"]: r for r in sets} | |
| neighbours = {a: {b for b in ids if b != a and by_id[a]["prompt"] == by_id[b]["prompt"]} for a in ids} | |
| k = min(RANKINGS_PER_LIST, len({r["prompt"] for r in sets})) | |
| return design_lists(ids, neighbours, n_lists, k) or [[] for _ in range(n_lists)] | |
| SEGMENTS = load_segments(DATASET_SHA, REPO_FILES, STIMULI) | |
| LISTS, CHECKS, NOT_IN_STUDY = build_lists(STIMULI) | |
| STUDY = [s for s in STIMULI if in_study(s)] | |
| EXPOSURE = Counter(i for l in LISTS for i in l) | |
| RANKING_SETS = load_groups(DATASET_SHA, REPO_FILES) | |
| SET_BY_ID = {r["set_id"]: r for r in RANKING_SETS} | |
| LIST_RANKINGS = assign_rankings(RANKING_SETS, len(LISTS)) | |
| (ROOT / "lists.json").write_text(json.dumps( | |
| {"n_lists": len(LISTS), "attention_checks": CHECKS, "not_in_study": NOT_IN_STUDY, | |
| "exposure": dict(sorted(EXPOSURE.items())), "lists": LISTS, | |
| "ranking_sets": RANKING_SETS, "list_rankings": LIST_RANKINGS}, indent=2)) | |
| print(f"{len(LISTS)} lists of {sorted({len(l) for l in LISTS})} images, {min(CHECKS_PER_LIST, len(CHECKS))} attention " | |
| f"checks each; {len(NOT_IN_STUDY)} stimuli not in the study (human_study = false); " | |
| f"{len(RANKING_SETS)} ranking groups, {len(LIST_RANKINGS[0])} per list") | |
| # ---------------------------------------------------------------- persistence | |
| scheduler = None | |
| if RESPONSES_REPO and HF_TOKEN: | |
| from huggingface_hub import CommitScheduler | |
| scheduler = CommitScheduler( | |
| repo_id=RESPONSES_REPO, repo_type="dataset", folder_path=DATA_DIR, | |
| path_in_repo="responses", every=2, private=True, token=HF_TOKEN, | |
| ) | |
| lock = threading.Lock() | |
| completed = Counter() | |
| pending = {} # session_id -> (list_id, started_at) | |
| def recount(): | |
| """Count completed sessions per list, from the responses repo and the local folder.""" | |
| seen = set() | |
| if RESPONSES_REPO and HF_TOKEN: | |
| try: | |
| from huggingface_hub import HfApi | |
| for f in HfApi(token=HF_TOKEN).list_repo_files(RESPONSES_REPO, repo_type="dataset"): | |
| if f.startswith("responses/list") and f.endswith(".json"): | |
| seen.add(f.split("/", 1)[1]) | |
| except Exception as e: # repo may not exist yet | |
| print("recount: could not list responses repo:", e) | |
| for p in DATA_DIR.glob("list*/*.json"): | |
| seen.add(f"{p.parent.name}/{p.name}") | |
| for f in seen: | |
| try: | |
| completed[int(f[4:6])] += 1 | |
| except ValueError: | |
| pass | |
| recount() | |
| def choose_list(): | |
| now = time.time() | |
| active = Counter(l for l, t in pending.values() if now - t < PENDING_TTL) | |
| return min(range(len(LISTS)), key=lambda k: (completed[k] + active[k], random.random())) | |
| # ---------------------------------------------------------------- API | |
| app = FastAPI(title="Partial prompt study") | |
| async def start(req: Request): | |
| body = await req.json() | |
| with lock: | |
| list_id = choose_list() | |
| session_id = uuid.uuid4().hex | |
| pending[session_id] = (list_id, time.time()) | |
| stimuli = [{k: BY_ID[i].get(k) for k in PUBLIC_FIELDS} for i in LISTS[list_id]] | |
| rankings = [{k: SET_BY_ID[r][k] for k in ("set_id", "prompt", "visual_style")} | |
| | {"images": [{k: im[k] for k in ("image_id", "file_name")} for im in SET_BY_ID[r]["images"]]} | |
| for r in LIST_RANKINGS[list_id]] | |
| return {"session_id": session_id, "list_id": list_id, "study_version": STUDY_VERSION, | |
| "stimuli": stimuli, "rankings": rankings, | |
| "prolific": bool(body.get("prolific_pid")), "contact": STUDY_CONTACT} | |
| async def submit(req: Request): | |
| body = await req.json() | |
| session_id = str(body.get("session_id", "")) | |
| if not session_id.isalnum() or len(session_id) > 64: | |
| raise HTTPException(400, "invalid session_id") | |
| list_id = int(body.get("list_id", -1)) | |
| if not 0 <= list_id < len(LISTS): | |
| raise HTTPException(400, "invalid list_id") | |
| # Attach the taxonomy labels server-side so the stored file is self-describing. | |
| for t in body.get("trials", []): | |
| s = BY_ID.get(t.get("sample_id")) | |
| if s: | |
| for k in LABEL_FIELDS: | |
| t[k] = s.get(k) | |
| # Selected segments (question 2 on segmented images): keep known ids, add their labels. | |
| known = SEGMENTS.get(s["image_id"], {}) | |
| if t.get("segments"): | |
| t["segments"] = [{"id": i, **known[i]} for i in t["segments"] if isinstance(i, int) and i in known] | |
| # Rankings: keep only known groups and their own images (image_id), in the order the participant | |
| # chose; the matching stimulus (sample_id, null for images rated only in the group) is added alongside. | |
| rankings = [] | |
| for r in body.get("rankings") or []: | |
| st = SET_BY_ID.get(r.get("set_id")) | |
| ims = {im["image_id"]: im for im in st["images"]} if st else {} | |
| order = r.get("order") or [] | |
| if st and sorted(order) == sorted(ims): | |
| rankings.append({**r, "prompt": st["prompt"], "visual_style": st["visual_style"], | |
| "order_sample_ids": [ims[i]["sample_id"] for i in order]}) | |
| body["rankings"] = rankings | |
| # Prolific IDs are never saved: only whether the participant came from Prolific, plus an optional | |
| # salted hash (PID_SALT) that lets the researchers match a Prolific submission to its session. | |
| participant = body.get("participant") or {} | |
| pid = participant.pop("prolific_pid", None) | |
| participant.pop("prolific_session_id", None) | |
| participant["source"] = "prolific" if pid else "direct" | |
| if pid and PID_SALT: | |
| participant["pid_hash"] = hashlib.sha256(f"{PID_SALT}:{pid}".encode()).hexdigest()[:16] | |
| body["participant"] = participant | |
| body["received_at"] = datetime.now(timezone.utc).isoformat() | |
| body["study_version"] = STUDY_VERSION | |
| body["dataset"] = {"repo": DATASET_REPO, "revision": DATASET_SHA, "split": DATASET_SPLIT} | |
| folder = DATA_DIR / f"list{list_id:02d}" | |
| folder.mkdir(parents=True, exist_ok=True) | |
| path = folder / f"{session_id}.json" | |
| ctx = scheduler.lock if scheduler else lock | |
| with ctx: | |
| new = not path.exists() | |
| path.write_text(json.dumps(body, ensure_ascii=False, indent=1)) | |
| with lock: | |
| pending.pop(session_id, None) | |
| if new: | |
| completed[list_id] += 1 | |
| code = COMPLETION_CODE if pid else None | |
| redirect = f"https://app.prolific.com/submissions/complete?cc={code}" if code else None | |
| return {"ok": True, "completion_code": code, "redirect": redirect} | |
| def status(key: str = ""): | |
| if ADMIN_KEY and key != ADMIN_KEY: | |
| raise HTTPException(403, "add ?key=ADMIN_KEY") | |
| now = time.time() | |
| return {"study_version": STUDY_VERSION, | |
| "dataset": f"{DATASET_REPO}@{DATASET_SHA[:8]} ({DATASET_SPLIT}, {len(STIMULI)} stimuli)", | |
| "n_lists": len(LISTS), "items_per_list": [len(l) for l in LISTS], | |
| "completed_per_list": {k: completed[k] for k in range(len(LISTS))}, | |
| "in_progress": sum(1 for _, t in pending.values() if now - t < PENDING_TTL), | |
| "saving_to": RESPONSES_REPO if scheduler else "local folder only (RESPONSES_REPO or HF_TOKEN missing)"} | |
| # ---------------------------------------------------------------- admin summary | |
| RESP_CACHE = ROOT / "responses_cache" # local copy of the responses repo, refreshed by /api/summary | |
| summary_lock = threading.Lock() | |
| session_cache = {"at": 0.0, "sessions": {}} | |
| def load_sessions(max_age=60): | |
| """Every saved session, keyed by 'listNN/<id>.json': the responses repo plus the local folder.""" | |
| with summary_lock: | |
| if time.time() - session_cache["at"] < max_age: | |
| return session_cache["sessions"] | |
| files = [] | |
| if scheduler: | |
| try: | |
| from huggingface_hub import snapshot_download | |
| snapshot_download(RESPONSES_REPO, repo_type="dataset", allow_patterns="responses/**", | |
| local_dir=RESP_CACHE, token=HF_TOKEN) | |
| except Exception as e: # repo may not exist yet | |
| print("summary: could not download responses repo:", e) | |
| files += RESP_CACHE.glob("responses/list*/*.json") | |
| files += DATA_DIR.glob("list*/*.json") # after the repo copy: local files may not be pushed yet | |
| sessions = {} | |
| for p in files: | |
| try: | |
| sessions[f"{p.parent.name}/{p.name}"] = json.loads(p.read_text()) | |
| except (OSError, ValueError) as e: # e.g. a file being written | |
| print(f"summary: skipping {p}: {e}") | |
| session_cache.update(at=time.time(), sessions=sessions) | |
| return sessions | |
| # Quality checks. analysis/collect_responses.py applies the same rules. Each failed attention check | |
| # counts once; a session is excluded once it fails EXCLUDE_AT checks, fewer are only a warning. | |
| FAST_TRIAL_MS = 1000 # an image answered faster than this... | |
| FAST_SHARE = 0.25 # ...for this share of the images or more | |
| SAME_ANSWER_SHARE = 0.99 # share of identical ratings across a session | |
| EXCLUDE_AT = 3 # failed checks that exclude a session | |
| # A check fails when the rating says the image matches. Only an absent object makes "Partly" (3) | |
| # wrong; a wrong colour, count or relation can honestly be rated 3, so those fail from "Mostly" (4). | |
| CHECK_FAIL_AT = {"absent": 3} | |
| CHECK_FAIL_AT_DEFAULT = 4 | |
| def failed_check(t): | |
| """Whether a trial is an attention check rated as matching. Uses the labels saved with the trial, | |
| so a session is judged on the stimuli it was shown even if the dataset changed since.""" | |
| s = t if "control_type" in t else BY_ID.get(t.get("sample_id"), {}) # older sessions: current labels | |
| if not is_attention_check(s): | |
| return False | |
| kind = s.get("control_type") or "absent" # the original check (E1, "a banana") is an absent object | |
| return t["rating"] >= CHECK_FAIL_AT.get(kind, CHECK_FAIL_AT_DEFAULT) | |
| def quality_flags(session): | |
| """The checks a session fails, one entry per failed attention-check stimulus.""" | |
| trials = [t for t in session.get("trials", []) if t.get("rating")] | |
| flags = ["attention_check" for t in trials if failed_check(t)] | |
| if trials and sum((t.get("rt") or 0) < FAST_TRIAL_MS for t in trials) / len(trials) >= FAST_SHARE: | |
| flags.append("too_fast") | |
| ratings = Counter(t["rating"] for t in trials) | |
| if trials and ratings.most_common(1)[0][1] / len(trials) >= SAME_ANSWER_SHARE: | |
| flags.append("same_answers") | |
| return flags | |
| def is_excluded(session): | |
| return len(quality_flags(session)) >= EXCLUDE_AT | |
| def median(values): | |
| v = sorted(values) | |
| return (v[len(v) // 2] + v[(len(v) - 1) // 2]) / 2 if v else None | |
| def mean(values): | |
| return sum(values) / len(values) if values else None | |
| def summarize(sessions): | |
| """Progress per list and response statistics. Excluded sessions (EXCLUDE_AT failed checks) are | |
| counted in the progress but left out of the response statistics.""" | |
| now = time.time() | |
| active = Counter(l for l, t in pending.values() if now - t < PENDING_TTL) | |
| done = Counter(s.get("list_id") for s in sessions) | |
| flags = {id(s): quality_flags(s) for s in sessions} | |
| included = [s for s in sessions if len(flags[id(s)]) < EXCLUDE_AT] | |
| trials = defaultdict(list) | |
| for s in included: | |
| for t in s.get("trials", []): | |
| if t.get("sample_id") in BY_ID and t.get("rating"): | |
| trials[t["sample_id"]].append(t) | |
| stimuli = [] | |
| for st in STUDY: | |
| ts = trials[st["sample_id"]] | |
| ratings = [t["rating"] for t in ts] | |
| stimuli.append({ | |
| **{k: st.get(k) for k in ("sample_id", "prompt", "file_name", "case_id", "category", | |
| "category_name", "subcategory", "role")}, | |
| "attention_check": is_attention_check(st), | |
| "n": len(ts), "mean": mean(ratings), | |
| "dist": [ratings.count(r) for r in range(1, 6)], | |
| "problem_rate": mean([t.get("problem", t.get("unwanted")) == "yes" for t in ts]), | |
| "median_rt_s": (median([t["rt"] for t in ts if t.get("rt")]) or 0) / 1000 or None, | |
| "marks": [[m["x"], m["y"]] for t in ts for m in t.get("marks", [])], | |
| "segments": Counter(seg["category"] for t in ts for seg in t.get("segments") or []).most_common(), | |
| }) | |
| by_cat = defaultdict(list) | |
| for st in STUDY: | |
| if not is_attention_check(st): | |
| by_cat[(st.get("category"), st.get("category_name"))] += trials[st["sample_id"]] | |
| categories = [{"category": c, "name": name, "n": len(ts), "mean": mean([t["rating"] for t in ts]), | |
| "problem_rate": mean([t.get("problem", t.get("unwanted")) == "yes" for t in ts])} | |
| for (c, name), ts in sorted(by_cat.items(), key=lambda kv: str(kv[0][0]))] | |
| by_pair = defaultdict(list) | |
| for st in STUDY: | |
| roles = list(st.get("pair_roles") or []) | |
| for i, pid in enumerate(st["pair_ids"]): | |
| ratings = [t["rating"] for t in trials[st["sample_id"]]] | |
| by_pair[pid].append({"sample_id": st["sample_id"], "prompt": st["prompt"], | |
| "pair_role": roles[i] if i < len(roles) else None, | |
| "n": len(ratings), "mean": mean(ratings)}) | |
| pairs = [{"pair_id": pid, "members": members} for pid, members in sorted(by_pair.items())] | |
| recent = sorted(sessions, key=lambda s: s.get("received_at") or "", reverse=True)[:100] | |
| return { | |
| "study_version": STUDY_VERSION, | |
| "dataset": f"{DATASET_REPO}@{DATASET_SHA[:8]} ({DATASET_SPLIT}, {len(STIMULI)} stimuli)", | |
| "saving_to": RESPONSES_REPO if scheduler else "local folder only (RESPONSES_REPO or HF_TOKEN missing)", | |
| "generated_at": datetime.now(timezone.utc).isoformat(), | |
| "totals": {"completed": len(sessions), "excluded": len(sessions) - len(included), | |
| "exclude_at": EXCLUDE_AT, | |
| "flagged": sum(0 < len(flags[id(s)]) < EXCLUDE_AT for s in sessions), | |
| "excluded_by": dict(Counter(f for s in sessions if len(flags[id(s)]) >= EXCLUDE_AT for f in flags[id(s)])), | |
| "in_progress": sum(active.values()), | |
| "median_minutes": (median([s["total_time_ms"] for s in sessions | |
| if s.get("total_time_ms")]) or 0) / 60000 or None}, | |
| "lists": [{"list_id": k, "items": len(LISTS[k]), "completed": done[k], "in_progress": active[k], | |
| # What the list shows, for the admin page: its images (checks marked) and its rankings. | |
| "stimuli": [{"sample_id": i, "prompt": BY_ID[i]["prompt"], "file_name": BY_ID[i]["file_name"], | |
| "check": is_attention_check(BY_ID[i])} for i in LISTS[k]], | |
| "rankings": [{"set_id": r, "prompt": SET_BY_ID[r]["prompt"], | |
| "images": [im["file_name"] for im in SET_BY_ID[r]["images"]]} | |
| for r in LIST_RANKINGS[k]]} | |
| for k in range(len(LISTS))], | |
| "categories": categories, "pairs": pairs, "stimuli": stimuli, "rankings": summarize_rankings(included), | |
| # No Prolific IDs here: the page only needs to tell sessions apart. | |
| "sessions": [{"session": str(s.get("session_id", ""))[:8], "list_id": s.get("list_id"), | |
| "received_at": s.get("received_at"), | |
| "minutes": (s.get("total_time_ms") or 0) / 60000 or None, | |
| "prolific": s.get("participant", {}).get("source") == "prolific" | |
| or bool(s.get("participant", {}).get("prolific_pid")), # sessions saved before IDs were dropped | |
| "english": s.get("participant", {}).get("english"), | |
| "t2i": s.get("participant", {}).get("t2i"), | |
| "flags": flags[id(s)], "excluded": len(flags[id(s)]) >= EXCLUDE_AT} | |
| for s in recent], | |
| } | |
| def summarize_rankings(sessions): | |
| """Per ranking group: how many times it was ranked, and each image's mean rank (1 = favourite).""" | |
| ranks = defaultdict(lambda: defaultdict(list)) | |
| for s in sessions: | |
| for r in s.get("rankings") or []: | |
| for k, i in enumerate(r.get("order") or [], 1): | |
| ranks[r.get("set_id")][i].append(k) | |
| out = [] | |
| for st in RANKING_SETS: | |
| got = ranks[st["set_id"]] | |
| images = [{**im, "case_id": (BY_ID.get(im["sample_id"]) or {}).get("case_id"), | |
| "n": len(got[im["image_id"]]), "mean_rank": mean(got[im["image_id"]]), | |
| "first": mean([k == 1 for k in got[im["image_id"]]])} | |
| for im in st["images"]] | |
| images.sort(key=lambda im: (im["mean_rank"] is None, im["mean_rank"] or 0)) | |
| out.append({**{k: st[k] for k in ("set_id", "prompt", "visual_style")}, | |
| "n": max((im["n"] for im in images), default=0), "images": images}) | |
| return out | |
| def summary(key: str = "", refresh: bool = False): | |
| # Unlike /api/status, this exposes response content, so it stays closed until ADMIN_KEY is set. | |
| if not ADMIN_KEY: | |
| raise HTTPException(403, "set the ADMIN_KEY secret on the Space to enable the dashboard") | |
| if key != ADMIN_KEY: | |
| raise HTTPException(403, "wrong key") | |
| return summarize(list(load_sessions(max_age=0 if refresh else 60).values())) | |
| # ---------------------------------------------------------------- participant feedback | |
| FEEDBACK_MIN_OTHERS = 5 # a stimulus is compared only once this many other participants rated it | |
| FEEDBACK_MIN_ITEMS = 3 # below this many comparable stimuli, no feedback is shown | |
| def feedback_for(session): | |
| """Compare a participant's ratings with the other participants', after their answers are saved. | |
| Attention checks are never used (that would reveal them), nor are excluded sessions. Returns None until enough other participants have rated this participant's stimuli. | |
| """ | |
| others = defaultdict(list) | |
| for s in load_sessions().values(): | |
| if s.get("session_id") == session.get("session_id") or is_excluded(s): | |
| continue | |
| for t in s.get("trials", []): | |
| if t.get("rating"): | |
| others[t.get("sample_id")].append(t["rating"]) | |
| rows = [] | |
| for t in session.get("trials", []): | |
| st = BY_ID.get(t.get("sample_id")) | |
| if st and not is_attention_check(st) and t.get("rating") and len(others[st["sample_id"]]) >= FEEDBACK_MIN_OTHERS: | |
| ratings = others[st["sample_id"]] | |
| rows.append((st, t["rating"], mean(ratings), len(ratings))) | |
| if len(rows) < FEEDBACK_MIN_ITEMS: | |
| return None | |
| st, rating, m, n = max(rows, key=lambda r: abs(r[1] - r[2])) | |
| return { | |
| "n_compared": len(rows), | |
| "mean_diff": round(mean([r - m for _, r, m, _ in rows]), 2), | |
| "n_close": sum(abs(r - m) <= 1 for _, r, m, _ in rows), | |
| # Only a real disagreement is worth showing. | |
| "most_different": {**{k: st.get(k) for k in PUBLIC_FIELDS}, "rating": rating, | |
| "others_mean": round(m, 1), "others_n": n} if abs(rating - m) >= 1 else None, | |
| } | |
| async def feedback(req: Request): | |
| body = await req.json() | |
| session_id, list_id = str(body.get("session_id", "")), body.get("list_id") | |
| if not session_id.isalnum() or len(session_id) > 64 or not isinstance(list_id, int): | |
| raise HTTPException(400, "invalid session") | |
| path = DATA_DIR / f"list{list_id:02d}" / f"{session_id}.json" | |
| if not path.exists(): # feedback only for saved sessions | |
| raise HTTPException(404, "session not found") | |
| return {"feedback": feedback_for(json.loads(path.read_text()))} | |
| def admin(): | |
| return FileResponse(ROOT / "static" / "admin.html") | |
| app.mount("/stimuli", StaticFiles(directory=STIM_DIR), name="stimuli") | |
| app.mount("/static", StaticFiles(directory=ROOT / "static"), name="static") | |
| def index(): | |
| return FileResponse(ROOT / "static" / "index.html") | |