| """Generate H&E descriptions with GPT-6.1 Sol on Azure OpenAI for images that have no real text (plan A: |
| one FIXED 3-shot prefix of real PathGen-1.6M captions -> prompt-cached; short caption; low reasoning effort). |
| |
| Targets (images on disk without paired text): |
| ge_patches : PathGen-1.6M gene-expression patches (brain*, gliomas*, idc*, saromas*, skin*) 1640 |
| slide_rep : PathGen-1.6M/slide_report/image_set/*.png 1 |
| orion : ORION H&E tiles (he/*.jpeg) 6335 |
| hemit : HEMIT H&E inputs (he/*/input/*) 55 |
| Config (env): AZURE_OPENAI_ENDPOINT, AZURE_OPENAI_API_KEY, AZURE_OPENAI_DEPLOYMENT, |
| optional AZURE_OPENAI_API_VERSION (if unset the v1 endpoint <endpoint>/openai/v1/ is used) |
| Usage: python 27_azure_caption.py --dry # list targets + 3-shot prompt, no API call |
| python 27_azure_caption.py --pilot 20 # 20 images, prints measured cost / image |
| python 27_azure_caption.py --budget 40 # everything; stops when the USD budget is reached |
| Output: cache/gen_text/captions.jsonl (resumable; one line per image: set, key, path, caption, usage) |
| Generated text is SYNTHETIC (seen only the H&E) -> never use it as test text; tag ttype='synthetic'. |
| """ |
| import argparse |
| import base64 |
| import concurrent.futures as cf |
| import glob |
| import io |
| import json |
| import os |
| import random |
| import threading |
| import time |
|
|
| import numpy as np |
| import pandas as pd |
| from PIL import Image, PngImagePlugin |
|
|
| from common import CACHE, ROOT |
|
|
| PngImagePlugin.MAX_TEXT_CHUNK = 64 * 1024 * 1024 |
| P = os.path.join(ROOT, "..") |
| OUT = os.path.join(CACHE, "gen_text") |
| os.makedirs(OUT, exist_ok=True) |
| PRICE = {"in": 2.0, "cached": 0.10, "out": 10.0} |
| MAXPX = 512 |
|
|
| ap = argparse.ArgumentParser() |
| ap.add_argument("--sets", nargs="+", default=["ge_patches", "slide_rep", "orion", "hemit"], |
| help="tcga_notext = TCGA images left without real text after 40_clean_text.py; dlpfc / idc = external test sets (all spots); tcga_calib = 1000 TCGA patches that HAVE real text (held-out patients of 31_), for calibrating synthetic vs real text") |
| ap.add_argument("--pilot", type=int, default=0) |
| ap.add_argument("--budget", type=float, default=40.0) |
| ap.add_argument("--workers", type=int, default=8) |
| ap.add_argument("--effort", default="low") |
| ap.add_argument("--dry", action="store_true") |
| args = ap.parse_args() |
|
|
|
|
| def targets(): |
| rows = [] |
| ge = ["brain_image", "brain_train_image", "brain_trainnew_image", "gliomas_test_image_new", "gliomas_train_image", |
| "idc_image", "idc_train_image", "saromas_test_image_new", "saromas_train_image_new", "skin_image", "skin_train_image"] |
| src = {"ge_patches": [f for d in ge for f in glob.glob(os.path.join(P, "PathGen-1.6M", d, "*"))], |
| "slide_rep": glob.glob(os.path.join(P, "PathGen-1.6M", "slide_report", "image_set", "*")), |
| "orion": glob.glob(os.path.join(P, "ORION", "ORIONCRC_dataset_tile_20x", "he", "*.jpeg")), |
| "hemit": glob.glob(os.path.join(P, "HEMIT", "he", "*", "input", "*"))} |
| if "tcga_calib" in args.sets: |
| I = pd.read_csv(os.path.join(CACHE, "allimg_meta.csv")) |
| T = pd.read_csv(os.path.join(CACHE, "alltxt_meta.csv")) |
| seen = set(pd.read_csv(os.path.join(CACHE, "pathgen_meta.csv")).query("split in ['train', 'val']").patient) |
| ev = sorted(set(T.img_idx) & set(I.index[~I.patient.isin(seen)])) |
| src["tcga_calib"] = list(I.image_path.values[np.random.default_rng(0).choice(ev, 1000, replace=False)]) |
| if "tcga_notext" in args.sets: |
| src["tcga_notext"] = list(pd.read_csv(os.path.join(CACHE, "notext_meta.csv")).image_path) |
| for s in ("dlpfc", "idc"): |
| if s in args.sets: |
| n = len(np.load(os.path.join(CACHE, "external", EXT[s]), mmap_mode="r")) |
| rows += [{"set": s, "key": f"external/{s}/{i}", "path": f"npy:{s}:{i}"} for i in range(n)] |
| for s in args.sets: |
| if s in EXT: |
| continue |
| for f in sorted(src[s]): |
| if f.lower().endswith((".png", ".jpg", ".jpeg", ".tif", ".tiff")): |
| rows.append({"set": s, "key": os.path.relpath(f, P), "path": f}) |
| return pd.DataFrame(rows) |
|
|
|
|
| EXT = {"dlpfc": "dlpfc_he_05um.npy", "idc": "idc_he.npy"} |
|
|
|
|
| def b64(path): |
| if path.startswith("npy:"): |
| _, st, i = path.split(":") |
| im = Image.fromarray(np.asarray(np.load(os.path.join(CACHE, "external", EXT[st]), mmap_mode="r")[int(i)])) |
| else: |
| im = Image.open(path).convert("RGB") |
| im.thumbnail((MAXPX, MAXPX)) |
| buf = io.BytesIO() |
| im.save(buf, format="JPEG", quality=90) |
| return "data:image/jpeg;base64," + base64.b64encode(buf.getvalue()).decode() |
|
|
|
|
| def few_shot(): |
| """3 fixed real PathGen-1.6M captions (40-60 words), different slides; same for every request -> cacheable.""" |
| T = pd.read_csv(os.path.join(CACHE, "alltxt_meta.csv")) |
| I = pd.read_csv(os.path.join(CACHE, "allimg_meta.csv")) |
| T = T[(T.ttype == "caption") & T.source.str.contains("pathgen1.6m")] |
| w = T.caption.str.split().str.len() |
| T = T[(w >= 40) & (w <= 60)] |
| rng = random.Random(0) |
| picks, seen = [], set() |
| for i in rng.sample(list(T.index), len(T)): |
| wsi = I.wsi_id[T.img_idx[i]] |
| if wsi[:12] in seen: |
| continue |
| try: |
| img = b64(I.image_path[T.img_idx[i]]) |
| except Exception: |
| continue |
| seen.add(wsi[:12]); picks.append((img, T.caption[i], T.key[i])) |
| if len(picks) == 3: |
| return picks |
|
|
|
|
| SYSTEM = ("You are a pathologist writing a short description of one H&E-stained histopathology image patch. " |
| "Describe only what is visible: tissue type and architecture, cell types and their density, nuclear " |
| "features, stroma, inflammation, vessels, necrosis or mitoses when present. Follow the style and length " |
| "of the examples (one paragraph, about 50 words). Do not mention the examples, staining quality, " |
| "magnification or patient information, and do not invent a diagnosis that the image does not support.") |
|
|
|
|
| def messages(shots, target_img): |
| m = [{"role": "system", "content": SYSTEM}] |
| for img, cap, _ in shots: |
| m.append({"role": "user", "content": [{"type": "text", "text": "Describe this H&E patch."}, |
| {"type": "image_url", "image_url": {"url": img, "detail": "auto"}}]}) |
| m.append({"role": "assistant", "content": cap}) |
| m.append({"role": "user", "content": [{"type": "text", "text": "Describe this H&E patch."}, |
| {"type": "image_url", "image_url": {"url": target_img, "detail": "auto"}}]}) |
| return m |
|
|
|
|
| def client(): |
| from openai import AzureOpenAI, OpenAI |
| ep, key = os.environ["AZURE_OPENAI_ENDPOINT"].rstrip("/"), os.environ["AZURE_OPENAI_API_KEY"] |
| if os.environ.get("AZURE_OPENAI_API_VERSION"): |
| return AzureOpenAI(azure_endpoint=ep, api_key=key, api_version=os.environ["AZURE_OPENAI_API_VERSION"]) |
| return OpenAI(base_url=ep + "/openai/v1/", api_key=key) |
|
|
|
|
| def cost(u): |
| cached = (u.get("prompt_tokens_details") or {}).get("cached_tokens") or 0 |
| return ((u["prompt_tokens"] - cached) * PRICE["in"] + cached * PRICE["cached"] + u["completion_tokens"] * PRICE["out"]) / 1e6 |
|
|
|
|
| D = targets() |
| shots = few_shot() |
| print(f"targets: {len(D)} " + json.dumps(D.set.value_counts().to_dict())) |
| print("3-shot examples (PathGen-1.6M keys):", [k for _, _, k in shots]) |
| fout = os.path.join(OUT, "captions.jsonl") |
| done = set() |
| spent = 0.0 |
| if os.path.exists(fout): |
| for line in open(fout): |
| r = json.loads(line) |
| done.add(r["key"]); spent += r.get("usd", 0) |
| todo = D[~D.key.isin(done)] |
| if args.pilot: |
| todo = todo.groupby("set", group_keys=False).apply(lambda g: g.sample(min(len(g), max(1, args.pilot // D.set.nunique())), random_state=0)) |
| print(f"already done {len(done)} (USD {spent:.2f}); to do now {len(todo)}") |
| json.dump({"system": SYSTEM, "shots": [{"key": k, "caption": c} for _, c, k in shots], "maxpx": MAXPX, |
| "effort": args.effort, "deployment": os.environ.get("AZURE_OPENAI_DEPLOYMENT")}, |
| open(os.path.join(OUT, "prompt.json"), "w"), indent=1) |
| if args.dry: |
| for _, c, k in shots: |
| print(f"- {k}: {c}") |
| raise SystemExit |
|
|
| cl, dep = client(), os.environ["AZURE_OPENAI_DEPLOYMENT"] |
| lock = threading.Lock() |
| stop = threading.Event() |
|
|
|
|
| def run(row): |
| if stop.is_set(): |
| return |
| for attempt in range(6): |
| try: |
| r = cl.chat.completions.create(model=dep, messages=messages(shots, b64(row.path)), |
| reasoning_effort=args.effort, max_completion_tokens=2000, |
| prompt_cache_key="he-caption-3shot-v1") |
| u = r.usage.model_dump() |
| rec = {"set": row.set, "key": row.key, "path": row.path, "caption": r.choices[0].message.content.strip(), |
| "finish": r.choices[0].finish_reason, "usage": u, "usd": cost(u), "ttype": "synthetic"} |
| break |
| except Exception as e: |
| err = repr(e)[:300] |
| if "content_filter" in err or attempt == 5: |
| rec = {"set": row.set, "key": row.key, "path": row.path, "caption": None, "error": err, "usd": 0} |
| break |
| time.sleep(2 ** attempt * 5) |
| global spent |
| with lock: |
| with open(fout, "a") as f: |
| f.write(json.dumps(rec) + "\n") |
| spent += rec["usd"] |
| if spent >= args.budget: |
| stop.set() |
|
|
|
|
| t0 = time.time() |
| with cf.ThreadPoolExecutor(args.workers) as ex: |
| for i, _ in enumerate(ex.map(run, [r for r in todo.itertuples()])): |
| if i % 200 == 0: |
| print(f"{i}/{len(todo)} USD {spent:.2f} {time.time() - t0:.0f}s", flush=True) |
| R = pd.read_json(fout, lines=True) |
| ok = R[R.caption.notna()] |
| print(f"done {len(ok)} ok / {len(R)} total; failures {R.caption.isna().sum()}; USD spent {R.usd.sum():.2f}; " |
| f"USD/image {ok.usd.mean():.4f} -> projected for all {len(D)}: {ok.usd.mean() * len(D):.0f}") |
| if len(ok): |
| u = pd.json_normalize(ok.usage) |
| print(u.filter(regex="prompt_tokens$|completion_tokens$|cached_tokens|reasoning_tokens").mean().round(0).to_string()) |
| if stop.is_set(): |
| print(f"STOPPED: budget {args.budget} USD reached") |
|
|
