Update app.py

app.py

@@ -1,73 +1,46 @@
 import io
 import time
-import numpy as np
+import base64
 import torch
-import torch.nn.functional as F
 from PIL import Image
-from huggingface_hub import hf_hub_download
-from torchvision import transforms
+from transformers import AutoProcessor, AutoModelForCausalLM
 from fastapi import FastAPI, HTTPException, UploadFile, File
 from fastapi.middleware.cors import CORSMiddleware
 from contextlib import asynccontextmanager
 
-# ─── 28 Class Labels لـ Sapiens ──────────────────────────────────
-SEG_CLASSES = [
-    "Background", "Apparel", "Face Neck", "Hair", "Left Foot",
-    "Left Hand", "Left Lower Arm", "Left Lower Leg", "Left Shoe",
-    "Left Sock", "Left Upper Arm", "Left Upper Leg", "Lower Clothing",
-    "Right Foot", "Right Hand", "Right Lower Arm", "Right Lower Leg",
-    "Right Shoe", "Right Sock", "Right Upper Arm", "Right Upper Leg",
-    "Torso", "Upper Clothing", "Lower Lip", "Upper Lip", "Lower Teeth",
-    "Upper Teeth", "Tongue"
-]
+MODEL_ID = "Qwen/Qwen3.5-0.8B"
 
-# ─── Classes التي تدل على وجود إنسان ─────────────────────────────
-HUMAN_CLASSES = set(range(1, 28))  # كل شيء ما عدا Background (0)
+VQA_QUESTION = (
+    "Is there a woman or any part of a woman's body in this image? "
+    "Answer yes or no only. /no_think"
+)
 
 MODEL_DATA = {}
 
-def preprocess(image: Image.Image) -> torch.Tensor:
-    transform = transforms.Compose([
-        transforms.Resize((1024, 768)),
-        transforms.ToTensor(),
-        transforms.Normalize(
-            mean=[0.485, 0.456, 0.406],
-            std=[0.229, 0.224, 0.225]
-        )
-    ])
-    return transform(image).unsqueeze(0)
-
-def postprocess(output: torch.Tensor, original_size: tuple) -> np.ndarray:
-    # output shape: [1, 28, H, W]
-    seg_map = F.interpolate(
-        output, size=original_size, mode="bilinear", align_corners=False
-    )
-    seg_map = torch.argmax(seg_map, dim=1).squeeze(0).cpu().numpy()
-    return seg_map.astype(np.uint8)
-
 @asynccontextmanager
 async def lifespan(app: FastAPI):
-    print("📥 Downloading sapiens-seg-1b TorchScript model (~4.72 GB)...")
+    print(f"📥 Loading {MODEL_ID}...")
     start = time.time()
 
-    model_path = hf_hub_download(
-        repo_id="facebook/sapiens-seg-1b-torchscript",
-        filename="sapiens_1b_goliath_best_goliath_mIoU_7994_epoch_151_torchscript.pt2"
+    MODEL_DATA["processor"] = AutoProcessor.from_pretrained(
+        MODEL_ID,
+        trust_remote_code=True
     )
-    print(f"✅ Downloaded in {time.time()-start:.1f}s")
-
-    print("📦 Loading model into memory...")
-    load_start = time.time()
-    model = torch.jit.load(model_path, map_location="cpu")
-    model.eval()
-    MODEL_DATA["model"] = model
-    print(f"✅ Model ready in {time.time()-load_start:.1f}s")
+    MODEL_DATA["model"] = AutoModelForCausalLM.from_pretrained(
+        MODEL_ID,
+        torch_dtype=torch.float32,
+        trust_remote_code=True,
+        attn_implementation="eager",
+        device_map="cpu"
+    ).eval()
+
+    print(f"✅ Model ready in {time.time()-start:.1f}s")
     yield
     MODEL_DATA.clear()
 
 app = FastAPI(
-    title="Human Body Segmentation - Sapiens-1B",
-    description="Meta Sapiens-seg-1b | 28 class body part segmentation",
+    title="Female Detection API - Qwen3.5-0.8B",
+    description="Qwen3.5-0.8B VLM | VQA",
     version="1.0.0",
     lifespan=lifespan
 )
@@ -84,6 +57,15 @@ app.add_middleware(
 def health():
     return {"status": "ok", "model_loaded": "model" in MODEL_DATA}
 
+def decide(answer: str) -> tuple[str, str]:
+    a = answer.strip().lower()
+    if a == "no" or a.startswith("no"):
+        return "allow", "model_answered_no"
+    elif "yes" in a:
+        return "block", "model_answered_yes"
+    else:
+        return "block", "unexpected_answer_blocked_for_safety"
+
 @app.post("/analyze")
 async def analyze_image(file: UploadFile = File(...)):
 
@@ -91,54 +73,67 @@ async def analyze_image(file: UploadFile = File(...)):
         raise HTTPException(status_code=400, detail="الملف ليس صورة")
 
     try:
-        image = Image.open(io.BytesIO(await file.read())).convert("RGB")
+        image_bytes = await file.read()
+        image = Image.open(io.BytesIO(image_bytes)).convert("RGB")
     except Exception as e:
         raise HTTPException(status_code=400, detail=f"خطأ في قراءة الصورة: {str(e)}")
 
     try:
-        model = MODEL_DATA["model"]
-        original_size = (image.height, image.width)
-
-        # ─── Preprocessing ────────────────────────────────────────
-        tensor = preprocess(image)
-
-        # ─── Inference ────────────────────────────────────────────
-        start_time = time.time()
-        with torch.inference_mode():
-            output = model(tensor)
-        elapsed = round(time.time() - start_time, 2)
-
-        # ─── Postprocessing ───────────────────────────────────────
-        seg_map = postprocess(output, original_size)
-
-        # ─── استخراج الـ classes الموجودة ─────────────────────────
-        unique_classes = set(np.unique(seg_map).tolist())
-        detected_parts = [
-            SEG_CLASSES[c] for c in sorted(unique_classes)
-            if c in HUMAN_CLASSES
+        processor = MODEL_DATA["processor"]
+        model     = MODEL_DATA["model"]
+
+        # ─── Qwen3.5 يستخدم نفس طريقة Qwen3-VL ──────────────────
+        # الصورة ترسل كـ type: "image" مباشرة في content
+        messages = [
+            {
+                "role": "user",
+                "content": [
+                    {
+                        "type": "image",
+                        "image": image   # PIL Image مباشرة
+                    },
+                    {
+                        "type": "text",
+                        "text": VQA_QUESTION
+                    }
+                ]
+            }
         ]
 
-        # ─── حساب نسبة كل جزء من الصورة ──────────────────────────
-        total_pixels = seg_map.size
-        parts_coverage = {}
-        for c in sorted(unique_classes):
-            if c in HUMAN_CLASSES:
-                count = int(np.sum(seg_map == c))
-                percentage = round((count / total_pixels) * 100, 2)
-                if percentage > 0.1:  # تجاهل أقل من 0.1%
-                    parts_coverage[SEG_CLASSES[c]] = f"{percentage}%"
+        # ─── apply_chat_template ──────────────────────────────────
+        inputs = processor.apply_chat_template(
+            messages,
+            tokenize=True,
+            add_generation_prompt=True,
+            return_dict=True,
+            return_tensors="pt"
+        )
 
-        human_detected = len(detected_parts) > 0
-        summary = f"yes detected human body parts: {', '.join(detected_parts)}" \
-                  if human_detected else "no detected human body"
+        start_time = time.time()
+        with torch.no_grad():
+            generated_ids = model.generate(
+                **inputs,
+                max_new_tokens=20,
+                do_sample=False,
+                temperature=None,
+                top_p=None,
+            )
+
+        # ─── استخراج الإجابة الجديدة فقط ─────────────────────────
+        input_len  = inputs["input_ids"].shape[1]
+        new_tokens = generated_ids[:, input_len:]
+        answer     = processor.decode(new_tokens[0], skip_special_tokens=True).strip()
+
+        elapsed          = round(time.time() - start_time, 2)
+        decision, reason = decide(answer)
 
         return {
-            "summary":         summary,
-            "detected":        human_detected,
-            "parts_found":     detected_parts,
-            "parts_coverage":  parts_coverage,
-            "execution_time":  elapsed,
-            "status":          "success"
+            "decision":       decision,
+            "reason":         reason,
+            "vqa_answer":     answer,
+            "question":       VQA_QUESTION,
+            "execution_time": elapsed,
+            "status":         "success"
         }
 
     except Exception as e: