





from typing import Dict, List, Optional, Union

import numpy as np

from transformers.utils import is_vision_available
from transformers.utils.generic import TensorType

from ...image_processing_utils import BaseImageProcessor, BatchFeature, get_size_dict
from ...image_transforms import (
    center_crop,
    get_resize_output_image_size,
    normalize,
    rescale,
    resize,
    to_channel_dimension_format,
)
from ...image_utils import (
    IMAGENET_STANDARD_MEAN,
    IMAGENET_STANDARD_STD,
    ChannelDimension,
    ImageInput,
    PILImageResampling,
    is_valid_image,
    to_numpy_array,
    valid_images,
)
from ...utils import logging


if is_vision_available():
  import PIL

SCREAMING_SNAKE_CASE__       :			Optional[Any]			      =						logging.get_logger(__name__)






def       _A  (      lowerCamelCase   ):
   if isinstance(lowerCamelCase							, (list, tuple)   ) and isinstance(videos[0]							, (list, tuple)   ) and is_valid_image(videos[0][0]   ):
      return videos

   elif isinstance(lowerCamelCase							, (list, tuple)   ) and is_valid_image(videos[0]   ):
      return [videos]

   elif is_valid_image(lowerCamelCase   ):
      return [[videos]]

   raise ValueError(F"""Could not make batched video from {videos}"""   )







class   __lowerCAmelCase			(					_UpperCamelCase  ):
   _UpperCamelCase :						List[str]     		=				["""pixel_values"""]





   def __init__( self					,			snake_case = True					,			snake_case = None					,			snake_case = PILImageResampling.BILINEAR					,			snake_case = True					,			snake_case = None					,			snake_case = True					,			snake_case = 1 / 255					,			snake_case = True					,			snake_case = True					,			snake_case = None					,			snake_case = None					,			**snake_case					,			)				->  None:
      """simple docstring"""





      super().__init__(**snake_case      )
      a__	:						List[Any]      		=  size if size is not None else {"shortest_edge": 256}
      a__	:						List[Any]      		=  get_size_dict(snake_case					,			default_to_square=snake_case      )
      a__	:						Tuple      		=  crop_size if crop_size is not None else {"height": 224, "width": 224}
      a__	:						int      		=  get_size_dict(snake_case					,			param_name="crop_size"      )

      a__	:						Optional[int]      		=  do_resize
      a__	:						str      		=  size
      a__	:						Tuple      		=  do_center_crop
      a__	:						Dict      		=  crop_size
      a__	:						Union[str, Any]      		=  resample
      a__	:						Union[str, Any]      		=  do_rescale
      a__	:						Union[str, Any]      		=  rescale_factor
      a__	:						str      		=  offset
      a__	:						Tuple      		=  do_normalize
      a__	:						Union[str, Any]      		=  image_mean if image_mean is not None else IMAGENET_STANDARD_MEAN
      a__	:						Optional[Any]      		=  image_std if image_std is not None else IMAGENET_STANDARD_STD





   def 	_snake_case ( self					,			snake_case					,			snake_case					,			snake_case = PILImageResampling.BILINEAR					,			snake_case = None					,			**snake_case					,			)				->  np.ndarray:
      """simple docstring"""





      a__	:						Tuple      		=  get_size_dict(snake_case					,			default_to_square=snake_case      )
      if "shortest_edge" in size:
         a__	:						str      		=  get_resize_output_image_size(snake_case					,			size["shortest_edge"]					,			default_to_square=snake_case      )
      elif "height" in size and "width" in size:
         a__	:						Dict      		=  (size["height"], size["width"])
      else:
         raise ValueError(F"""Size must have 'height' and 'width' or 'shortest_edge' as keys. Got {size.keys()}"""      )
      return resize(snake_case					,			size=snake_case					,			resample=snake_case					,			data_format=snake_case					,			**snake_case      )





   def 	_snake_case ( self					,			snake_case					,			snake_case					,			snake_case = None					,			**snake_case					,			)				->  np.ndarray:
      """simple docstring"""





      a__	:						Union[str, Any]      		=  get_size_dict(snake_case      )
      if "height" not in size or "width" not in size:
         raise ValueError(F"""Size must have 'height' and 'width' as keys. Got {size.keys()}"""      )
      return center_crop(snake_case					,			size=(size["height"], size["width"])					,			data_format=snake_case					,			**snake_case      )





   def 	_snake_case ( self					,			snake_case					,			snake_case					,			snake_case = True					,			snake_case = None					,			**snake_case					,			)				->  List[str]:
      """simple docstring"""





      a__	:						Tuple      		=  image.astype(np.floataa      )
      if offset:
         a__	:						Tuple      		=  image - (scale / 2)
      return rescale(snake_case					,			scale=snake_case					,			data_format=snake_case					,			**snake_case      )





   def 	_snake_case ( self					,			snake_case					,			snake_case					,			snake_case					,			snake_case = None					,			**snake_case					,			)				->  np.ndarray:
      """simple docstring"""





      return normalize(snake_case					,			mean=snake_case					,			std=snake_case					,			data_format=snake_case					,			**snake_case      )





   def 	_snake_case ( self					,			snake_case					,			snake_case = None					,			snake_case = None					,			snake_case = None					,			snake_case = None					,			snake_case = None					,			snake_case = None					,			snake_case = None					,			snake_case = None					,			snake_case = None					,			snake_case = None					,			snake_case = None					,			snake_case = ChannelDimension.FIRST					,			)				->  np.ndarray:
      """simple docstring"""





      if do_resize and size is None or resample is None:
         raise ValueError("Size and resample must be specified if do_resize is True."      )

      if do_center_crop and crop_size is None:
         raise ValueError("Crop size must be specified if do_center_crop is True."      )

      if do_rescale and rescale_factor is None:
         raise ValueError("Rescale factor must be specified if do_rescale is True."      )

      if do_normalize and (image_mean is None or image_std is None):
         raise ValueError("Image mean and std must be specified if do_normalize is True."      )

      if offset and not do_rescale:
         raise ValueError("For offset, do_rescale must also be set to True."      )

      # All transformations expect numpy arrays.
      a__	:						Any      		=  to_numpy_array(snake_case      )

      if do_resize:
         a__	:						Optional[int]      		=  self.resize(image=snake_case					,			size=snake_case					,			resample=snake_case      )

      if do_center_crop:
         a__	:						Optional[Any]      		=  self.center_crop(snake_case					,			size=snake_case      )

      if do_rescale:
         a__	:						Union[str, Any]      		=  self.rescale(image=snake_case					,			scale=snake_case					,			offset=snake_case      )

      if do_normalize:
         a__	:						Optional[int]      		=  self.normalize(image=snake_case					,			mean=snake_case					,			std=snake_case      )

      a__	:						int      		=  to_channel_dimension_format(snake_case					,			snake_case      )
      return image





   def 	_snake_case ( self					,			snake_case					,			snake_case = None					,			snake_case = None					,			snake_case = None					,			snake_case = None					,			snake_case = None					,			snake_case = None					,			snake_case = None					,			snake_case = None					,			snake_case = None					,			snake_case = None					,			snake_case = None					,			snake_case = None					,			snake_case = ChannelDimension.FIRST					,			**snake_case					,			)				->  PIL.Image.Image:
      """simple docstring"""





      a__	:						Union[str, Any]      		=  do_resize if do_resize is not None else self.do_resize
      a__	:						Union[str, Any]      		=  resample if resample is not None else self.resample
      a__	:						List[Any]      		=  do_center_crop if do_center_crop is not None else self.do_center_crop
      a__	:						str      		=  do_rescale if do_rescale is not None else self.do_rescale
      a__	:						Tuple      		=  rescale_factor if rescale_factor is not None else self.rescale_factor
      a__	:						int      		=  offset if offset is not None else self.offset
      a__	:						int      		=  do_normalize if do_normalize is not None else self.do_normalize
      a__	:						List[str]      		=  image_mean if image_mean is not None else self.image_mean
      a__	:						Tuple      		=  image_std if image_std is not None else self.image_std

      a__	:						Dict      		=  size if size is not None else self.size
      a__	:						Tuple      		=  get_size_dict(snake_case					,			default_to_square=snake_case      )
      a__	:						Any      		=  crop_size if crop_size is not None else self.crop_size
      a__	:						List[Any]      		=  get_size_dict(snake_case					,			param_name="crop_size"      )

      if not valid_images(snake_case      ):
         raise ValueError(
             "Invalid image type. Must be of type PIL.Image.Image, numpy.ndarray, "
             "torch.Tensor, tf.Tensor or jax.ndarray."      )

      a__	:						Dict      		=  make_batched(snake_case      )

      a__	:						Dict      		=  [
          [
              self._preprocess_image(
                  image=snake_case					,			do_resize=snake_case					,			size=snake_case					,			resample=snake_case					,			do_center_crop=snake_case					,			crop_size=snake_case					,			do_rescale=snake_case					,			rescale_factor=snake_case					,			offset=snake_case					,			do_normalize=snake_case					,			image_mean=snake_case					,			image_std=snake_case					,			data_format=snake_case					,			)
              for img in video
          ]
          for video in videos
      ]

      a__	:						int      		=  {"pixel_values": videos}
      return BatchFeature(data=snake_case					,			tensor_type=snake_case      )








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]]      )

