test

handler.py

@@ -1,49 +1,317 @@
-from typing import List, Any
+# from typing import List, Any
+# import torch
+# from diffusers import StableCascadePriorPipeline, StableCascadeDecoderPipeline
+
+# # Configurar el dispositivo para ejecutar el modelo
+# device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
+# if device.type != 'cuda':
+#     raise ValueError("Se requiere ejecutar en GPU")
+
+# # Configurar el tipo de dato mixto basado en la capacidad de la GPU
+# dtype = torch.bfloat16 if torch.cuda.get_device_capability(device.index)[0] >= 8 else torch.float16
+
+import cv2
+import numpy as np
+import PIL
+from PIL import Image
+import diffusers
+from diffusers.models import ControlNetModel
+from depth_anything.dpt import DepthAnything
+from depth_anything.util.transform import Resize, NormalizeImage, PrepareForNet
+from diffusers.utils import load_image
+from insightface.app import FaceAnalysis
+
 import torch
-from diffusers import StableCascadePriorPipeline, StableCascadeDecoderPipeline
+from pipeline_stable_diffusion_xl_instantid_full import StableDiffusionXLInstantIDPipeline, draw_kps
+from diffusers.pipelines.controlnet.multicontrolnet import MultiControlNetModel
+
+from controlnet_aux import OpenposeDetector
+from depth_anything.dpt import DepthAnything
+
+import torch.nn.functional as F
+from torchvision.transforms import Compose
+
+from huggingface_hub import hf_hub_download
+
+hf_hub_download(repo_id="InstantX/InstantID", filename="ControlNetModel/config.json", local_dir="./checkpoints")
+hf_hub_download(
+    repo_id="InstantX/InstantID",
+    filename="ControlNetModel/diffusion_pytorch_model.safetensors",
+    local_dir="./checkpoints",
+)
+hf_hub_download(repo_id="InstantX/InstantID", filename="ip-adapter.bin", local_dir="./checkpoints")
 
-# Configurar el dispositivo para ejecutar el modelo
 device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
 if device.type != 'cuda':
     raise ValueError("Se requiere ejecutar en GPU")
 
-# Configurar el tipo de dato mixto basado en la capacidad de la GPU
-dtype = torch.bfloat16 if torch.cuda.get_device_capability(device.index)[0] >= 8 else torch.float16
+dtype = torch.float16 if str(device).__contains__("cuda") else torch.float32
 
 class EndpointHandler():
     def __init__(self):
-        # Inicializar aquí si es necesario
-        pass
+        face_adapter = f"./checkpoints/ip-adapter.bin"
+        controlnet_path = f"./checkpoints/ControlNetModel"
 
-    def __call__(self, data: Any) -> List[Any]:
-        # Configurar el número de imágenes por prompt
-        num_images_per_prompt = 1
+        transform = Compose([
+            Resize(
+                width=518,
+                height=518,
+                resize_target=False,
+                keep_aspect_ratio=True,
+                ensure_multiple_of=14,
+                resize_method='lower_bound',
+                image_interpolation_method=cv2.INTER_CUBIC,
+            ),
+            NormalizeImage(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]),
+            PrepareForNet(),
+        ])
 
-        # Cargar los modelos con el tipo de dato y dispositivo correctos
-        prior = StableCascadePriorPipeline.from_pretrained("stabilityai/stable-cascade-prior", torch_dtype=dtype).to(device)
-        decoder = StableCascadeDecoderPipeline.from_pretrained("stabilityai/stable-cascade", torch_dtype=dtype).to(device)
+        self.controlnet_identitynet = ControlNetModel.from_pretrained(
+            controlnet_path, torch_dtype=dtype
+        )
 
-        prompt = data.get("inputs", "Una imagen interesante")  # Asegúrate de pasar un prompt adecuado
-        negative_prompt = data.get("negative_prompt", "")
+        pretrained_model_name_or_path = "wangqixun/YamerMIX_v8"
 
-        prior_output = prior(
-            prompt=prompt,
-            height=512,
-            width=512,
-            negative_prompt=negative_prompt,
-            guidance_scale=7.5,
-            num_inference_steps=50,
-            num_images_per_prompt=num_images_per_prompt,
+        self.pipe = StableDiffusionXLInstantIDPipeline.from_pretrained(
+        pretrained_model_name_or_path,
+        controlnet=[self.controlnet_identitynet],
+        torch_dtype=dtype,
+        safety_checker=None,
+        feature_extractor=None,
+        ).to(device)
+
+
+        self.pipe.scheduler = diffusers.EulerDiscreteScheduler.from_config(
+            self.pipe.scheduler.config
+        )
+
+        # load and disable LCM
+        self.pipe.load_lora_weights("latent-consistency/lcm-lora-sdxl")
+        self.pipe.disable_lora()
+
+        self.pipe.cuda()
+        self.pipe.load_ip_adapter_instantid(face_adapter)
+        self.pipe.image_proj_model.to("cuda")
+        self.pipe.unet.to("cuda")
+
+
+        # controlnet-pose/canny/depth
+        controlnet_pose_model = "thibaud/controlnet-openpose-sdxl-1.0"
+        controlnet_canny_model = "diffusers/controlnet-canny-sdxl-1.0"
+        controlnet_depth_model = "diffusers/controlnet-depth-sdxl-1.0-small"
+
+        controlnet_pose = ControlNetModel.from_pretrained(
+            controlnet_pose_model, torch_dtype=dtype
+        ).to(device)
+        controlnet_canny = ControlNetModel.from_pretrained(
+            controlnet_canny_model, torch_dtype=dtype
+        ).to(device)
+        controlnet_depth = ControlNetModel.from_pretrained(
+            controlnet_depth_model, torch_dtype=dtype
+        ).to(device)
+
+        def get_canny_image(image, t1=100, t2=200):
+            image = cv2.cvtColor(np.array(image), cv2.COLOR_RGB2BGR)
+            edges = cv2.Canny(image, t1, t2)
+            return Image.fromarray(edges, "L")
+        
+        def get_depth_map(image):
+        
+            image = np.array(image) / 255.0
+
+            h, w = image.shape[:2]
+
+            image = transform({'image': image})['image']
+            image = torch.from_numpy(image).unsqueeze(0).to("cuda")
+
+            with torch.no_grad():
+                depth = depth_anything(image)
+
+            depth = F.interpolate(depth[None], (h, w), mode='bilinear', align_corners=False)[0, 0]
+            depth = (depth - depth.min()) / (depth.max() - depth.min()) * 255.0
+
+            depth = depth.cpu().numpy().astype(np.uint8)
+
+            depth_image = Image.fromarray(depth)
+
+            return depth_image
+        
+        self.controlnet_map = {
+            "pose": controlnet_pose,
+            "canny": get_canny_image,
+            "depth": controlnet_depth,
+        }
+
+        openpose = OpenposeDetector.from_pretrained("lllyasviel/ControlNet")
+        depth_anything = DepthAnything.from_pretrained('LiheYoung/depth_anything_vitl14').to(device).eval()
+
+
+        self.controlnet_map_fn = {
+            "pose": openpose,
+            "canny": get_canny_image,
+            "depth": get_depth_map,
+        }
+
+        self.app = FaceAnalysis(
+            name="antelopev2",
+            root="./",
+            providers=["CPUExecutionProvider"],
         )
+        self.app.prepare(ctx_id=0, det_size=(640, 640))
+        
+    def __call__(self):
+        self.pipe.scheduler = diffusers.LCMScheduler.from_config(self.pipe.scheduler.config)
+        self.pipe.enable_lora()
+        
+        adapter_strength_ratio = 0.8
+        identitynet_strength_ratio = 0.8
+        pose_strength = 0.4
+        canny_strength = 0.3
+        depth_strength = 0.5
+        controlnet_selection = ["pose", "canny", "depth"]
+        
+        face_image_path = "./kaifu_resize.png"
+        pose_image_path = "./pose.jpg"
+        
+        def convert_from_cv2_to_image(img: np.ndarray) -> Image:
+            return Image.fromarray(cv2.cvtColor(img, cv2.COLOR_BGR2RGB))
+
+        def convert_from_image_to_cv2(img: Image) -> np.ndarray:
+            return cv2.cvtColor(np.array(img), cv2.COLOR_RGB2BGR)
+
+        # check if the input is valid
+        # if face_image_path is None:
+        #     raise gr.Error(
+        #         f"Cannot find any input face image! Please upload the face image"
+        #     )
+        #  check the prompt
+        # if prompt is None:
+        prompt = "a person"
+        negative_prompt=""
+        
+        # apply the style template
+        # prompt, negative_prompt = apply_style(style_name, prompt, negative_prompt)
+
+        face_image = load_image(face_image_path)
+        face_image = resize_img(face_image, max_side=1024)
+        face_image_cv2 = convert_from_image_to_cv2(face_image)
+        height, width, _ = face_image_cv2.shape
+
+        # Extract face features
+        face_info = self.app.get(face_image_cv2)
 
-        decoder_output = decoder(
-            image_embeddings=prior_output["image_embeddings"].half(),
+        print("error si no hay face")
+        # if len(face_info) == 0:
+        #     raise gr.Error(
+        #         f"Unable to detect a face in the image. Please upload a different photo with a clear face."
+        #     )
+
+        face_info = sorted(
+            face_info,
+            key=lambda x: (x["bbox"][2] - x["bbox"][0]) * x["bbox"][3] - x["bbox"][1],
+        )[
+            -1
+        ]  # only use the maximum face
+        
+        
+        def resize_img(
+            input_image,
+            max_side=1280,
+            min_side=1024,
+            size=None,
+            pad_to_max_side=False,
+            mode=PIL.Image.BILINEAR,
+            base_pixel_number=64,
+        ):
+            w, h = input_image.size
+            if size is not None:
+                w_resize_new, h_resize_new = size
+            else:
+                ratio = min_side / min(h, w)
+                w, h = round(ratio * w), round(ratio * h)
+                ratio = max_side / max(h, w)
+                input_image = input_image.resize([round(ratio * w), round(ratio * h)], mode)
+                w_resize_new = (round(ratio * w) // base_pixel_number) * base_pixel_number
+                h_resize_new = (round(ratio * h) // base_pixel_number) * base_pixel_number
+            input_image = input_image.resize([w_resize_new, h_resize_new], mode)
+
+            if pad_to_max_side:
+                res = np.ones([max_side, max_side, 3], dtype=np.uint8) * 255
+                offset_x = (max_side - w_resize_new) // 2
+                offset_y = (max_side - h_resize_new) // 2
+                res[
+                    offset_y : offset_y + h_resize_new, offset_x : offset_x + w_resize_new
+                ] = np.array(input_image)
+                input_image = Image.fromarray(res)
+            return input_image        
+        
+        face_emb = face_info["embedding"]
+        face_kps = draw_kps(convert_from_cv2_to_image(face_image_cv2), face_info["kps"])
+        img_controlnet = face_image
+        if pose_image_path is not None:
+            pose_image = load_image(pose_image_path)
+            pose_image = resize_img(pose_image, max_side=1024)
+            img_controlnet = pose_image
+            pose_image_cv2 = convert_from_image_to_cv2(pose_image)
+
+            face_info = self.app.get(pose_image_cv2)
+
+            # get error if no face is detected 
+            # if len(face_info) == 0:
+            #     raise gr.Error(
+            #         f"Cannot find any face in the reference image! Please upload another person image"
+            #     )
+
+            face_info = face_info[-1]
+            face_kps = draw_kps(pose_image, face_info["kps"])
+
+            width, height = face_kps.size
+
+        control_mask = np.zeros([height, width, 3])
+        x1, y1, x2, y2 = face_info["bbox"]
+        x1, y1, x2, y2 = int(x1), int(y1), int(x2), int(y2)
+        control_mask[y1:y2, x1:x2] = 255
+        control_mask = Image.fromarray(control_mask.astype(np.uint8))
+
+        if len(controlnet_selection) > 0:
+            controlnet_scales = {
+                "pose": pose_strength,
+                "canny": canny_strength,
+                "depth": depth_strength,
+            }
+            self.pipe.controlnet = MultiControlNetModel(
+                [self.controlnet_identitynet]
+                + [self.controlnet_map[s] for s in controlnet_selection]
+            )
+            control_scales = [float(identitynet_strength_ratio)] + [
+                controlnet_scales[s] for s in controlnet_selection
+            ]
+            control_images = [face_kps] + [
+                self.controlnet_map_fn[s](img_controlnet).resize((width, height))
+                for s in controlnet_selection
+            ]
+        else:
+            self.pipe.controlnet = self.controlnet_identitynet
+            control_scales = float(identitynet_strength_ratio)
+            control_images = face_kps
+
+        generator = torch.Generator(device=device.type).manual_seed(3)
+
+        print("Start inference...")
+
+        self.pipe.set_ip_adapter_scale(adapter_strength_ratio)
+        images = self.pipe(
             prompt=prompt,
             negative_prompt=negative_prompt,
+            image_embeds=face_emb,
+            image=control_images,
+            control_mask=control_mask,
+            controlnet_conditioning_scale=control_scales,
+            num_inference_steps=30,
             guidance_scale=7.5,
-            output_type="pil",
-            num_inference_steps=20
-        )
+            height=height,
+            width=width,
+            generator=generator,
+        ).images
 
-        # Asumiendo que quieres retornar la primera imagen
-        return [decoder_output.images[0]]
+        return images[0]

requirements.txt

@@ -1,7 +1,18 @@
-safetensors 
-opencv-python
-controlnet_hinter==0.0.5
-git+https://github.com/kashif/diffusers.git@diffusers-yield-callback
-https://gradio-builds.s3.amazonaws.com/aabb08191a7d94d2a1e9ff87b0d3c3987cd519c5/gradio-4.18.0-py3-none-any.whl
+diffusers==0.25.1
+torch==2.0.0
+torchvision==0.15.1
+transformers==4.37.1
 accelerate
-transformers
+safetensors
+einops
+onnxruntime-gpu
+spaces==0.19.4
+omegaconf
+peft
+huggingface-hub==0.20.2
+opencv-python
+insightface
+gradio
+controlnet_aux
+gdown
+peft