#!/usr/bin/env python3
"""
LocateAnything WebUI — a small dependency-free (stdlib only) web front-end for an
ax-llm `serve` instance running the LocateAnything-3B grounding/detection model.
Run:
AXLLM_SERVE_URL=http://127.0.0.1:8010 \
AXLLM_IMAGE_DIR=/path/to/sample_images \
python3 locateanything_webui.py --port 7861
then open http://localhost:7861
The backend serves the UI, lists sample thumbnails from AXLLM_IMAGE_DIR, and proxies
detection to the ax-llm OpenAI-compatible /v1/chat/completions endpoint with stream=true,
parsing complete .. / .. tokens and re-emitting them to the browser
as clean SSE events (status / box / done) for real-time incremental drawing.
"""
import os, sys, json, re, argparse, mimetypes, urllib.request, urllib.error
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
SERVE_URL = os.environ.get("AXLLM_SERVE_URL", "http://127.0.0.1:8010").rstrip("/")
MODEL = os.environ.get("AXLLM_MODEL", "AXERA-TECH/LocateAnything-3B")
IMAGE_DIR = os.environ.get("AXLLM_IMAGE_DIR", "")
IMG_EXT = (".jpg", ".jpeg", ".png", ".bmp", ".webp")
# ----------------------------------------------------------------------------- token parsing
REF_RE = re.compile(r"(.*?)")
BOX_RE = re.compile(r"((?:<\d+>)+)")
COORD_RE = re.compile(r"<(\d+)>")
def parse_stream_into_events(buf, state):
"""Consume as many complete / elements from the front of `buf` as possible,
yielding ('box', label, coords) events; returns leftover buffer."""
events = []
while True:
rm = REF_RE.search(buf)
bm = BOX_RE.search(buf)
cands = []
if rm: cands.append((rm.start(), "ref", rm))
if bm: cands.append((bm.start(), "box", bm))
if not cands:
break
cands.sort(key=lambda c: c[0])
_, kind, m = cands[0]
if kind == "ref":
state["label"] = m.group(1)
buf = buf[m.end():]
else:
coords = [int(x) for x in COORD_RE.findall(m.group(1))]
if len(coords) in (2, 4):
events.append((state.get("label", ""), coords))
buf = buf[m.end():]
return buf, events
def _clean_categories(categories):
if isinstance(categories, str):
categories = categories.replace("\n", ",").split(",")
if not isinstance(categories, list):
return []
return [str(c).strip() for c in categories if str(c).strip()]
def _clean_phrase(req):
phrase = req.get("phrase", "")
phrase = str(phrase).strip()
return phrase or "object"
def _normalize_task(req, categories):
task = str(req.get("task", "") or "").strip()
legacy = {"detection": "object_detection", "grounding": "phrase_grounding_multi", "ocr": "scene_text_detection"}
task = legacy.get(task, task)
valid = {
"object_detection", "phrase_grounding_single", "phrase_grounding_multi",
"text_grounding", "scene_text_detection", "document_layout",
"gui_grounding_box", "gui_grounding_point", "pointing",
}
if task in valid:
return task
if categories:
return "object_detection"
return "custom_prompt" if req.get("prompt") else "object_detection"
def build_task_prompt(task, *, category=None, categories=None, phrase="object", fallback_prompt=""):
categories = categories or []
category_text = str(category).strip() if category is not None else ", ".join(categories)
category_text = category_text or "object"
phrase = str(phrase).strip() or "object"
if task == "object_detection":
return "Locate all the instances that matches the following description: " + category_text + "."
if task == "phrase_grounding_single":
return "Locate a single instance that matches the following description: " + phrase + "."
if task == "phrase_grounding_multi":
return "Locate all the instances that match the following description: " + phrase + "."
if task == "text_grounding":
return "Please locate the text referred as " + phrase + "."
if task == "scene_text_detection":
return "Detect all the text in box format."
if task == "document_layout":
return "Detect all the objects in the image that belong to the category set: " + category_text + "."
if task == "gui_grounding_box":
return "Locate the region that matches the following description: " + phrase + "."
if task in ("gui_grounding_point", "pointing"):
return "Point to: " + phrase + "."
return str(fallback_prompt or "Locate all the instances that matches the following description: object.")
def build_task_queries(req):
categories = _clean_categories(req.get("categories"))
task = _normalize_task(req, categories)
if task in ("object_detection", "document_layout"):
if categories:
return [(c, build_task_prompt(task, category=c, categories=categories)) for c in categories]
return [(None, build_task_prompt(task, categories=categories, fallback_prompt=req.get("prompt", "")))]
if task == "custom_prompt":
return [(None, str(req.get("prompt", "")))]
return [(None, build_task_prompt(task, phrase=_clean_phrase(req), fallback_prompt=req.get("prompt", "")))]
# ----------------------------------------------------------------------------- HTML/CSS/JS
PAGE = r"""
LocateAnything
Samples
LocateAnything
64
⬆ Upload image
Idle
0 targets
encoding image
Pick a sample above or upload an image, then press Detect
Use this image’s categories?
This image is tagged with:
"""
# ----------------------------------------------------------------------------- server
def list_images(d):
out = []
if d and os.path.isdir(d):
for n in sorted(os.listdir(d)):
if n.lower().endswith(IMG_EXT):
out.append(n)
return out[:48]
def load_tags(d):
"""Per-image presets from /tags.json. Each value is either a bare category
list (legacy) or {"tags": [...], "phrase": "..."}. Returns {name: {"tags", "phrase"}}."""
if not d:
return {}
try:
with open(os.path.join(d, "tags.json"), encoding="utf-8") as f:
m = json.load(f)
except Exception:
return {}
out = {}
for k, v in m.items():
if isinstance(v, list):
out[k] = {"tags": v, "phrase": ""}
elif isinstance(v, dict):
t = v.get("tags"); p = v.get("phrase")
out[k] = {"tags": t if isinstance(t, list) else [],
"phrase": p if isinstance(p, str) else ""}
return out
class Handler(BaseHTTPRequestHandler):
def log_message(self, *a): pass
def _send(self, code, ctype, body, extra=None):
self.send_response(code); self.send_header("Content-Type", ctype)
if isinstance(body, str): body = body.encode("utf-8")
self.send_header("Content-Length", str(len(body)))
for k, v in (extra or {}).items(): self.send_header(k, v)
self.end_headers(); self.wfile.write(body)
def do_GET(self):
path = self.path.split("?")[0]
if path == "/" or path == "/index.html":
return self._send(200, "text/html; charset=utf-8", PAGE, {"Cache-Control": "no-store"})
if path == "/api/thumbs":
names = list_images(IMAGE_DIR)
tags = load_tags(IMAGE_DIR)
data = [{"src": "/thumb/" + n, "name": n,
"tags": tags.get(n, {}).get("tags", []),
"phrase": tags.get(n, {}).get("phrase", "")} for n in names]
return self._send(200, "application/json", json.dumps(data))
if path.startswith("/thumb/"):
name = os.path.basename(path[len("/thumb/"):])
fp = os.path.join(IMAGE_DIR, name)
if IMAGE_DIR and os.path.isfile(fp) and name.lower().endswith(IMG_EXT):
ctype = mimetypes.guess_type(fp)[0] or "image/jpeg"
with open(fp, "rb") as f: return self._send(200, ctype, f.read(), {"Cache-Control": "max-age=3600"})
return self._send(404, "text/plain", "not found")
return self._send(404, "text/plain", "not found")
def do_POST(self):
if self.path.split("?")[0] != "/api/detect":
return self._send(404, "text/plain", "not found")
try:
n = int(self.headers.get("Content-Length", "0"))
req = json.loads(self.rfile.read(n).decode("utf-8"))
except Exception as e:
return self._send(400, "text/plain", "bad request: %s" % e)
image_b64 = req.get("image", "")
max_tokens = int(req.get("max_tokens", 512))
# One detection per category keeps per-category labels + colors in the UI.
queries = build_task_queries(req)
# open SSE to client
self.send_response(200)
self.send_header("Content-Type", "text/event-stream")
self.send_header("Cache-Control", "no-cache")
self.send_header("X-Accel-Buffering", "no")
self.end_headers()
self.close_connection = True
def emit(obj):
self.wfile.write(("data: " + json.dumps(obj) + "\n\n").encode("utf-8")); self.wfile.flush()
emit({"type": "status", "phase": "encoding"})
started = False; count = 0
try:
for forced_label, prompt in queries:
body = {"model": MODEL, "stream": True, "temperature": 0, "max_tokens": max_tokens,
"messages": [{"role": "user", "content": [
{"type": "image_url", "image_url": {"url": "data:image/jpeg;base64," + image_b64}},
{"type": "text", "text": prompt}]}]}
up = urllib.request.Request(SERVE_URL + "/v1/chat/completions",
data=json.dumps(body).encode("utf-8"),
headers={"Content-Type": "application/json"})
resp = urllib.request.urlopen(up, timeout=600)
state = {"label": ""}; buf = ""
for raw in resp:
line = raw.decode("utf-8", "ignore").strip()
if not line.startswith("data:"): continue
d = line[5:].strip()
if d == "[DONE]": break
try: o = json.loads(d)
except Exception: continue
piece = (o.get("choices", [{}])[0].get("delta", {}) or {}).get("content") or ""
if not piece: continue
if not started: emit({"type": "status", "phase": "generating"}); started = True
buf += piece
buf, evs = parse_stream_into_events(buf, state)
for lab, coords in evs:
count += 1
emit({"type": "box", "label": (forced_label if forced_label is not None else lab), "box": coords, "index": count})
emit({"type": "done", "count": count})
except (BrokenPipeError, ConnectionResetError):
pass
except Exception as e:
try: emit({"type": "error", "message": str(e)})
except Exception: pass
class Server(ThreadingHTTPServer):
daemon_threads = True
def main():
global SERVE_URL, IMAGE_DIR, MODEL
ap = argparse.ArgumentParser()
ap.add_argument("--host", default="0.0.0.0")
ap.add_argument("--port", type=int, default=7861)
ap.add_argument("--serve-url", default=SERVE_URL)
ap.add_argument("--image-dir", default=IMAGE_DIR)
ap.add_argument("--model", default=MODEL)
a = ap.parse_args()
SERVE_URL, IMAGE_DIR, MODEL = a.serve_url.rstrip("/"), a.image_dir, a.model
print("LocateAnything WebUI")
print(" serve : %s (model=%s)" % (SERVE_URL, MODEL))
print(" images: %s (%d found)" % (IMAGE_DIR or "(none)", len(list_images(IMAGE_DIR))))
print(" open : http://localhost:%d" % a.port)
Server((a.host, a.port), Handler).serve_forever()
if __name__ == "__main__":
main()