import base64
import gc
import io
import json
import os
import re
IS_ZERO_GPU = os.environ.get("SPACES_ZERO_GPU") is not None
# ZeroGPU requires `spaces` to be imported before any CUDA-initializing package (importing torch /
# diffusers touches torch.cuda), so this block must stay above the torch/diffusers imports below.
if IS_ZERO_GPU:
import spaces
else:
class _NoSpaces:
@staticmethod
def GPU(*args, **kwargs):
if len(args) == 1 and callable(args[0]) and not kwargs:
return args[0]
def deco(fn):
return fn
return deco
spaces = _NoSpaces()
import gradio as gr # noqa: E402
import torch # noqa: E402
from diffusers import ModularPipeline # noqa: E402
# Captioning logic lives in the published modular block; the Space is just a UI + model host over it.
BLOCK_REPO = "OzzyGT/ideogram4_caption_blocks"
# Selectable checkpoints fed to the block (no on-the-fly quantization — the SDNQ repos are already quantized):
# bf16 full ~15GB (ZeroGPU / >=24GB GPU), SDNQ 8-bit ~9GB, SDNQ 4-bit ~6GB (both fit a 16GB GPU).
MODELS = {
"bf16 (full · google/gemma-4-E4B-it)": "google/gemma-4-E4B-it",
"SDNQ 8-bit (OzzyGT)": "OzzyGT/gemma_4_E4B_it_sdnq_dynamic_8bit",
"SDNQ 4-bit (OzzyGT)": "OzzyGT/gemma_4_E4B_it_sdnq_dynamic_4bit",
}
# bf16 on ZeroGPU (48GB, fastest); 8-bit locally (better quality than 4-bit, still fits a 16GB GPU).
DEFAULT_PRECISION = (
"bf16 (full · google/gemma-4-E4B-it)" if IS_ZERO_GPU else "SDNQ 8-bit (OzzyGT)"
)
# Ideogram 4's native caption schema (from the official ideogram-oss/ideogram4 docs). Key ORDER matters for
# quality, bbox is [ymin, xmin, ymax, xmax] in 0-1000 normalized coords, colors are UPPERCASE #RRGGBB. The
# verbatim example anchors the exact format.
DEFAULT_INSTRUCTION = (
"You are a captioner for the Ideogram 4 image model. Look at the image and output ONE JSON object (no prose, "
"no markdown fences) that reconstructs it in Ideogram 4's native caption schema. Keep keys in the exact order "
"shown.\n\n"
"SCHEMA (keys in order):\n"
"- high_level_description: one or two sentences summarizing the whole image (subject + medium/style).\n"
"- style_description: an object with EXACTLY ONE of `photo` or `art_style`:\n"
" photo caption order: aesthetics, lighting, photo, medium, color_palette\n"
" non-photo caption order: aesthetics, lighting, medium, art_style, color_palette\n"
' color_palette: optional, up to 16 UPPERCASE hex colors (e.g. "#1B1B2F").\n'
"- compositional_deconstruction: {background, elements}.\n"
" background: the global setting/backdrop.\n"
" elements: list; keys in order per type:\n"
" obj: type, bbox, desc, color_palette\n"
" text: type, bbox, text, desc, color_palette\n\n"
"RULES:\n"
"- bbox is [ymin, xmin, ymax, xmax] in 0-1000 normalized coordinates (origin top-left), independent of "
"resolution. bbox and per-element color_palette are optional (<=5 colors per element).\n"
"- Main subject first. A coherent subject (one person/animal/object) is exactly ONE element; its parts go in "
"its `desc`. `desc` is identity-first and concrete (color, material, pose, distinguishing features).\n"
"- Commit to one concrete value each (no 'various'/'or similar'). `text` elements carry the literal in-image "
"characters, verbatim. Colors UPPERCASE #RRGGBB. Describe only what is visible; do not invent.\n\n"
"EXAMPLE (format reference only):\n"
'{"high_level_description":"A medium-shot photograph of Formula 1 driver Max Verstappen wearing his Red Bull '
"Racing suit and cap, smiling as he holds his helmet and talks to a man in a white shirt and black vest at a "
'race track.","style_description":{"aesthetics":"saturated primary colors, rule of thirds, joyful and '
'triumphant","lighting":"overcast daylight, diffused, soft subtle shadows","photo":"shallow depth of field, '
'sharp focus, eye-level, telephoto","medium":"photograph"},"compositional_deconstruction":{"background":"An '
"out-of-focus racing paddock; blurred figures, a purple and white structure with a red F1 logo on the left, "
'daylight, sky not visible.","elements":[{"type":"obj","bbox":[55,642,1000,937],"desc":"An older man in '
"profile facing left, grey hair, fair skin, white long-sleeved button-down shirt under a navy blue quilted "
'vest, a slight smile."}]}}'
)
DEVICE = (
torch.device("cuda")
if (IS_ZERO_GPU or torch.cuda.is_available())
else torch.device("cpu")
)
# One pipeline resident at a time: ZeroGPU serves the bf16 default; locally the user picks one and switching reloads.
_current = {"id": None, "pipe": None}
def _load(checkpoint):
if "sdnq" in checkpoint.lower():
import sdnq # noqa: F401 registers the SDNQ backend so prebuilt quantized checkpoints load
pipe = ModularPipeline.from_pretrained(BLOCK_REPO, trust_remote_code=True)
pipe.load_components(
pretrained_model_name_or_path=checkpoint, torch_dtype=torch.bfloat16
)
pipe.to(DEVICE)
return pipe
def get_pipe(precision):
"""Return the caption pipeline, (re)loading the selected checkpoint if it changed (one resident at a time)."""
if IS_ZERO_GPU:
precision = DEFAULT_PRECISION # ZeroGPU serves the bf16 default only, whatever the caller passes
checkpoint = MODELS[precision]
if _current["id"] != checkpoint:
_current.update(id=None, pipe=None)
gc.collect()
if torch.cuda.is_available():
torch.cuda.empty_cache()
_current.update(id=checkpoint, pipe=_load(checkpoint))
print(f"loaded {checkpoint}")
return _current["pipe"]
# Two paths:
# ZeroGPU -> the platform hands out the GPU only for the duration of a @spaces.GPU call and takes
# it right back, so there's nothing to "hog". We just preload bf16 at module scope (the
# canonical ZeroGPU pattern) and skip the whole load/unload + quantization machinery.
# Local -> we hold a real GPU, so this support app must not squat on VRAM while the user runs the
# Ideogram generation app. Nothing loads until the Load button; Unload frees it again.
if IS_ZERO_GPU:
get_pipe(DEFAULT_PRECISION)
@spaces.GPU(duration=120)
def load_model(precision):
"""Load the selected model into VRAM (local only; nothing is resident until this runs)."""
get_pipe(precision)
return f"✅ Loaded: {precision}"
def unload_model():
"""Evict the resident model from VRAM and RAM so it stops hogging the GPU (local only)."""
_current.update(id=None, pipe=None)
gc.collect()
if torch.cuda.is_available():
torch.cuda.empty_cache()
return "⚪ No model loaded"
def _parse_json(text: str):
"""Parse the model output; return (dict|None, repaired: bool).
repaired=True means the JSON was incomplete (unbalanced brackets) and we closed it to parse —
a sign the output was cut short. Robust to markdown fences, stray text/braces, and control
chars in strings (strict=False).
"""
text = text.strip()
text = re.sub(r"^```(?:json)?\s*", "", text)
text = re.sub(r"\s*```$", "", text).strip()
start = text.find("{")
if start == -1:
return None, False
frag = text[start:]
# 1) first '{' .. last '}' (handles the common clean case + trailing prose without braces)
try:
obj = json.loads(frag[: frag.rfind("}") + 1], strict=False)
if isinstance(obj, dict):
return obj, False
except json.JSONDecodeError:
pass
# 2) decode just the first complete object, ignoring any trailing content (stray braces, prose)
try:
obj, _ = json.JSONDecoder(strict=False).raw_decode(frag)
if isinstance(obj, dict):
return obj, False
except json.JSONDecodeError:
pass
# 3) repair truncated/short output (e.g. a missing final '}') by balancing brackets, then parse
try:
obj = json.loads(_balance_json(frag), strict=False)
return (obj, True) if isinstance(obj, dict) else (None, False)
except json.JSONDecodeError:
return None, False
def _balance_json(frag: str) -> str:
"""Close unbalanced strings/brackets in a truncated JSON fragment so it can be parsed."""
stack, in_str, esc = [], False, False
for ch in frag:
if in_str:
if esc:
esc = False
elif ch == "\\":
esc = True
elif ch == '"':
in_str = False
elif ch == '"':
in_str = True
elif ch in "{[":
stack.append(ch)
elif ch == "}" and stack and stack[-1] == "{":
stack.pop()
elif ch == "]" and stack and stack[-1] == "[":
stack.pop()
out = frag
if in_str:
out += '"' # close a string cut off mid-value
out = out.rstrip()
if out.endswith(","):
out = out[:-1] # drop a dangling comma before the closers
for ch in reversed(stack):
out += "}" if ch == "{" else "]"
return out
def _parse_palette(text: str) -> list:
"""Parse a free-form list of hex colors into normalized UPPERCASE #RRGGBB (max 16)."""
out = []
for tok in re.split(r"[,\s]+", (text or "").strip()):
tok = tok.strip().upper().lstrip("#")
if re.fullmatch(r"[0-9A-F]{6}", tok):
out.append("#" + tok)
return out[:16]
_push_counter = {"v": 0}
def _nonce() -> int:
_push_counter["v"] += 1
return _push_counter["v"]
def _data_url(image, max_side=1400) -> str:
"""Downscaled JPEG data URL for the canvas. Full-res PNG of a real photo can be tens of MB,
which bloats the #bbox-in payload and makes the JS JSON.parse fail (so nothing renders). A
display-sized JPEG keeps it small; bboxes are 0-1000 normalized so resolution doesn't matter."""
img = image.convert("RGB")
w, h = img.size
if max(w, h) > max_side:
scale = max_side / max(w, h)
img = img.resize((max(1, round(w * scale)), max(1, round(h * scale))))
buf = io.BytesIO()
img.save(buf, format="JPEG", quality=85)
return "data:image/jpeg;base64," + base64.b64encode(buf.getvalue()).decode()
def _ensure_ids(state):
"""Give every element a stable _id (used to reconcile canvas edits back to elements)."""
cd = state.get("compositional_deconstruction") or {}
els = cd.get("elements") or []
n = state.get("_next_id", 0)
for el in els:
if not el.get("_id"):
el["_id"] = f"e{n}"
n += 1
state["_next_id"] = n
return state
def _elements(state) -> list:
cd = (state or {}).get("compositional_deconstruction") or {}
return cd.get("elements") or []
# ---- JSON export (schema-clean, ordered) ---------------------------------------------------------
def _clean_element(el: dict) -> dict:
"""Serialize one element in Ideogram4 key order, dropping empty optionals."""
out = {"type": el.get("type", "obj")}
if el.get("bbox"):
out["bbox"] = el["bbox"]
if el.get("type") == "text":
out["text"] = el.get("text", "")
if el.get("desc"):
out["desc"] = el["desc"]
else:
if el.get("desc"):
out["desc"] = el["desc"]
if el.get("color_palette"):
out["color_palette"] = el["color_palette"]
return out
def _export(state) -> dict:
"""Normalize the working state into the clean Ideogram4 JSON object."""
state = state or {}
out = {}
if state.get("high_level_description"):
out["high_level_description"] = state["high_level_description"]
if state.get("style_description"):
out["style_description"] = state["style_description"]
cd = state.get("compositional_deconstruction") or {}
cdd = {}
if cd.get("background"):
cdd["background"] = cd["background"]
cdd["elements"] = [_clean_element(e) for e in (cd.get("elements") or [])]
out["compositional_deconstruction"] = cdd
return out
def _dumps(state) -> str:
return json.dumps(_export(state), indent=2, ensure_ascii=False)
def canvas_payload(image, state, include_image=True) -> str:
"""Build the JSON payload for the JS editor (#bbox-in): the source image + the full element list
(type / bbox / desc / text / palette), each keyed by a stable id. The JS renders both the boxes
and the right-side cards from this. include_image=False = elements only (no image reload)."""
_ensure_ids(state)
elements = []
for el in _elements(state):
bbox = el.get("bbox")
if not (isinstance(bbox, (list, tuple)) and len(bbox) == 4):
bbox = [350, 350, 650, 650] # give box-less elements a default so they're editable
ymin, xmin, ymax, xmax = bbox
elements.append(
{
"id": el["_id"],
"type": el.get("type", "obj"),
"ymin": ymin, "xmin": xmin, "ymax": ymax, "xmax": xmax,
"desc": el.get("desc", ""),
"text": el.get("text", ""),
"colors": el.get("color_palette") or [],
}
)
payload = {"nonce": _nonce(), "elements": elements}
payload["style_colors"] = (state.get("style_description") or {}).get("color_palette") or []
payload["image"] = _data_url(image) if (include_image and image is not None) else None
return json.dumps(payload)
def reconcile_canvas(state, out_json):
"""Fold JS editor edits (#bbox-out) back into element state by stable id, preserving top-level
fields. The JS owns the element list, so this just rebuilds compositional_deconstruction.elements
and returns the new state (no re-render signalling needed — the JS already drew everything)."""
try:
data = json.loads(out_json or "{}")
except json.JSONDecodeError:
return state
out_els = data.get("elements") or []
by_id = {el.get("_id"): el for el in _elements(state) if el.get("_id")}
new_elems = []
for oe in out_els:
el = dict(by_id.get(oe.get("id")) or {})
el["_id"] = oe.get("id")
el["type"] = oe.get("type", "obj")
el["bbox"] = oe.get("bbox")
el["desc"] = oe.get("desc", "")
el["text"] = oe.get("text", "")
el["color_palette"] = _parse_palette(" ".join(oe.get("colors") or []))
new_elems.append(el)
new_state = dict(state or {})
# Style palette is edited on the canvas side too — fold it back, preserving other style fields.
if "style_colors" in data:
style = dict(new_state.get("style_description") or {})
colors = _parse_palette(" ".join(data.get("style_colors") or []))
if colors:
style["color_palette"] = colors
else:
style.pop("color_palette", None)
if style:
new_state["style_description"] = style
else:
new_state.pop("style_description", None)
cd = dict(new_state.get("compositional_deconstruction") or {})
cd["elements"] = new_elems
cd.setdefault("background", "")
new_state["compositional_deconstruction"] = cd
return new_state
def sync_from_state(image, state):
"""From the caption state: return (state-with-ids, top fields..., canvas payload)."""
state = dict(state or {})
cd = dict(state.get("compositional_deconstruction") or {})
cd["elements"] = [dict(e) for e in (cd.get("elements") or [])]
state["compositional_deconstruction"] = cd
_ensure_ids(state)
style = state.get("style_description") or {}
if "art_style" in style and "photo" not in style:
kind, kind_val = "art_style", style.get("art_style", "")
else:
kind, kind_val = "photo", style.get("photo", "")
background = cd.get("background", "")
return (
state,
state.get("high_level_description", ""),
background,
kind,
style.get("aesthetics", ""),
style.get("lighting", ""),
style.get("medium", ""),
kind_val,
canvas_payload(image, state, include_image=True),
)
def rebuild_from_fields(
state, hld, background, kind, aesthetics, lighting, medium, kind_val
):
"""Write the non-label fields back into the caption state (preserving elements + the JS-managed
color palette) and refresh JSON."""
state = dict(state or {})
state["high_level_description"] = hld
# Key order matters for quality: photo -> aesthetics, lighting, photo, medium, color_palette;
# non-photo -> aesthetics, lighting, medium, art_style, color_palette.
colors = (state.get("style_description") or {}).get("color_palette") or [] # owned by the canvas
style = {}
if aesthetics:
style["aesthetics"] = aesthetics
if lighting:
style["lighting"] = lighting
if kind == "photo":
if kind_val:
style["photo"] = kind_val
if medium:
style["medium"] = medium
else:
if medium:
style["medium"] = medium
if kind_val:
style["art_style"] = kind_val
if colors:
style["color_palette"] = colors
if style:
state["style_description"] = style
else:
state.pop("style_description", None)
cd = dict(state.get("compositional_deconstruction") or {})
cd["background"] = background
cd.setdefault("elements", [])
state["compositional_deconstruction"] = cd
return state, _dumps(state)
def apply_json(image, text):
"""Parse the raw-JSON tab back into state, refreshing every field + the canvas."""
data, _ = _parse_json(text)
if data is None:
raise gr.Error("Invalid JSON — could not parse.")
return sync_from_state(image, data) # (state, fields..., payload)
@spaces.GPU(duration=120)
def caption(
image,
instruction,
precision,
max_new_tokens,
temperature,
):
if image is None:
raise gr.Error("An image is required.")
pipe = get_pipe(precision)
text = pipe(
image=image,
instruction=instruction,
max_new_tokens=int(max_new_tokens),
temperature=float(temperature),
output="caption_raw",
)
data, repaired = _parse_json(text)
if data is None:
return {}, text, (
"⚠️ Couldn't parse the model output as JSON — it may have been truncated at the token "
"limit. Raise **Max new tokens** and re-run, or fix it in the JSON tab and click Apply JSON."
)
# Ended naturally but dropped trailing brackets = complete content we just closed.
status = (
"✅ Caption ready (minor auto-repair of the closing brackets)."
if repaired
else "✅ Caption ready."
)
return data, _dumps(data), status
CSS = """
#caption-btn {
background: linear-gradient(180deg, #2563eb 0%, #1e3a8a 100%) !important;
color: #fff !important; border: none !important;
box-shadow: inset 0 1px 0 rgba(255,255,255,0.15), 0 1px 2px rgba(0,0,0,0.25) !important;
}
#load-btn {
background: linear-gradient(180deg, #15803d 0%, #14532d 100%) !important;
color: #fff !important; border: none !important;
box-shadow: inset 0 1px 0 rgba(255,255,255,0.15), 0 1px 2px rgba(0,0,0,0.25) !important;
}
#unload-btn {
background: linear-gradient(180deg, #b91c1c 0%, #7f1d1d 100%) !important;
color: #fff !important; border: none !important;
box-shadow: inset 0 1px 0 rgba(255,255,255,0.15), 0 1px 2px rgba(0,0,0,0.25) !important;
}
#caption-btn, #load-btn, #unload-btn { min-height: 56px !important; height: 56px !important; }
/* ---- Scrollless, full-viewport layout: the page never scrolls; only the fields column does. ---- */
html, body { height: 100%; margin: 0; overflow: hidden; }
gradio-app { height: var(--app-h, 100vh); overflow: hidden; }
.gradio-container {
height: var(--app-h, 100vh) !important;
max-width: 100% !important;
padding: 6px 14px 4px !important;
overflow: hidden !important;
}
#app-title { margin: 0 0 2px !important; }
#app-title h1 { font-size: 1.2rem !important; margin: 0 !important; }
#app-title small {
font-size: 0.68rem; line-height: 1.35;
color: var(--body-text-color-subdued, #9aa0ac);
}
/* The right-hand editor column is the only thing that scrolls (like the desktop app). */
#fields-col {
max-height: calc(var(--app-h, 100vh) - 118px);
overflow-y: auto;
overflow-x: hidden;
padding-right: 6px;
}
#canvas-col { min-height: 0; }
/* Theme tokens for the custom editor — light by default, dark when Gradio adds .dark to