from __future__ import annotations
import base64
import html
import json
import os
from uuid import uuid4
from io import BytesIO
import gradio as gr
from PIL import Image
from grounding_demo.catalog import ROOT, load_cases
from grounding_demo.client import GroundingClient
from grounding_demo.console import TASK_HELP, compile_query, metrics_html, query_from_prompt, targets_html, task_help_html
from grounding_demo.protocol import parse_gam, parse_reference_boxes
from grounding_demo.render import draw_primitives
ROUTE_META = {
"grounding": ("Grounding", "Open-vocabulary object localization"),
"dense": ("Dense", "Dense instance localization"),
"referring": ("Referring", "Language-guided referring"),
"dense_point": ("Dense Point", "Dense point prediction"),
"referring_point_in_mask": ("Referring Point in Mask", "Point inside a referred target"),
"dense_point_in_mask": ("Dense Point in Mask", "Point inside multiple targets"),
"visual_prompt": ("Visual Prompt", "Exemplar-guided localization"),
"ocr": ("OCR", "Text recognition and localization"),
"gui": ("GUI", "Interface element grounding"),
}
CASES = load_cases()
BY_ID = {row["id"]: row for row in CASES}
BY_ROUTE = {route: [row for row in CASES if row["route"] == route] for route in ROUTE_META}
def route_choices(route):
return [(row["title"], row["id"]) for row in BY_ROUTE[route]]
def gallery_items(route):
return [
(str(ROOT / row["image"]), case_caption(row, i))
for i, row in enumerate(BY_ROUTE[route], 1)
]
def case_caption(row, index):
return f"{ROUTE_META[row['route']][0]} {index:02d}"
def input_view(path, canonical, route):
"""Show only user-supplied reference boxes; the client still receives raw path."""
references = parse_reference_boxes(canonical) if route == "visual_prompt" else ()
if not references or not path:
return path
with Image.open(path) as source:
return draw_primitives(source, (), references, "INPUT · example boxes")
def query_update(canonical, route):
return gr.update(value=query_from_prompt(canonical, route), placeholder=TASK_HELP[route][3])
def case_meta(row):
return (
"
TRAINING SOURCE"
f"{html.escape(row['source'])}IMAGE SHA256"
f"{row.get('image_sha256', '')[:16]}"
"Curated research examples for qualitative exploration, not a held-out benchmark. Inference receives only the image and task prompt.
YOUR IMAGEUploaded image"
"Select a task and describe the visual evidence to extract.
",
)
def choose_upload(path, route, canonical):
return (
gr.update(value=None),
gr.update(selected_index=None),
*upload_image(path, route, canonical),
)
def edit_query(text, route, path):
return (
compile_query(text, route),
input_view(path, compile_query(text, route), route),
metrics_html(ROUTE_META[route][0]),
targets_html([]),
"[]",
"",
status_html("Prompt updated · Run grounding to refresh the overlay"),
)
def result_bundle(path, canonical, route, raw, elapsed=None, usage=None):
parsed = parse_gam(raw, route)
items = [dict(label=p.label, coordinates=list(p.coordinates)) for p in parsed.primitives]
valid_items = items if parsed.valid else []
image = Image.open(path).convert("RGB")
source_bytes = BytesIO()
image.save(source_bytes, format="PNG")
image_source = "data:image/png;base64," + base64.b64encode(source_bytes.getvalue()).decode()
rendered = (
draw_primitives(
image,
parsed.primitives,
parse_reference_boxes(canonical),
)
if parsed.valid
else image
)
state = "Completed"
if not parsed.valid:
state = "Protocol error"
message = (
f"Complete · {len(valid_items)} visual targets"
if parsed.valid
else "Output validation failed · " + "; ".join(parsed.errors)
)
return dict(
image=rendered,
metrics=metrics_html(ROUTE_META[route][0], state, len(valid_items), elapsed, usage),
targets=targets_html(valid_items, image_source=image_source),
normalized=json.dumps(valid_items, ensure_ascii=False, indent=2),
raw=raw,
status=status_html(
message, "error" if not parsed.valid else "ready"
),
)
def display_bundle(bundle):
return tuple(
bundle[key] for key in ("image", "metrics", "targets", "normalized", "raw", "status")
)
def inference_error(exc):
timed_out = "timeout" in type(exc).__name__.lower()
return (
"Request timed out. Check the connection and retry; your image and prompt are preserved."
if timed_out else
"Inference unavailable. Check the connection and retry; your image and prompt are preserved."
)
def run_inference(path, query_text, route):
if not path:
raise gr.Error("Choose an example or upload an image first.")
canonical = compile_query(query_text, route)
try:
raw, elapsed, usage = GroundingClient().infer(path, canonical)
bundle = result_bundle(path, canonical, route, raw, elapsed, usage)
except Exception as exc:
message = inference_error(exc)
return (
path,
metrics_html(ROUTE_META[route][0], "API error"),
targets_html([]),
"[]",
f"{type(exc).__name__}: {exc}",
status_html(message, "error"),
)
return display_bundle(bundle)
RUN_LABEL = "Run Grounding"
def run_inference_stream(path, query_text, route):
"""Expose real request state without an opaque progress mask or synthetic steps."""
if not path:
raise gr.Error("Choose an example or upload an image first.")
yield (
path,
metrics_html(ROUTE_META[route][0], "Awaiting inference"),
targets_html([], pending=True),
"[]",
"",
status_html("Awaiting inference", "busy"),
gr.update(value="Grounding in progress…", interactive=False),
)
try:
result = run_inference(path, query_text, route)
except Exception as exc:
result = (
path,
metrics_html(ROUTE_META[route][0], "API error"),
targets_html([]),
"[]",
f"{type(exc).__name__}: {exc}",
status_html(inference_error(exc), "error"),
)
yield (*result, gr.update(value=RUN_LABEL, interactive=True))
def run_console(path, query_text, route):
# Identifies this request even when a proxy coalesces busy + final SSE events.
run_id = uuid4().hex
for state in run_inference_stream(path, query_text, route):
state = list(state)
state[5] = state[5].replace("data-phase=", f"data-run-id='{run_id}' data-phase=", 1)
yield (*state, state[-1])
def service_status():
ok, detail = GroundingClient().health()
return (
f"
"
f"{'Inference ready' if ok else 'Inference offline'}"
f"{'Model loaded. Ready to process your image.' if ok else 'Explore examples while the inference service is unavailable.'}
Model"
+ html.escape(os.getenv("GROUNDING_MODEL", "grounding-pi-vlm"))
+ "Max output tokens"
+ html.escape(os.getenv("GROUNDING_MAX_TOKENS", "1024"))
+ "Read-only runtime configuration. Only the current image and task prompt enter inference.
"
)
run_btn = gr.Button(RUN_LABEL, variant="primary", elem_id="run-button")
with gr.Column(elem_id="service-panel"):
health = gr.HTML("
""",
elem_id="canvas-chrome",
)
with gr.Row(elem_id="query-row"):
query = gr.Textbox(
value=query_from_prompt(first_case["prompt"], first_route),
label="What would you like to locate?",
show_label=False,
lines=2,
max_lines=4,
placeholder=TASK_HELP[first_route][3],
elem_id="query-dock",
)
gr.HTML("
Enter to run · Shift + Enter for a new line
")
mobile_run = gr.Button(RUN_LABEL, variant="primary", elem_id="mobile-run-button")
with gr.Column(elem_id="targets-panel"):
targets = gr.HTML(targets_html([]), elem_id="targets-content")
with gr.Row(elem_id="review-row", equal_height=True):
with gr.Column(scale=4, min_width=320, elem_id="sandbox-panel"):
gr.HTML(
"
Explore Examples
Curated visual scenarios
"
)
gallery = gr.Gallery(
value=gallery_items(first_route),
columns=3,
rows=1,
height=170,
show_label=False,
allow_preview=False,
object_fit="cover",
selected_index=2,
interactive=False,
elem_id="case-gallery",
)
with gr.Accordion("Data provenance", open=False, elem_id="source-details"):
meta = gr.HTML(case_meta(first_case))
with gr.Column(scale=6, min_width=420, elem_id="review-summary"):
with gr.Row(elem_id="result-columns", equal_height=True):
with gr.Column(scale=4, min_width=200, elem_id="metrics-panel"):
gr.HTML("
Inference Metrics
")
metrics = gr.HTML(metrics_html(ROUTE_META[first_route][0]))
with gr.Column(scale=6, min_width=240, elem_id="sidebar"):
gr.HTML(
"
Target Detail
Select a category below the canvas to magnify its original pixels.