import os import gc import base64 import json import ast import re from io import BytesIO import torch import spaces import numpy as np from PIL import Image, ImageDraw, ImageFont import supervision as sv from typing import Iterable import gradio as gr from gradio import Server from fastapi.responses import HTMLResponse from threading import Thread from transformers import ( Qwen3_5ForConditionalGeneration, AutoProcessor, TextIteratorStreamer, ) # ------------------------------------------------------------------ # Config & Constants # ------------------------------------------------------------------ DEVICE = "cuda" if torch.cuda.is_available() else "cpu" DTYPE = torch.bfloat16 if torch.cuda.is_available() and torch.cuda.is_bf16_supported() else torch.float16 MODEL_NAME = "Qwen/Qwen3.8-27B" BRIGHT_YELLOW = sv.Color(r=255, g=230, b=0) DARK_OUTLINE = sv.Color(r=40, g=40, b=40) BLACK = sv.Color(r=0, g=0, b=0) WHITE = sv.Color(r=255, g=255, b=255) # Spatial path colors SPATIAL_LINE = (255, 69, 0) # OrangeRed SPATIAL_DOT = (255, 69, 0) SPATIAL_RING = (255, 255, 255) SPATIAL_LABEL_BG = (80, 20, 0) SPATIAL_LABEL_TXT = (255, 255, 255) SPATIAL_ARROW = (255, 165, 0) # Orange for direction # ------------------------------------------------------------------ # Model Loading # ------------------------------------------------------------------ print(f"Loading model: {MODEL_NAME} ...") qwen_model = Qwen3_5ForConditionalGeneration.from_pretrained( MODEL_NAME, torch_dtype=DTYPE, device_map=DEVICE, attn_implementation="kernels-community/flash-attn2@v3", ).eval() qwen_processor = AutoProcessor.from_pretrained(MODEL_NAME) print("Model loaded.") # ------------------------------------------------------------------ # Examples Config # ------------------------------------------------------------------ EXAMPLES_CONFIG = [ {"image": "examples/1.jpg", "prompt": "Detect the yellow car that is parked.", "mode": "Detect"}, {"image": "examples/2.jpg", "prompt": "Point to all the red cars.", "mode": "Point"}, {"image": "examples/3.jpg", "prompt": "Map a path from the door to the lamp.", "mode": "Spatial"}, ] def make_thumb_b64(path, max_dim=220): if not os.path.exists(path): return "" try: img = Image.open(path).convert("RGB") img.thumbnail((max_dim, max_dim), Image.LANCZOS) buf = BytesIO() img.save(buf, format="JPEG", quality=65) return f"data:image/jpeg;base64,{base64.b64encode(buf.getvalue()).decode()}" except Exception as e: return "" def encode_full_image(path): if not os.path.exists(path): return "" try: with open(path, "rb") as f: data = f.read() ext = path.rsplit(".", 1)[-1].lower() mime = {"jpg": "image/jpeg", "jpeg": "image/jpeg", "png": "image/png", "webp": "image/webp"}.get(ext, "image/jpeg") return f"data:{mime};base64,{base64.b64encode(data).decode()}" except Exception as e: return "" def build_client_config(): examples = [] for i, ex in enumerate(EXAMPLES_CONFIG): examples.append({ "idx": i, "thumb": make_thumb_b64(ex["image"]), "prompt": ex["prompt"], "mode": ex["mode"], }) return {"examples": examples, "modes": ["Detect", "Point", "Spatial"], "default_mode": "Detect"} print("Building client config…") CLIENT_CONFIG = build_client_config() # ------------------------------------------------------------------ # Helpers # ------------------------------------------------------------------ def safe_parse_json(text: str): text = re.sub(r"```(json)?", "", text).strip() match = re.search(r'(\[.*\]|\{.*\})', text, re.DOTALL) if match: json_str = match.group(1) json_str_clean = re.sub(r',\s*([}\]])', r'\1', json_str) try: return json.loads(json_str_clean) except json.JSONDecodeError: try: return ast.literal_eval(json_str_clean) except Exception: pass text_clean = re.sub(r',\s*([}\]])', r'\1', text) try: return json.loads(text_clean) except json.JSONDecodeError: pass try: return ast.literal_eval(text_clean) except Exception: pass return [] def _extract_point(item: dict): if not isinstance(item, dict): return None for k in ["point_2d", "point", "points", "coordinate", "coordinates", "xy"]: if k in item and isinstance(item[k], (list, tuple)) and len(item[k]) == 2: return item[k] return None def _extract_bbox(item: dict): if not isinstance(item, dict): return None for k in ["bbox_2d", "bbox", "box", "bounding_box", "xyxy"]: if k in item and isinstance(item[k], (list, tuple)) and len(item[k]) == 4: return item[k] return None def _load_font(size: int = 16): size = max(6, int(size)) try: return ImageFont.truetype("arial.ttf", size) except (IOError, OSError): try: return ImageFont.truetype("DejaVuSans.ttf", size) except (IOError, OSError): return ImageFont.load_default() def pil_to_b64_png(image: Image.Image) -> str: buf = BytesIO() image.save(buf, format="PNG") return f"data:image/png;base64,{base64.b64encode(buf.getvalue()).decode()}" def annotate_image(image: Image.Image, result: dict, point_radius: int = 6, box_thickness: int = 2, text_scale: float = 0.5): if not isinstance(image, Image.Image) or not isinstance(result, dict): return image image = image.convert("RGB") ow, oh = image.size point_radius = max(1, int(point_radius)) box_thickness = max(1, int(box_thickness)) text_scale = max(0.1, float(text_scale)) text_thickness = max(1, round(text_scale * 2)) if "points" in result and result["points"]: pts = [[int(p["x"] * ow), int(p["y"] * oh)] for p in result["points"]] if not pts: return image kp = sv.KeyPoints(xy=np.array(pts).reshape(1, -1, 2)) scene = np.array(image.copy()) scene = sv.VertexAnnotator(radius=point_radius + 3, color=DARK_OUTLINE).annotate(scene=scene, key_points=kp) scene = sv.VertexAnnotator(radius=point_radius, color=BRIGHT_YELLOW).annotate(scene=scene, key_points=kp) labels = [p.get("label", "") for p in result["points"]] if any(labels): tb, vl = [], [] for i, p in enumerate(result["points"]): if labels[i]: cx, cy = int(p["x"] * ow), int(p["y"] * oh) tb.append([cx - 2, cy - 2, cx + 2, cy + 2]) vl.append(labels[i]) if tb: scene = sv.LabelAnnotator( color=BRIGHT_YELLOW, text_color=BLACK, text_scale=text_scale, text_thickness=text_thickness, text_padding=5, text_position=sv.Position.TOP_CENTER, color_lookup=sv.ColorLookup.INDEX, ).annotate(scene=scene, detections=sv.Detections(xyxy=np.array(tb)), labels=vl) return Image.fromarray(scene) if "objects" in result and result["objects"]: boxes, labels = [], [] for obj in result["objects"]: boxes.append([ obj.get("x_min", 0.0) * ow, obj.get("y_min", 0.0) * oh, obj.get("x_max", 0.0) * ow, obj.get("y_max", 0.0) * oh, ]) labels.append(obj.get("label", "object")) if not boxes: return image scene = np.array(image.copy()) h, w = scene.shape[:2] masks = np.zeros((len(boxes), h, w), dtype=bool) for i, box in enumerate(boxes): x1, y1 = max(0, int(box[0])), max(0, int(box[1])) x2, y2 = min(w, int(box[2])), min(h, int(box[3])) masks[i, y1:y2, x1:x2] = True dets = sv.Detections(xyxy=np.array(boxes), mask=masks) if len(dets) == 0: return image scene = sv.MaskAnnotator(color=BRIGHT_YELLOW, opacity=0.18, color_lookup=sv.ColorLookup.INDEX).annotate(scene=scene, detections=dets) scene = sv.BoxAnnotator(color=BRIGHT_YELLOW, thickness=box_thickness, color_lookup=sv.ColorLookup.INDEX).annotate(scene=scene, detections=dets) scene = sv.LabelAnnotator( color=BRIGHT_YELLOW, text_color=BLACK, text_scale=text_scale, text_thickness=text_thickness, text_padding=6, color_lookup=sv.ColorLookup.INDEX, ).annotate(scene=scene, detections=dets, labels=labels) return Image.fromarray(scene) return image def annotate_spatial_path(image: Image.Image, result: dict, dot_radius: int = 6, line_width: int = 4, text_scale: float = 0.5): if not isinstance(image, Image.Image) or not isinstance(result, dict): return image image = image.convert("RGB") w, h = image.size if "points" not in result or not result["points"]: return image dot_radius = max(1, int(dot_radius)) line_width = max(1, int(line_width)) text_scale = max(0.1, float(text_scale)) draw = ImageDraw.Draw(image, "RGBA") font_label = _load_font(16 * text_scale * 2) font_num = _load_font(14 * text_scale * 2) points = result["points"] pts = [(int(p["x"] * w), int(p["y"] * h)) for p in points] labels = [p.get("label", f"P{i+1}") for i, p in enumerate(points)] scale_ratio = dot_radius / 8.0 if len(pts) >= 2: for i in range(len(pts) - 1): draw.line([pts[i], pts[i+1]], fill=SPATIAL_LINE + (60,), width=line_width + 6) for i in range(len(pts) - 1): draw.line([pts[i], pts[i+1]], fill=SPATIAL_LINE, width=line_width) for i in range(len(pts) - 1): x1, y1 = pts[i] x2, y2 = pts[i+1] dx, dy = x2 - x1, y2 - y1 length = (dx * dx + dy * dy) ** 0.5 if length < 12: continue ux, uy = dx / length, dy / length offset = 18 * scale_ratio bx, by = x2 - ux * offset, y2 - uy * offset px, py = -uy, ux aw, ah = 7 * scale_ratio, 9 * scale_ratio p1 = (bx + px * aw, by + py * aw) p2 = (bx - px * aw, by - py * aw) p3 = (bx + ux * ah, by + uy * ah) draw.polygon([p1, p2, p3], fill=SPATIAL_ARROW) for i, (cx, cy) in enumerate(pts): halo_r = dot_radius + 8 ring_r = dot_radius + 3 draw.ellipse((cx - halo_r, cy - halo_r, cx + halo_r, cy + halo_r), fill=SPATIAL_LINE + (50,)) draw.ellipse((cx - ring_r, cy - ring_r, cx + ring_r, cy + ring_r), outline=SPATIAL_RING, width=max(1, round(3 * scale_ratio))) draw.ellipse((cx - dot_radius, cy - dot_radius, cx + dot_radius, cy + dot_radius), fill=SPATIAL_DOT, outline=SPATIAL_DOT) num_text = str(i + 1) nbbox = draw.textbbox((0, 0), num_text, font=font_num) nw = nbbox[2] - nbbox[0] nh = nbbox[3] - nbbox[1] nx, ny = cx - nw // 2, cy - nh // 2 - 1 draw.text((nx, ny), num_text, fill=SPATIAL_RING, font=font_num) for i, (cx, cy) in enumerate(pts): label = f"{i+1}. {labels[i]}" bbox = draw.textbbox((0, 0), label, font=font_label) tw, th = bbox[2] - bbox[0], bbox[3] - bbox[1] lx, ly = cx + dot_radius + 10, cy - th - 8 pad = 5 draw.rectangle((lx - pad, ly - pad, lx + tw + pad, ly + th + pad), fill=SPATIAL_LABEL_BG, outline=SPATIAL_LINE, width=1) draw.text((lx, ly), label, fill=SPATIAL_LABEL_TXT, font=font_label) n_pts = len(pts) legend_text = f"Spatial map · {n_pts} waypoints · path length {len(pts)-1} segments" legend_font = _load_font(13 * text_scale * 2) bbox = draw.textbbox((0, 0), legend_text, font=legend_font) tw, th = bbox[2] - bbox[0], bbox[3] - bbox[1] fx, fy = 10, h - th - 22 draw.rectangle((fx - 8, fy - 6, fx + tw + 16, fy + th + 10), fill=SPATIAL_LABEL_BG + (220,)) draw.text((fx, fy), legend_text, fill=SPATIAL_LABEL_TXT, font=legend_font) return image # ------------------------------------------------------------------ # Gradio Server (Server mode): FastAPI + Gradio queue/API engine # ------------------------------------------------------------------ app = Server(title="Qwen3.8-27B-Object-Detection") @app.mcp.tool(name="run_inference") @app.api(name="run_inference") @spaces.GPU(size="xlarge", duration=90) def infer( image_b64: str, mode: str, prompt: str, point_radius: int, box_thickness: int, text_scale: float, ) -> dict: """Runs object detection, point localization, or spatial mapping.""" gc.collect() torch.cuda.empty_cache() if not image_b64: raise gr.Error("Please upload an image.") if not prompt or prompt.strip() == "": raise gr.Error("Please provide a prompt.") try: header, data = image_b64.split(",", 1) pil_image = Image.open(BytesIO(base64.b64decode(data))).convert("RGB") except Exception as e: raise gr.Error(f"Invalid image data: {e}") pil_image.thumbnail((512, 512)) category = mode if category == "Detect": full_prompt = ( f"Provide bounding box coordinates for {prompt}. " f"Report strictly in JSON format as a list of objects with 'label' and " f"'bbox_2d' (xmin, ymin, xmax, ymax in 0-1000 scale)." ) elif category == "Point": full_prompt = ( f"Provide 2d point coordinates for {prompt}. " f"Report strictly in JSON format as a list of objects with 'label' and " f"'point_2d' (x, y in 0-1000 scale)." ) elif category == "Spatial": full_prompt = ( f"Identify the key spatial waypoints to map a path/route for: {prompt}. " f"Return the points in the order they should be connected along the path, " f"from start to end. Report in JSON format as a list of objects with " f"'label' (name of each waypoint) and 'point_2d' (x, y in 0-1000 scale). " f"Include 2-8 waypoints that best describe the spatial map." ) else: full_prompt = prompt messages = [{ "role": "user", "content": [ {"type": "image", "image": pil_image}, {"type": "text", "text": full_prompt}, ], }] text = qwen_processor.apply_chat_template( messages, tokenize=False, add_generation_prompt=True ) inputs = qwen_processor( text=[text], images=[pil_image], return_tensors="pt", padding=True, ).to(qwen_model.device) streamer = TextIteratorStreamer( qwen_processor.tokenizer, skip_prompt=True, skip_special_tokens=True, timeout=120, ) thread = Thread( target=qwen_model.generate, kwargs=dict( **inputs, streamer=streamer, max_new_tokens=2048, use_cache=True, do_sample=False, ), ) thread.start() full_text = "" for tok in streamer: full_text += tok thread.join() result_text = full_text result_image = pil_image.copy() if category == "Point": parsed = safe_parse_json(full_text) if isinstance(parsed, dict): for k in ["points", "keypoints", "point"]: if k in parsed and isinstance(parsed[k], list): parsed = parsed[k] break else: for v in parsed.values(): if isinstance(v, list): parsed = v; break else: parsed = [] result = {"points": []} if isinstance(parsed, list): for item in parsed: pt = _extract_point(item) if pt: x, y = pt result["points"].append({"label": item.get("label", ""), "x": x / 1000.0, "y": y / 1000.0}) if result["points"]: result_image = annotate_image(pil_image.copy(), result, point_radius, box_thickness, text_scale) result_text = json.dumps(result, indent=2) else: result_text = f"Could not extract any points.\nRaw model output:\n{full_text}" elif category == "Detect": parsed = safe_parse_json(full_text) if isinstance(parsed, dict): for k in ["objects", "detections", "bboxes", "boxes", "results"]: if k in parsed and isinstance(parsed[k], list): parsed = parsed[k] break else: for v in parsed.values(): if isinstance(v, list): parsed = v; break else: parsed = [] result = {"objects": []} if isinstance(parsed, list): for item in parsed: bbox = _extract_bbox(item) if bbox: xmin, ymin, xmax, ymax = bbox result["objects"].append({ "label": item.get("label", "object"), "x_min": xmin / 1000.0, "y_min": ymin / 1000.0, "x_max": xmax / 1000.0, "y_max": ymax / 1000.0, }) if result["objects"]: result_image = annotate_image(pil_image.copy(), result, point_radius, box_thickness, text_scale) result_text = json.dumps(result, indent=2) else: result_text = f"Could not extract any objects.\nRaw model output:\n{full_text}" elif category == "Spatial": parsed = safe_parse_json(full_text) if isinstance(parsed, dict): for k in ["points", "waypoints", "path", "route", "nodes", "map"]: if k in parsed and isinstance(parsed[k], list): parsed = parsed[k] break else: for v in parsed.values(): if isinstance(v, list): parsed = v; break else: parsed = [] result = {"points": []} if isinstance(parsed, list): for item in parsed: pt = _extract_point(item) if pt: x, y = pt result["points"].append({"label": item.get("label", "waypoint"), "x": x / 1000.0, "y": y / 1000.0}) if result["points"]: wp_lines = "\n".join(f" {i+1}. {p['label']} → ({p['x']:.3f}, {p['y']:.3f})" for i, p in enumerate(result["points"])) result_text = ( f"Spatial map generated.\n" f"Waypoints ({len(result['points'])}):\n{wp_lines}\n" f"Path segments: {max(0, len(result['points']) - 1)}" ) result_image = annotate_spatial_path(pil_image.copy(), result, point_radius, box_thickness * 2, text_scale) else: result_text = f"Could not extract any spatial waypoints.\nRaw model output:\n{full_text}" return {"image": pil_to_b64_png(result_image), "text": result_text} @app.api(name="load_example", queue=False) def load_example(idx: float) -> dict: try: i = int(idx) except (ValueError, TypeError): i = -1 if i < 0 or i >= len(EXAMPLES_CONFIG): return {"image": "", "prompt": "", "mode": "Detect", "name": "", "status": "error"} ex = EXAMPLES_CONFIG[i] b64 = encode_full_image(ex["image"]) return { "image": b64, "prompt": ex["prompt"], "mode": ex["mode"], "name": os.path.basename(ex["image"]), "status": "ok" if b64 else "error" } @app.get("/api/config") def client_config(): return CLIENT_CONFIG @app.get("/", response_class=HTMLResponse) async def homepage(): html_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "index.html") with open(html_path, "r", encoding="utf-8") as f: return f.read() if __name__ == "__main__": app.launch(show_error=True, mcp_server=True)