





'''simple docstring'''


import os
from glob import glob

import imageio
import torch
import torchvision
import wandb
from img_processing import custom_to_pil, loop_post_process, preprocess, preprocess_vqgan
from loaders import load_vqgan
from PIL import Image
from torch import nn

from transformers import CLIPModel, CLIPTokenizerFast
from utils import get_device, get_timestamp, show_pil
class 				a_       :





     def __init__(			self		: Dict     ,				lowercase		: str = "cpu"     ,				lowercase		: str = "openai/clip-vit-large-patch14"						):





        """simple docstring"""




        lowercase_      :Dict         =			device
        lowercase_      :List[Any]         =			CLIPTokenizerFast.from_pretrained(lowercase						)
        lowercase_      :Union[str, Any]         =			[0.48_14_54_66, 0.4_57_82_75, 0.40_82_10_73]
        lowercase_      :List[Any]         =			[0.26_86_29_54, 0.26_13_02_58, 0.27_57_77_11]
        lowercase_      :Tuple         =			torchvision.transforms.Normalize(self.image_mean     ,				self.image_std						)
        lowercase_      :Dict         =			torchvision.transforms.Resize(224						)
        lowercase_      :Any         =			torchvision.transforms.CenterCrop(224						)





     def 							lowercase__				(			self		: Union[str, Any]     ,				lowercase		: Optional[int]						):





        """simple docstring"""




        lowercase_      :Any         =			self.resize(lowercase						)
        lowercase_      :List[str]         =			self.center_crop(lowercase						)
        lowercase_      :List[str]         =			self.normalize(lowercase						)
        return images







     def __call__(			self		: Union[str, Any]     ,				lowercase		: List[str]=None     ,				lowercase		: Optional[Any]=None     ,				**lowercase		: int						):





        """simple docstring"""




        lowercase_      :str         =			self.tokenizer(text=lowercase     ,				**lowercase						)
        lowercase_      :Optional[Any]         =			self.preprocess_img(lowercase						)
        lowercase_      :List[str]         =			{key: value.to(self.device						) for (key, value) in encoding.items()}
        return encoding







class 				a_       (    nn.Module							):





     def __init__(			self		: List[Any]     ,				lowercase		: List[str]=10     ,				lowercase		: str=0.01     ,				lowercase		: Union[str, Any]=None     ,				lowercase		: int=None     ,				lowercase		: Optional[int]=None     ,				lowercase		: Union[str, Any]=None     ,				lowercase		: Any=None     ,				lowercase		: Optional[int]=None     ,				lowercase		: Optional[Any]=False     ,				lowercase		: int=True     ,				lowercase		: Union[str, Any]="image"     ,				lowercase		: List[Any]=True     ,				lowercase		: Union[str, Any]=False     ,				lowercase		: int=False     ,				lowercase		: Tuple=False     ,				):





        """simple docstring"""




        super().__init__()
        lowercase_      :Tuple         =			None
        lowercase_      :Union[str, Any]         =			device if device else get_device()
        if vqgan:
           lowercase_      :int         =			vqgan
        else:
           lowercase_      :int         =			load_vqgan(self.device     ,				conf_path=lowercase     ,				ckpt_path=lowercase						)
        self.vqgan.eval()
        if clip:
           lowercase_      :Union[str, Any]         =			clip
        else:
           lowercase_      :Optional[Any]         =			CLIPModel.from_pretrained("openai/clip-vit-base-patch32"						)
        self.clip.to(self.device						)
        lowercase_      :List[str]         =			ProcessorGradientFlow(device=self.device						)

        lowercase_      :int         =			iterations
        lowercase_      :Union[str, Any]         =			lr
        lowercase_      :Optional[int]         =			log
        lowercase_      :Any         =			make_grid
        lowercase_      :Dict         =			return_val
        lowercase_      :Optional[Any]         =			quantize
        lowercase_      :Tuple         =			self.vqgan.decoder.z_shape





     def 							lowercase__				(			self		: str     ,				lowercase		: Optional[Any]=None     ,				lowercase		: List[Any]=None     ,				lowercase		: List[Any]=5     ,				lowercase		: Any=True						):





        """simple docstring"""




        lowercase_      :Tuple         =			[]
        if output_path is None:
           lowercase_      :Union[str, Any]         =			"./animation.gif"
        if input_path is None:
           lowercase_      :Optional[Any]         =			self.save_path
        lowercase_      :str         =			sorted(glob(input_path + "/*"						)						)
        if not len(lowercase						):
           raise ValueError(
               "No images found in save path, aborting (did you pass save_intermediate=True to the generate"
               " function?)"						)
        if len(lowercase						) == 1:
           print("Only one image found in save path, (did you pass save_intermediate=True to the generate function?)"						)
        lowercase_      :Optional[int]         =			total_duration / len(lowercase						)
        lowercase_      :Union[str, Any]         =			[frame_duration] * len(lowercase						)
        if extend_frames:
           lowercase_      :List[Any]         =			1.5
           lowercase_      :List[str]         =			3
        for file_name in paths:
           if file_name.endswith(".png"						):
              images.append(imageio.imread(lowercase						)						)
        imageio.mimsave(lowercase     ,				lowercase     ,				duration=lowercase						)
        print(F'gif saved to {output_path}'						)





     def 							lowercase__				(			self		: int     ,				lowercase		: Tuple=None     ,				lowercase		: Tuple=None						):





        """simple docstring"""




        if not (path or img):
           raise ValueError("Input either path or tensor"						)
        if img is not None:
           raise NotImplementedError
        lowercase_      :Tuple         =			preprocess(Image.open(lowercase						)     ,				target_image_size=256						).to(self.device						)
        lowercase_      :Union[str, Any]         =			preprocess_vqgan(lowercase						)
        lowercase_      ,							*lowercase_      :List[Any]         =			self.vqgan.encode(lowercase						)
        return z





     def 							lowercase__				(			self		: List[Any]     ,				lowercase		: Any						):





        """simple docstring"""




        lowercase_      :Tuple         =			self.latent.detach().requires_grad_()
        lowercase_      :List[str]         =			base_latent + transform_vector
        if self.quantize:
           lowercase_      ,							*lowercase_      :List[str]         =			self.vqgan.quantize(lowercase						)
        else:
           lowercase_      :str         =			trans_latent
        return self.vqgan.decode(lowercase						)





     def 							lowercase__				(			self		: List[Any]     ,				lowercase		: Union[str, Any]     ,				lowercase		: List[str]     ,				lowercase		: List[Any]=None						):





        """simple docstring"""




        lowercase_      :Union[str, Any]         =			self.clip_preprocessor(text=lowercase     ,				images=lowercase     ,				return_tensors="pt"     ,				padding=lowercase						)
        lowercase_      :Tuple         =			self.clip(**lowercase						)
        lowercase_      :List[Any]         =			clip_outputs.logits_per_image
        if weights is not None:
           lowercase_      :Any         =			similarity_logits * weights
        return similarity_logits.sum()





     def 							lowercase__				(			self		: List[Any]     ,				lowercase		: str     ,				lowercase		: List[str]     ,				lowercase		: Union[str, Any]						):





        """simple docstring"""




        lowercase_      :List[Any]         =			self._get_clip_similarity(pos_prompts["prompts"]     ,				lowercase     ,				weights=(1 / pos_prompts["weights"])						)
        if neg_prompts:
           lowercase_      :Any         =			self._get_clip_similarity(neg_prompts["prompts"]     ,				lowercase     ,				weights=neg_prompts["weights"]						)
        else:
           lowercase_      :Any         =			torch.tensor([1]     ,				device=self.device						)
        lowercase_      :Optional[Any]         =			-torch.log(lowercase						) + torch.log(lowercase						)
        return loss





     def 							lowercase__				(			self		: int     ,				lowercase		: Optional[Any]     ,				lowercase		: Any     ,				lowercase		: Tuple						):





        """simple docstring"""




        lowercase_      :Union[str, Any]         =			torch.randn_like(self.latent     ,				requires_grad=lowercase     ,				device=self.device						)
        lowercase_      :Optional[Any]         =			torch.optim.Adam([vector]     ,				lr=self.lr						)

        for i in range(self.iterations						):
           optim.zero_grad()
           lowercase_      :Any         =			self._add_vector(lowercase						)
           lowercase_      :str         =			loop_post_process(lowercase						)
           lowercase_      :List[Any]         =			self._get_CLIP_loss(lowercase     ,				lowercase     ,				lowercase						)
           print("CLIP loss"     ,				lowercase						)
           if self.log:
              wandb.log({"CLIP Loss": clip_loss}						)
           clip_loss.backward(retain_graph=lowercase						)
           optim.step()
           if self.return_val == "image":
              yield custom_to_pil(transformed_img[0]						)
           else:
              yield vector





     def 							lowercase__				(			self		: Any     ,				lowercase		: Optional[Any]     ,				lowercase		: List[str]     ,				lowercase		: str						):





        """simple docstring"""




        wandb.init(reinit=lowercase     ,				project="face-editor"						)
        wandb.config.update({"Positive Prompts": positive_prompts}						)
        wandb.config.update({"Negative Prompts": negative_prompts}						)
        wandb.config.update({"lr": self.lr, "iterations": self.iterations}						)
        if image_path:
           lowercase_      :List[str]         =			Image.open(lowercase						)
           lowercase_      :Optional[int]         =			image.resize((256, 256)						)
           wandb.log("Original Image"     ,				wandb.Image(lowercase						)						)





     def 							lowercase__				(			self		: Optional[Any]     ,				lowercase		: Tuple						):





        """simple docstring"""




        if not prompts:
           return []
        lowercase_      :Any         =			[]
        lowercase_      :Dict         =			[]
        if isinstance(lowercase     ,				lowercase						):
           lowercase_      :Dict         =			[prompt.strip() for prompt in prompts.split("|"						)]
        for prompt in prompts:
           if isinstance(lowercase     ,				(tuple, list)						):
              lowercase_      :List[Any]         =			prompt[0]
              lowercase_      :Dict         =			float(prompt[1]						)
           elif ":" in prompt:
              lowercase_      ,							lowercase_      :int         =			prompt.split(":"						)
              lowercase_      :Tuple         =			float(lowercase						)
           else:
              lowercase_      :Dict         =			prompt
              lowercase_      :Optional[int]         =			1.0
           processed_prompts.append(lowercase						)
           weights.append(lowercase						)
        return {
            "prompts": processed_prompts,
            "weights": torch.tensor(lowercase     ,				device=self.device						),
        }







     def 							lowercase__				(			self		: Union[str, Any]     ,				lowercase		: Any     ,				lowercase		: List[str]=None     ,				lowercase		: Any=None     ,				lowercase		: Union[str, Any]=True     ,				lowercase		: Any=False     ,				lowercase		: Optional[int]=True     ,				lowercase		: List[Any]=True     ,				lowercase		: Dict=None     ,				):





        """simple docstring"""




        if image_path:
           lowercase_      :Dict         =			self._get_latent(lowercase						)
        else:
           lowercase_      :List[Any]         =			torch.randn(self.latent_dim     ,				device=self.device						)
        if self.log:
           self._init_logging(lowercase     ,				lowercase     ,				lowercase						)

        assert pos_prompts, "You must provide at least one positive prompt."
        lowercase_      :Optional[int]         =			self.process_prompts(lowercase						)
        lowercase_      :int         =			self.process_prompts(lowercase						)
        if save_final and save_path is None:
           lowercase_      :List[str]         =			os.path.join("./outputs/"     ,				"_".join(pos_prompts["prompts"]						)						)
        if not os.path.exists(lowercase						):
           os.makedirs(lowercase						)
        else:
           lowercase_      :Dict         =			save_path + "_" + get_timestamp()
           os.makedirs(lowercase						)
        lowercase_      :Tuple         =			save_path

        lowercase_      :Dict         =			self.vqgan.decode(self.latent						)[0]
        if show_intermediate:
           print("Original Image"						)
           show_pil(custom_to_pil(lowercase						)						)

        lowercase_      :List[Any]         =			loop_post_process(lowercase						)
        for iter, transformed_img in enumerate(self._optimize_CLIP(lowercase     ,				lowercase     ,				lowercase						)						):
           if show_intermediate:
              show_pil(lowercase						)
           if save_intermediate:
              transformed_img.save(os.path.join(self.save_path     ,				F'iter_{iter:03d}.png'						)						)
           if self.log:
              wandb.log({"Image": wandb.Image(lowercase						)}						)
        if show_final:
           show_pil(lowercase						)
        if save_final:
           transformed_img.save(os.path.join(self.save_path     ,				F'iter_{iter:03d}_final.png'						)						)









'''simple docstring'''


from typing import Any, Dict, List, Union

from ..utils import add_end_docstrings, is_torch_available, is_vision_available, logging, requires_backends
from .base import PIPELINE_INIT_ARGS, ChunkPipeline


if is_vision_available():
 from PIL import Image

 from ..image_utils import load_image

if is_torch_available():
 import torch

 from transformers.modeling_outputs import BaseModelOutput

 from ..models.auto.modeling_auto import MODEL_FOR_ZERO_SHOT_OBJECT_DETECTION_MAPPING

lowerCAmelCase :					Union[str, Any] 							=logging.get_logger(__name__)
@add_end_docstrings(_lowerCAmelCase							)
class 				a_       (    _lowerCAmelCase							):





     def __init__(			self		: List[Any]     ,				**lowercase		: Optional[int]						):





        """simple docstring"""




        super().__init__(**lowercase						)

        if self.framework == "tf":
           raise ValueError(F'The {self.__class__} is only available in PyTorch.'						)

        requires_backends(self     ,				"vision"						)
        self.check_model_type(lowercase						)





     def __call__(			self		: Tuple     ,				lowercase		: Union[str, "Image.Image", List[Dict[str, Any]]]     ,				lowercase		: Union[str, List[str]] = None     ,				**lowercase		: str     ,				):





        """simple docstring"""




        if "text_queries" in kwargs:
           lowercase_      :List[Any]         =			kwargs.pop("text_queries"						)

        if isinstance(lowercase     ,				(str, Image.Image)						):
           lowercase_      :List[str]         =			{"image": image, "candidate_labels": candidate_labels}
        else:
           lowercase_      :Optional[Any]         =			image
        lowercase_      :str         =			super().__call__(lowercase     ,				**lowercase						)
        return results





     def 							lowercase__				(			self		: Optional[int]     ,				**lowercase		: List[str]						):





        """simple docstring"""




        lowercase_      :Tuple         =			{}
        if "threshold" in kwargs:
           lowercase_      :Dict         =			kwargs["threshold"]
        if "top_k" in kwargs:
           lowercase_      :Optional[Any]         =			kwargs["top_k"]
        return {}, {}, postprocess_params





     def 							lowercase__				(			self		: List[Any]     ,				lowercase		: Dict						):





        """simple docstring"""




        lowercase_      :Any         =			load_image(inputs["image"]						)
        lowercase_      :List[str]         =			inputs["candidate_labels"]
        if isinstance(lowercase     ,				lowercase						):
           lowercase_      :Union[str, Any]         =			candidate_labels.split(","						)

        lowercase_      :Tuple         =			torch.tensor([[image.height, image.width]]     ,				dtype=torch.intaa						)
        for i, candidate_label in enumerate(lowercase						):
           lowercase_      :Union[str, Any]         =			self.tokenizer(lowercase     ,				return_tensors=self.framework						)
           lowercase_      :Tuple         =			self.image_processor(lowercase     ,				return_tensors=self.framework						)
           yield {
               "is_last": i == len(lowercase						) - 1,
               "target_size": target_size,
               "candidate_label": candidate_label,
               **text_inputs,
               **image_features,
           }





     def 							lowercase__				(			self		: List[str]     ,				lowercase		: List[str]						):





        """simple docstring"""




        lowercase_      :Dict         =			model_inputs.pop("target_size"						)
        lowercase_      :str         =			model_inputs.pop("candidate_label"						)
        lowercase_      :List[Any]         =			model_inputs.pop("is_last"						)

        lowercase_      :Optional[Any]         =			self.model(**lowercase						)

        lowercase_      :str         =			{"target_size": target_size, "candidate_label": candidate_label, "is_last": is_last, **outputs}
        return model_outputs





     def 							lowercase__				(			self		: Optional[int]     ,				lowercase		: List[str]     ,				lowercase		: List[str]=0.1     ,				lowercase		: Optional[int]=None						):





        """simple docstring"""




        lowercase_      :Dict         =			[]
        for model_output in model_outputs:
           lowercase_      :int         =			model_output["candidate_label"]
           lowercase_      :str         =			BaseModelOutput(lowercase						)
           lowercase_      :List[Any]         =			self.image_processor.post_process_object_detection(
               outputs=lowercase     ,				threshold=lowercase     ,				target_sizes=model_output["target_size"]						)[0]

           for index in outputs["scores"].nonzero():
              lowercase_      :Optional[int]         =			outputs["scores"][index].item()
              lowercase_      :int         =			self._get_bounding_box(outputs["boxes"][index][0]						)

              lowercase_      :int         =			{"score": score, "label": label, "box": box}
              results.append(lowercase						)

        lowercase_      :Dict         =			sorted(lowercase     ,				key=lambda lowercase						: x["score"]     ,				reverse=lowercase						)
        if top_k:
           lowercase_      :List[str]         =			results[:top_k]

        return results







     def 							lowercase__				(			self		: Union[str, Any]     ,				lowercase		: "torch.Tensor"						):





        """simple docstring"""




        if self.framework != "pt":
           raise ValueError("The ZeroShotObjectDetectionPipeline is only available in PyTorch."						)
        lowercase_      ,							lowercase_      ,							lowercase_      ,							lowercase_      :List[str]         =			box.int().tolist()
        lowercase_      :List[Any]         =			{
            "xmin": xmin,
            "ymin": ymin,
            "xmax": xmax,
            "ymax": ymax,
        }
        return bbox


