Update app.py

app.py

@@ -1,43 +1,220 @@
+# app.py — InstantID SDXL (version stable Hugging Face)
+# --------------------------------------------
+# ⚙️ Nécessite :
+#   - dossier ./instantid/ avec :
+#       pipeline_stable_diffusion_xl_instantid.py
+#       ip_adapter/ (avec les fichiers du repo GitHub)
+#   - requirements.txt avec numpy==1.26.4, onnxruntime==1.16.3
+# --------------------------------------------
+
+import os, sys, traceback, importlib.util
+
+# --- Sécuriser l'environnement ---
+os.environ.setdefault("OMP_NUM_THREADS", "1")
+os.environ.setdefault("HF_HUB_ENABLE_HF_TRANSFER", "1")
+
+# --- S'assurer que Python trouve ./instantid/ip_adapter ---
+sys.path.insert(0, os.path.abspath("./instantid"))
+
+import torch, gradio as gr
+from PIL import Image, ImageOps, ImageDraw
 from huggingface_hub import hf_hub_download
-import os
+from diffusers.models import ControlNetModel
+
+DEVICE = "cuda" if torch.cuda.is_available() else "cpu"
+DTYPE  = torch.float16 if DEVICE == "cuda" else torch.float32
 
+# --- Répertoires / fichiers ---
 ASSETS_REPO = "InstantX/InstantID"
 CHECKPOINTS_DIR  = "./checkpoints"
 CN_LOCAL_DIR     = os.path.join(CHECKPOINTS_DIR, "ControlNetModel")
 IP_ADAPTER_LOCAL = os.path.join(CHECKPOINTS_DIR, "ip-adapter.bin")
 
+# ------------------------------------------------------------
+# Téléchargement robuste (détecte les fichiers vides)
+# ------------------------------------------------------------
 def safe_download(repo, filename, local_dir, min_bytes, label):
-    """Télécharge avec reprise, puis vérifie la taille. Re-télécharge si besoin."""
     os.makedirs(local_dir, exist_ok=True)
-    # Si fichier présent mais trop petit => le supprimer (corrompu)
     local_path = os.path.join(local_dir, filename)
+    # Supprimer fichier incomplet
     if os.path.exists(local_path) and os.path.getsize(local_path) < min_bytes:
-        try:
-            os.remove(local_path)
-        except Exception:
-            pass
-    # Téléchargement (avec reprise)
+        print(f"⚠️ {label} corrompu ({os.path.getsize(local_path)} bytes) → suppression")
+        os.remove(local_path)
+    # Télécharger / reprendre
     path = hf_hub_download(
         repo_id=repo,
         filename=filename,
         local_dir=local_dir,
         resume_download=True,
-        force_download=not os.path.exists(os.path.join(local_dir, filename)),
+        force_download=not os.path.exists(local_path),
     )
-    # Vérification de taille
     size = os.path.getsize(path)
+    print(f"✅ {label} téléchargé ({size/1e6:.1f} MB)")
     if size < min_bytes:
-        raise RuntimeError(
-            f"Téléchargement incomplet de {label} ({size} bytes). "
-            f"Relance le Space; si ça persiste, vérifie la connectivité HF."
-        )
+        raise RuntimeError(f"Téléchargement incomplet de {label}.")
     return path
 
 def ensure_assets_or_download():
     os.makedirs(CHECKPOINTS_DIR, exist_ok=True)
     os.makedirs(CN_LOCAL_DIR, exist_ok=True)
-    # ControlNet IdentityNet (~2.5 Go)
     safe_download(ASSETS_REPO, "ControlNetModel/config.json", CHECKPOINTS_DIR, 1000, "IdentityNet config")
     safe_download(ASSETS_REPO, "ControlNetModel/diffusion_pytorch_model.safetensors", CHECKPOINTS_DIR, 100_000_000, "IdentityNet weights")
-    # ip-adapter (~1.6 Go)
     safe_download(ASSETS_REPO, "ip-adapter.bin", CHECKPOINTS_DIR, 100_000_000, "ip-adapter")
+
+# ------------------------------------------------------------
+# Import dynamique de la pipeline InstantID
+# ------------------------------------------------------------
+def import_pipeline_or_fail():
+    pipeline_file = "./instantid/pipeline_stable_diffusion_xl_instantid.py"
+    if not os.path.exists(pipeline_file):
+        raise RuntimeError("❌ Fichier pipeline manquant dans ./instantid/")
+    spec = importlib.util.spec_from_file_location("instantid_pipeline", pipeline_file)
+    mod  = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(mod)
+    # Chercher la classe InstantID
+    for name, obj in vars(mod).items():
+        if isinstance(obj, type) and "InstantID" in name and hasattr(obj, "from_pretrained"):
+            print(f"✅ Pipeline trouvée : {name}")
+            return obj
+    raise RuntimeError("❌ Aucune classe pipeline InstantID trouvée dans le fichier.")
+
+# ------------------------------------------------------------
+# draw_kps local (remplace la dépendance d'origine)
+# ------------------------------------------------------------
+def draw_kps_local(img_pil, kps):
+    w, h = img_pil.size
+    out = Image.new("RGB", (w, h), "white")
+    d = ImageDraw.Draw(out)
+    r = max(2, min(w, h)//100)
+    for (x, y) in kps:
+        d.ellipse((x - r, y - r, x + r, y + r), fill="black")
+    return out
+
+# ------------------------------------------------------------
+# Chargement du modèle complet
+# ------------------------------------------------------------
+load_logs = []
+try:
+    SDXLInstantID = import_pipeline_or_fail()
+    ensure_assets_or_download()
+
+    BASE_MODEL = "stabilityai/stable-diffusion-xl-base-1.0"
+    load_logs.append(f"Chargement base: {BASE_MODEL}")
+
+    controlnet_identitynet = ControlNetModel.from_pretrained(CN_LOCAL_DIR, torch_dtype=DTYPE)
+    pipe = SDXLInstantID.from_pretrained(
+        BASE_MODEL,
+        controlnet=[controlnet_identitynet],
+        torch_dtype=DTYPE,
+        safety_checker=None,
+        feature_extractor=None,
+    ).to(DEVICE)
+
+    pipe.load_ip_adapter_instantid(IP_ADAPTER_LOCAL)
+    if DEVICE == "cuda":
+        if hasattr(pipe, "image_proj_model"): pipe.image_proj_model.to("cuda")
+        if hasattr(pipe, "unet"): pipe.unet.to("cuda")
+
+    load_logs.append("✅ InstantID prêt.")
+except Exception:
+    load_logs += ["❌ ERREUR au chargement:", traceback.format_exc()]
+    pipe = None
+
+if pipe is None:
+    raise RuntimeError("Échec de chargement du pipeline.\n" + "\n".join(load_logs))
+
+# ------------------------------------------------------------
+# Détection visage (InsightFace)
+# ------------------------------------------------------------
+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):
+    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é.")
+    face = faces[-1]
+    return draw_kps_local(pil_img, face["kps"])
+
+# ------------------------------------------------------------
+# Inference
+# ------------------------------------------------------------
+def generate(face_image, prompt, negative_prompt, identity_strength, adapter_strength, steps, cfg, width, height, seed):
+    try:
+        if face_image is None:
+            return None, "Merci d'ajouter une photo visage.", "\n".join(load_logs)
+        gen = None if seed is None or int(seed) < 0 else torch.Generator(device=DEVICE).manual_seed(int(seed))
+        face = ImageOps.exif_transpose(face_image).convert("RGB")
+        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)
+        if hasattr(pipe, "set_ip_adapter_scale"):
+            pipe.set_ip_adapter_scale(float(adapter_strength))
+        images = pipe(
+            prompt=prompt.strip(),
+            negative_prompt=(negative_prompt or "").strip(),
+            image=kps_img,
+            controlnet_conditioning_scale=float(identity_strength),
+            num_inference_steps=int(steps),
+            guidance_scale=float(cfg),
+            width=int(width),
+            height=int(height),
+            generator=gen,
+        ).images
+        return images[0], "", "\n".join(load_logs)
+    except torch.cuda.OutOfMemoryError as oom:
+        return None, "CUDA OOM: baisse la résolution.", "\n".join(load_logs)
+    except Exception:
+        return None, "Erreur:\n"+traceback.format_exc(), "\n".join(load_logs)
+
+# ------------------------------------------------------------
+# Interface Gradio
+# ------------------------------------------------------------
+EX_PROMPT = (
+    "one piece style, Eiichiro Oda style, anime portrait, upper body, pirate outfit, "
+    "clean lineart, cel shading, vibrant colors, expressive eyes, symmetrical face, "
+    "dynamic lighting, simple background, high detail"
+)
+EX_NEG = (
+    "low quality, worst quality, lowres, blurry, noisy, watermark, text, logo, jpeg artifacts, "
+    "bad anatomy, distorted eyes, deformed, multiple faces, nsfw"
+)
+
+with gr.Blocks(css="footer{display:none !important}") as demo:
+    gr.Markdown("# 🏴‍☠️ One Piece – InstantID SDXL")
+
+    with gr.Row():
+        with gr.Column():
+            face_image = gr.Image(type="pil", label="Photo visage", height=360)
+            prompt = gr.Textbox(label="Prompt", value=EX_PROMPT)
+            negative = gr.Textbox(label="Negative Prompt", value=EX_NEG)
+            identity_strength = gr.Slider(0.2, 1.5, 0.95, 0.05, label="Fidélité visage")
+            adapter_strength  = gr.Slider(0.2, 1.5, 0.85, 0.05, label="Détails anime")
+            steps  = gr.Slider(10, 60, 30, 1, label="Steps")
+            cfg    = gr.Slider(0.1, 12.0, 5.5, 0.1, label="CFG")
+            width  = gr.Dropdown(choices=[576, 640, 704, 768, 896], value=704, label="Largeur")
+            height = gr.Dropdown(choices=[704, 768, 896, 1024], value=896, label="Hauteur")
+            seed   = gr.Number(value=-1, label="Seed (-1 aléatoire)")
+            btn    = gr.Button("🎨 Générer", variant="primary")
+
+        with gr.Column():
+            out_image = gr.Image(label="Résultat", interactive=False)
+            err_box   = gr.Textbox(label="Erreurs", visible=False)
+            log_box   = gr.Textbox(label="Logs", value="\n".join(load_logs), lines=10)
+
+    def wrap(*args):
+        img, err, logs = generate(*args)
+        return img, gr.update(visible=bool(err), value=err), gr.update(value=logs)
+
+    btn.click(
+        wrap,
+        inputs=[face_image, prompt, negative, identity_strength, adapter_strength, steps, cfg, width, height, seed],
+        outputs=[out_image, err_box, log_box],
+    )
+
+demo.queue()
+if __name__ == "__main__":
+    demo.launch(server_name="0.0.0.0", server_port=7860, ssr_mode=False)