import os import io import sys import base64 import uuid # rebuild trigger 2026-09-02 fix CambrianConfig import import time import asyncio import logging import tempfile import mimetypes from contextvars import ContextVar from contextlib import contextmanager, redirect_stdout, redirect_stderr, nullcontext from typing import Optional, List # Matikan progress bar SEBELUM import hub/transformers os.environ.setdefault("HF_HUB_DISABLE_PROGRESS_BARS", "1") # Inject HF_TOKEN dari cache / env try: _tok_path = os.path.expanduser("~/.cache/huggingface/token") if os.path.exists(_tok_path): _tok = open(_tok_path).read().strip() if _tok: os.environ.setdefault("HF_TOKEN", _tok) os.environ.setdefault("HUGGING_FACE_HUB_TOKEN", _tok) # also check /tmp/sandiyo_token (for Sandiyo account) if os.path.exists("/tmp/sandiyo_token"): _tok2 = open("/tmp/sandiyo_token").read().strip() if _tok2: os.environ.setdefault("HF_TOKEN", _tok2) os.environ.setdefault("HUGGING_FACE_HUB_TOKEN", _tok2) except Exception: pass # ZeroGPU shim harus paling awal try: import spaces # noqa: F401 except ImportError: spaces = None if spaces is not None: @spaces.GPU(inference=False) def _activate_zero_gpu(): pass from fastapi import FastAPI, HTTPException, Request from fastapi.responses import JSONResponse from pydantic import BaseModel import uvicorn import aiohttp import gradio as gr import torch import numpy as np from PIL import Image, ImageSequence # LongVU imports (vendored) from longvu.builder import load_pretrained_model from longvu.constants import DEFAULT_IMAGE_TOKEN, IMAGE_TOKEN_INDEX from longvu.conversation import conv_templates, SeparatorStyle from longvu.mm_datautils import process_images, tokenizer_image_token from longvu.mm_utils import KeywordsStoppingCriteria # ==================== CONFIG ==================== MODEL_PATH = "Vision-CAIR/LongVU_Qwen2_7B" MODEL_NAME = "cambrian_qwen" DEFAULT_PROMPT = "Describe this video in detail." DEFAULT_MAX_NEW_TOKENS = 256 MAX_CONCURRENT_INFERENCES = int(os.environ.get("MAX_CONCURRENT_INFERENCES", 1)) MAX_INFERENCE_WAITERS = int(os.environ.get("MAX_INFERENCE_WAITERS", 20)) DOWNLOAD_TIMEOUT = int(os.environ.get("DOWNLOAD_TIMEOUT", 30)) MAX_FRAMES = int(os.environ.get("MAX_FRAMES", 32)) MAX_FILE_MB = int(os.environ.get("MAX_FILE_MB", 15)) # ==================== REQUEST CONTEXT ==================== request_id_ctx: ContextVar[str] = ContextVar("request_id", default="-") logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class RequestContextFilter(logging.Filter): def filter(self, record): record.request_id = request_id_ctx.get() return True def setup_request_logging(): log_filter = RequestContextFilter() formatter = logging.Formatter("%(asctime)s | %(levelname)-7s | [req:%(request_id)s] | %(message)s") for handler in logging.getLogger().handlers: handler.addFilter(log_filter) handler.setFormatter(formatter) setup_request_logging() for _noisy in ("httpx", "httpcore", "huggingface_hub", "urllib3", "filelock", "transformers"): logging.getLogger(_noisy).setLevel(logging.WARNING) import warnings warnings.filterwarnings("ignore", message=".*unauthenticated.*") warnings.filterwarnings("ignore", message=".*not documented.*") warnings.filterwarnings("ignore", message=".*standardize_rope.*") warnings.filterwarnings("ignore", message=".*Unrecognized keys.*") warnings.filterwarnings("ignore", message=".*generation flags.*") warnings.filterwarnings("ignore", message=".*AttentionMaskConverter.*") warnings.filterwarnings("ignore", message=".*deprecated.*") warnings.filterwarnings("ignore", message=".*Special tokens.*") warnings.filterwarnings("ignore", message=".*attention mask.*") warnings.filterwarnings("ignore", message=".*pad_token_id.*") warnings.filterwarnings("ignore", category=ResourceWarning) warnings.filterwarnings("ignore", category=UserWarning) warnings.filterwarnings("ignore", category=FutureWarning) warnings.filterwarnings("ignore") _orig_unraisable = sys.unraisablehook def _quiet_unraisable(unraisable): try: msg = str(unraisable.exc_value) if unraisable.exc_value else "" if any(p in msg for p in ("Invalid file descriptor", "BaseEventLoop")): return except Exception: pass return _orig_unraisable(unraisable) sys.unraisablehook = _quiet_unraisable class _NoiseLogFilter(logging.Filter): _PATTERNS = ("not documented","standardize_rope_params","Unrecognized keys","Invalid file descriptor","BaseEventLoop","unauthenticated requests","SiglipVisionModel","UNEXPECTED","generation flags","AttentionMaskConverter","deprecated and will be removed","FutureWarning","Special tokens","attention mask","pad_token_id","eos_token_id") def filter(self, record): try: msg = record.getMessage() except Exception: return True return not any(p in msg for p in self._PATTERNS) for _h in logging.getLogger().handlers: _h.addFilter(_NoiseLogFilter()) for _lname in ("transformers", "huggingface_hub", "httpx", "asyncio"): lg = logging.getLogger(_lname) for _h in lg.handlers: _h.addFilter(_NoiseLogFilter()) for _lg in list(logging.Logger.manager.loggerDict.values()): if isinstance(_lg, logging.Logger): for _h in getattr(_lg, "handlers", []): _h.addFilter(_NoiseLogFilter()) _orig_emit = logging.StreamHandler.emit def _filtered_emit(self, record): if any(p in record.getMessage() for p in _NoiseLogFilter._PATTERNS): return return _orig_emit(self, record) logging.StreamHandler.emit = _filtered_emit @contextmanager def request_scope(): rid = uuid.uuid4().hex[:8] token = request_id_ctx.set(rid) try: yield rid finally: request_id_ctx.reset(token) # ==================== GLOBALS ==================== MODEL = None TOKENIZER = None IMAGE_PROCESSOR = None CONTEXT_LEN = 2048 DEVICE = "cpu" INFERENCE_SEMAPHORE: asyncio.Semaphore | None = None _HTTP_SESSION: aiohttp.ClientSession | None = None _INFERENCE_ACTIVE = 0 _INFERENCE_WAITERS = 0 COTRACKER_MODEL = None _COTRACKER_LOCK = None # lazy init threading.Lock class ServerBusyError(Exception): pass def _get_cotracker(): global COTRACKER_MODEL, _COTRACKER_LOCK if COTRACKER_MODEL is not None: return COTRACKER_MODEL import threading if _COTRACKER_LOCK is None: _COTRACKER_LOCK = threading.Lock() with _COTRACKER_LOCK: if COTRACKER_MODEL is not None: return COTRACKER_MODEL try: import torch as _t # torch.hub akan download scaled_offline.pth dari HF (facebook/cotracker3) _m = _t.hub.load("facebookresearch/co-tracker", "cotracker3_offline", pretrained=True, trust_repo=True) _m = _m.to(DEVICE).eval() COTRACKER_MODEL = _m logger.info("CoTracker3 offline loaded") except Exception as e: logger.warning(f"CoTracker load failed, fallback cv2: {e}") COTRACKER_MODEL = None return COTRACKER_MODEL def _cotracker_hint(frames: List[Image.Image]) -> str: """Return textual motion hint from CoTracker grid tracks, or '' if fail.""" if len(frames) < 2: return "" m = _get_cotracker() if m is None: return "" try: import torch as _t import numpy as np # CoTracker expects (B,T,3,H,W) float, H/W ~256 # downscale untuk CPU cepat: 128 H = W = 128 arr = np.stack([np.array(f.resize((W, H), Image.BILINEAR)) for f in frames]) # T,H,W,3 video = _t.from_numpy(arr).permute(0, 3, 1, 2)[None].float().to(DEVICE) # 1,T,3,H,W # grid 5x5 =25 points with _t.inference_mode(): tracks, vis = m(video, grid_size=5) # tracks: 1,T,N,2 vis: 1,T,N # tracks in pixel coords of 128x128 tracks = tracks[0].cpu().numpy() # T,N,2 vis = vis[0].cpu().numpy() if vis is not None else None # hitung displacement first->last untuk titik visible T, N, _ = tracks.shape hints = [] # bagi 4 kuadran biar deskripsi spasial for qi, (qx, qy, qname) in enumerate([(32,32,"top-left"),(96,32,"top-right"),(32,96,"bottom-left"),(96,96,"bottom-right")]): # cari titik grid terdekat kuadran # grid points di CoTracker grid_size 5: spacing ~25px, cari nearest # sederhanakan: ambil subset tracks dan lihat rata2 pass # sederhanakan: rata2 displacement semua titik visible disp = tracks[-1] - tracks[0] # N,2 (dx,dy) if vis is not None: # hanya titik visible di kedua frame mask = (vis[0] > 0.5) & (vis[-1] > 0.5) if mask.sum() > 0: disp = disp[mask] if len(disp) == 0: return "" mean_dx = float(np.mean(disp[:, 0])) mean_dy = float(np.mean(disp[:, 1])) mean_mag = float(np.mean(np.sqrt(disp[:,0]**2 + disp[:,1]**2))) if mean_mag < 2: return " CoTracker: mostly static (mean disp {:.1f}px).".format(mean_mag) dirs = [] if mean_dx > 3: dirs.append("right") elif mean_dx < -3: dirs.append("left") if mean_dy > 3: dirs.append("down") elif mean_dy < -3: dirs.append("up") dir_txt = "-".join(dirs) if dirs else "mixed" return f" CoTracker: avg disp {mean_mag:.1f}px {dir_txt} (dx {mean_dx:+.0f}, dy {mean_dy:+.0f}, {len(disp)} tracks)." except Exception as e: logger.warning(f"CoTracker hint failed: {e}") return "" # ==================== NOISE FILTER ==================== _NOISE_PATTERNS = ("not documented","standardize_rope_params","Unrecognized keys","unauthenticated requests","Invalid file descriptor","BaseEventLoop","Exception ignored","it/s]","B/s]","SiglipVisionModel","UNEXPECTED","LOAD REPORT","generation flags","AttentionMaskConverter","FutureWarning","deprecated and will be removed","Special tokens","attention mask","pad_token_id","eos_token_id") class _NoiseFilter: def __init__(self, orig): self.orig = orig self._buf = "" def write(self, s): if not s: return 0 self._buf += s while "\n" in self._buf or "\r" in self._buf: nl = self._buf.find("\n") cr = self._buf.find("\r") if nl == -1: idx = cr elif cr == -1: idx = nl else: idx = min(nl, cr) line = self._buf[: idx + 1] self._buf = self._buf[idx + 1 :] if not any(p in line for p in _NOISE_PATTERNS): try: self.orig.write(line) except Exception: pass return len(s) def flush(self): if self._buf and not any(p in self._buf for p in _NOISE_PATTERNS): try: self.orig.write(self._buf) except Exception: pass self._buf = "" try: self.orig.flush() except Exception: pass def isatty(self): try: return self.orig.isatty() except Exception: return False def fileno(self): try: return self.orig.fileno() except Exception: return 1 def __getattr__(self, name): return getattr(self.orig, name) @contextmanager def noise_filtered(): with redirect_stdout(_NoiseFilter(sys.stdout)), redirect_stderr(_NoiseFilter(sys.stderr)): yield # ==================== MODEL LOADER ==================== def load_model(): global MODEL, TOKENIZER, IMAGE_PROCESSOR, CONTEXT_LEN logger.info(f"Loading {MODEL_PATH} model on CPU...") try: with noise_filtered(): tokenizer, model, image_processor, context_len = load_pretrained_model( MODEL_PATH, None, MODEL_NAME, device="cpu", device_map={"": "cpu"} ) model.eval() TOKENIZER = tokenizer MODEL = model IMAGE_PROCESSOR = image_processor CONTEXT_LEN = context_len logger.info("Model loaded successfully on cpu!") except Exception as e: logger.error(f"Failed to load model: {str(e)}", exc_info=True) MODEL = None TOKENIZER = None IMAGE_PROCESSOR = None def ensure_model_loaded(): if MODEL is None or TOKENIZER is None: load_model() return MODEL is not None # ==================== UTILS ==================== def get_semaphore() -> asyncio.Semaphore: global INFERENCE_SEMAPHORE if INFERENCE_SEMAPHORE is None: INFERENCE_SEMAPHORE = asyncio.Semaphore(MAX_CONCURRENT_INFERENCES) return INFERENCE_SEMAPHORE async def get_http_session() -> aiohttp.ClientSession: global _HTTP_SESSION if _HTTP_SESSION is None or _HTTP_SESSION.closed: _HTTP_SESSION = aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=DOWNLOAD_TIMEOUT), connector=aiohttp.TCPConnector(limit=100), headers={"User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36"}) return _HTTP_SESSION def extract_frames_from_bytes(data: bytes, hint: str = "") -> List[Image.Image]: """Decode bytes (image/gif/webp/mp4) ke list PIL frames. Support semua type via single fungsi.""" # Check size limit if len(data) > MAX_FILE_MB * 1024 * 1024: raise ValueError(f"file too large: {len(data)} bytes > {MAX_FILE_MB}MB") # Try PIL first (handles jpg/png/gif/webp animated) try: im = Image.open(io.BytesIO(data)) # Hanya GIF/WEBP yang dianggap animated - MPO/JPEG dengan n_frames>1 (thumbnail) diabaikan fmt = getattr(im, "format", None) try: n_frames = getattr(im, "n_frames", 1) except Exception: n_frames = 1 is_animated = False if fmt in ("GIF", "WEBP") and (getattr(im, "is_animated", False) or n_frames > 1): is_animated = True frames: List[Image.Image] = [] if is_animated and n_frames > 1: for frame in ImageSequence.Iterator(im): f = frame.convert("RGB") frames.append(f.copy()) if len(frames) >= 64: # cap awal break # Pastikan semua frame ukuran sama (gif kadang beda), resize ke frame pertama if len(frames) > 1: base_size = frames[0].size frames = [f if f.size == base_size else f.resize(base_size, Image.BILINEAR) for f in frames] else: # Single image (termasuk MPO yang punya 2 frame thumbnail) if im.mode != "RGB": im = im.convert("RGB") frames = [im] if frames: return frames except Exception as e: # PIL failed, coba fallback decord/imageio untuk mp4 pass # Fallback: coba decord untuk mp4/video try: from decord import VideoReader, cpu # Tulis ke temp file suffix = ".mp4" if hint: # hint bisa extension dari URL if "." in hint: suffix = "." + hint.split(".")[-1].split("?")[0].lower() with tempfile.NamedTemporaryFile(suffix=suffix, delete=False) as tf: tf.write(data) tf_path = tf.name try: vr = VideoReader(tf_path, ctx=cpu(0), num_threads=1) frames = [] # Sample uniformly max 64 total = len(vr) indices = np.linspace(0, total-1, min(total, 64), dtype=int) for idx in indices: img = vr[int(idx)].asnumpy() frames.append(Image.fromarray(img).convert("RGB")) return frames finally: try: os.unlink(tf_path) except Exception: pass except Exception: pass # Fallback imageio try: import imageio.v3 as iio # imageio can read gif/webp/mp4 # Coba baca semua frames try: # For gif/webp imgs = iio.imread(io.BytesIO(data), index=None, plugin="pillow") if isinstance(imgs, np.ndarray): if imgs.ndim == 3: # single image HWC return [Image.fromarray(imgs).convert("RGB")] elif imgs.ndim == 4: # THWC return [Image.fromarray(imgs[i]).convert("RGB") for i in range(min(imgs.shape[0], 64))] except Exception: pass # Try pyav for mp4 try: import av with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as tf: tf.write(data) tf_path = tf.name try: container = av.open(tf_path) frames = [] for frame in container.decode(video=0): img = frame.to_ndarray(format="rgb24") frames.append(Image.fromarray(img).convert("RGB")) if len(frames) >= 64: break return frames finally: try: os.unlink(tf_path) except Exception: pass except Exception: pass except Exception: pass raise ValueError("unsupported media or decode failed") def sample_frames(frames: List[Image.Image], max_frames: int = MAX_FRAMES) -> List[Image.Image]: """Adaptive sampler: uniform + motion-aware. Untuk gif/webp pendek (<=max) return semua.""" n = len(frames) if n <= max_frames: return frames # Cepat: downscale ke 32x32 grayscale untuk hitung difference murah small = [] for f in frames: try: small.append(np.array(f.resize((32, 32), Image.BILINEAR).convert("L"), dtype=np.int16)) except Exception: small.append(np.zeros((32, 32), dtype=np.int16)) # motion score = mean abs diff vs prev (frame 0 skor besar biar kepilih) scores = [1e9] # frame 0 selalu kepilih for i in range(1, n): scores.append(float(np.mean(np.abs(small[i].astype(np.int32) - small[i-1].astype(np.int32))))) # Selalu keep first & last untuk timeline, sisanya pick skor tertinggi # Jika semua skor rendah (video static), fallback ke uniform if max(scores[1:]) < 5: # threshold static indices = np.linspace(0, n-1, max_frames, dtype=int) return [frames[int(i)] for i in indices] # ambil top (max_frames-2) skor tertinggi + first/last, lalu sort # pastikan distribusi: bagi timeline jadi bucket biar tidak klaster # hybrid: 50% uniform, 50% motion top k_motion = max_frames // 2 k_uniform = max_frames - k_motion uniform_idx = set(np.linspace(0, n-1, k_uniform, dtype=int).tolist()) # top motion (exclude 0/last yang sudah di uniform) ranked = sorted(range(1, n-1), key=lambda i: scores[i], reverse=True) motion_idx = set(ranked[:k_motion]) # gabung + selalu keep 0 & n-1 chosen = sorted(uniform_idx | motion_idx | {0, n-1}) # jika masih kurang dari max_frames, tambah top berikutnya sampai cukup for idx in ranked: if len(chosen) >= max_frames: break if idx not in chosen: chosen.append(idx) chosen = sorted(chosen)[:max_frames] return [frames[i] for i in chosen] def decode_b64_to_bytes(b64_str: str) -> bytes: if b64_str.startswith("data:"): # data:image/gif;base64,... or data:video/mp4;base64,... b64_str = b64_str.split(",", 1)[1] # handle URL-safe? # add padding if needed b64_str = b64_str.strip() missing_padding = len(b64_str) % 4 if missing_padding: b64_str += "=" * (4 - missing_padding) return base64.b64decode(b64_str) async def resolve_media_to_frames(media_input: str) -> List[Image.Image]: """Auto-detect URL vs base64/dataURI -> frames""" media_input = media_input.strip() if not media_input: raise HTTPException(status_code=400, detail="image_input is required") data: bytes hint = "" if media_input.startswith("http://") or media_input.startswith("https://"): hint = media_input session = await get_http_session() async with session.get(media_input, headers={"User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", "Accept": "*/*"}) as resp: if resp.status != 200: raise HTTPException(status_code=400, detail=f"download_failed: HTTP {resp.status}") # check content-length cl = resp.headers.get("Content-Length") if cl and int(cl) > MAX_FILE_MB * 1024 * 1024: raise HTTPException(status_code=400, detail=f"file too large > {MAX_FILE_MB}MB") data = await resp.read() if len(data) > MAX_FILE_MB * 1024 * 1024: raise HTTPException(status_code=400, detail=f"file too large > {MAX_FILE_MB}MB") else: # assume base64 try: data = await asyncio.to_thread(decode_b64_to_bytes, media_input) except Exception as e: raise HTTPException(status_code=400, detail=f"invalid_image: {str(e)[:100]}") hint = "base64" try: frames = await asyncio.to_thread(extract_frames_from_bytes, data, hint) except Exception as e: raise HTTPException(status_code=400, detail=f"decode_failed: {str(e)[:100]}") if not frames: raise HTTPException(status_code=400, detail="no frames decoded") frames = sample_frames(frames, MAX_FRAMES) return frames def _solve_6grid(frames: List[Image.Image]) -> str | None: """Deterministik 6grid 300x100 20-frame: CoTracker per-cell vs header, fallback moments.""" try: if len(frames) != 20: return None W,H = frames[0].size if W != 300 or H != 100: return None import cv2, numpy as np header_h = int(H*0.25) grid_h = H - header_h cell_h = grid_h // 2 cell_w = W // 3 # Coba CoTracker dulu (paling presisi untuk translasi/rotasi) ct = _get_cotracker() if ct is not None: try: import torch as _t # siapkan header track (kanan 80px) def ct_tracks(crops): Hc,Wc = 128,128 arr=np.stack([np.array(f.resize((Wc,Hc), Image.BILINEAR)) for f in crops]) # T,H,W,3 video=_t.from_numpy(arr).permute(0,3,1,2)[None].float().to(DEVICE) # 1,T,3,128,128 with _t.inference_mode(): tracks, vis = ct(video, grid_size=5) # 1,T,N,2 return tracks[0].cpu().numpy(), (vis[0].cpu().numpy() if vis is not None else None) header_crops=[f.crop((W-80,0,W,header_h)) for f in frames] h_tracks, h_vis = ct_tracks(header_crops) # h_tracks: T,N,2 # hitung per-cell tracks dan bandingkan via DTW/cosine best=None; best_score=-1e9 h_disp=np.mean(np.sqrt(((h_tracks[-1]-h_tracks[0])**2).sum(axis=1))) h_traj=h_tracks.mean(axis=1) # T,2 mean trajectory for idx in range(1,7): r=(idx-1)//3; cc=(idx-1)%3 x0=cc*cell_w; y0=header_h+r*cell_h crops=[f.crop((x0,y0,x0+cell_w,y0+cell_h)) for f in frames] c_tracks, _ = ct_tracks(crops) c_traj=c_tracks.mean(axis=1) # similarity: Pearson correlation x+y + magnitude ratio def corr(a,b): if np.std(a)<1e-3 or np.std(b)<1e-3: return 0 return np.corrcoef(a,b)[0,1] cx=corr(h_traj[:,0], c_traj[:,0]); cy=corr(h_traj[:,1], c_traj[:,1]) c_disp=np.mean(np.sqrt(((c_tracks[-1]-c_tracks[0])**2).sum(axis=1))) mag_sim=1-abs(h_disp-c_disp)/(max(h_disp,c_disp)+1e-3) score=(cx+cy)/2*0.7 + mag_sim*0.3 # logger.info(f"6grid ct cell {idx} corr {cx:.2f},{cy:.2f} mag {c_disp:.1f} vs h {h_disp:.1f} score {score:.2f}") if score>best_score: best_score=score; best=idx if best is not None: logger.info(f"6grid CoTracker header disp {h_disp:.1f} -> best {best} score {best_score:.2f}") return str(best) except Exception as e: logger.warning(f"6grid CoTracker failed, fallback moments: {e}") # Fallback moments (lama) header_h = int(H*0.25) grid_h = H - header_h cell_h = grid_h // 2 cell_w = W // 3 # helper: untuk tiap region (list of crops per frame), hitung centroid + angle per frame def region_moments(crops): cents, angles, areas = [], [], [] for f in crops: g = np.array(f.convert("L")) # icon vs white bg: <200 _, th = cv2.threshold(g, 200, 255, cv2.THRESH_BINARY_INV) # clean small noise kernel = np.ones((2,2), np.uint8) th = cv2.morphologyEx(th, cv2.MORPH_OPEN, kernel) M = cv2.moments(th) if M["m00"] < 20: cents.append(None); angles.append(None); areas.append(0) continue cx = M["m10"]/M["m00"]; cy = M["m01"]/M["m00"] cents.append((cx, cy)) # angle via fitEllipse or moments # use mu20 mu02 mu11 mu20 = M["mu20"]/M["m00"] if M["m00"] else 0 mu02 = M["mu02"]/M["m00"] if M["m00"] else 0 mu11 = M["mu11"]/M["m00"] if M["m00"] else 0 ang = 0.5*np.arctan2(2*mu11, mu20-mu02)*180/np.pi angles.append(ang) areas.append(M["m00"]) return cents, angles, areas # header icon region: cari bbox moving via diff, sudah ada header_h, tapi Q ada di x~240-295 # Kita ambil header crops full lalu cari header Q bbox via variance (icon moves) header_crops = [f.crop((0,0,W,header_h)) for f in frames] # cari bbox header icon via diff vs median (seperti sebelumnya) tapi fokus ke x>180 # gunakan header_crops untuk moments tapi mask ke area kanan # Untuk header, kita crop kanan 100px header_icon_crops = [f.crop((W-80,0,W,header_h)) for f in frames] h_cents, h_angles, h_areas = region_moments(header_icon_crops) # filter None h_valid = [(c,a) for c,a in zip(h_cents,h_angles) if c is not None and a is not None] if len(h_valid) < 5: return None # hitung motion type header: translasi vs rotasi h_cents_f = [c for c in h_cents if c is not None] h_angles_f = [a for a in h_angles if a is not None] h_dx = max(c[0] for c in h_cents_f) - min(c[0] for c in h_cents_f) h_dy = max(c[1] for c in h_cents_f) - min(c[1] for c in h_cents_f) h_trans = np.hypot(h_dx, h_dy) h_rot = max(h_angles_f) - min(h_angles_f) h_rot_std = float(np.std(h_angles_f)) # h_is_rot = rotasi dominan jika angle std >15 deg dan trans kecil h_is_rot = h_rot_std > 12 and h_trans < 8 # untuk setiap cell, hitung similarly best = None best_score = -1 for idx in range(1,7): r=(idx-1)//3; c=(idx-1)%3 x0=c*cell_w; y0=header_h+r*cell_h crops=[f.crop((x0,y0,x0+cell_w,y0+cell_h)) for f in frames] cents, angles, _ = region_moments(crops) cents_f=[x for x in cents if x is not None] angles_f=[a for a in angles if a is not None] if len(cents_f)<5: continue dx = max(x[0] for x in cents_f)-min(x[0] for x in cents_f) dy = max(x[1] for x in cents_f)-min(x[1] for x in cents_f) trans=np.hypot(dx,dy) rot_std=float(np.std(angles_f)) if angles_f else 0 is_rot=rot_std>12 and trans<8 # similarity: header rot vs cell rot, atau header trans vs cell trans if h_is_rot == is_rot: # keduanya rotasi atau keduanya translasi → score tinggi # fine score: untuk rotasi, cocokkan rot_std, untuk translasi cocokkan arah if h_is_rot: score = 10 - abs(rot_std - h_rot_std) # semakin dekat std semakin tinggi else: # translasi: arah dominan # hitung arah h vs cell via dx/dy h_dir=np.arctan2(h_dy, h_dx) if h_trans>1 else 0 c_dir=np.arctan2(dy, dx) if trans>1 else 0 ang_diff=abs(h_dir-c_dir) ang_diff=min(ang_diff, 2*np.pi-ang_diff) score = 10 - ang_diff*2 - abs(trans-h_trans)*0.2 else: score = -10 # logger.debug # print(f"cell {idx} trans {trans:.1f} rot_std {rot_std:.1f} is_rot {is_rot} score {score:.1f}") if score > best_score: best_score=score; best=idx if best is not None: logger.info(f"6grid solver: header {'rot' if h_is_rot else 'trans'} trans {h_trans:.1f} rot_std {h_rot_std:.1f} -> best cell {best} score {best_score:.1f}") return str(best) return None except Exception as e: logger.warning(f"6grid solver failed: {e}", exc_info=True) return None def describe_sync(frames: List[Image.Image], prompt: str, max_new_tokens: int) -> str: if MODEL is None or TOKENIZER is None or IMAGE_PROCESSOR is None: raise RuntimeError("model_not_loaded") # Fast-path untuk 6grid challenge: deterministik, bypass LLM if "GRID LAYOUT" in prompt and len(frames) > 1: grid_ans = _solve_6grid(frames) if grid_ans is not None: return grid_ans # Safety: samakan ukuran semua frame (MPO thumbnail bisa beda) if len(frames) > 1: base_size = frames[0].size frames = [f if f.size == base_size else f.resize(base_size, Image.BILINEAR) for f in frames] # Convert frames to numpy array for process_images # LongVU expects video as np stack (T, H, W, C) uint8 video_np = np.stack([np.array(f) for f in frames]) image_sizes = [frames[0].size[::-1]] # (H,W) ??? PIL size is (W,H), model expects (H,W) # Actually example: image_sizes = [video[0].shape[:2]] -> (H,W) # PIL size is (W,H), so use np shape # Use video_np[0].shape[:2] image_sizes = [video_np[0].shape[:2]] with nullcontext(): # process_images returns list of tensors per vision tower, each (T, C, H, W) half video_tensors = process_images(video_np, IMAGE_PROCESSOR, MODEL.config) # LongVU example does unsqueeze(0) to add batch dim: (T,C,H,W) -> (1,T,C,H,W) if isinstance(video_tensors, list): video_tensors = [t.unsqueeze(0).to(MODEL.device) if hasattr(t, 'to') else t.unsqueeze(0) for t in video_tensors] else: video_tensors = video_tensors.unsqueeze(0).to(MODEL.device) video_tensors = [video_tensors] # Build prompt — untuk animated (>1 frame) tambahkan motion cues dari optical flow qs = prompt.strip() motion_hint = "" if len(frames) > 1: # 1) CoTracker dense (paling presisi) 2) Farneback + centroid fallback motion_hint = _cotracker_hint(frames) if not motion_hint: try: import cv2 # type: ignore small = [np.array(f.resize((64, 64), Image.BILINEAR).convert("L")) for f in frames] mags, dirs = [], [] for i in range(1, len(small)): flow = cv2.calcOpticalFlowFarneback(small[i-1], small[i], None, 0.5, 3, 15, 3, 5, 1.2, 0) mag, ang = cv2.cartToPolar(flow[..., 0], flow[..., 1]) mags.append(float(np.mean(mag))) dirs.append(float(np.median(ang) * 180 / np.pi)) avg_mag = float(np.mean(mags)) if mags else 0 # fallback centroid untuk sparse object (dot) dimana Farneback noise centroid_hint = "" try: cents = [] for s in small: # threshold putih di atas 200 _, th = cv2.threshold(s, 200, 255, cv2.THRESH_BINARY) M = cv2.moments(th) if M["m00"] > 10: cx = M["m10"] / M["m00"] cy = M["m01"] / M["m00"] cents.append((cx, cy)) if len(cents) >= 2: dx = cents[-1][0] - cents[0][0] dy = cents[-1][1] - cents[0][1] if abs(dx) > 5 or abs(dy) > 5: dir_txt = [] if dx > 5: dir_txt.append("right") elif dx < -5: dir_txt.append("left") if dy > 5: dir_txt.append("down") elif dy < -5: dir_txt.append("up") centroid_hint = f" Object centroid moves {'-'.join(dir_txt)} ({dx:+.0f}px, {dy:+.0f}px)." except Exception: pass if avg_mag < 0.8 and not centroid_hint: motion_hint = " Motion: mostly static with minor changes." else: bins = {"right": 0, "down": 0, "left": 0, "up": 0} for a in dirs: if 315 <= a or a < 45: bins["right"] += 1 elif 45 <= a < 135: bins["down"] += 1 elif 135 <= a < 225: bins["left"] += 1 else: bins["up"] += 1 dom = max(bins, key=bins.get) if dirs else "unknown" motion_hint = f" Motion: avg flow {avg_mag:.1f}px, dominant {dom} ({bins.get(dom,0)}/{len(dirs)})," if centroid_hint: motion_hint += centroid_hint else: motion_hint += " object/camera movement present." except Exception: motion_hint = " Motion: changes detected between frames." if motion_hint and motion_hint.lower().strip() not in qs.lower(): qs = qs + motion_hint if "frame" not in qs.lower(): qs = qs + " Describe frame-by-frame temporal order." # LongVU uses DEFAULT_IMAGE_TOKEN qs = DEFAULT_IMAGE_TOKEN + "\n" + qs conv = conv_templates["qwen"].copy() conv.append_message(conv.roles[0], qs) conv.append_message(conv.roles[1], None) prompt_text = conv.get_prompt() input_ids = tokenizer_image_token(prompt_text, TOKENIZER, IMAGE_TOKEN_INDEX, return_tensors="pt").unsqueeze(0).to(MODEL.device) stop_str = conv.sep if conv.sep_style != SeparatorStyle.TWO else conv.sep2 keywords = [stop_str] stopping_criteria = KeywordsStoppingCriteria(keywords, TOKENIZER, input_ids) with torch.inference_mode(): output_ids = MODEL.generate( input_ids, images=video_tensors, image_sizes=image_sizes, do_sample=True, temperature=0.2, top_p=None, max_new_tokens=max_new_tokens, use_cache=True, stopping_criteria=[stopping_criteria], ) # Upstream LongVU decodes full output_ids (with inputs_embeds, output = new tokens only) # Match upstream: batch_decode full output_ids, skip_special, strip text = TOKENIZER.batch_decode(output_ids, skip_special_tokens=True)[0].strip() # 5.42: if output includes prompt, strip prompt prefix # upstream juga pakai remove_after_last_dot; kita pakai strip saja if prompt_text in text: text = text.split(prompt_text)[-1].strip() return text.strip() async def run_inference(frames: List[Image.Image], prompt: str, max_new_tokens: int) -> str: global _INFERENCE_ACTIVE, _INFERENCE_WAITERS if _INFERENCE_WAITERS >= MAX_INFERENCE_WAITERS: raise ServerBusyError() _INFERENCE_WAITERS += 1 acquired = False try: async with get_semaphore(): _INFERENCE_WAITERS -= 1 acquired = True _INFERENCE_ACTIVE += 1 start = time.perf_counter() try: result = await asyncio.to_thread(describe_sync, frames, prompt, max_new_tokens) logger.info(f"Inference done ({(time.perf_counter()-start)*1000:.0f}ms, tokens={max_new_tokens}, frames={len(frames)})") return result finally: _INFERENCE_ACTIVE -= 1 finally: if not acquired: _INFERENCE_WAITERS -= 1 def health() -> dict: return { "status": "healthy" if MODEL is not None else "unhealthy", "model_loaded": MODEL is not None, "device": DEVICE, "timestamp": time.strftime("%Y-%m-%dT%H:%M:%S%z"), "load": { "inference_active": _INFERENCE_ACTIVE, "inference_waiters": _INFERENCE_WAITERS, "max_concurrent_inferences": MAX_CONCURRENT_INFERENCES, "max_inference_waiters": MAX_INFERENCE_WAITERS, }, } # ==================== FASTAPI ==================== fastapi_app = FastAPI(title="Describe API - LongVU", version="2.0") @fastapi_app.exception_handler(ServerBusyError) async def server_busy_handler(_, __): return JSONResponse({"success": False, "error": "server_busy", "status_code": 429}, status_code=429, headers={"Retry-After": "2"}) @fastapi_app.exception_handler(HTTPException) async def http_exc_handler(_, exc): return JSONResponse({"success": False, "error": exc.detail, "status_code": exc.status_code}, status_code=exc.status_code) @fastapi_app.exception_handler(Exception) async def global_exc_handler(_, exc): logger.error(f"Unhandled error: {exc}", exc_info=True) return JSONResponse({"success": False, "error": "Internal server error"}, status_code=500) @fastapi_app.get("/health") async def health_endpoint(): return health() class SolveRequest(BaseModel): image_input: str prompt: str = DEFAULT_PROMPT max_new_tokens: int = DEFAULT_MAX_NEW_TOKENS @fastapi_app.post("/solve") async def solve_endpoint(request: SolveRequest): with request_scope() as rid: logger.info("Solve request received") if MODEL is None or TOKENIZER is None: raise HTTPException(status_code=503, detail="model_loading") if not request.image_input or not request.image_input.strip(): raise HTTPException(status_code=400, detail="image_input is required") if not request.prompt or not request.prompt.strip(): raise HTTPException(status_code=400, detail="prompt is required") frames = await resolve_media_to_frames(request.image_input.strip()) logger.info(f"Media resolved: frames={len(frames)}, size={frames[0].size}") text = await run_inference(frames, request.prompt.strip(), request.max_new_tokens) if not text: text = "" logger.warning("Model returned empty output") logger.info("Solve completed") return {"success": True, "text": text} # ==================== GRADIO ==================== gradio_demo = gr.Blocks(title="Describe API - LongVU", analytics_enabled=False) with gradio_demo: gr.Markdown("## Describe API - LongVU (image / gif / webp / mp4)\nSupport semua type via `POST /solve` dengan `image_input` URL atau base64.") with gr.Row(): inp_media = gr.File(label="media (image/gif/webp/mp4) - upload", file_count="single") inp_prompt = gr.Textbox(label="prompt", value=DEFAULT_PROMPT) inp_tokens = gr.Slider(minimum=16, maximum=1024, value=DEFAULT_MAX_NEW_TOKENS, step=16, label="max_new_tokens") out_text = gr.Textbox(label="result") btn = gr.Button("Run /solve") async def gradio_solve(file_obj, prompt: str, max_new_tokens: float | None) -> str: with request_scope() as rid: try: if MODEL is None or TOKENIZER is None: raise gr.Error("model_loading") if file_obj is None: raise gr.Error("file is required (image/gif/webp/mp4)") if not prompt or not prompt.strip(): raise gr.Error("prompt is required") tokens = DEFAULT_MAX_NEW_TOKENS if max_new_tokens is None else int(max_new_tokens) # gr.File di Gradio 5 bisa kirim str path atau dict {"path": "...", "_type": "gradio.FileData"} if isinstance(file_obj, str): file_path = file_obj elif isinstance(file_obj, dict): file_path = file_obj.get("path") or file_obj.get("name") or file_obj.get("url") or str(file_obj) elif hasattr(file_obj, "name"): file_path = file_obj.name else: file_path = str(file_obj) try: with open(file_path, "rb") as f: data = f.read() except Exception as e: logger.error(f"Gradio read failed {file_path}: {e}", exc_info=True) raise gr.Error(f"read failed: {e}") hint = file_path frames = await asyncio.to_thread(extract_frames_from_bytes, data, hint) frames = sample_frames(frames, MAX_FRAMES) logger.info(f"Gradio media: frames={len(frames)}, hint={hint}") result = await run_inference(frames, prompt.strip(), tokens) logger.info(f"Gradio result len={len(result) if result else 0} preview={(result[:80] if result else '')!r}") if not result: raise gr.Error("empty result from model") return result except gr.Error: raise except Exception as e: logger.error(f"Gradio solve unhandled: {e}", exc_info=True) raise gr.Error(f"solve failed: {str(e)[:200]}") btn.click(gradio_solve, inputs=[inp_media, inp_prompt, inp_tokens], outputs=out_text, api_name="solve", concurrency_limit=None) # Advanced: URL/base64 juga didukung via API POST /solve (JSON), tidak perlu UI tambahan gradio_demo.queue(max_size=20, default_concurrency_limit=1) # ==================== RUN ==================== if __name__ == "__main__": import threading app_obj, _, _ = gradio_demo.launch(ssr_mode=False, prevent_thread_lock=True) logger.info("launch() returned - server live") try: app_obj.add_api_route("/health", health, methods=["GET"]) app_obj.add_api_route("/solve", solve_endpoint, methods=["POST"]) logger.info("Clean /health dan /solve registered pada live app") except Exception as e: logger.warning(f"Gagal daftar route: {e}", exc_info=True) threading.Thread(target=ensure_model_loaded, daemon=True).start() threading.Event().wait()