

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    :Optional[Any]      =			logging.get_logger(__name__)  # pylint: disable=invalid-name

class   _a					(			snake_case_				):




      """simple docstring"""

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





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





      def 		__A	(   self							:				Any	):
            self.enable_attention_slicing(UpperCAmelCase	)







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

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

            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(UpperCAmelCase						,			UpperCAmelCase	) or callback_steps <= 0)
            ):
                  raise ValueError(
                      f'''`callback_steps` has to be a positive integer but is {callback_steps} of type'''
                      f''' {type(UpperCAmelCase	)}.'''	)

            # get prompt text embeddings
            A_						  =				self.tokenizer(
                UpperCAmelCase						,			padding="max_length"						,			max_length=self.tokenizer.model_max_length						,			return_tensors="pt"						,			)
            A_						  =				text_inputs.input_ids

            if text_input_ids.shape[-1] > self.tokenizer.model_max_length:
                  A_						  =				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}'''	)
                  A_						  =				text_input_ids[:, : self.tokenizer.model_max_length]

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

            # duplicate text embeddings for each generation per prompt, using mps friendly method
            A_  ,     A_  ,     A_						  =				text_embeddings.shape
            A_						  =				text_embeddings.repeat(1						,			UpperCAmelCase						,			1	)
            A_						  =				text_embeddings.view(bs_embed * num_images_per_prompt						,			UpperCAmelCase						,			-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.
            A_						  =				guidance_scale > 1.0
            # get unconditional embeddings for classifier free guidance
            if do_classifier_free_guidance:
                  A_						  =				42
                  if negative_prompt is None:
                        A_						  =				[""]
                  elif type(UpperCAmelCase	) is not type(UpperCAmelCase	):
                        raise TypeError(
                            f'''`negative_prompt` should be the same type to `prompt`, but got {type(UpperCAmelCase	)} !='''
                            f''' {type(UpperCAmelCase	)}.'''	)
                  elif isinstance(UpperCAmelCase						,			UpperCAmelCase	):
                        A_						  =				[negative_prompt]
                  elif batch_size != len(UpperCAmelCase	):
                        raise ValueError(
                            f'''`negative_prompt`: {negative_prompt} has batch size {len(UpperCAmelCase	)}, but `prompt`:'''
                            f''' {prompt} has batch size {batch_size}. Please make sure that passed `negative_prompt` matches'''
                            " the batch size of `prompt`."	)
                  else:
                        A_						  =				negative_prompt

                  A_						  =				text_input_ids.shape[-1]
                  A_						  =				self.tokenizer(
                      UpperCAmelCase						,			padding="max_length"						,			max_length=UpperCAmelCase						,			truncation=UpperCAmelCase						,			return_tensors="pt"						,			)
                  A_						  =				self.text_encoder(uncond_input.input_ids.to(self.device	)	)[0]

                  # duplicate unconditional embeddings for each generation per prompt, using mps friendly method
                  A_						  =				uncond_embeddings.shape[1]
                  A_						  =				uncond_embeddings.repeat(UpperCAmelCase						,			UpperCAmelCase						,			1	)
                  A_						  =				uncond_embeddings.view(batch_size * num_images_per_prompt						,			UpperCAmelCase						,			-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
                  A_						  =				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`.
            A_						  =				(batch_size * num_images_per_prompt, self.unet.config.in_channels, height // 8, width // 8)
            A_						  =				(batch_size * num_images_per_prompt, self.unet.config.in_channels, 64, 64)
            A_						  =				text_embeddings.dtype
            if latents is None:
                  if self.device.type == "mps":
                        # randn does not exist on mps
                        A_						  =				torch.randn(
                            UpperCAmelCase						,			generator=UpperCAmelCase						,			device="cpu"						,			dtype=UpperCAmelCase	).to(self.device	)
                        A_						  =				torch.randn(UpperCAmelCase						,			generator=UpperCAmelCase						,			device="cpu"						,			dtype=UpperCAmelCase	).to(
                            self.device	)
                  else:
                        A_						  =				torch.randn(
                            UpperCAmelCase						,			generator=UpperCAmelCase						,			device=self.device						,			dtype=UpperCAmelCase	)
                        A_						  =				torch.randn(UpperCAmelCase						,			generator=UpperCAmelCase						,			device=self.device						,			dtype=UpperCAmelCase	)
            else:
                  if latents_reference.shape != latents_shape:
                        raise ValueError(f'''Unexpected latents shape, got {latents.shape}, expected {latents_shape}'''	)
                  A_						  =				latents_reference.to(self.device	)
                  A_						  =				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
            A_						  =				(latents_shape[3] - latents_shape_reference[3]) // 2
            A_						  =				(latents_shape[2] - latents_shape_reference[2]) // 2
            A_						  =				latents_shape_reference[3] if dx >= 0 else latents_shape_reference[3] + 2 * dx
            A_						  =				latents_shape_reference[2] if dy >= 0 else latents_shape_reference[2] + 2 * dy
            A_						  =				0 if dx < 0 else dx
            A_						  =				0 if dy < 0 else dy
            A_						  =				max(-dx						,			0	)
            A_						  =				max(-dy						,			0	)
            # import pdb
            # pdb.set_trace()
            A_						  =				latents_reference[:, :, dy : dy + h, dx : dx + w]

            # set timesteps
            self.scheduler.set_timesteps(UpperCAmelCase	)

            # Some schedulers like PNDM have timesteps as arrays
            # It's more optimized to move all timesteps to correct device beforehand
            A_						  =				self.scheduler.timesteps.to(self.device	)

            # scale the initial noise by the standard deviation required by the scheduler
            A_						  =				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]
            A_						  =				"eta" in set(inspect.signature(self.scheduler.step	).parameters.keys()	)
            A_						  =				{}
            if accepts_eta:
                  A_						  =				eta

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

                  # predict the noise residual
                  A_						  =				self.unet(UpperCAmelCase						,			UpperCAmelCase						,			encoder_hidden_states=UpperCAmelCase	).sample

                  # perform guidance
                  if do_classifier_free_guidance:
                        A_  ,     A_						  =				noise_pred.chunk(2	)
                        A_						  =				noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

                  # compute the previous noisy sample x_t -> x_t-1
                  A_						  =				self.scheduler.step(UpperCAmelCase						,			UpperCAmelCase						,			UpperCAmelCase						,			**UpperCAmelCase	).prev_sample

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

            A_						  =				1 / 0.18_215 * latents
            A_						  =				self.vae.decode(UpperCAmelCase	).sample

            A_						  =				(image / 2 + 0.5).clamp(0						,			1	)

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

            if self.safety_checker is not None:
                  A_						  =				self.feature_extractor(self.numpy_to_pil(UpperCAmelCase	)						,			return_tensors="pt"	).to(
                      self.device	)
                  A_  ,     A_						  =				self.safety_checker(
                      images=UpperCAmelCase						,			clip_input=safety_checker_input.pixel_values.to(text_embeddings.dtype	)	)
            else:
                  A_						  =				None

            if output_type == "pil":
                  A_						  =				self.numpy_to_pil(UpperCAmelCase	)

            if not return_dict:
                  return (image, has_nsfw_concept)

            return StableDiffusionPipelineOutput(images=UpperCAmelCase						,			nsfw_content_detected=UpperCAmelCase	)




"""simple docstring"""







import json
from typing import List, Optional, Tuple

from tokenizers import normalizers

from ...tokenization_utils_fast import PreTrainedTokenizerFast
from ...utils import logging
from .tokenization_squeezebert import SqueezeBertTokenizer


a_							=    logging.get_logger(__name__)

a_							=    {"""vocab_file""": """vocab.txt""", """tokenizer_file""": """tokenizer.json"""}

a_							=    {
    """vocab_file""": {
        """squeezebert/squeezebert-uncased""": (
            """https://huggingface.co/squeezebert/squeezebert-uncased/resolve/main/vocab.txt"""
        ),
        """squeezebert/squeezebert-mnli""": """https://huggingface.co/squeezebert/squeezebert-mnli/resolve/main/vocab.txt""",
        """squeezebert/squeezebert-mnli-headless""": (
            """https://huggingface.co/squeezebert/squeezebert-mnli-headless/resolve/main/vocab.txt"""
        ),
    },
    """tokenizer_file""": {
        """squeezebert/squeezebert-uncased""": (
            """https://huggingface.co/squeezebert/squeezebert-uncased/resolve/main/tokenizer.json"""
        ),
        """squeezebert/squeezebert-mnli""": (
            """https://huggingface.co/squeezebert/squeezebert-mnli/resolve/main/tokenizer.json"""
        ),
        """squeezebert/squeezebert-mnli-headless""": (
            """https://huggingface.co/squeezebert/squeezebert-mnli-headless/resolve/main/tokenizer.json"""
        ),
    },
}

a_							=    {
    """squeezebert/squeezebert-uncased""": 5_12,
    """squeezebert/squeezebert-mnli""": 5_12,
    """squeezebert/squeezebert-mnli-headless""": 5_12,
}


a_							=    {
    """squeezebert/squeezebert-uncased""": {"""do_lower_case""": True},
    """squeezebert/squeezebert-mnli""": {"""do_lower_case""": True},
    """squeezebert/squeezebert-mnli-headless""": {"""do_lower_case""": True},
}







class 					A_(SCREAMING_SNAKE_CASE_      ):







			"""simple docstring"""



			a_       :			Union[str, Any]												=			VOCAB_FILES_NAMES
			a_       :			List[str]												=			PRETRAINED_VOCAB_FILES_MAP
			a_       :			Optional[Any]												=			PRETRAINED_INIT_CONFIGURATION
			a_       :			Tuple												=			PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
			a_       :			Optional[int]												=			SqueezeBertTokenizer
			def __init__(	self				,   A=None				,   A=None				,   A=True				,   A="[UNK]"				,   A="[SEP]"				,   A="[PAD]"				,   A="[CLS]"				,   A="[MASK]"				,   A=True				,   A=None				,   **A				,   ):
						super().__init__(
						    A				,   tokenizer_file=A				,   do_lower_case=A				,   unk_token=A				,   sep_token=A				,   pad_token=A				,   cls_token=A				,   mask_token=A				,   tokenize_chinese_chars=A				,   strip_accents=A				,   **A				,   )

						_lowerCamelCase					:					Union[str, Any]	       =    json.loads(self.backend_tokenizer.normalizer.__getstate__()							)
						if (
						    normalizer_state.get('lowercase'				,   A							) != do_lower_case
						    or normalizer_state.get('strip_accents'				,   A							) != strip_accents
						    or normalizer_state.get('handle_chinese_chars'				,   A							) != tokenize_chinese_chars
						):
									_lowerCamelCase					:					List[Any]	       =    getattr(A				,   normalizer_state.pop('type'							)							)
									_lowerCamelCase					:					Optional[Any]	       =    do_lower_case
									_lowerCamelCase					:					str	       =    strip_accents
									_lowerCamelCase					:					Any	       =    tokenize_chinese_chars
									_lowerCamelCase					:					int	       =    normalizer_class(**A							)

						_lowerCamelCase					:					Dict	       =    do_lower_case
			def    _lowerCAmelCase		(	self				,   A				,   A=None							):
						_lowerCamelCase					:					Dict	       =    [self.cls_token_id] + token_ids_a + [self.sep_token_id]

						if token_ids_a:
									output += token_ids_a + [self.sep_token_id]

						return output
			def    _lowerCAmelCase		(	self				,   A				,   A = None							):
						_lowerCamelCase					:					Union[str, Any]	       =    [self.sep_token_id]
						_lowerCamelCase					:					Optional[Any]	       =    [self.cls_token_id]
						if token_ids_a is None:
									return len(cls + token_ids_a + sep							) * [0]
						return len(cls + token_ids_a + sep							) * [0] + len(token_ids_a + sep							) * [1]

			def    _lowerCAmelCase		(	self				,   A				,   A = None							):
						_lowerCamelCase					:					str	       =    self._tokenizer.model.save(A				,   name=A							)
						return tuple(A							)





