Update app.py

app.py

@@ -1,78 +1,96 @@
-# app.py — InstantID SDXL (Option 1, robuste aux erreurs OMP + hub)
-# - Tu uploades SEULEMENT la pipeline .py (texte) dans ./instantid/
-# - Les poids sont téléchargés auto depuis InstantX/InstantID (repo Model)
-
+# app.py — InstantID SDXL (import local résilient + draw_kps local)
 import os, traceback, importlib.util
 
-# --- Sécuriser OMP_NUM_THREADS (éviter libgomp error) ---
+# OMP robuste
 val = os.environ.get("OMP_NUM_THREADS", "")
 try:
     if val == "" or int(val) <= 0:
         os.environ["OMP_NUM_THREADS"] = "1"
 except Exception:
     os.environ["OMP_NUM_THREADS"] = "1"
-
 os.environ.setdefault("HF_HUB_ENABLE_HF_TRANSFER", "1")
 
 import torch, gradio as gr
-from PIL import Image, ImageOps
-from huggingface_hub import hf_hub_download   # <- pas de HfHubHTTPError (compat large)
+from PIL import Image, ImageOps, ImageDraw
+from huggingface_hub import hf_hub_download
 from diffusers.models import ControlNetModel
 
 DEVICE = "cuda" if torch.cuda.is_available() else "cpu"
 DTYPE  = torch.float16 if DEVICE == "cuda" else torch.float32
 
-# -------- Repo Model qui contient les poids (public) --------
+# -------- Repo MODEL pour télécharger les poids --------
 ASSETS_REPO = "InstantX/InstantID"
 
-# -------- Chemins locaux attendus dans le Space --------
+# -------- Fichiers attendus localement (TEXTE uniquement) --------
 PIPE_CANDIDATES = [
     "./instantid/pipeline_stable_diffusion_xl_instantid.py",
     "./instantid/pipeline_stable_diffusion_xl_instantid_full.py",
 ]
-CHECKPOINTS_DIR   = "./checkpoints"
-CN_LOCAL_DIR      = os.path.join(CHECKPOINTS_DIR, "ControlNetModel")
-IP_ADAPTER_LOCAL  = os.path.join(CHECKPOINTS_DIR, "ip-adapter.bin")
 
-# -------- Utilitaires --------
+CHECKPOINTS_DIR  = "./checkpoints"
+CN_LOCAL_DIR     = os.path.join(CHECKPOINTS_DIR, "ControlNetModel")
+IP_ADAPTER_LOCAL = os.path.join(CHECKPOINTS_DIR, "ip-adapter.bin")
+
+# ---------- import dynamique tolérant ----------
 def import_pipeline_or_fail():
     pipeline_file = next((p for p in PIPE_CANDIDATES if os.path.exists(p)), None)
     if pipeline_file is None:
         raise RuntimeError(
             "Pipeline InstantID introuvable.\n"
-            "➡️ Uploade l’un de ces fichiers (TEXTE uniquement) dans ton Space :\n"
-            "   - instantid/pipeline_stable_diffusion_xl_instantid.py\n"
-            "   - instantid/pipeline_stable_diffusion_xl_instantid_full.py\n"
-            "(Ne mets PAS de .safetensors dans le Space — ils seront téléchargés automatiquement)."
+            "➡️ Place un des fichiers (texte) dans ./instantid/ :\n"
+            "   - pipeline_stable_diffusion_xl_instantid.py\n"
+            "   - pipeline_stable_diffusion_xl_instantid_full.py\n"
         )
     spec = importlib.util.spec_from_file_location("instantid_pipeline", pipeline_file)
     mod  = importlib.util.module_from_spec(spec)
     spec.loader.exec_module(mod)
-    SDXLInstantID = getattr(mod, "StableDiffusionXLInstantIDPipeline", None)
-    draw_kps = getattr(mod, "draw_kps", None)
-    if SDXLInstantID is None or draw_kps is None:
-        raise RuntimeError("Le fichier pipeline ne contient pas StableDiffusionXLInstantIDPipeline/draw_kps.")
-    return SDXLInstantID, draw_kps
 
+    # Cherche une classe dont le nom contient 'InstantID' et qui expose from_pretrained
+    for name, obj in vars(mod).items():
+        if isinstance(obj, type) and "InstantID" in name and hasattr(obj, "from_pretrained"):
+            return obj  # classe pipeline
+    # Si rien trouvé, affiche les classes disponibles
+    avail = [n for n, o in vars(mod).items() if isinstance(o, type)]
+    raise RuntimeError("Aucune classe pipeline 'InstantID' trouvée. Classes dispo: " + ", ".join(avail))
+
+# ---------- téléchargement des poids depuis le repo MODEL ----------
 def ensure_assets_or_download():
     os.makedirs(CHECKPOINTS_DIR, exist_ok=True)
     os.makedirs(CN_LOCAL_DIR, exist_ok=True)
-    # ControlNet IdentityNet
     if not os.path.isfile(os.path.join(CN_LOCAL_DIR, "config.json")):
         hf_hub_download(ASSETS_REPO, "ControlNetModel/config.json", local_dir=CHECKPOINTS_DIR)
     if not os.path.isfile(os.path.join(CN_LOCAL_DIR, "diffusion_pytorch_model.safetensors")):
         hf_hub_download(ASSETS_REPO, "ControlNetModel/diffusion_pytorch_model.safetensors", local_dir=CHECKPOINTS_DIR)
-    # ip-adapter
     if not os.path.isfile(IP_ADAPTER_LOCAL):
         hf_hub_download(ASSETS_REPO, "ip-adapter.bin", local_dir=CHECKPOINTS_DIR)
 
-# -------- Chargement pipeline --------
+# ---------- draw_kps local (remplace la dépendance du fichier) ----------
+def draw_kps_local(img_pil: Image.Image, kps):
+    """
+    kps: ndarray shape (5,2) insightface 'antelopev2' (yeux G/D, nez, bouche G/D).
+    On dessine de petits ronds noirs sur fond blanc (format attendu par IdentityNet).
+    """
+    w, h = img_pil.size
+    out = Image.new("RGB", (w, h), "white")
+    d = ImageDraw.Draw(out)
+    r = max(2, min(w, h) // 100)  # rayon adaptatif
+    try:
+        import numpy as np
+        pts = kps if isinstance(kps, (list, tuple)) else np.array(kps)
+    except Exception:
+        pts = kps
+    for p in pts:
+        x, y = float(p[0]), float(p[1])
+        d.ellipse((x - r, y - r, x + r, y + r), fill="black")
+    return out
+
+# ---------- Chargement pipeline ----------
 load_logs = []
 try:
-    SDXLInstantID, draw_kps = import_pipeline_or_fail()
+    SDXLInstantID = import_pipeline_or_fail()
     ensure_assets_or_download()
 
-    BASE_MODEL = "stabilityai/stable-diffusion-xl-base-1.0"  # tu peux remplacer par un SDXL plus “anime”
+    BASE_MODEL = "stabilityai/stable-diffusion-xl-base-1.0"  # changeable vers un SDXL anime
 
     load_logs.append("Chargement ControlNet IdentityNet…")
     controlnet_identitynet = ControlNetModel.from_pretrained(CN_LOCAL_DIR, torch_dtype=DTYPE)
@@ -86,11 +104,10 @@ try:
         feature_extractor=None,
     ).to(DEVICE)
 
-    # ip-adapter d’InstantID
     if hasattr(pipe, "load_ip_adapter_instantid"):
         pipe.load_ip_adapter_instantid(IP_ADAPTER_LOCAL)
     else:
-        raise RuntimeError("La méthode load_ip_adapter_instantid est absente de cette pipeline.")
+        raise RuntimeError("Cette pipeline ne fournit pas load_ip_adapter_instantid().")
 
     if DEVICE == "cuda":
         if hasattr(pipe, "image_proj_model"): pipe.image_proj_model.to("cuda")
@@ -104,21 +121,21 @@ except Exception:
 if pipe is None:
     raise RuntimeError("Échec chargement pipeline InstantID.\n" + "\n".join(load_logs))
 
-# -------- InsightFace (landmarks) --------
+# ---------- InsightFace pour landmarks ----------
 from insightface.app import FaceAnalysis
 fa = FaceAnalysis(name="antelopev2", root="./", providers=["CPUExecutionProvider"])
 fa.prepare(ctx_id=0, det_size=(640, 640))
 
-def extract_kps_image(pil_img: Image.Image):
+def kps_image_from_face(pil_img: Image.Image):
     import numpy as np, cv2
     img_cv2 = cv2.cvtColor(np.array(pil_img.convert("RGB")), cv2.COLOR_RGB2BGR)
     faces = fa.get(img_cv2)
     if not faces:
-        raise ValueError("Aucun visage détecté. Utilise un portrait net (visage centré).")
+        raise ValueError("Aucun visage détecté. Utilise une photo portrait nette (visage centré).")
     face = faces[-1]
-    return draw_kps(pil_img, face["kps"])
+    return draw_kps_local(pil_img, face["kps"])
 
-# -------- Inference --------
+# ---------- Inference ----------
 def generate(face_image, prompt, negative_prompt, identity_strength, adapter_strength, steps, cfg, width, height, seed):
     try:
         if face_image is None:
@@ -130,7 +147,7 @@ def generate(face_image, prompt, negative_prompt, identity_strength, adapter_str
         ms = min(face.size); x=(face.width-ms)//2; y=(face.height-ms)//2
         face_sq = face.crop((x, y, x+ms, y+ms)).resize((512,512), Image.Resampling.LANCZOS)
 
-        kps_img = extract_kps_image(face_sq)
+        kps_img = kps_image_from_face(face_sq)
 
         if hasattr(pipe, "set_ip_adapter_scale"):
             pipe.set_ip_adapter_scale(float(adapter_strength))
@@ -154,7 +171,7 @@ def generate(face_image, prompt, negative_prompt, identity_strength, adapter_str
     except Exception:
         return None, "Erreur:\n"+traceback.format_exc(), "\n".join(load_logs)
 
-# -------- UI --------
+# ---------- UI ----------
 EX_PROMPT = (
     "one piece style, Eiichiro Oda style, anime portrait, upper body, pirate outfit, straw hat, "
     "clean lineart, cel shading, vibrant colors, expressive eyes, symmetrical face, looking at camera, "
@@ -166,7 +183,7 @@ EX_NEG = (
 )
 
 with gr.Blocks(css="footer{display:none !important}") as demo:
-    gr.Markdown("# 🏴‍☠️ One Piece — InstantID (SDXL) — Poids auto depuis HF")
+    gr.Markdown("# 🏴‍☠️ One Piece — InstantID (SDXL) — Import local robuste")
 
     with gr.Row():
         with gr.Column():