Update app.py

app.py

@@ -1,222 +1,43 @@
-# app.py — InstantID SDXL (import local résilient + draw_kps local)
-import os, sys, traceback, importlib.util
-
-# 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")
-
-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
-from diffusers.models import ControlNetModel
+import os
 
-DEVICE = "cuda" if torch.cuda.is_available() else "cpu"
-DTYPE  = torch.float16 if DEVICE == "cuda" else torch.float32
-
-# -------- Repo MODEL pour télécharger les poids --------
 ASSETS_REPO = "InstantX/InstantID"
-
-# -------- 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")
 
-# ---------- 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:
+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)
+    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)
+    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)),
+    )
+    # Vérification de taille
+    size = os.path.getsize(path)
+    if size < min_bytes:
         raise RuntimeError(
-            "Pipeline InstantID introuvable.\n"
-            "➡️ Place un des fichiers (texte) dans ./instantid/ :\n"
-            "   - pipeline_stable_diffusion_xl_instantid.py\n"
-            "   - pipeline_stable_diffusion_xl_instantid_full.py\n"
+            f"Téléchargement incomplet de {label} ({size} bytes). "
+            f"Relance le Space; si ça persiste, vérifie la connectivité HF."
         )
-    spec = importlib.util.spec_from_file_location("instantid_pipeline", pipeline_file)
-    mod  = importlib.util.module_from_spec(spec)
-    spec.loader.exec_module(mod)
+    return path
 
-    # 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)
-    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)
-    if not os.path.isfile(IP_ADAPTER_LOCAL):
-        hf_hub_download(ASSETS_REPO, "ip-adapter.bin", local_dir=CHECKPOINTS_DIR)
-
-# ---------- 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 = import_pipeline_or_fail()
-    ensure_assets_or_download()
-
-    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)
-
-    load_logs.append(f"Chargement pipeline InstantID (base={BASE_MODEL})…")
-    pipe = SDXLInstantID.from_pretrained(
-        BASE_MODEL,
-        controlnet=[controlnet_identitynet],
-        torch_dtype=DTYPE,
-        safety_checker=None,
-        feature_extractor=None,
-    ).to(DEVICE)
-
-    if hasattr(pipe, "load_ip_adapter_instantid"):
-        pipe.load_ip_adapter_instantid(IP_ADAPTER_LOCAL)
-    else:
-        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")
-        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 chargement pipeline InstantID.\n" + "\n".join(load_logs))
-
-# ---------- 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 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 une photo portrait nette (visage centré).")
-    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 = kps_image_from_face(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,  # landmarks pour IdentityNet
-            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:
-        msg = "CUDA OOM: baisse résolution (ex: 640×768 → 576×704), steps 24–28, CFG 5–6."
-        return None, f"{msg}\n{oom}", "\n".join(load_logs)
-    except Exception:
-        return None, "Erreur:\n"+traceback.format_exc(), "\n".join(load_logs)
-
-# ---------- 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, "
-    "dynamic lighting, simple background, high detail"
-)
-EX_NEG = (
-    "low quality, worst quality, lowres, blurry, noisy, watermark, text, logo, jpeg artifacts, "
-    "bad anatomy, distorted eyes, cross-eye, asymmetrical eyes, deformed, multiple faces, nsfw"
-)
-
-with gr.Blocks(css="footer{display:none !important}") as demo:
-    gr.Markdown("# 🏴‍☠️ One Piece — InstantID (SDXL) — Import local robuste")
-
-    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, lines=3)
-            negative = gr.Textbox(label="Negative Prompt", value=EX_NEG, lines=3)
-
-            identity_strength = gr.Slider(0.2, 1.5, value=0.95, step=0.05, label="IdentityNet strength (fidélité)")
-            adapter_strength  = gr.Slider(0.2, 1.5, value=0.85, step=0.05, label="Adapter strength (détails)")
-            steps  = gr.Slider(10, 60, value=30, step=1, label="Steps")
-            cfg    = gr.Slider(0.1, 12.0, value=5.5, step=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(ssr_mode=False, server_name="0.0.0.0", server_port=7860)
+    # 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")