


import inspect
from typing import Callable, List, Optional, Union

import torch
from transformers import CLIPImageProcessor, CLIPTextModel, CLIPTokenizer

from diffusers import DiffusionPipeline
from diffusers.models import AutoencoderKL, UNetaDConditionModel
from diffusers.pipelines.stable_diffusion import StableDiffusionPipelineOutput
from diffusers.pipelines.stable_diffusion.safety_checker import StableDiffusionSafetyChecker
from diffusers.schedulers import DDIMScheduler, LMSDiscreteScheduler, PNDMScheduler
from diffusers.utils import logging


a_ :Tuple  							=			logging.get_logger(__name__)  # pylint: disable=invalid-name

class 							lowercase  (						_UpperCAmelCase			):







     def __init__(						self :						Dict   ,   _lowercase :						AutoencoderKL   ,   _lowercase :						CLIPTextModel   ,   _lowercase :						CLIPTokenizer   ,   _lowercase :						UNetaDConditionModel   ,   _lowercase :						Union[DDIMScheduler, PNDMScheduler, LMSDiscreteScheduler]   ,   _lowercase :						StableDiffusionSafetyChecker   ,   _lowercase :						CLIPImageProcessor   ,   ):
           super().__init__()
           self.register_modules(
               vae=_lowercase   ,   text_encoder=_lowercase   ,   tokenizer=_lowercase   ,   unet=_lowercase   ,   scheduler=_lowercase   ,   safety_checker=_lowercase   ,   feature_extractor=_lowercase   ,   )







     def 							lowercase__       (						self :						Tuple   ,   _lowercase :						Optional[Union[str, int]] = "auto"      ):
           if slice_size == "auto":
                 # half the attention head size is usually a good trade-off between
                 # speed and memory
                 SCREAMING_SNAKE_CASE__       :  Tuple						     =		self.unet.config.attention_head_dim // 2
           self.unet.set_attention_slice(_lowercase      )







     def 							lowercase__       (						self :						List[Any]      ):
           self.enable_attention_slicing(_lowercase      )







     @torch.no_grad()
     def __call__(						self :						Any   ,   _lowercase :						Union[str, List[str]]   ,   _lowercase :						int = 5_12   ,   _lowercase :						int = 5_12   ,   _lowercase :						int = 50   ,   _lowercase :						float = 7.5   ,   _lowercase :						Optional[Union[str, List[str]]] = None   ,   _lowercase :						Optional[int] = 1   ,   _lowercase :						float = 0.0   ,   _lowercase :						Optional[torch.Generator] = None   ,   _lowercase :						Optional[torch.FloatTensor] = None   ,   _lowercase :						Optional[str] = "pil"   ,   _lowercase :						bool = True   ,   _lowercase :						Optional[Callable[[int, int, torch.FloatTensor], None]] = None   ,   _lowercase :						int = 1   ,   _lowercase :						Optional[torch.FloatTensor] = None   ,   **_lowercase :						Dict   ,   ):

           if isinstance(_lowercase   ,   _lowercase      ):
                 SCREAMING_SNAKE_CASE__       :  str						     =		1
           elif isinstance(_lowercase   ,   _lowercase      ):
                 SCREAMING_SNAKE_CASE__       :  Any						     =		len(_lowercase      )
           else:
                 raise ValueError(f"""`prompt` has to be of type `str` or `list` but is {type(_lowercase      )}"""      )

           if height % 8 != 0 or width % 8 != 0:
                 raise ValueError(f"""`height` and `width` have to be divisible by 8 but are {height} and {width}."""      )

           if (callback_steps is None) or (
               callback_steps is not None and (not isinstance(_lowercase   ,   _lowercase      ) or callback_steps <= 0)
           ):
                 raise ValueError(
                     f"""`callback_steps` has to be a positive integer but is {callback_steps} of type"""
                     f""" {type(_lowercase      )}."""      )

           # get prompt text embeddings
           SCREAMING_SNAKE_CASE__       :  Any						     =		self.tokenizer(
               _lowercase   ,   padding='''max_length'''   ,   max_length=self.tokenizer.model_max_length   ,   return_tensors='''pt'''   ,   )
           SCREAMING_SNAKE_CASE__       :  Optional[Any]						     =		text_inputs.input_ids

           if text_input_ids.shape[-1] > self.tokenizer.model_max_length:
                 SCREAMING_SNAKE_CASE__       :  Tuple						     =		self.tokenizer.batch_decode(text_input_ids[:, self.tokenizer.model_max_length :]      )
                 logger.warning(
                     '''The following part of your input was truncated because CLIP can only handle sequences up to'''
                     f""" {self.tokenizer.model_max_length} tokens: {removed_text}"""      )
                 SCREAMING_SNAKE_CASE__       :  int						     =		text_input_ids[:, : self.tokenizer.model_max_length]

           if text_embeddings is None:
                 SCREAMING_SNAKE_CASE__       :  Optional[Any]						     =		self.text_encoder(text_input_ids.to(self.device      )      )[0]

           # duplicate text embeddings for each generation per prompt, using mps friendly method
           SCREAMING_SNAKE_CASE__     ,       SCREAMING_SNAKE_CASE__     ,       SCREAMING_SNAKE_CASE__       :  Optional[Any]						     =		text_embeddings.shape
           SCREAMING_SNAKE_CASE__       :  Any						     =		text_embeddings.repeat(1   ,   _lowercase   ,   1      )
           SCREAMING_SNAKE_CASE__       :  Tuple						     =		text_embeddings.view(bs_embed * num_images_per_prompt   ,   _lowercase   ,   -1      )

           # here `guidance_scale` is defined analog to the guidance weight `w` of equation (2)
           # of the Imagen paper: https://arxiv.org/pdf/2205.11487.pdf . `guidance_scale = 1`
           # corresponds to doing no classifier free guidance.
           SCREAMING_SNAKE_CASE__       :  Optional[int]						     =		guidance_scale > 1.0
           # get unconditional embeddings for classifier free guidance
           if do_classifier_free_guidance:
                 SCREAMING_SNAKE_CASE__       :  List[str]
                 if negative_prompt is None:
                       SCREAMING_SNAKE_CASE__       :  Any						     =		['''''']
                 elif type(_lowercase      ) is not type(_lowercase      ):
                       raise TypeError(
                           f"""`negative_prompt` should be the same type to `prompt`, but got {type(_lowercase      )} !="""
                           f""" {type(_lowercase      )}."""      )
                 elif isinstance(_lowercase   ,   _lowercase      ):
                       SCREAMING_SNAKE_CASE__       :  List[str]						     =		[negative_prompt]
                 elif batch_size != len(_lowercase      ):
                       raise ValueError(
                           f"""`negative_prompt`: {negative_prompt} has batch size {len(_lowercase      )}, but `prompt`:"""
                           f""" {prompt} has batch size {batch_size}. Please make sure that passed `negative_prompt` matches"""
                           ''' the batch size of `prompt`.'''      )
                 else:
                       SCREAMING_SNAKE_CASE__       :  Any						     =		negative_prompt

                 SCREAMING_SNAKE_CASE__       :  Tuple						     =		text_input_ids.shape[-1]
                 SCREAMING_SNAKE_CASE__       :  List[Any]						     =		self.tokenizer(
                     _lowercase   ,   padding='''max_length'''   ,   max_length=_lowercase   ,   truncation=_lowercase   ,   return_tensors='''pt'''   ,   )
                 SCREAMING_SNAKE_CASE__       :  List[str]						     =		self.text_encoder(uncond_input.input_ids.to(self.device      )      )[0]

                 # duplicate unconditional embeddings for each generation per prompt, using mps friendly method
                 SCREAMING_SNAKE_CASE__       :  int						     =		uncond_embeddings.shape[1]
                 SCREAMING_SNAKE_CASE__       :  List[Any]						     =		uncond_embeddings.repeat(_lowercase   ,   _lowercase   ,   1      )
                 SCREAMING_SNAKE_CASE__       :  Optional[int]						     =		uncond_embeddings.view(batch_size * num_images_per_prompt   ,   _lowercase   ,   -1      )

                 # For classifier free guidance, we need to do two forward passes.
                 # Here we concatenate the unconditional and text embeddings into a single batch
                 # to avoid doing two forward passes
                 SCREAMING_SNAKE_CASE__       :  List[str]						     =		torch.cat([uncond_embeddings, text_embeddings]      )

           # get the initial random noise unless the user supplied it

           # Unlike in other pipelines, latents need to be generated in the target device
           # for 1-to-1 results reproducibility with the CompVis implementation.
           # However this currently doesn't work in `mps`.
           SCREAMING_SNAKE_CASE__       :  Union[str, Any]						     =		(batch_size * num_images_per_prompt, self.unet.config.in_channels, height // 8, width // 8)
           SCREAMING_SNAKE_CASE__       :  str						     =		(batch_size * num_images_per_prompt, self.unet.config.in_channels, 64, 64)
           SCREAMING_SNAKE_CASE__       :  Optional[Any]						     =		text_embeddings.dtype
           if latents is None:
                 if self.device.type == "mps":
                       # randn does not exist on mps
                       SCREAMING_SNAKE_CASE__       :  str						     =		torch.randn(
                           _lowercase   ,   generator=_lowercase   ,   device='''cpu'''   ,   dtype=_lowercase      ).to(self.device      )
                       SCREAMING_SNAKE_CASE__       :  List[str]						     =		torch.randn(_lowercase   ,   generator=_lowercase   ,   device='''cpu'''   ,   dtype=_lowercase      ).to(
                           self.device      )
                 else:
                       SCREAMING_SNAKE_CASE__       :  str						     =		torch.randn(
                           _lowercase   ,   generator=_lowercase   ,   device=self.device   ,   dtype=_lowercase      )
                       SCREAMING_SNAKE_CASE__       :  Optional[int]						     =		torch.randn(_lowercase   ,   generator=_lowercase   ,   device=self.device   ,   dtype=_lowercase      )
           else:
                 if latents_reference.shape != latents_shape:
                       raise ValueError(f"""Unexpected latents shape, got {latents.shape}, expected {latents_shape}"""      )
                 SCREAMING_SNAKE_CASE__       :  Optional[int]						     =		latents_reference.to(self.device      )
                 SCREAMING_SNAKE_CASE__       :  List[str]						     =		latents.to(self.device      )

           # This is the key part of the pipeline where we
           # try to ensure that the generated images w/ the same seed
           # but different sizes actually result in similar images
           SCREAMING_SNAKE_CASE__       :  List[Any]						     =		(latents_shape[3] - latents_shape_reference[3]) // 2
           SCREAMING_SNAKE_CASE__       :  Union[str, Any]						     =		(latents_shape[2] - latents_shape_reference[2]) // 2
           SCREAMING_SNAKE_CASE__       :  Optional[int]						     =		latents_shape_reference[3] if dx >= 0 else latents_shape_reference[3] + 2 * dx
           SCREAMING_SNAKE_CASE__       :  Optional[Any]						     =		latents_shape_reference[2] if dy >= 0 else latents_shape_reference[2] + 2 * dy
           SCREAMING_SNAKE_CASE__       :  str						     =		0 if dx < 0 else dx
           SCREAMING_SNAKE_CASE__       :  int						     =		0 if dy < 0 else dy
           SCREAMING_SNAKE_CASE__       :  Any						     =		max(-dx   ,   0      )
           SCREAMING_SNAKE_CASE__       :  Optional[int]						     =		max(-dy   ,   0      )
           # import pdb
           # pdb.set_trace()
           SCREAMING_SNAKE_CASE__       :  List[Any]						     =		latents_reference[:, :, dy : dy + h, dx : dx + w]

           # set timesteps
           self.scheduler.set_timesteps(_lowercase      )

           # Some schedulers like PNDM have timesteps as arrays
           # It's more optimized to move all timesteps to correct device beforehand
           SCREAMING_SNAKE_CASE__       :  Union[str, Any]						     =		self.scheduler.timesteps.to(self.device      )

           # scale the initial noise by the standard deviation required by the scheduler
           SCREAMING_SNAKE_CASE__       :  Any						     =		latents * self.scheduler.init_noise_sigma

           # prepare extra kwargs for the scheduler step, since not all schedulers have the same signature
           # eta (η) is only used with the DDIMScheduler, it will be ignored for other schedulers.
           # eta corresponds to η in DDIM paper: https://arxiv.org/abs/2010.02502
           # and should be between [0, 1]
           SCREAMING_SNAKE_CASE__       :  Optional[int]						     =		'''eta''' in set(inspect.signature(self.scheduler.step      ).parameters.keys()      )
           SCREAMING_SNAKE_CASE__       :  Any						     =		{}
           if accepts_eta:
                 SCREAMING_SNAKE_CASE__       :  Union[str, Any]						     =		eta

           for i, t in enumerate(self.progress_bar(_lowercase      )      ):
                 # expand the latents if we are doing classifier free guidance
                 SCREAMING_SNAKE_CASE__       :  Dict						     =		torch.cat([latents] * 2      ) if do_classifier_free_guidance else latents
                 SCREAMING_SNAKE_CASE__       :  str						     =		self.scheduler.scale_model_input(_lowercase   ,   _lowercase      )

                 # predict the noise residual
                 SCREAMING_SNAKE_CASE__       :  Optional[Any]						     =		self.unet(_lowercase   ,   _lowercase   ,   encoder_hidden_states=_lowercase      ).sample

                 # perform guidance
                 if do_classifier_free_guidance:
                       SCREAMING_SNAKE_CASE__     ,       SCREAMING_SNAKE_CASE__       :  List[Any]						     =		noise_pred.chunk(2      )
                       SCREAMING_SNAKE_CASE__       :  str						     =		noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

                 # compute the previous noisy sample x_t -> x_t-1
                 SCREAMING_SNAKE_CASE__       :  List[Any]						     =		self.scheduler.step(_lowercase   ,   _lowercase   ,   _lowercase   ,   **_lowercase      ).prev_sample

                 # call the callback, if provided
                 if callback is not None and i % callback_steps == 0:
                       callback(_lowercase   ,   _lowercase   ,   _lowercase      )

           SCREAMING_SNAKE_CASE__       :  Dict						     =		1 / 0.18215 * latents
           SCREAMING_SNAKE_CASE__       :  Dict						     =		self.vae.decode(_lowercase      ).sample

           SCREAMING_SNAKE_CASE__       :  List[Any]						     =		(image / 2 + 0.5).clamp(0   ,   1      )

           # we always cast to float32 as this does not cause significant overhead and is compatible with bfloat16
           SCREAMING_SNAKE_CASE__       :  int						     =		image.cpu().permute(0   ,   2   ,   3   ,   1      ).float().numpy()

           if self.safety_checker is not None:
                 SCREAMING_SNAKE_CASE__       :  Any						     =		self.feature_extractor(self.numpy_to_pil(_lowercase      )   ,   return_tensors='''pt'''      ).to(
                     self.device      )
                 SCREAMING_SNAKE_CASE__     ,       SCREAMING_SNAKE_CASE__       :  Any						     =		self.safety_checker(
                     images=_lowercase   ,   clip_input=safety_checker_input.pixel_values.to(text_embeddings.dtype      )      )
           else:
                 SCREAMING_SNAKE_CASE__       :  Optional[int]						     =		None

           if output_type == "pil":
                 SCREAMING_SNAKE_CASE__       :  int						     =		self.numpy_to_pil(_lowercase      )

           if not return_dict:
                 return (image, has_nsfw_concept)

           return StableDiffusionPipelineOutput(images=_lowercase   ,   nsfw_content_detected=_lowercase      )














'''simple docstring'''







# Copyright 2021 The HuggingFace Inc. team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from packaging import version

from .. import __version__
from .constants import IMAGENET_DEFAULT_MEAN, IMAGENET_DEFAULT_STD, IMAGENET_STANDARD_MEAN, IMAGENET_STANDARD_STD
from .doc import (
    add_code_sample_docstrings,
    add_end_docstrings,
    add_start_docstrings,
    add_start_docstrings_to_model_forward,
    copy_func,
    replace_return_docstrings,
)
from .generic import (
    ContextManagers,
    ExplicitEnum,
    ModelOutput,
    PaddingStrategy,
    TensorType,
    add_model_info_to_auto_map,
    cached_property,
    can_return_loss,
    expand_dims,
    find_labels,
    flatten_dict,
    infer_framework,
    is_jax_tensor,
    is_numpy_array,
    is_tensor,
    is_tf_symbolic_tensor,
    is_tf_tensor,
    is_torch_device,
    is_torch_dtype,
    is_torch_tensor,
    reshape,
    squeeze,
    strtobool,
    tensor_size,
    to_numpy,
    to_py_obj,
    transpose,
    working_or_temp_dir,
)
from .hub import (
    CLOUDFRONT_DISTRIB_PREFIX,
    DISABLE_TELEMETRY,
    HF_MODULES_CACHE,
    HUGGINGFACE_CO_PREFIX,
    HUGGINGFACE_CO_RESOLVE_ENDPOINT,
    PYTORCH_PRETRAINED_BERT_CACHE,
    PYTORCH_TRANSFORMERS_CACHE,
    S3_BUCKET_PREFIX,
    TRANSFORMERS_CACHE,
    TRANSFORMERS_DYNAMIC_MODULE_NAME,
    EntryNotFoundError,
    PushToHubMixin,
    RepositoryNotFoundError,
    RevisionNotFoundError,
    cached_file,
    default_cache_path,
    define_sagemaker_information,
    download_url,
    extract_commit_hash,
    get_cached_models,
    get_file_from_repo,
    get_full_repo_name,
    has_file,
    http_user_agent,
    is_offline_mode,
    is_remote_url,
    move_cache,
    send_example_telemetry,
    try_to_load_from_cache,
)
from .import_utils import (
    ENV_VARS_TRUE_AND_AUTO_VALUES,
    ENV_VARS_TRUE_VALUES,
    TORCH_FX_REQUIRED_VERSION,
    USE_JAX,
    USE_TF,
    USE_TORCH,
    DummyObject,
    OptionalDependencyNotAvailable,
    _LazyModule,
    ccl_version,
    direct_transformers_import,
    get_torch_version,
    is_accelerate_available,
    is_apex_available,
    is_bitsandbytes_available,
    is_bsa_available,
    is_coloredlogs_available,
    is_cython_available,
    is_datasets_available,
    is_decord_available,
    is_detectrona_available,
    is_faiss_available,
    is_flax_available,
    is_ftfy_available,
    is_in_notebook,
    is_ipex_available,
    is_jieba_available,
    is_jumanpp_available,
    is_kenlm_available,
    is_keras_nlp_available,
    is_librosa_available,
    is_natten_available,
    is_ninja_available,
    is_onnx_available,
    is_openai_available,
    is_optimum_available,
    is_pandas_available,
    is_peft_available,
    is_phonemizer_available,
    is_protobuf_available,
    is_psutil_available,
    is_pyanvml_available,
    is_pyctcdecode_available,
    is_pytesseract_available,
    is_pytest_available,
    is_pytorch_quantization_available,
    is_rjieba_available,
    is_sacremoses_available,
    is_safetensors_available,
    is_sagemaker_dp_enabled,
    is_sagemaker_mp_enabled,
    is_scipy_available,
    is_sentencepiece_available,
    is_seqio_available,
    is_sklearn_available,
    is_soundfile_availble,
    is_spacy_available,
    is_speech_available,
    is_sudachi_available,
    is_tensorflow_probability_available,
    is_tensorflow_text_available,
    is_tfaonnx_available,
    is_tf_available,
    is_timm_available,
    is_tokenizers_available,
    is_torch_available,
    is_torch_bfaa_available,
    is_torch_bfaa_cpu_available,
    is_torch_bfaa_gpu_available,
    is_torch_compile_available,
    is_torch_cuda_available,
    is_torch_fx_available,
    is_torch_fx_proxy,
    is_torch_mps_available,
    is_torch_neuroncore_available,
    is_torch_tensorrt_fx_available,
    is_torch_tfaa_available,
    is_torch_tpu_available,
    is_torchaudio_available,
    is_torchdistx_available,
    is_torchdynamo_available,
    is_torchvision_available,
    is_training_run_on_sagemaker,
    is_vision_available,
    requires_backends,
    torch_only_method,
)


__lowerCAmelCase        =							"pytorch_model.bin"
__lowerCAmelCase        =							"pytorch_model.bin.index.json"
__lowerCAmelCase        =							"adapter_config.json"
__lowerCAmelCase        =							"adapter_model.bin"
__lowerCAmelCase        =							"adapter_model.safetensors"
__lowerCAmelCase        =							"tf_model.h5"
__lowerCAmelCase        =							"tf_model.h5.index.json"
__lowerCAmelCase        =							"model.ckpt"
__lowerCAmelCase        =							"flax_model.msgpack"
__lowerCAmelCase        =							"flax_model.msgpack.index.json"
__lowerCAmelCase        =							"model.safetensors"
__lowerCAmelCase        =							"model.safetensors.index.json"
__lowerCAmelCase        =							"config.json"
__lowerCAmelCase        =							"preprocessor_config.json"
__lowerCAmelCase        =							FEATURE_EXTRACTOR_NAME
__lowerCAmelCase        =							"generation_config.json"
__lowerCAmelCase        =							"modelcard.json"

__lowerCAmelCase        =							"▁"
__lowerCAmelCase        =							SENTENCEPIECE_UNDERLINE  # Kept for backward compatibility

__lowerCAmelCase        =							[
    [[0, 1, 0, 1], [1, 0, 0, 1]]
] * 2  # Needs to have 0s and 1s only since XLM uses it for langs too.
__lowerCAmelCase        =							[[7, 6, 0, 0, 1], [1, 2, 3, 0, 0], [0, 0, 0, 4, 5]]
__lowerCAmelCase        =							[[1, 1, 1, 1, 1], [1, 1, 1, 0, 0], [0, 0, 0, 1, 1]]


def       __UpperCamelCase					(      lowercase_       :  Optional[int]   ):
       """simple docstring"""




       if version.parse(lowercase_   ) < version.parse(lowercase_   ):
              if "dev" in min_version:
                     a_						     =			(
                         'This example requires a source install from HuggingFace Transformers (see '
                         '`https://huggingface.co/docs/transformers/installation#install-from-source`),'
                     )
              else:
                     a_						     =			F'This example requires a minimum version of {min_version},'
              error_message += F' but the version found is {__version__}.\n'
              raise ImportError(
                  error_message
                  + 'Check out https://github.com/huggingface/transformers/tree/main/examples#important-note for the examples corresponding to other '
                  'versions of HuggingFace Transformers.'   )




