





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

SCREAMING_SNAKE_CASE__       :			List[str]			      =						logging.get_logger(__name__)







@add_end_docstrings(_UpperCamelCase  )
class   __lowerCAmelCase			(					_UpperCamelCase  ):





   def __init__( self					,			**snake_case      )				->  Dict:
      """simple docstring"""





      super().__init__(**snake_case      )

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

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





   def __call__( self					,			snake_case					,			snake_case = None					,			**snake_case					,			)				->  List[Any]:
      """simple docstring"""





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

      if isinstance(snake_case					,			(str, Image.Image)      ):
         a__	:						Any      		=  {"image": image, "candidate_labels": candidate_labels}
      else:
         a__	:						Tuple      		=  image
      a__	:						Tuple      		=  super().__call__(snake_case					,			**snake_case      )
      return results





   def 	_snake_case ( self					,			**snake_case      )				->  Optional[Any]:
      """simple docstring"""





      a__	:						Union[str, Any]      		=  {}
      if "threshold" in kwargs:
         a__	:						int      		=  kwargs["threshold"]
      if "top_k" in kwargs:
         a__	:						Tuple      		=  kwargs["top_k"]
      return {}, {}, postprocess_params





   def 	_snake_case ( self					,			snake_case      )				->  Tuple:
      """simple docstring"""





      a__	:						Any      		=  load_image(inputs["image"]      )
      a__	:						int      		=  inputs["candidate_labels"]
      if isinstance(snake_case					,			snake_case      ):
         a__	:						List[str]      		=  candidate_labels.split(","      )

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





   def 	_snake_case ( self					,			snake_case      )				->  Union[str, Any]:
      """simple docstring"""





      a__	:						str      		=  model_inputs.pop("target_size"      )
      a__	:						int      		=  model_inputs.pop("candidate_label"      )
      a__	:						Dict      		=  model_inputs.pop("is_last"      )

      a__	:						Tuple      		=  self.model(**snake_case      )

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





   def 	_snake_case ( self					,			snake_case					,			snake_case=0.1					,			snake_case=None      )				->  Any:
      """simple docstring"""





      a__	:						Tuple      		=  []
      for model_output in model_outputs:
         a__	:						int      		=  model_output["candidate_label"]
         a__	:						str      		=  BaseModelOutput(snake_case      )
         a__	:						int      		=  self.image_processor.post_process_object_detection(
             outputs=snake_case					,			threshold=snake_case					,			target_sizes=model_output["target_size"]      )[0]

         for index in outputs["scores"].nonzero():
            a__	:						Tuple      		=  outputs["scores"][index].item()
            a__	:						str      		=  self._get_bounding_box(outputs["boxes"][index][0]      )

            a__	:						List[Any]      		=  {"score": score, "label": label, "box": box}
            results.append(snake_case      )

      a__	:						Union[str, Any]      		=  sorted(snake_case					,			key=lambda snake_case      : x["score"]					,			reverse=snake_case      )
      if top_k:
         a__	:						str      		=  results[:top_k]

      return results





   def 	_snake_case ( self					,			snake_case      )				->  Dict[str, int]:
      """simple docstring"""





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








from typing import Any, Callable, Dict, List, Optional, Union

import torch
from transformers import CLIPImageProcessor, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DiffusionPipeline,
    LMSDiscreteScheduler,
    PNDMScheduler,
    StableDiffusionPipeline,
    UNetaDConditionModel,
)
from diffusers.pipelines.stable_diffusion import StableDiffusionPipelineOutput
from diffusers.pipelines.stable_diffusion.safety_checker import StableDiffusionSafetyChecker


SCREAMING_SNAKE_CASE__       :			Dict			      =						"""CompVis/stable-diffusion-v1-1"""
SCREAMING_SNAKE_CASE__       :			Dict			      =						"""CompVis/stable-diffusion-v1-2"""
SCREAMING_SNAKE_CASE__       :			Tuple			      =						"""CompVis/stable-diffusion-v1-3"""
SCREAMING_SNAKE_CASE__       :			str			      =						"""CompVis/stable-diffusion-v1-4"""







class   __lowerCAmelCase			(					_UpperCamelCase  ):





   def __init__( self					,			snake_case					,			snake_case					,			snake_case					,			snake_case					,			snake_case					,			snake_case					,			snake_case					,			snake_case = True					,			)				->  Any:
      """simple docstring"""





      super()._init_()

      a__	:						Union[str, Any]      		=  StableDiffusionPipeline.from_pretrained(snake_case      )
      a__	:						Optional[Any]      		=  StableDiffusionPipeline.from_pretrained(snake_case      )
      a__	:						Optional[Any]      		=  StableDiffusionPipeline.from_pretrained(snake_case      )
      a__	:						int      		=  StableDiffusionPipeline(
          vae=snake_case					,			text_encoder=snake_case					,			tokenizer=snake_case					,			unet=snake_case					,			scheduler=snake_case					,			safety_checker=snake_case					,			feature_extractor=snake_case					,			requires_safety_checker=snake_case					,			)

      self.register_modules(pipelinea=self.pipea					,			pipelinea=self.pipea					,			pipelinea=self.pipea					,			pipelinea=self.pipea      )





   @property
   def 	_snake_case ( self      )				->  Dict[str, Any]:
      """simple docstring"""





      return {k: getattr(self					,			snake_case      ) for k in self.config.keys() if not k.startswith("_"      )}





   def 	_snake_case ( self					,			snake_case = "auto"      )				->  Optional[Any]:
      """simple docstring"""





      if slice_size == "auto":
         # half the attention head size is usually a good trade-off between
         # speed and memory
         a__	:						List[Any]      		=  self.unet.config.attention_head_dim // 2
      self.unet.set_attention_slice(snake_case      )





   def 	_snake_case ( self      )				->  Tuple:
      """simple docstring"""





      self.enable_attention_slicing(snake_case      )





   @torch.no_grad()
   def 	_snake_case ( self					,			snake_case					,			snake_case = 512					,			snake_case = 512					,			snake_case = 50					,			snake_case = 7.5					,			snake_case = None					,			snake_case = 1					,			snake_case = 0.0					,			snake_case = None					,			snake_case = None					,			snake_case = "pil"					,			snake_case = True					,			snake_case = None					,			snake_case = 1					,			**snake_case					,			)				->  Optional[int]:
      """simple docstring"""





      return self.pipea(
          prompt=snake_case					,			height=snake_case					,			width=snake_case					,			num_inference_steps=snake_case					,			guidance_scale=snake_case					,			negative_prompt=snake_case					,			num_images_per_prompt=snake_case					,			eta=snake_case					,			generator=snake_case					,			latents=snake_case					,			output_type=snake_case					,			return_dict=snake_case					,			callback=snake_case					,			callback_steps=snake_case					,			**snake_case					,			)





   @torch.no_grad()
   def 	_snake_case ( self					,			snake_case					,			snake_case = 512					,			snake_case = 512					,			snake_case = 50					,			snake_case = 7.5					,			snake_case = None					,			snake_case = 1					,			snake_case = 0.0					,			snake_case = None					,			snake_case = None					,			snake_case = "pil"					,			snake_case = True					,			snake_case = None					,			snake_case = 1					,			**snake_case					,			)				->  Dict:
      """simple docstring"""





      return self.pipea(
          prompt=snake_case					,			height=snake_case					,			width=snake_case					,			num_inference_steps=snake_case					,			guidance_scale=snake_case					,			negative_prompt=snake_case					,			num_images_per_prompt=snake_case					,			eta=snake_case					,			generator=snake_case					,			latents=snake_case					,			output_type=snake_case					,			return_dict=snake_case					,			callback=snake_case					,			callback_steps=snake_case					,			**snake_case					,			)





   @torch.no_grad()
   def 	_snake_case ( self					,			snake_case					,			snake_case = 512					,			snake_case = 512					,			snake_case = 50					,			snake_case = 7.5					,			snake_case = None					,			snake_case = 1					,			snake_case = 0.0					,			snake_case = None					,			snake_case = None					,			snake_case = "pil"					,			snake_case = True					,			snake_case = None					,			snake_case = 1					,			**snake_case					,			)				->  Union[str, Any]:
      """simple docstring"""





      return self.pipea(
          prompt=snake_case					,			height=snake_case					,			width=snake_case					,			num_inference_steps=snake_case					,			guidance_scale=snake_case					,			negative_prompt=snake_case					,			num_images_per_prompt=snake_case					,			eta=snake_case					,			generator=snake_case					,			latents=snake_case					,			output_type=snake_case					,			return_dict=snake_case					,			callback=snake_case					,			callback_steps=snake_case					,			**snake_case					,			)





   @torch.no_grad()
   def 	_snake_case ( self					,			snake_case					,			snake_case = 512					,			snake_case = 512					,			snake_case = 50					,			snake_case = 7.5					,			snake_case = None					,			snake_case = 1					,			snake_case = 0.0					,			snake_case = None					,			snake_case = None					,			snake_case = "pil"					,			snake_case = True					,			snake_case = None					,			snake_case = 1					,			**snake_case					,			)				->  Dict:
      """simple docstring"""





      return self.pipea(
          prompt=snake_case					,			height=snake_case					,			width=snake_case					,			num_inference_steps=snake_case					,			guidance_scale=snake_case					,			negative_prompt=snake_case					,			num_images_per_prompt=snake_case					,			eta=snake_case					,			generator=snake_case					,			latents=snake_case					,			output_type=snake_case					,			return_dict=snake_case					,			callback=snake_case					,			callback_steps=snake_case					,			**snake_case					,			)





   @torch.no_grad()
   def 	_snake_case ( self					,			snake_case					,			snake_case = 512					,			snake_case = 512					,			snake_case = 50					,			snake_case = 7.5					,			snake_case = None					,			snake_case = 1					,			snake_case = 0.0					,			snake_case = None					,			snake_case = None					,			snake_case = "pil"					,			snake_case = True					,			snake_case = None					,			snake_case = 1					,			**snake_case					,			)				->  Dict:
      """simple docstring"""





      a__	:						Any      		=  "cuda" if torch.cuda.is_available() else "cpu"
      self.to(snake_case      )

      # Checks if the height and width are divisible by 8 or not
      if height % 8 != 0 or width % 8 != 0:
         raise ValueError(F"""`height` and `width` must be divisible by 8 but are {height} and {width}."""      )

      # Get first result from Stable Diffusion Checkpoint v1.1
      a__	:						Any      		=  self.textaimg_sda_a(
          prompt=snake_case					,			height=snake_case					,			width=snake_case					,			num_inference_steps=snake_case					,			guidance_scale=snake_case					,			negative_prompt=snake_case					,			num_images_per_prompt=snake_case					,			eta=snake_case					,			generator=snake_case					,			latents=snake_case					,			output_type=snake_case					,			return_dict=snake_case					,			callback=snake_case					,			callback_steps=snake_case					,			**snake_case					,			)

      # Get first result from Stable Diffusion Checkpoint v1.2
      a__	:						List[Any]      		=  self.textaimg_sda_a(
          prompt=snake_case					,			height=snake_case					,			width=snake_case					,			num_inference_steps=snake_case					,			guidance_scale=snake_case					,			negative_prompt=snake_case					,			num_images_per_prompt=snake_case					,			eta=snake_case					,			generator=snake_case					,			latents=snake_case					,			output_type=snake_case					,			return_dict=snake_case					,			callback=snake_case					,			callback_steps=snake_case					,			**snake_case					,			)

      # Get first result from Stable Diffusion Checkpoint v1.3
      a__	:						Optional[Any]      		=  self.textaimg_sda_a(
          prompt=snake_case					,			height=snake_case					,			width=snake_case					,			num_inference_steps=snake_case					,			guidance_scale=snake_case					,			negative_prompt=snake_case					,			num_images_per_prompt=snake_case					,			eta=snake_case					,			generator=snake_case					,			latents=snake_case					,			output_type=snake_case					,			return_dict=snake_case					,			callback=snake_case					,			callback_steps=snake_case					,			**snake_case					,			)

      # Get first result from Stable Diffusion Checkpoint v1.4
      a__	:						Dict      		=  self.textaimg_sda_a(
          prompt=snake_case					,			height=snake_case					,			width=snake_case					,			num_inference_steps=snake_case					,			guidance_scale=snake_case					,			negative_prompt=snake_case					,			num_images_per_prompt=snake_case					,			eta=snake_case					,			generator=snake_case					,			latents=snake_case					,			output_type=snake_case					,			return_dict=snake_case					,			callback=snake_case					,			callback_steps=snake_case					,			**snake_case					,			)

      # Get all result images into a single list and pass it via StableDiffusionPipelineOutput for final result
      return StableDiffusionPipelineOutput([resa[0], resa[0], resa[0], resa[0]]      )

