

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

import torch
from transformers import (
    CLIPImageProcessor,
    CLIPTextModel,
    CLIPTokenizer,
    WhisperForConditionalGeneration,
    WhisperProcessor,
)

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


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






class 						__magic_name__    (       _snake_case    ):


 def __init__(						self		: Optional[Any]			,						lowerCAmelCase__		: WhisperForConditionalGeneration			,						lowerCAmelCase__		: WhisperProcessor			,						lowerCAmelCase__		: AutoencoderKL			,						lowerCAmelCase__		: CLIPTextModel			,						lowerCAmelCase__		: CLIPTokenizer			,						lowerCAmelCase__		: UNetaDConditionModel			,						lowerCAmelCase__		: Union[DDIMScheduler, PNDMScheduler, LMSDiscreteScheduler]			,						lowerCAmelCase__		: StableDiffusionSafetyChecker			,						lowerCAmelCase__		: CLIPImageProcessor			,						)    ->      int:
   super().__init__()

   if safety_checker is None:
     logger.warning(
         f"You have disabled the safety checker for {self.__class__} by passing `safety_checker=None`. Ensure"
         " that you abide to the conditions of the Stable Diffusion license and do not expose unfiltered"
         " results in services or applications open to the public. Both the diffusers team and Hugging Face"
         " strongly recommend to keep the safety filter enabled in all public facing circumstances, disabling"
         " it only for use-cases that involve analyzing network behavior or auditing its results. For more"
         " information, please have a look at https://github.com/huggingface/diffusers/pull/254 ."    )

   self.register_modules(
       speech_model=lowerCAmelCase__			,						speech_processor=lowerCAmelCase__			,						vae=lowerCAmelCase__			,						text_encoder=lowerCAmelCase__			,						tokenizer=lowerCAmelCase__			,						unet=lowerCAmelCase__			,						scheduler=lowerCAmelCase__			,						feature_extractor=lowerCAmelCase__			,						)


 def _UpperCamelCase	(						self		: List[Any]			,						lowerCAmelCase__		: Optional[Union[str, int]] = "auto"    )    ->      int:
   if slice_size == "auto":
     UpperCAmelCase					      =  self.unet.config.attention_head_dim // 2
   self.unet.set_attention_slice(lowerCAmelCase__    )


 def _UpperCamelCase	(						self		: Union[str, Any]    )    ->      Union[str, Any]:
   self.enable_attention_slicing(lowerCAmelCase__    )


 @torch.no_grad()
 def __call__(						self		: Dict			,						lowerCAmelCase__		: Optional[Any]			,						lowerCAmelCase__		: List[str]=1_6_0_0_0			,						lowerCAmelCase__		: int = 5_1_2			,						lowerCAmelCase__		: int = 5_1_2			,						lowerCAmelCase__		: int = 5_0			,						lowerCAmelCase__		: float = 7.5			,						lowerCAmelCase__		: Optional[Union[str, List[str]]] = None			,						lowerCAmelCase__		: Optional[int] = 1			,						lowerCAmelCase__		: float = 0.0			,						lowerCAmelCase__		: Optional[torch.Generator] = None			,						lowerCAmelCase__		: Optional[torch.FloatTensor] = None			,						lowerCAmelCase__		: Optional[str] = "pil"			,						lowerCAmelCase__		: bool = True			,						lowerCAmelCase__		: Optional[Callable[[int, int, torch.FloatTensor], None]] = None			,						lowerCAmelCase__		: int = 1			,						**lowerCAmelCase__		: Any			,						)    ->      Optional[Any]:
   UpperCAmelCase					      =  self.speech_processor.feature_extractor(
       lowerCAmelCase__			,						return_tensors="pt"			,						sampling_rate=lowerCAmelCase__    ).input_features.to(self.device    )
   UpperCAmelCase					      =  self.speech_model.generate(lowerCAmelCase__			,						max_length=4_8_0_0_0_0    )

   UpperCAmelCase					      =  self.speech_processor.tokenizer.batch_decode(lowerCAmelCase__			,						skip_special_tokens=lowerCAmelCase__			,						normalize=lowerCAmelCase__    )[
       0
   ]

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

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

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

   if text_input_ids.shape[-1] > self.tokenizer.model_max_length:
     UpperCAmelCase					      =  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}"    )
     UpperCAmelCase					      =  text_input_ids[:, : self.tokenizer.model_max_length]
   UpperCAmelCase					      =  self.text_encoder(text_input_ids.to(self.device    )    )[0]

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

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

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

   # set timesteps
   self.scheduler.set_timesteps(lowerCAmelCase__    )

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

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

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

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

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

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

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

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

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

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

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

   if not return_dict:
     return image

   return StableDiffusionPipelineOutput(images=lowerCAmelCase__			,						nsfw_content_detected=lowerCAmelCase__    )






import json
import os
import unittest

from transformers import BatchEncoding, LEDTokenizer, LEDTokenizerFast
from transformers.models.led.tokenization_led import VOCAB_FILES_NAMES
from transformers.testing_utils import require_tokenizers, require_torch
from transformers.utils import cached_property

from ...test_tokenization_common import TokenizerTesterMixin






@require_tokenizers
class 						__magic_name__    (       _snake_case	,						unittest.TestCase    ):
 UpperCAmelCase							    =						LEDTokenizer
 UpperCAmelCase							    =						LEDTokenizerFast
 UpperCAmelCase							    =						True


 def _UpperCamelCase	(						self		: Tuple    )    ->      Union[str, Any]:
   super().setUp()
   UpperCAmelCase					      =  [
       "l",
       "o",
       "w",
       "e",
       "r",
       "s",
       "t",
       "i",
       "d",
       "n",
       "\u0120",
       "\u0120l",
       "\u0120n",
       "\u0120lo",
       "\u0120low",
       "er",
       "\u0120lowest",
       "\u0120newer",
       "\u0120wider",
       "<unk>",
   ]
   UpperCAmelCase					      =  dict(zip(lowerCAmelCase__			,						range(len(lowerCAmelCase__    )    )    )    )
   UpperCAmelCase					      =  ["#version: 0.2", "\u0120 l", "\u0120l o", "\u0120lo w", "e r", ""]
   UpperCAmelCase					      =  {"unk_token": "<unk>"}

   UpperCAmelCase					      =  os.path.join(self.tmpdirname			,						VOCAB_FILES_NAMES["vocab_file"]    )
   UpperCAmelCase					      =  os.path.join(self.tmpdirname			,						VOCAB_FILES_NAMES["merges_file"]    )
   with open(self.vocab_file			,						"w"			,						encoding="utf-8"    ) as fp:
     fp.write(json.dumps(lowerCAmelCase__    ) + "\n"    )
   with open(self.merges_file			,						"w"			,						encoding="utf-8"    ) as fp:
     fp.write("\n".join(lowerCAmelCase__    )    )


 def _UpperCamelCase	(						self		: Union[str, Any]			,						**lowerCAmelCase__		: Optional[int]    )    ->      Optional[int]:
   kwargs.update(self.special_tokens_map    )
   return self.tokenizer_class.from_pretrained(self.tmpdirname			,						**lowerCAmelCase__    )


 def _UpperCamelCase	(						self		: str			,						**lowerCAmelCase__		: str    )    ->      Optional[int]:
   kwargs.update(self.special_tokens_map    )
   return self.rust_tokenizer_class.from_pretrained(self.tmpdirname			,						**lowerCAmelCase__    )


 def _UpperCamelCase	(						self		: List[str]			,						lowerCAmelCase__		: List[Any]    )    ->      List[Any]:
   return "lower newer", "lower newer"


 @cached_property
 def _UpperCamelCase	(						self		: Dict    )    ->      str:
   return LEDTokenizer.from_pretrained("allenai/led-base-16384"    )


 @cached_property
 def _UpperCamelCase	(						self		: int    )    ->      Tuple:
   return LEDTokenizerFast.from_pretrained("allenai/led-base-16384"    )


 @require_torch
 def _UpperCamelCase	(						self		: Tuple    )    ->      List[str]:
   UpperCAmelCase					      =  ["A long paragraph for summarization.", "Another paragraph for summarization."]
   UpperCAmelCase					      =  [0, 2_5_0, 2_5_1, 1_7_8_1_8, 1_3, 3_9_1_8_6, 1_9_3_8, 4, 2]

   for tokenizer in [self.default_tokenizer, self.default_tokenizer_fast]:
     UpperCAmelCase					      =  tokenizer(lowerCAmelCase__			,						max_length=len(lowerCAmelCase__    )			,						padding=lowerCAmelCase__			,						return_tensors="pt"    )
     self.assertIsInstance(lowerCAmelCase__			,						lowerCAmelCase__    )

     self.assertEqual((2, 9)			,						batch.input_ids.shape    )
     self.assertEqual((2, 9)			,						batch.attention_mask.shape    )
     UpperCAmelCase					      =  batch.input_ids.tolist()[0]
     self.assertListEqual(lowerCAmelCase__			,						lowerCAmelCase__    )


 @require_torch
 def _UpperCamelCase	(						self		: Union[str, Any]    )    ->      List[Any]:
   UpperCAmelCase					      =  ["A long paragraph for summarization.", "Another paragraph for summarization."]
   for tokenizer in [self.default_tokenizer, self.default_tokenizer_fast]:
     UpperCAmelCase					      =  tokenizer(lowerCAmelCase__			,						padding=lowerCAmelCase__			,						return_tensors="pt"    )
     self.assertIn("input_ids"			,						lowerCAmelCase__    )
     self.assertIn("attention_mask"			,						lowerCAmelCase__    )
     self.assertNotIn("labels"			,						lowerCAmelCase__    )
     self.assertNotIn("decoder_attention_mask"			,						lowerCAmelCase__    )


 @require_torch
 def _UpperCamelCase	(						self		: int    )    ->      int:
   UpperCAmelCase					      =  [
       "Summary of the text.",
       "Another summary.",
   ]
   for tokenizer in [self.default_tokenizer, self.default_tokenizer_fast]:
     UpperCAmelCase					      =  tokenizer(text_target=lowerCAmelCase__			,						max_length=3_2			,						padding="max_length"			,						return_tensors="pt"    )
     self.assertEqual(3_2			,						targets["input_ids"].shape[1]    )


 @require_torch
 def _UpperCamelCase	(						self		: Any    )    ->      int:
   for tokenizer in [self.default_tokenizer, self.default_tokenizer_fast]:
     UpperCAmelCase					      =  tokenizer(
         ["I am a small frog" * 1_0_2_4, "I am a small frog"]			,						padding=lowerCAmelCase__			,						truncation=lowerCAmelCase__			,						return_tensors="pt"    )
     self.assertIsInstance(lowerCAmelCase__			,						lowerCAmelCase__    )
     self.assertEqual(batch.input_ids.shape			,						(2, 5_1_2_2)    )


 @require_torch
 def _UpperCamelCase	(						self		: Dict    )    ->      Tuple:
   UpperCAmelCase					      =  ["A long paragraph for summarization."]
   UpperCAmelCase					      =  [
       "Summary of the text.",
   ]
   for tokenizer in [self.default_tokenizer, self.default_tokenizer_fast]:
     UpperCAmelCase					      =  tokenizer(lowerCAmelCase__			,						return_tensors="pt"    )
     UpperCAmelCase					      =  tokenizer(text_target=lowerCAmelCase__			,						return_tensors="pt"    )
     UpperCAmelCase					      =  inputs["input_ids"]
     UpperCAmelCase					      =  targets["input_ids"]
     self.assertTrue((input_ids[:, 0] == tokenizer.bos_token_id).all().item()    )
     self.assertTrue((labels[:, 0] == tokenizer.bos_token_id).all().item()    )
     self.assertTrue((input_ids[:, -1] == tokenizer.eos_token_id).all().item()    )
     self.assertTrue((labels[:, -1] == tokenizer.eos_token_id).all().item()    )


 @require_torch
 def _UpperCamelCase	(						self		: Optional[int]    )    ->      Optional[Any]:
   for tokenizer in [self.default_tokenizer, self.default_tokenizer_fast]:
     UpperCAmelCase					      =  ["Summary of the text.", "Another summary."]
     UpperCAmelCase					      =  [[0, 0, 0, 0, 0, 0, 0], [0, 0, 0, 0, 0, -1, -1]]

     UpperCAmelCase					      =  tokenizer(lowerCAmelCase__			,						padding=lowerCAmelCase__    )
     UpperCAmelCase					      =  [[0] * len(lowerCAmelCase__    ) for x in encoded_output["input_ids"]]
     UpperCAmelCase					      =  tokenizer.pad(lowerCAmelCase__    )
     self.assertSequenceEqual(outputs["global_attention_mask"]			,						lowerCAmelCase__    )


 def _UpperCamelCase	(						self		: List[str]    )    ->      int:
   pass


 def _UpperCamelCase	(						self		: Optional[Any]    )    ->      Union[str, Any]:
   for tokenizer, pretrained_name, kwargs in self.tokenizers_list:
     with self.subTest(f"{tokenizer.__class__.__name__} ({pretrained_name})"    ):
       UpperCAmelCase					      =  self.rust_tokenizer_class.from_pretrained(lowerCAmelCase__			,						**lowerCAmelCase__    )
       UpperCAmelCase					      =  self.tokenizer_class.from_pretrained(lowerCAmelCase__			,						**lowerCAmelCase__    )
       UpperCAmelCase					      =  "A, <mask> AllenNLP sentence."
       UpperCAmelCase					      =  tokenizer_r.encode_plus(lowerCAmelCase__			,						add_special_tokens=lowerCAmelCase__			,						return_token_type_ids=lowerCAmelCase__    )
       UpperCAmelCase					      =  tokenizer_p.encode_plus(lowerCAmelCase__			,						add_special_tokens=lowerCAmelCase__			,						return_token_type_ids=lowerCAmelCase__    )
       self.assertEqual(sum(tokens_r["token_type_ids"]    )			,						sum(tokens_p["token_type_ids"]    )    )
       self.assertEqual(
           sum(tokens_r["attention_mask"]    ) / len(tokens_r["attention_mask"]    )			,						sum(tokens_p["attention_mask"]    ) / len(tokens_p["attention_mask"]    )			,						)

       UpperCAmelCase					      =  tokenizer_r.convert_ids_to_tokens(tokens_r["input_ids"]    )
       UpperCAmelCase					      =  tokenizer_p.convert_ids_to_tokens(tokens_p["input_ids"]    )
       self.assertSequenceEqual(tokens_p["input_ids"]			,						[0, 2_5_0, 6, 5_0_2_6_4, 3_8_2_3, 4_8_7, 2_1_9_9_2, 3_6_4_5, 4, 2]    )
       self.assertSequenceEqual(tokens_r["input_ids"]			,						[0, 2_5_0, 6, 5_0_2_6_4, 3_8_2_3, 4_8_7, 2_1_9_9_2, 3_6_4_5, 4, 2]    )

       self.assertSequenceEqual(
           lowerCAmelCase__			,						["<s>", "A", ",", "<mask>", "ĠAllen", "N", "LP", "Ġsentence", ".", "</s>"]    )
       self.assertSequenceEqual(
           lowerCAmelCase__			,						["<s>", "A", ",", "<mask>", "ĠAllen", "N", "LP", "Ġsentence", ".", "</s>"]    )



