InstantID: OpenVINO を使用したゼロショット ID 保持生成#

from pathlib import Path 
import sys 

repo_dir = Path("InstantID") 

if not repo_dir.exists():
     !git clone https://github.com/InstantID/InstantID.git 

sys.path.append(str(repo_dir))

%pip install -q "openvino>=2023.3.0" opencv-python transformers diffusers accelerate gdown "scikit-image>=0.19.2" "gradio>=4.19" "nncf>=2.9.0" "datasets>=2.14.6" "peft==0.6.2"

MODELS_DIR = Path("models") 
face_detector_path = MODELS_DIR / "antelopev2" / "scrfd_10g_bnkps.onnx" 
face_embeddings_path = MODELS_DIR / "antelopev2" / "glintr100.onnx"

from zipfile import ZipFile 
import gdown 

archive_file = Path("antelopev2.zip") 

if not face_detector_path.exists() or face_embeddings_path.exists(): 
    if not archive_file.exists(): 
        gdown.download( 
            "https://drive.google.com/uc?id=18wEUfMNohBJ4K3Ly5wpTejPfDzp-8fI8", 
            str(archive_file), 
        ) 
    with ZipFile(archive_file, "r") as zip_face_models: 
        zip_face_models.extractall(MODELS_DIR)

import cv2 
import numpy as np 
from skimage import transform as trans 

def softmax(z): 
    assert len(z.shape) == 2 
    s = np.max(z, axis=1) 
    s = s[:, np.newaxis] # necessary step to do broadcasting 
    e_x = np.exp(z - s) 
    div = np.sum(e_x, axis=1) 
    div = div[:, np.newaxis] # dito 
    return e_x / div 

def distance2bbox(points, distance, max_shape=None): 
        """
        Decode distance prediction to bounding box.
        Args: 
        points (Tensor): Shape (n, 2), [x, y]. 
        distance (Tensor): Distance from the given point to 4 
            boundaries (left, top, right, bottom). 
        max_shape (tuple): Shape of the image. 
        Returns: 
        Tensor: Decoded bboxes.
        """
    x1 = points[:, 0] - distance[:, 0] 
    y1 = points[:, 1] - distance[:, 1] 
    x2 = points[:, 0] + distance[:, 2] 
    y2 = points[:, 1] + distance[:, 3] 
    if max_shape is not None: 
        x1 = x1.clamp(min=0, max=max_shape[1]) 
        y1 = y1.clamp(min=0, max=max_shape[0]) 
        x2 = x2.clamp(min=0, max=max_shape[1]) 
        y2 = y2.clamp(min=0, max=max_shape[0]) 
    return np.stack([x1, y1, x2, y2], axis=-1) 

def distance2kps(points, distance, max_shape=None): 
    """Decode distance prediction to bounding box.
    Args: 
        points (Tensor): Shape (n, 2), [x, y]. 
        distance (Tensor): Distance from the given point to 4 
            boundaries (left, top, right, bottom). 
        max_shape (tuple): Shape of the image. 
    Returns: 
        Tensor: Decoded bboxes.
    """
    preds = [] 
    for i in range(0, distance.shape[1], 2): 
        px = points[:, i % 2] + distance[:, i] 
        py = points[:, i % 2 + 1] + distance[:, i + 1] 
        if max_shape is not None: 
            px = px.clamp(min=0, max=max_shape[1]) 
            py = py.clamp(min=0, max=max_shape[0]) 
        preds.append(px) 
        preds.append(py) 
    return np.stack(preds, axis=-1) 

def prepare_input(image, std, mean, reverse_channels=True): 
    normalized_image = (image.astype(np.float32) - mean) / std 
    if reverse_channels: 
        normalized_image = normalized_image[:, :, ::-1] 
    input_tensor = np.expand_dims(np.transpose(normalized_image, (2, 0, 1)), 0) 
    return input_tensor 

class RetinaFace: 
    def __init__(self, ov_model): 
        self.taskname = "detection" 
        self.ov_model = ov_model 
        self.center_cache = {} 
        self.nms_thresh = 0.4 
        self.det_thresh = 0.5 
        self._init_vars() 

    def _init_vars(self): 
        self.input_size = (640, 640) 
        outputs = self.ov_model.outputs 
        self.input_mean = 127.5 
        self.input_std = 128.0 
        # print(self.output_names) 
        # assert len(outputs)==10 or len(outputs)==15 
        self.use_kps = False 
        self._anchor_ratio = 1.0 
        self._num_anchors = 1 
        if len(outputs) == 6: 
            self.fmc = 3 
            self._feat_stride_fpn = [8, 16, 32] 
            self._num_anchors = 2 
        elif len(outputs) == 9: 
            self.fmc = 3 
            self._feat_stride_fpn = [8, 16, 32] 
            self._num_anchors = 2 
            self.use_kps = True 
        elif len(outputs) == 10: 
            self.fmc = 5 
            self._feat_stride_fpn = [8, 16, 32, 64, 128] 
            self._num_anchors = 1 
        elif len(outputs) == 15: 
            self.fmc = 5 
            self._feat_stride_fpn = [8, 16, 32, 64, 128] 
            self._num_anchors = 1 
            self.use_kps = True 

def prepare(self, **kwargs): 
    nms_thresh = kwargs.get("nms_thresh", None) 
    if nms_thresh is not None: 
        self.nms_thresh = nms_thresh 
    det_thresh = kwargs.get("det_thresh", None) 
    if det_thresh is not None: 
        self.det_thresh = det_thresh 
    input_size = kwargs.get("input_size", None) 
    if input_size is not None: 
        if self.input_size is not None: 
            print("warning: det_size is already set in detection model, ignore") 
        else: 
            self.input_size = input_size 

def forward(self, img, threshold): 
    scores_list = [] 
    bboxes_list = [] 
    kpss_list = [] 
    blob = prepare_input(img, self.input_mean, self.input_std, True) 
    net_outs = self.ov_model(blob) 

    input_height = blob.shape[2] 
    input_width = blob.shape[3] 
    fmc = self.fmc 
    for idx, stride in enumerate(self._feat_stride_fpn): 
        scores = net_outs[idx] 
        bbox_preds = net_outs[idx + fmc] 
        bbox_preds = bbox_preds * stride 
        if self.use_kps: 
            kps_preds = net_outs[idx + fmc * 2] * stride 
        height = input_height // stride 
        width = input_width // stride 
        key = (height, width, stride) 
        if key in self.center_cache: 
            anchor_centers = self.center_cache[key] 
        else: 
            anchor_centers = np.stack(np.mgrid[:height, :width][::-1], axis=-1).astype(np.float32) 
            anchor_centers = (anchor_centers * stride).reshape((-1, 2)) 
            if self._num_anchors > 1: 
                anchor_centers = np.stack([anchor_centers] * self._num_anchors, axis=1).reshape((-1, 2)) 
            if len(self.center_cache) < 100: 
                self.center_cache[key] = anchor_centers 

        pos_inds = np.where(scores >= threshold)[0] 
        bboxes = distance2bbox(anchor_centers, bbox_preds) 
        pos_scores = scores[pos_inds] 
        pos_bboxes = bboxes[pos_inds] 
        scores_list.append(pos_scores) 
        bboxes_list.append(pos_bboxes) 
        if self.use_kps: 
            kpss = distance2kps(anchor_centers, kps_preds) 
            # kpss = kps_preds 
            kpss = kpss.reshape((kpss.shape[0], -1, 2)) 
            pos_kpss = kpss[pos_inds] 
            kpss_list.append(pos_kpss) 
    return scores_list, bboxes_list, kpss_list 

    def detect(self, img,     input_size=None, max_num=0, metric="default"): 
        assert input_size is not None or self.input_size is not None 
        input_size = self.input_size 
        if input_size is None else input_size 

            im_ratio = float(img.shape[0]) / img.shape[1] 
            model_ratio = float(input_size[1]) / input_size[0] 
        if im_ratio > model_ratio: 
            new_height = input_size[1] 
            new_width = int(new_height / im_ratio) 
        else: 
            new_width = input_size[0] 
            new_height = int(new_width * im_ratio) 
        det_scale = float(new_height) / img.shape[0] 
        resized_img = cv2.resize(img, (new_width, new_height)) 
        det_img = np.zeros((input_size[1], input_size[0], 3), dtype=np.uint8) 
        det_img[:new_height, :new_width, :]= resized_img 

        scores_list, bboxes_list, kpss_list = self.forward(det_img, self.det_thresh) 

        scores = np.vstack(scores_list) 
        scores_ravel = scores.ravel() 
        order = scores_ravel.argsort()[::-1] 
        bboxes = np.vstack(bboxes_list) / det_scale 
        if self.use_kps: 
            kpss = np.vstack(kpss_list) / det_scale 
        pre_det = np.hstack((bboxes, scores)).astype(np.float32, copy=False) 
        pre_det = pre_det[order,     :] 
        keep = self.nms(pre_det) 
        det = pre_det[keep, :] 
        if self.use_kps: 
            kpss = kpss[order, :, :] 
            kpss = kpss[keep, :, :] 
        else: 
            kpss = None 
        if max_num > 0 and det.shape[0] > max_num: 
            area = (det[:, 2] - det[:, 0]) * (det[:, 3] - det[:, 1]) 
            img_center = img.shape[0] // 2, img.shape[1] // 2 
            offsets = np.vstack( 
                [ 
                    (det[:, 0] + det[:, 2]) / 2 - img_center[1], 
                    (det[:, 1] + det[:, 3]) / 2 - img_center[0], 
                ] 
            ) 
            offset_dist_squared = np.sum(np.power(offsets, 2.0), 0) 
            if metric == "max": 
                values = area 
            else: 
                values = area - offset_dist_squared * 2.0 # some extra weight on the centering 
            bindex = np.argsort(values)[::-1] # some extra weight on the centering 
            bindex = bindex[0:max_num] 
            det = det[bindex, :] 
            if kpss is not None: 
                kpss = kpss[bindex, :] 
        return det, kpss 

    def nms(self, dets): 
        thresh = self.nms_thresh 
        x1 = dets[:, 0] 
        y1 = dets[:, 1] 
        x2 = dets[:, 2] 
        y2 = dets[:, 3] 
        scores = dets[:, 4] 

        areas = (x2 - x1 + 1) * (y2 - y1 + 1) 
        order = scores.argsort()[::-1] 

        keep = [] 
        while order.size > 0: 
            i = order[0] 
            keep.append(i) 
            xx1 = np.maximum(x1[i], x1[order[1:]]) 
            yy1 = np.maximum(y1[i], y1[order[1:]]) 
            xx2 = np.minimum(x2[i], x2[order[1:]]) 
            yy2 = np.minimum(y2[i], y2[order[1:]]) 

            w = np.maximum(0.0, xx2 - xx1 + 1) 
            h = np.maximum(0.0, yy2 - yy1 + 1) 
            inter = w * h 
            ovr = inter / (areas[i] + areas[order[1:]]- inter) 

            inds = np.where(ovr <= thresh)[0] 
            order = order[inds + 1] 

        return keep 
arcface_dst = np.array( 
    [ 
        [38.2946, 51.6963], 
        [73.5318, 51.5014], 
        [56.0252, 71.7366], 
        [41.5493, 92.3655], 
        [70.7299, 92.2041], 
    ], 

dtype=np.float32, ) def estimate_norm(lmk, image_size=112, mode="arcface"): 
    assert lmk.shape == (5, 2) 
    assert image_size % 112 == 0 or image_size % 128 == 0 
    if image_size % 112 == 0: 
        ratio = float(image_size) / 112.0 
        diff_x = 0 
    else: 
        ratio = float(image_size) / 128.0 
        diff_x = 8.0 * ratio 
    dst = arcface_dst * ratio 
    dst[:, 0] += diff_x 
    tform = trans.SimilarityTransform() 
    tform.estimate(lmk, dst) 
    M = tform.params[0:2, :] 
    return M 

def norm_crop(img, landmark, image_size=112, mode="arcface"):     M = estimate_norm(landmark, image_size, mode) 
    warped = cv2.warpAffine(img, M, (image_size, image_size), borderValue=0.0) 
    return warped 

class FaceEmbeddings: 
    def __init__(self, ov_model): 
        self.ov_model = ov_model 
        self.taskname = "recognition」 
        input_mean = 127.5 
        input_std = 127.5 
        self.input_mean = input_mean 
        self.input_std = input_std 
        input_shape = self.ov_model.inputs[0].partial_shape 
        self.input_size = (input_shape[3].get_length(), input_shape[2].get_length()) 
        self.input_shape = input_shape 

    def get(self, img, kps): 
        aimg = norm_crop(img, landmark=kps, image_size=self.input_size[0]) 
        embedding = self.get_feat(aimg).flatten() 
        return embedding 

    def get_feat(self, imgs): 
        if not isinstance(imgs, list): 
            imgs = [imgs] 
        input_size = self.input_size 
        blob = np.concatenate([prepare_input(cv2.resize(img, input_size), self.input_mean, self.input_std, True) for img in imgs]) 
        net_out = self.ov_model(blob)[0] 
        return net_out 

    def forward(self, batch_data): 
        blob = (batch_data - self.input_mean) / self.input_std 
        net_out = self.ov_model(blob)[0] 
        return net_out 

class OVFaceAnalysis: 
    def __init__(self, detect_model, embedding_model): 
        self.det_model = RetinaFace(detect_model) 
        self.embed_model = FaceEmbeddings(embedding_model) 

    def get(self, img, max_num=0): 
        bboxes, kpss = self.det_model.detect(img, max_num=max_num, metric="default") 
        if bboxes.shape[0] == 0: 
            return [] 
        ret = [] 
        for i in range(bboxes.shape[0]): 
            bbox = bboxes[i, 0:4] 
            det_score = bboxes[i, 4] 
            kps = None 
            if kpss is not None: 
                kps = kpss[i] 
            embedding = self.embed_model.get(img, kps) 
            ret.append({"bbox": bbox, "score": det_score, "kps": kps, "embedding": embedding}) 
        return ret

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");

from diffusers.models import ControlNetModel 
from diffusers import LCMScheduler 
from pipeline_stable_diffusion_xl_instantid import StableDiffusionXLInstantIDPipeline 

import torch 
from PIL 
import Image import gc 

ov_controlnet_path = MODELS_DIR / "controlnet.xml" 
ov_unet_path = MODELS_DIR / "unet.xml" 
ov_vae_decoder_path = MODELS_DIR / "vae_decoder.xml" 
ov_text_encoder_path = MODELS_DIR / "text_encoder.xml" 
ov_text_encoder_2_path = MODELS_DIR / "text_encoder_2.xml" 
ov_image_proj_encoder_path = MODELS_DIR / "image_proj_model.xml" 

required_pipeline_parts = [ 
    ov_controlnet_path, 
    ov_unet_path, 
    ov_vae_decoder_path, 
    ov_text_encoder_path, 
    ov_text_encoder_2_path, 
    ov_image_proj_encoder_path, 
] 

def load_pytorch_pipeline(sdxl_id="stabilityai/stable-diffusion-xl-base-1.0"):     # prepare models under ./checkpoints 
    face_adapter = Path("checkpoints/ip-adapter.bin") 
    controlnet_path = Path("checkpoints/ControlNetModel") 

    # IdentityNet をロード 
    controlnet = ControlNetModel.from_pretrained(controlnet_path) 

    pipe = StableDiffusionXLInstantIDPipeline.from_pretrained(sdxl_id, controlnet=controlnet) 

    # アダプターをロード 
    pipe.load_ip_adapter_instantid(face_adapter) 
    # load lcm lora 
    pipe.load_lora_weights("latent-consistency/lcm-lora-sdxl") 
    pipe.fuse_lora() 
    scheduler = LCMScheduler.from_config(pipe.scheduler.config) 
    pipe.set_ip_adapter_scale(0.8) 

    controlnet, unet, vae = pipe.controlnet, pipe.unet, pipe.vae 
    text_encoder, text_encoder_2, tokenizer, tokenizer_2 = ( 
        pipe.text_encoder, 
        pipe.text_encoder_2, 
        pipe.tokenizer, 
        pipe.tokenizer_2, 
    ) 
    image_proj_model = pipe.image_proj_model 
    return ( 
        controlnet, 
        unet, 
        vae, 
        text_encoder, 
        text_encoder_2, 
        tokenizer, 
        tokenizer_2, 
         image_proj_model, 
        scheduler, 
    ) 

load_torch_models = any([not path.exists() for path in required_pipeline_parts]) 

if load_torch_models:     ( 
        controlnet, 
        unet, 
        vae, 
        text_encoder, 
        text_encoder_2, 
        tokenizer, 
        tokenizer_2, 
        image_proj_model, 
        scheduler, 
    ) = load_pytorch_pipeline() 
    tokenizer.save_pretrained(MODELS_DIR / "tokenizer") 
    tokenizer_2.save_pretrained(MODELS_DIR / "tokenizer_2") 
    scheduler.save_pretrained(MODELS_DIR / "scheduler") 
else:     ( 
        controlnet, 
        unet, 
        vae, 
        text_encoder, 
        text_encoder_2, 
        tokenizer, 
        tokenizer_2, 
        image_proj_model, 
        scheduler, 
    ) = (None, None, None, None, None, None, None, None, None) 

gc.collect();

import numpy as np 
from diffusers import StableDiffusionXLControlNetPipeline 
from diffusers.pipelines.stable_diffusion_xl import StableDiffusionXLPipelineOutput 
from typing import Any, Callable, Dict, List, Optional, Tuple, Union 

import numpy as np 
import torch 

from diffusers.image_processor import PipelineImageInput, VaeImageProcessor 

class OVStableDiffusionXLInstantIDPipeline(StableDiffusionXLControlNetPipeline): 
    def __init__( 
        self, 
        text_encoder, 
        text_encoder_2, 
        image_proj_model, 
        controlnet, 
        unet, 
        vae_decoder, 
        tokenizer, 
        tokenizer_2, 
        scheduler, 
    ): 
        self.text_encoder = text_encoder 
        self.text_encoder_2 = text_encoder_2 
        self.tokenizer = tokenizer 
        self.tokenizer_2 = tokenizer_2 
        self.image_proj_model = image_proj_model 
        self.controlnet = controlnet 
        self.unet = unet 
        self.vae_decoder = vae_decoder 
        self.scheduler = scheduler 
        self.image_proj_model_in_features = 512 
        self.vae_scale_factor = 8 
        self.vae_scaling_factor = 0.13025 
        self.image_processor = VaeImageProcessor(vae_scale_factor=self.vae_scale_factor, do_convert_rgb=True) 
        self.control_image_processor = VaeImageProcessor( 
            vae_scale_factor=self.vae_scale_factor, 
            do_convert_rgb=True, 
            do_normalize=False, 
        ) 
        self._internal_dict = {} 
        self._progress_bar_config = {} 

    def _encode_prompt_image_emb(self, prompt_image_emb, num_images_per_prompt, do_classifier_free_guidance): 
        if isinstance(prompt_image_emb, torch.Tensor): 
            prompt_image_emb = prompt_image_emb.clone().detach() 
        else: 
            prompt_image_emb = torch.tensor(prompt_image_emb)
 
        prompt_image_emb = prompt_image_emb.reshape([1, -1, self.image_proj_model_in_features]) 

        if do_classifier_free_guidance: 
            prompt_image_emb = torch.cat([torch.zeros_like(prompt_image_emb), prompt_image_emb], dim=0) 
        else: 
            prompt_image_emb = torch.cat([prompt_image_emb], dim=0) 
        prompt_image_emb = self.image_proj_model(prompt_image_emb)[0] 

        bs_embed, seq_len, _ = prompt_image_emb.shape 
        prompt_image_emb = np.tile(prompt_image_emb, (1, num_images_per_prompt, 1)) 
        prompt_image_emb = prompt_image_emb.reshape(bs_embed * num_images_per_prompt, seq_len, -1) 

        return prompt_image_emb 

    def __call__( 
        self, 
        prompt: Union[str, List[str]] = None, 
        prompt_2: Optional[Union[str, List[str]]] = None, 
        image: PipelineImageInput = None, 
        height: Optional[int] = None, 
        width: Optional[int] = None, 
        num_inference_steps: int = 50, 
        guidance_scale: float = 5.0, 
        negative_prompt: Optional[Union[str, List[str]]] = None, 
        negative_prompt_2: Optional[Union[str, List[str]]] = None, 
        num_images_per_prompt: Optional[int] = 1, eta: float = 0.0, 
        generator: Optional[Union[torch.Generator, List[torch.Generator]]] = None, 
        latents: Optional[torch.FloatTensor] = None, 
        prompt_embeds: Optional[torch.FloatTensor] = None, 
        negative_prompt_embeds: Optional[torch.FloatTensor] = None, 
        pooled_prompt_embeds: Optional[torch.FloatTensor] = None, 
        negative_pooled_prompt_embeds: Optional[torch.FloatTensor] = None, 
        image_embeds: Optional[torch.FloatTensor] = None, 
        output_type: Optional[str] = "pil", 
        return_dict: bool = True, 
        cross_attention_kwargs: Optional[Dict[str, Any]] = None, 
        controlnet_conditioning_scale: Union[float, List[float]] = 1.0, 
        guess_mode: bool = False, 
        control_guidance_start: Union[float, List[float]] = 0.0, 
        control_guidance_end: Union[float, List[float]] = 1.0, 
        original_size: Tuple[int, int] = None, 
        crops_coords_top_left: Tuple[int, int] = (0, 0), 
        target_size: Tuple[int, int] = None, 
        negative_original_size: Optional[Tuple[int, int]] = None, 
        negative_crops_coords_top_left: Tuple[int, int] = (0, 0), 
        negative_target_size: Optional[Tuple[int, int]] = None, 
        clip_skip: Optional[int] = None, 
        callback_on_step_end: Optional[Callable[[int, int, Dict], None]] = None, 
        callback_on_step_end_tensor_inputs: List[str] = ["latents"], 
        # IP アダプター 
        ip_adapter_scale=None, 
        **kwargs, 
    ): 
        r""" 
        The call function to the pipeline for generation.
        Args: 
            prompt (`str` or `List[str]`, *optional*): 
                The prompt or prompts to guide image generation.If not defined, you need to pass `prompt_embeds`. 
            prompt_2 (`str` or `List[str]`, *optional*): 
                The prompt or prompts to be sent to `tokenizer_2` and `text_encoder_2`.If not defined, `prompt` is 
                used in both text-encoders. 
            image (`torch.FloatTensor`, `PIL.Image.Image`, `np.ndarray`, `List[torch.FloatTensor]`, `List[PIL.Image.Image]`, `List[np.ndarray]`,: 
            `List[List[torch.FloatTensor]]`, `List[List[np.ndarray]]` or `List[List[PIL.Image.Image]]`): 
                The ControlNet input condition to provide guidance to the `unet` for generation.If the type is 
                specified as `torch.FloatTensor`, it is passed to ControlNet as is.`PIL.Image.Image` can also be 
                accepted as an image. The dimensions of the output image defaults to `image`'s dimensions. If height__module.unet.up_blocks.0.upsamplers.0.conv.base_layer/aten::_convolu 
                and/or width are passed, `image` is resized accordingly.If multiple ControlNets are specified in 
                `init`, images must be passed as a list such that each element of the list can be correctly batched for 
                input to a single ControlNet. 
            height (`int`, *optional*, defaults to `self.unet.config.sample_size * self.vae_scale_factor`): 
                The height in pixels of the generated image.Anything below 512 pixels won't work well for 
                [stabilityai/stable-diffusion-xl-base-1.0](https://huggingface.co/stabilityai/stable-diffusion-xl-base-1.0) 
                and checkpoints that are not specifically fine-tuned on low resolutions. 
            width (`int`, *optional*, defaults to `self.unet.config.sample_size * self.vae_scale_factor`): 
                The width in pixels of the generated image.Anything below 512 pixels won't work well for 
                [stabilityai/stable-diffusion-xl-base-1.0](https://huggingface.co/stabilityai/stable-diffusion-xl-base-1.0) 
                and checkpoints that are not specifically fine-tuned on low resolutions. 
            num_inference_steps (`int`, *optional*, defaults to 50): 
                The number of denoising steps.More denoising steps usually lead to a higher quality image at the 
                expense of slower inference. 
            guidance_scale (`float`, *optional*, defaults to 5.0):
                A higher guidance scale value encourages the model to generate images closely linked to the text 
                `prompt` at the expense of lower image quality.Guidance scale is enabled when `guidance_scale > 1`. 
            negative_prompt (`str` or `List[str]`, *optional*): 
                The prompt or prompts to guide what to not include in image generation.If not defined, you need to 
                pass `negative_prompt_embeds` instead.Ignored when not using guidance (`guidance_scale < 1`). 
            negative_prompt_2 (`str` or `List[str]`, *optional*): 
                The prompt or prompts to guide what to not include in image generation.This is sent to `tokenizer_2` 
                and `text_encoder_2`.If not defined, `negative_prompt` is used in both text-encoders. 
            num_images_per_prompt (`int`, *optional*, defaults to 1):
                The number of images to generate per prompt. 
            eta (`float`, *optional*, defaults to 0.0): 
                Corresponds to parameter eta (η) from the [DDIM](https://arxiv.org/abs/2010.02502) paper.Only applies 
                to the [`~schedulers.DDIMScheduler`], and is ignored in other schedulers. 
            generator (`torch.Generator` or `List[torch.Generator]`, *optional*):
                A [`torch.Generator`](https://pytorch.org/docs/stable/generated/torch.Generator.html) to make 
                generation deterministic. 
            latents (`torch.FloatTensor`, *optional*):
                Pre-generated noisy latents sampled from a Gaussian distribution, to be used as inputs for image 
                generation. Can be used to tweak the same generation with different prompts. If not provided, a latents 
                tensor is generated by sampling using the supplied random `generator`. 
            prompt_embeds (`torch.FloatTensor`, *optional*): 
                Pre-generated text embeddings. Can be used to easily tweak text inputs (prompt weighting). If not 
                provided, text embeddings are generated from the `prompt` input argument. 
            negative_prompt_embeds (`torch.FloatTensor`, *optional*): 
                Pre-generated negative text embeddings. Can be used to easily tweak text inputs (prompt weighting).  If 
                not provided, `negative_prompt_embeds` are generated from the `negative_prompt` input argument. 
            pooled_prompt_embeds (`torch.FloatTensor`, *optional*): 
                Pre-generated pooled text embeddings.Can be used to easily tweak text inputs (prompt weighting).If 
                not provided, pooled text embeddings are generated from `prompt` input argument. 
            negative_pooled_prompt_embeds (`torch.FloatTensor`, *optional*): 
                Pre-generated negative pooled text embeddings.Can be used to easily tweak text inputs (prompt 
                weighting).If not provided, pooled `negative_prompt_embeds` are generated from `negative_prompt` input 
                argument. 
            image_embeds (`torch.FloatTensor`, *optional*):
                Pre-generated image embeddings. 
            output_type (`str`, *optional*, defaults to `"pil"`): 
                The output format of the generated image.Choose between `PIL.Image` or `np.array`. 
            return_dict (`bool`, *optional*, defaults to `True`):
                Whether or not to return a [`~pipelines.stable_diffusion.StableDiffusionPipelineOutput`] instead of a 
                plain tuple. 
            controlnet_conditioning_scale (`float` or `List[float]`, *optional*, defaults to 1.0):
                The outputs of the ControlNet are multiplied by `controlnet_conditioning_scale` before they are added 
                to the residual in the original `unet`.
                If multiple ControlNets are specified in `init`, you can set 
                the corresponding scale as a list. 
            control_guidance_start (`float` or `List[float]`, *optional*, defaults to 0.0):
                The percentage of total steps at which the ControlNet starts applying. 
            control_guidance_end (`float` or `List[float]`, *optional*, defaults to 1.0):
                The percentage of total steps at which the ControlNet stops applying. 
            original_size (`Tuple[int]`, *optional*, defaults to (1024, 1024)): 
                If `original_size` is not the same as `target_size` the image will appear to be down- or upsampled. 
                `original_size` defaults to `(height, width)` if not specified.Part of SDXL's micro-conditioning as 
                explained in section 2.2 of 
                [https://huggingface.co/papers/2307.01952](https://huggingface.co/papers/2307.01952). 
            crops_coords_top_left (`Tuple[int]`, *optional*, defaults to (0, 0)):
                `crops_coords_top_left` can be used to generate an image that appears to be "cropped" from the position 
                `crops_coords_top_left` downwards.
                Favorable, well-centered images are usually achieved by setting 
                `crops_coords_top_left` to (0, 0).
                Part of SDXL's micro-conditioning as explained in section 2.2 of 
                [https://huggingface.co/papers/2307.01952](https://huggingface.co/papers/2307.01952). 
            target_size (`Tuple[int]`, *optional*, defaults to (1024, 1024)): 
                For most cases, `target_size` should be set to the desired height and width of the generated image.
                If not specified it will default to `(height, width)`.
                Part of SDXL's micro-conditioning as explained in 
                section 2.2 of [https://huggingface.co/papers/2307.01952](https://huggingface.co/papers/2307.01952). 
            negative_original_size (`Tuple[int]`, *optional*, defaults to (1024, 1024)): 
                To negatively condition the generation process based on a specific image resolution.
                 Part of SDXL's 
                micro-conditioning as explained in section 2.2 of 
                [https://huggingface.co/papers/2307.01952](https://huggingface.co/papers/2307.01952).For more 
                information, refer toencode_pro this issue thread: https://github.com/huggingface/diffusers/issues/4208. 
            negative_crops_coords_top_left (`Tuple[int]`, *optional*, defaults to (0, 0)): 
                To negatively condition the generation process based on a specific crop coordinates.Part of SDXL's 
                micro-conditioning as explained in section 2.2 of 
                [https://huggingface.co/papers/2307.01952](https://huggingface.co/papers/2307.01952). For more 
                information, refer to this issue thread: https://github.com/huggingface/diffusers/issues/4208. 
            negative_target_size (`Tuple[int]`, *optional*, defaults to (1024, 1024)):  
                To negatively condition the generation process based on a target image resolution.It should be as same 
                as the `target_size` for most cases.Part of SDXL's micro-conditioning as explained in section 2.2 of 
                [https://huggingface.co/papers/2307.01952](https://huggingface.co/papers/2307.01952).For more 
                information, refer to this issue thread: https://github.com/huggingface/diffusers/issues/4208. 
            clip_skip (`int`, *optional*): 
                Number of layers to be skipped from CLIP while computing the prompt embeddings.A value of 1 means that 
                the output of the pre-final layer will be used for computing the prompt embeddings.
            Examples: 
        Returns: 
            [`~pipelines.stable_diffusion.StableDiffusionPipelineOutput`] or `tuple`: 
            If `return_dict` is `True`, [`~pipelines.stable_diffusion.StableDiffusionPipelineOutput`] is returned, 
            otherwise a `tuple` is returned containing the output images.
        """ 

        do_classifier_free_guidance = guidance_scale >= 1.0 
        # 制御ガイダンスのフォーマットを揃える 
        if not isinstance(control_guidance_start, list) and isinstance(control_guidance_end, list): 
            control_guidance_start = len(control_guidance_end) * [control_guidance_start] 
        elif not isinstance(control_guidance_end, list) and isinstance(control_guidance_start, list): 
            control_guidance_end = len(control_guidance_start) * [control_guidance_end] 
        elif not isinstance(control_guidance_start, list) and not isinstance(control_guidance_end, list): 
            control_guidance_start, control_guidance_end = ( 
                [control_guidance_start], 
                [control_guidance_end], 
            ) 

        # 2. 呼び出しパラメーターを定義 
        if prompt is not None and isinstance(prompt, str): 
            batch_size = 1 
        elif prompt is not None and isinstance(prompt, list): 
            batch_size = len(prompt) 
        else: 
            batch_size = prompt_embeds.shape[0] 

        ( 
            prompt_embeds, 
            negative_prompt_embeds, 
            pooled_prompt_embeds, 
            negative_pooled_prompt_embeds, 
        ) = self.encode_prompt( 
            prompt, 
            prompt_2, 
            num_images_per_prompt, 
            do_classifier_free_guidance, 
            negative_prompt, 
            negative_prompt_2, 
            prompt_embeds=prompt_embeds, 
            negative_prompt_embeds=negative_prompt_embeds, 
            pooled_prompt_embeds=pooled_prompt_embeds, 
            negative_pooled_prompt_embeds=negative_pooled_prompt_embeds, 
            lora_scale=None, 
            clip_skip=clip_skip, 
        ) 

        # 3.2 画像のエンコードプロンプト 
        prompt_image_emb = self._encode_prompt_image_emb(image_embeds, num_images_per_prompt, do_classifier_free_guidance) 

        # 4. 画像を準備 
        image = self.prepare_image( 
            image=image, 
            width=width, 
            height=height, 
            batch_size=batch_size * num_images_per_prompt, 
            num_images_per_prompt=num_images_per_prompt, 
            do_classifier_free_guidance=do_classifier_free_guidance, 
            guess_mode=guess_mode, 
        ) 
        height, width = image.shape[-2:] 
        # 5.  タイムステップを準備 
        self.scheduler.set_timesteps(num_inference_steps) 
        timesteps = self.scheduler.timesteps 

        # 6. 潜在変数を準備 
        num_channels_latents = 4 
        latents = self.prepare_latents( 
            int(batch_size) * int(num_images_per_prompt), 
            int(num_channels_latents), 
            int(height), 
            int(width), 
            dtype=torch.float32, 
            device=torch.device("cpu"), 
            generator=generator, 
            latents=latents, 
        ) 

        # 7. 追加のステップ kwargs を準備 
        extra_step_kwargs = self.prepare_extra_step_kwargs(generator, eta) 
        # 7.1 どのコントロールネットを保持するかを示すテンソルを作成 
        controlnet_keep = [] 
        for i in range(len(timesteps)): 
            keeps = [1.0 - float(i / len(timesteps) < s or (i + 1) / len(timesteps) > e) for s, e in zip(control_guidance_start, control_guidance_end)] 
            controlnet_keep.append(keeps) 

        # 7.2 追加された時間 ID と埋め込みを準備 
        if isinstance(image, list): 
            original_size = original_size or image[0].shape[-2:] 
        else: 
            original_size = original_size or image.shape[-2:] 
        target_size = target_size or (height, width) 

        add_text_embeds = pooled_prompt_embeds 
        if self.text_encoder_2 is None: 
            text_encoder_projection_dim = pooled_prompt_embeds.shape[-1] 
        else: 
            text_encoder_projection_dim = 1280 

        add_time_ids = self._get_add_time_ids( 
            original_size, 
            crops_coords_top_left, 
            target_size, 
            text_encoder_projection_dim=text_encoder_projection_dim, 
        ) 

        if negative_original_size is not None and negative_target_size is not None: 
            negative_add_time_ids = self._get_add_time_ids( 
                negative_original_size, 
                negative_crops_coords_top_left, 
                negative_target_size, 
                text_encoder_projection_dim=text_encoder_projection_dim, 
            ) 
        else: 
            negative_add_time_ids = add_time_ids 

        if do_classifier_free_guidance: 
            prompt_embeds = np.concatenate([negative_prompt_embeds, prompt_embeds], axis=0) 
            add_text_embeds = np.concatenate([negative_pooled_prompt_embeds, 
            add_text_embeds], axis=0) add_time_ids = np.concatenate([negative_add_time_ids, add_time_ids], axis=0) 

        add_time_ids = np.tile(add_time_ids, (batch_size * num_images_per_prompt, 1)) 
        encoder_hidden_states = np.concatenate([prompt_embeds, prompt_image_emb], axis=1) 

        # 8. ノイズ除去ループ 
        with self.progress_bar(total=num_inference_steps) as progress_bar: 
            for i, t in enumerate(timesteps):
                # 分類器のフリーガイダンスを行う場合は潜在変数を拡張 
                latent_model_input = torch.cat([latents] * 2) if do_classifier_free_guidance else latents 
                latent_model_input = self.scheduler.scale_model_input(latent_model_input, t) 

                # コントロールネット推論 
                control_model_input = latent_model_input 

                cond_scale = controlnet_conditioning_scale 

                controlnet_outputs = self.controlnet( 
                    [ 
                        control_model_input, 
                        t, 
                        prompt_image_emb, 
                        image, 
                        cond_scale, 
                        add_text_embeds, 
                        add_time_ids, 
                    ] 
                ) 

                controlnet_additional_blocks = list(controlnet_outputs.values()) 

                # ノイズ残留を予測 
                noise_pred = self.unet( 
                    [ 
                        latent_model_input, 
                        t, 
                        encoder_hidden_states, *controlnet_additional_blocks, 
                        add_text_embeds, 
                        add_time_ids, 
                    ] )[0] 

                    # ガイダンスを実行 
                    if do_classifier_free_guidance: 
                        noise_pred_uncond, noise_pred_text = noise_pred[0], noise_pred[1] 
                        noise_pred = noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond) 

                    # 前のノイズサンプルを計算 x_t -> x_t-1 
                    latents = self.scheduler.step( 
                        torch.from_numpy(noise_pred), 
                        t, 
                        latents, 
                        **extra_step_kwargs, 
                        return_dict=False, 
                    )[0] 
                    progress_bar.update() 

        if not output_type == "latent": 
            image = self.vae_decoder(latents / self.vae_scaling_factor)[0] 
        else: 
            image = latents 

        if not output_type == "latent": 
            image = self.image_processor.postprocess(torch.from_numpy(image), output_type=output_type) 

        if not return_dict: 
            return (image,) 

        return StableDiffusionXLPipelineOutput(images=image) 

    def encode_prompt( 
        self, 
        prompt: str, 
        prompt_2: Optional[str] = None, 
        num_images_per_prompt: int = 1, 
        do_classifier_free_guidance: bool = True, 
        negative_prompt: Optional[str] = None, 
        negative_prompt_2: Optional[str] = None, 
        prompt_embeds: Optional[torch.FloatTensor] = None, 
        negative_prompt_embeds: Optional[torch.FloatTensor] = None, 
        pooled_prompt_embeds: Optional[torch.FloatTensor] = None, 
        negative_pooled_prompt_embeds: Optional[torch.FloatTensor] = None, 
        lora_scale: Optional[float] = None, 
        clip_skip: Optional[int] = None, 
    ): 
        r""" 
        Encodes the prompt into text encoder hidden states.
        Args: 
            prompt (`str` or `List[str]`, *optional*): 
                prompt to be encoded 
            prompt_2 (`str` or `List[str]`, *optional*): 
                The prompt or prompts to be sent to the `tokenizer_2` and `text_encoder_2`.If not defined, `prompt` is 
                used in both text-encoders 
            num_images_per_prompt (`int`): 
                number of images that should be generated per prompt 
            do_classifier_free_guidance (`bool`): 
                whether to use classifier free guidance or not 
            negative_prompt (`str` or `List[str]`, *optional*): 
                The prompt or prompts not to guide the image generation.If not defined, one has to pass 
                `negative_prompt_embeds` instead.Ignored when not using guidance (i.e., ignored if `guidance_scale` is 
                less than `1`). 
            negative_prompt_2 (`str` or `List[str]`, *optional*):
                The prompt or prompts not to guide the image generation to be sent to `tokenizer_2` and 
                `text_encoder_2`.If not defined, `negative_prompt` is used in both text-encoders 
            prompt_embeds (`torch.FloatTensor`, *optional*): 
                Pre-generated text embeddings.Can be used to easily tweak text inputs, *e.g.* prompt weighting.If not 
                provided, text embeddings will be generated from `prompt` input argument. 
            negative_prompt_embeds (`torch.FloatTensor`, *optional*): 
                Pre-generated negative text embeddings.Can be used to easily tweak text inputs, *e.g.* prompt 
                weighting.If not provided, negative_prompt_embeds will be generated from `negative_prompt` input 
                argument. 
            pooled_prompt_embeds (`torch.FloatTensor`, *optional*): 
                Pre-generated pooled text embeddings.Can be used to easily tweak text inputs, *e.g.* prompt weighting.
                If not provided, pooled text embeddings will be generated from `prompt` input argument. 
            negative_pooled_prompt_embeds (`torch.FloatTensor`, *optional*): 
                Pre-generated negative pooled text embeddings.Can be used to easily tweak text inputs, *e.g.* prompt 
                weighting.If not provided, pooled negative_prompt_embeds will be generated from `negative_prompt` 
                input argument. 
            lora_scale (`float`, *optional*):
                A lora scale that will be applied to all LoRA layers of the text encoder if LoRA layers are loaded. 
            clip_skip (`int`, *optional*): 
                Number of layers to be skipped from CLIP while computing the prompt embeddings.A value of 1 means that 
                the output of the pre-final layer will be used for computing the prompt embeddings.
        """ 
        prompt = [prompt] if isinstance(prompt, str) else prompt 

        if prompt is not None: 
            batch_size = len(prompt) 
        else: 
            batch_size = prompt_embeds.shape[0] 

        # トークナイザーとテキスト・エンコーダーを定義 
        tokenizers = [self.tokenizer, self.tokenizer_2] if self.tokenizer is not None else [self.tokenizer_2] 
        text_encoders = [self.text_encoder, self.text_encoder_2] if self.text_encoder is not None else [self.text_encoder_2] 

        if prompt_embeds is None: 
            prompt_2 = prompt_2 or prompt 
            prompt_2 = [prompt_2] if isinstance(prompt_2, str) else prompt_2 

            # テキスト反転: 必要に応じてマルチベクトル・トークンを処理 
            prompt_embeds_list = [] 
            prompts = [prompt, prompt_2] 
            for prompt, tokenizer, text_encoder in zip(prompts, tokenizers, text_encoders): 
                text_inputs = tokenizer( 
                    prompt, 
                    padding="max_length", 
                    max_length=tokenizer.model_max_length, 
                    truncation=True, 
                    return_tensors="pt", 
                ) 

                text_input_ids = text_inputs.input_ids 

                prompt_embeds = text_encoder(text_input_ids) 

                # 関心を持っているのは、最終的なテキスト・エンコーダーのプールされた出力だけです 
                pooled_prompt_embeds = prompt_embeds[0] 
                hidden_states = list(prompt_embeds.values())[1:] 
                if clip_skip is None: 
                    prompt_embeds = hidden_states[-2] 
                else:
                    # "2" SDXL は常に最後から 2 番目のレイヤーからインデックスを作成するため 
                    prompt_embeds = hidden_states[-(clip_skip + 2)] 

                prompt_embeds_list.append(prompt_embeds) 

            prompt_embeds = np.concatenate(prompt_embeds_list, axis=-1) 

        # 分類器の無条件埋め込みを取得するフリーガイダンス 
        zero_out_negative_prompt = negative_prompt is None 
        if do_classifier_free_guidance and negative_prompt_embeds is None and zero_out_negative_prompt: 
            negative_prompt_embeds = np.zeros_like(prompt_embeds) 
            negative_pooled_prompt_embeds = np.zeros_like(pooled_prompt_embeds) 
        elif do_classifier_free_guidance and negative_prompt_embeds is None: 
            negative_prompt = negative_prompt or "" 
            negative_prompt_2 = negative_prompt_2 or negative_prompt 

            # 文字列をリストに正規化 
            negative_prompt = batch_size * [negative_prompt] if isinstance(negative_prompt, str) else negative_prompt 
            negative_prompt_2 = batch_size * [negative_prompt_2] if isinstance(negative_prompt_2, str) else negative_prompt_2 

            uncond_tokens: List[str] 
            if prompt is not None and type(prompt) is not type(negative_prompt): 
                raise TypeError(f"`negative_prompt` should be the same type to `prompt`, but got {type(negative_prompt)} !=" f" {type(prompt)}.") 
            elif batch_size != len(negative_prompt): 
                raise ValueError( 
                    f"`negative_prompt`: {negative_prompt} has batch size {len(negative_prompt)}, but `prompt`:" 
                    f" {prompt} has batch size {batch_size}.Please make sure that passed `negative_prompt` matches" 
                    " the batch size of `prompt`."                 ) 
            else: 
                uncond_tokens = [negative_prompt, negative_prompt_2] 

            negative_prompt_embeds_list = [] 
            for negative_prompt, tokenizer, text_encoder in zip(uncond_tokens, tokenizers, text_encoders): 
                max_length = prompt_embeds.shape[1] 
                uncond_input = tokenizer( 
                    negative_prompt, 
                    padding="max_length", 
                    max_length=max_length, 
                    truncation=True, 
                    return_tensors="pt", 
                ) 

                negative_prompt_embeds = text_encoder(uncond_input.input_ids) 
                # 関心を持っているのは、最終的なテキスト・エンコーダーのプールされた出力だけ 

                negative_pooled_prompt_embeds = negative_prompt_embeds[0] 
                hidden_states = list(negative_prompt_embeds.values())[1:] 
                negative_prompt_embeds = hidden_states[-2] 
                negative_prompt_embeds_list.append(negative_prompt_embeds) 

            negative_prompt_embeds = np.concatenate(negative_prompt_embeds_list, axis=-1) 

        bs_embed, seq_len, _ = prompt_embeds.shape 
        # mps フレンドリーな方法を使用して、プロンプトごとに各世代のテキスト埋め込みを複製 

        prompt_embeds = np.tile(prompt_embeds, (1, num_images_per_prompt, 1)) 
        prompt_embeds = prompt_embeds.reshape(bs_embed * num_images_per_prompt, seq_len, -1) 

        if do_classifier_free_guidance:
            # mps フレンドリーな方法を使用して、プロンプトごとに各世代の無条件埋め込みを複製 
            seq_len = negative_prompt_embeds.shape[1] 
            negative_prompt_embeds = np.tile(negative_prompt_embeds, (1, num_images_per_prompt, 1)) 
            negative_prompt_embeds = negative_prompt_embeds.reshape(batch_size * num_images_per_prompt, seq_len, -1) 
        pooled_prompt_embeds = np.tile(pooled_prompt_embeds, (1, num_images_per_prompt)).reshape(bs_embed * num_images_per_prompt, -1) 
        if do_classifier_free_guidance: 
            negative_pooled_prompt_embeds = np.tile(negative_pooled_prompt_embeds, (1, num_images_per_prompt)).reshape(bs_embed * num_images_per_prompt, -1) 

        return ( 
            prompt_embeds, 
            negative_prompt_embeds, 
            pooled_prompt_embeds, 
            negative_pooled_prompt_embeds, 
        ) 

    def prepare_image( 
        self, 
        image, 
        width, 
        height, 
        batch_size, 
        num_images_per_prompt, 
        do_classifier_free_guidance=False, 
        guess_mode=False, 
    ): 
        image = self.control_image_processor.preprocess(image, height=height, width=width).to(dtype=torch.float32) 
        image_batch_size = image.shape[0] 

        if image_batch_size == 1: 
            repeat_by = batch_size 
        else:
            # 画像バッチサイズはプロンプト・バッチ・サイズと同じ 
            repeat_by = num_images_per_prompt 

        image = image.repeat_interleave(repeat_by, dim=0) 

        if do_classifier_free_guidance and not guess_mode: 
            image = torch.cat([image] * 2) 

        return image 

    def _get_add_time_ids( 
        self, 
        original_size, 
        crops_coords_top_left, 
        target_size, 
        text_encoder_projection_dim, 
    ): 
        add_time_ids = list(original_size + crops_coords_top_left + target_size) 
        add_time_ids = torch.tensor([add_time_ids]) 
        return add_time_ids

skip_for_device = "GPU" in device.value 
to_quantize = widgets.Checkbox(value=not skip_for_device, description="Quantization", disabled=skip_for_device) 
to_quantize

# `skip_kernel_extension` モジュールを取得 
import requests 

r = requests.get( 

url="https://raw.githubusercontent.com/openvinotoolkit/openvino_notebooks/latest/utils/skip_kernel_extension.py", 
) 
open("skip_kernel_extension.py", "w").write(r.text) 

int8_pipe = None 

%load_ext skip_kernel_extension