






from logging import getLogger
from typing import Any, Callable, List, Optional, Union

import numpy as np
import PIL
import torch

from ...schedulers import DDPMScheduler
from ..onnx_utils import ORT_TO_NP_TYPE, OnnxRuntimeModel
from ..pipeline_utils import ImagePipelineOutput
from . import StableDiffusionUpscalePipeline


snake_case			 =						getLogger(__name__)


snake_case			 =						4
snake_case			 =						7

snake_case			 =						{
    """float16""": torch.floataa,
    """float32""": torch.floataa,
}


def      lowerCamelCase__	(   lowercase ):

  """simple docstring"""






  if isinstance(lowercase  ,   torch.Tensor ):
    return image
  elif isinstance(lowercase  ,   PIL.Image.Image ):
    SCREAMING_SNAKE_CASE				:					Optional[int]		  =			[image]

  if isinstance(image[0]  ,   PIL.Image.Image ):
    SCREAMING_SNAKE_CASE  ,     SCREAMING_SNAKE_CASE				:					int		  =			image[0].size
    SCREAMING_SNAKE_CASE  ,     SCREAMING_SNAKE_CASE				:					Tuple		  =			(x - x % 64 for x in (w, h))  # resize to integer multiple of 32

    SCREAMING_SNAKE_CASE				:					Tuple		  =			[np.array(i.resize((w, h) ) )[None, :] for i in image]
    SCREAMING_SNAKE_CASE				:					List[str]		  =			np.concatenate(lowercase  ,   axis=0 )
    SCREAMING_SNAKE_CASE				:					Any		  =			np.array(lowercase ).astype(np.floataa ) / 255.0
    SCREAMING_SNAKE_CASE				:					Dict		  =			image.transpose(0  ,   3  ,   1  ,   2 )
    SCREAMING_SNAKE_CASE				:					Dict		  =			2.0 * image - 1.0
    SCREAMING_SNAKE_CASE				:					Tuple		  =			torch.from_numpy(lowercase )
  elif isinstance(image[0]  ,   torch.Tensor ):
    SCREAMING_SNAKE_CASE				:					Union[str, Any]		  =			torch.cat(lowercase  ,   dim=0 )

  return image


class        SCREAMING_SNAKE_CASE	(	lowerCAmelCase    ):
     '''simple docstring'''


     def __init__( self				:		int    ,       UpperCAmelCase_				:		OnnxRuntimeModel    ,       UpperCAmelCase_				:		OnnxRuntimeModel    ,       UpperCAmelCase_				:		Any    ,       UpperCAmelCase_				:		OnnxRuntimeModel    ,       UpperCAmelCase_				:		DDPMScheduler    ,       UpperCAmelCase_				:		Any    ,       UpperCAmelCase_				:		int = 350    ,       ):
       super().__init__(
           vae=UpperCAmelCase_    ,       text_encoder=UpperCAmelCase_    ,       tokenizer=UpperCAmelCase_    ,       unet=UpperCAmelCase_    ,       low_res_scheduler=UpperCAmelCase_    ,       scheduler=UpperCAmelCase_    ,       safety_checker=UpperCAmelCase_    ,       feature_extractor=UpperCAmelCase_    ,       watermarker=UpperCAmelCase_    ,       max_noise_level=UpperCAmelCase_    ,       )


     def __call__( self				:		str    ,       UpperCAmelCase_				:		Union[str, List[str]]    ,       UpperCAmelCase_				:		Union[torch.FloatTensor, PIL.Image.Image, List[PIL.Image.Image]]    ,       UpperCAmelCase_				:		int = 75    ,       UpperCAmelCase_				:		float = 9.0    ,       UpperCAmelCase_				:		int = 20    ,       UpperCAmelCase_				:		Optional[Union[str, List[str]]] = None    ,       UpperCAmelCase_				:		Optional[int] = 1    ,       UpperCAmelCase_				:		float = 0.0    ,       UpperCAmelCase_				:		Optional[Union[torch.Generator, List[torch.Generator]]] = None    ,       UpperCAmelCase_				:		Optional[torch.FloatTensor] = None    ,       UpperCAmelCase_				:		Optional[np.ndarray] = None    ,       UpperCAmelCase_				:		Optional[np.ndarray] = None    ,       UpperCAmelCase_				:		Optional[str] = "pil"    ,       UpperCAmelCase_				:		bool = True    ,       UpperCAmelCase_				:		Optional[Callable[[int, int, torch.FloatTensor], None]] = None    ,       UpperCAmelCase_				:		Optional[int] = 1    ,       ):
       self.check_inputs(UpperCAmelCase_    ,       UpperCAmelCase_    ,       UpperCAmelCase_    ,       UpperCAmelCase_					)

       # 2. Define call parameters
       if prompt is not None and isinstance(UpperCAmelCase_    ,       UpperCAmelCase_					):
         SCREAMING_SNAKE_CASE				:					str		  =			1
       elif prompt is not None and isinstance(UpperCAmelCase_    ,       UpperCAmelCase_					):
         SCREAMING_SNAKE_CASE				:					Optional[Any]		  =			len(UpperCAmelCase_					)
       else:
         SCREAMING_SNAKE_CASE				:					List[Any]		  =			prompt_embeds.shape[0]

       SCREAMING_SNAKE_CASE				:					Dict		  =			self._execution_device
       # 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				:					Tuple		  =			guidance_scale > 1.0

       # 3. Encode input prompt
       SCREAMING_SNAKE_CASE				:					Tuple		  =			self._encode_prompt(
           UpperCAmelCase_    ,       UpperCAmelCase_    ,       UpperCAmelCase_    ,       UpperCAmelCase_    ,       UpperCAmelCase_    ,       prompt_embeds=UpperCAmelCase_    ,       negative_prompt_embeds=UpperCAmelCase_    ,       )

       SCREAMING_SNAKE_CASE				:					str		  =			ORT_TO_PT_TYPE[str(text_embeddings.dtype					)]

       # 4. Preprocess image
       SCREAMING_SNAKE_CASE				:					List[Any]		  =			preprocess(UpperCAmelCase_					)
       SCREAMING_SNAKE_CASE				:					Union[str, Any]		  =			image.cpu()

       # 5. set timesteps
       self.scheduler.set_timesteps(UpperCAmelCase_    ,       device=UpperCAmelCase_					)
       SCREAMING_SNAKE_CASE				:					Union[str, Any]		  =			self.scheduler.timesteps

       # 5. Add noise to image
       SCREAMING_SNAKE_CASE				:					Union[str, Any]		  =			torch.tensor([noise_level]    ,       dtype=torch.long    ,       device=UpperCAmelCase_					)
       SCREAMING_SNAKE_CASE				:					int		  =			torch.randn(image.shape    ,       generator=UpperCAmelCase_    ,       device=UpperCAmelCase_    ,       dtype=UpperCAmelCase_					)
       SCREAMING_SNAKE_CASE				:					Optional[int]		  =			self.low_res_scheduler.add_noise(UpperCAmelCase_    ,       UpperCAmelCase_    ,       UpperCAmelCase_					)

       SCREAMING_SNAKE_CASE				:					Dict		  =			2 if do_classifier_free_guidance else 1
       SCREAMING_SNAKE_CASE				:					Optional[int]		  =			np.concatenate([image] * batch_multiplier * num_images_per_prompt					)
       SCREAMING_SNAKE_CASE				:					Any		  =			np.concatenate([noise_level] * image.shape[0]					)

       # 6. Prepare latent variables
       SCREAMING_SNAKE_CASE  ,     SCREAMING_SNAKE_CASE				:					List[str]		  =			image.shape[2:]
       SCREAMING_SNAKE_CASE				:					List[Any]		  =			self.prepare_latents(
           batch_size * num_images_per_prompt    ,       UpperCAmelCase_    ,       UpperCAmelCase_    ,       UpperCAmelCase_    ,       UpperCAmelCase_    ,       UpperCAmelCase_    ,       UpperCAmelCase_    ,       UpperCAmelCase_    ,       )

       # 7. Check that sizes of image and latents match
       SCREAMING_SNAKE_CASE				:					List[Any]		  =			image.shape[1]
       if NUM_LATENT_CHANNELS + num_channels_image != NUM_UNET_INPUT_CHANNELS:
         raise ValueError(
             "Incorrect configuration settings! The config of `pipeline.unet` expects"
             f''' {NUM_UNET_INPUT_CHANNELS} but received `num_channels_latents`: {NUM_LATENT_CHANNELS} +'''
             f''' `num_channels_image`: {num_channels_image} '''
             f''' = {NUM_LATENT_CHANNELS+num_channels_image}. Please verify the config of'''
             " `pipeline.unet` or your `image` input."					)

       # 8. Prepare extra step kwargs. TODO: Logic should ideally just be moved out of the pipeline
       SCREAMING_SNAKE_CASE				:					Optional[int]		  =			self.prepare_extra_step_kwargs(UpperCAmelCase_    ,       UpperCAmelCase_					)

       SCREAMING_SNAKE_CASE				:					Optional[Any]		  =			next(
           (input.type for input in self.unet.model.get_inputs() if input.name == "timestep")    ,       "tensor(float)"					)
       SCREAMING_SNAKE_CASE				:					Union[str, Any]		  =			ORT_TO_NP_TYPE[timestep_dtype]

       # 9. Denoising loop
       SCREAMING_SNAKE_CASE				:					Any		  =			len(UpperCAmelCase_					) - num_inference_steps * self.scheduler.order
       with self.progress_bar(total=UpperCAmelCase_					) as progress_bar:
         for i, t in enumerate(UpperCAmelCase_					):
           # expand the latents if we are doing classifier free guidance
           SCREAMING_SNAKE_CASE				:					Any		  =			np.concatenate([latents] * 2					) if do_classifier_free_guidance else latents

           # concat latents, mask, masked_image_latents in the channel dimension
           SCREAMING_SNAKE_CASE				:					str		  =			self.scheduler.scale_model_input(UpperCAmelCase_    ,       UpperCAmelCase_					)
           SCREAMING_SNAKE_CASE				:					str		  =			np.concatenate([latent_model_input, image]    ,       axis=1					)

           # timestep to tensor
           SCREAMING_SNAKE_CASE				:					Dict		  =			np.array([t]    ,       dtype=UpperCAmelCase_					)

           # predict the noise residual
           SCREAMING_SNAKE_CASE				:					List[Any]		  =			self.unet(
               sample=UpperCAmelCase_    ,       timestep=UpperCAmelCase_    ,       encoder_hidden_states=UpperCAmelCase_    ,       class_labels=noise_level.astype(np.intaa					)    ,       )[0]

           # perform guidance
           if do_classifier_free_guidance:
             SCREAMING_SNAKE_CASE  ,     SCREAMING_SNAKE_CASE				:					Union[str, Any]		  =			np.split(UpperCAmelCase_    ,       2					)
           SCREAMING_SNAKE_CASE				:					Union[str, Any]		  =			noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

           # compute the previous noisy sample x_t -> x_t-1
           SCREAMING_SNAKE_CASE				:					Union[str, Any]		  =			self.scheduler.step(
               torch.from_numpy(UpperCAmelCase_					)    ,       UpperCAmelCase_    ,       UpperCAmelCase_    ,       **UpperCAmelCase_					).prev_sample

           # call the callback, if provided
           if i == len(UpperCAmelCase_					) - 1 or ((i + 1) > num_warmup_steps and (i + 1) % self.scheduler.order == 0):
             progress_bar.update()
             if callback is not None and i % callback_steps == 0:
               callback(UpperCAmelCase_    ,       UpperCAmelCase_    ,       UpperCAmelCase_					)

        # 10. Post-processing
       SCREAMING_SNAKE_CASE				:					Optional[Any]		  =			self.decode_latents(latents.float()					)

       # 11. Convert to PIL
       if output_type == "pil":
         SCREAMING_SNAKE_CASE				:					str		  =			self.numpy_to_pil(UpperCAmelCase_					)

       if not return_dict:
         return (image,)

       return ImagePipelineOutput(images=UpperCAmelCase_					)


     def  _A				( self				:		List[Any]    ,       UpperCAmelCase_				:		Dict					):
       SCREAMING_SNAKE_CASE				:					Dict		  =			1 / 0.08_333 * latents
       SCREAMING_SNAKE_CASE				:					List[Any]		  =			self.vae(latent_sample=UpperCAmelCase_					)[0]
       SCREAMING_SNAKE_CASE				:					Optional[Any]		  =			np.clip(image / 2 + 0.5    ,       0    ,       1					)
       SCREAMING_SNAKE_CASE				:					Tuple		  =			image.transpose((0, 2, 3, 1)					)
       return image






     def  _A				( self				:		Any    ,       UpperCAmelCase_				:		Union[str, List[str]]    ,       UpperCAmelCase_				:		Any    ,       UpperCAmelCase_				:		Optional[int]    ,       UpperCAmelCase_				:		bool    ,       UpperCAmelCase_				:		Optional[str]    ,       UpperCAmelCase_				:		Optional[torch.FloatTensor] = None    ,       UpperCAmelCase_				:		Optional[torch.FloatTensor] = None    ,       ):
       if prompt is not None and isinstance(UpperCAmelCase_    ,       UpperCAmelCase_					):
         SCREAMING_SNAKE_CASE				:					Optional[int]		  =			1
       elif prompt is not None and isinstance(UpperCAmelCase_    ,       UpperCAmelCase_					):
         SCREAMING_SNAKE_CASE				:					str		  =			len(UpperCAmelCase_					)
       else:
         SCREAMING_SNAKE_CASE				:					Optional[int]		  =			prompt_embeds.shape[0]

       if prompt_embeds is None:
         SCREAMING_SNAKE_CASE				:					int		  =			self.tokenizer(
             UpperCAmelCase_    ,       padding="max_length"    ,       max_length=self.tokenizer.model_max_length    ,       truncation=UpperCAmelCase_    ,       return_tensors="pt"    ,       )
         SCREAMING_SNAKE_CASE				:					Tuple		  =			text_inputs.input_ids
         SCREAMING_SNAKE_CASE				:					List[Any]		  =			self.tokenizer(UpperCAmelCase_    ,       padding="longest"    ,       return_tensors="pt"					).input_ids

         if untruncated_ids.shape[-1] >= text_input_ids.shape[-1] and not torch.equal(
             UpperCAmelCase_    ,       UpperCAmelCase_					):
           SCREAMING_SNAKE_CASE				:					Any		  =			self.tokenizer.batch_decode(
               untruncated_ids[:, self.tokenizer.model_max_length - 1 : -1]					)
           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}'''					)

         # no positional arguments to text_encoder
         SCREAMING_SNAKE_CASE				:					Tuple		  =			self.text_encoder(
             input_ids=text_input_ids.int().to(UpperCAmelCase_					)    ,       )
         SCREAMING_SNAKE_CASE				:					Optional[Any]		  =			prompt_embeds[0]

       SCREAMING_SNAKE_CASE  ,     SCREAMING_SNAKE_CASE  ,     SCREAMING_SNAKE_CASE				:					List[Any]		  =			prompt_embeds.shape
       # duplicate text embeddings for each generation per prompt, using mps friendly method
       SCREAMING_SNAKE_CASE				:					List[Any]		  =			prompt_embeds.repeat(1    ,       UpperCAmelCase_					)
       SCREAMING_SNAKE_CASE				:					str		  =			prompt_embeds.reshape(bs_embed * num_images_per_prompt    ,       UpperCAmelCase_    ,       -1					)

       # get unconditional embeddings for classifier free guidance
       if do_classifier_free_guidance and negative_prompt_embeds is None:
         SCREAMING_SNAKE_CASE				:					List[str]
         if negative_prompt is None:
           SCREAMING_SNAKE_CASE				:					List[Any]		  =			[""] * batch_size
         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_					):
           SCREAMING_SNAKE_CASE				:					Optional[int]		  =			[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:
           SCREAMING_SNAKE_CASE				:					Tuple		  =			negative_prompt

         SCREAMING_SNAKE_CASE				:					int		  =			text_input_ids.shape[-1]
         SCREAMING_SNAKE_CASE				:					int		  =			self.tokenizer(
             UpperCAmelCase_    ,       padding="max_length"    ,       max_length=UpperCAmelCase_    ,       truncation=UpperCAmelCase_    ,       return_tensors="pt"    ,       )

         # if hasattr(uncond_input, "attention_mask"):
         #     attention_mask = uncond_input.attention_mask.to(device)
         # else:
         #     attention_mask = None

         SCREAMING_SNAKE_CASE				:					Any		  =			self.text_encoder(
             input_ids=uncond_input.input_ids.int().to(UpperCAmelCase_					)    ,       )
         SCREAMING_SNAKE_CASE				:					Dict		  =			uncond_embeddings[0]

       if do_classifier_free_guidance:
         SCREAMING_SNAKE_CASE				:					int		  =			uncond_embeddings.shape[1]
         # duplicate unconditional embeddings for each generation per prompt, using mps friendly method
         SCREAMING_SNAKE_CASE				:					Tuple		  =			uncond_embeddings.repeat(1    ,       UpperCAmelCase_					)
         SCREAMING_SNAKE_CASE				:					Tuple		  =			uncond_embeddings.reshape(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
         SCREAMING_SNAKE_CASE				:					Tuple		  =			np.concatenate([uncond_embeddings, prompt_embeds]					)

       return prompt_embeds














import unittest

from transformers import SpeechaTextaConfig
from transformers.testing_utils import is_torch_available, require_torch, torch_device

from ...generation.test_utils import GenerationTesterMixin
from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, ids_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
   import torch

   from transformers.models.speech_to_text_a.modeling_speech_to_text_a import (
       SpeechaTextaDecoder,
       SpeechaTextaForCausalLM,
   )


@require_torch
class        SCREAMING_SNAKE_CASE	:
     '''simple docstring'''


     def __init__( self				:		List[str]    ,       UpperCAmelCase_				:		Any    ,       UpperCAmelCase_				:		List[Any]=99    ,       UpperCAmelCase_				:		Tuple=13    ,       UpperCAmelCase_				:		Dict=16    ,       UpperCAmelCase_				:		int=7    ,       UpperCAmelCase_				:		Optional[Any]=True    ,       UpperCAmelCase_				:		List[Any]=True    ,       UpperCAmelCase_				:		Any=True    ,       UpperCAmelCase_				:		List[str]=False    ,       UpperCAmelCase_				:		Tuple=True    ,       UpperCAmelCase_				:		Tuple=2    ,       UpperCAmelCase_				:		List[str]=32    ,       UpperCAmelCase_				:		Union[str, Any]=4    ,       UpperCAmelCase_				:		Any=4    ,       UpperCAmelCase_				:		Optional[int]=30    ,       UpperCAmelCase_				:		Union[str, Any]=0    ,       UpperCAmelCase_				:		Union[str, Any]=1    ,       UpperCAmelCase_				:		Any=2    ,       UpperCAmelCase_				:		Dict=None    ,       ):
       SCREAMING_SNAKE_CASE				:					Any		  =			parent
       SCREAMING_SNAKE_CASE				:					Tuple		  =			batch_size
       SCREAMING_SNAKE_CASE				:					str		  =			decoder_seq_length
       # For common tests
       SCREAMING_SNAKE_CASE				:					Tuple		  =			self.decoder_seq_length
       SCREAMING_SNAKE_CASE				:					str		  =			is_training
       SCREAMING_SNAKE_CASE				:					Dict		  =			use_attention_mask
       SCREAMING_SNAKE_CASE				:					int		  =			use_labels

       SCREAMING_SNAKE_CASE				:					Tuple		  =			vocab_size
       SCREAMING_SNAKE_CASE				:					int		  =			d_model
       SCREAMING_SNAKE_CASE				:					List[str]		  =			d_model
       SCREAMING_SNAKE_CASE				:					int		  =			decoder_layers
       SCREAMING_SNAKE_CASE				:					List[Any]		  =			decoder_layers
       SCREAMING_SNAKE_CASE				:					List[Any]		  =			decoder_ffn_dim
       SCREAMING_SNAKE_CASE				:					Tuple		  =			decoder_attention_heads
       SCREAMING_SNAKE_CASE				:					Dict		  =			decoder_attention_heads
       SCREAMING_SNAKE_CASE				:					Optional[int]		  =			eos_token_id
       SCREAMING_SNAKE_CASE				:					List[Any]		  =			bos_token_id
       SCREAMING_SNAKE_CASE				:					List[Any]		  =			pad_token_id
       SCREAMING_SNAKE_CASE				:					List[str]		  =			decoder_start_token_id
       SCREAMING_SNAKE_CASE				:					List[Any]		  =			use_cache
       SCREAMING_SNAKE_CASE				:					Tuple		  =			max_position_embeddings

       SCREAMING_SNAKE_CASE				:					Any		  =			None
       SCREAMING_SNAKE_CASE				:					Tuple		  =			decoder_seq_length
       SCREAMING_SNAKE_CASE				:					int		  =			2
       SCREAMING_SNAKE_CASE				:					Optional[Any]		  =			1


     def  _A				( self				:		Optional[int]					):
       SCREAMING_SNAKE_CASE				:					Optional[Any]		  =			ids_tensor([self.batch_size, self.decoder_seq_length]    ,       self.vocab_size					)

       SCREAMING_SNAKE_CASE				:					str		  =			None
       if self.use_attention_mask:
         SCREAMING_SNAKE_CASE				:					Tuple		  =			ids_tensor([self.batch_size, self.decoder_seq_length]    ,       vocab_size=2					)

       SCREAMING_SNAKE_CASE				:					str		  =			None
       if self.use_labels:
         SCREAMING_SNAKE_CASE				:					List[str]		  =			ids_tensor([self.batch_size, self.decoder_seq_length]    ,       self.vocab_size					)

       SCREAMING_SNAKE_CASE				:					str		  =			SpeechaTextaConfig(
           vocab_size=self.vocab_size    ,       d_model=self.d_model    ,       decoder_layers=self.decoder_layers    ,       decoder_ffn_dim=self.decoder_ffn_dim    ,       decoder_attention_heads=self.decoder_attention_heads    ,       eos_token_id=self.eos_token_id    ,       bos_token_id=self.bos_token_id    ,       use_cache=self.use_cache    ,       pad_token_id=self.pad_token_id    ,       decoder_start_token_id=self.decoder_start_token_id    ,       max_position_embeddings=self.max_position_embeddings    ,       )

       return (
           config,
           input_ids,
           attention_mask,
           lm_labels,
       )


     def  _A				( self				:		Optional[Any]    ,       UpperCAmelCase_				:		int    ,       UpperCAmelCase_				:		Dict    ,       UpperCAmelCase_				:		int    ,       UpperCAmelCase_				:		Any    ,       ):
       SCREAMING_SNAKE_CASE				:					str		  =			True
       SCREAMING_SNAKE_CASE				:					Optional[int]		  =			SpeechaTextaDecoder(config=UpperCAmelCase_					).to(UpperCAmelCase_					).eval()
       SCREAMING_SNAKE_CASE				:					Tuple		  =			input_ids[:2]

       input_ids[input_ids == 0] += 1
       # first forward pass
       SCREAMING_SNAKE_CASE				:					Union[str, Any]		  =			model(UpperCAmelCase_    ,       use_cache=UpperCAmelCase_					)
       SCREAMING_SNAKE_CASE				:					List[str]		  =			model(UpperCAmelCase_					)
       SCREAMING_SNAKE_CASE				:					Union[str, Any]		  =			model(UpperCAmelCase_    ,       use_cache=UpperCAmelCase_					)

       self.parent.assertTrue(len(UpperCAmelCase_					) == len(UpperCAmelCase_					)					)
       self.parent.assertTrue(len(UpperCAmelCase_					) == len(UpperCAmelCase_					) + 1					)

       SCREAMING_SNAKE_CASE				:					List[Any]		  =			outputs["past_key_values"]

       # create hypothetical next token and extent to next_input_ids
       SCREAMING_SNAKE_CASE				:					str		  =			ids_tensor((2, 1)    ,       config.vocab_size - 1					) + 1

       # append to next input_ids and
       SCREAMING_SNAKE_CASE				:					str		  =			torch.cat([input_ids, next_tokens]    ,       dim=-1					)

       print(UpperCAmelCase_					)

       SCREAMING_SNAKE_CASE				:					List[str]		  =			model(UpperCAmelCase_					)["last_hidden_state"]
       SCREAMING_SNAKE_CASE				:					Union[str, Any]		  =			model(UpperCAmelCase_    ,       past_key_values=UpperCAmelCase_					)["last_hidden_state"]

       # select random slice
       SCREAMING_SNAKE_CASE				:					List[Any]		  =			ids_tensor((1,)    ,       output_from_past.shape[-1]					).item()
       SCREAMING_SNAKE_CASE				:					str		  =			output_from_no_past[:, next_input_ids.shape[-1] - 1, random_slice_idx].detach()
       SCREAMING_SNAKE_CASE				:					int		  =			output_from_past[:, 0, random_slice_idx].detach()

       # test that outputs are equal for slice
       assert torch.allclose(UpperCAmelCase_    ,       UpperCAmelCase_    ,       atol=1E-3					)






     def  _A				( self				:		str					):
       SCREAMING_SNAKE_CASE				:					int		  =			self.prepare_config_and_inputs()
       (
           (
           SCREAMING_SNAKE_CASE
       )  ,     (
           SCREAMING_SNAKE_CASE
       )  ,     (
           SCREAMING_SNAKE_CASE
       )  ,     (
           SCREAMING_SNAKE_CASE
       )  ,     
       )				:					Optional[Any]		  =			config_and_inputs

       SCREAMING_SNAKE_CASE				:					Optional[Any]		  =			{
           "input_ids": input_ids,
           "attention_mask": attention_mask,
       }
       return config, inputs_dict





@require_torch
class        SCREAMING_SNAKE_CASE	(	lowerCAmelCase					,						lowerCAmelCase					,						lowerCAmelCase					,						unittest.TestCase    ):
     '''simple docstring'''


     UpperCamelCase_							:						int								=		(SpeechaTextaDecoder, SpeechaTextaForCausalLM) if is_torch_available() else ()
     UpperCamelCase_							:						Tuple								=		(SpeechaTextaForCausalLM,) if is_torch_available() else ()
     UpperCamelCase_							:						int								=		{'''text-generation''': SpeechaTextaForCausalLM} if is_torch_available() else {}
     UpperCamelCase_							:						Tuple								=		True
     UpperCamelCase_							:						Union[str, Any]								=		False


     def  _A				( self				:		Tuple    ,       ):
       SCREAMING_SNAKE_CASE				:					Any		  =			SpeechaTextaStandaloneDecoderModelTester(self    ,       is_training=UpperCAmelCase_					)
       SCREAMING_SNAKE_CASE				:					Optional[int]		  =			ConfigTester(self    ,       config_class=UpperCAmelCase_					)


     def  _A				( self				:		Optional[Any]					):
       pass


     def  _A				( self				:		str					):
       pass


     def  _A				( self				:		Dict					):
       pass


     def  _A				( self				:		Optional[Any]					):
       self.config_tester.run_common_tests()


     def  _A				( self				:		Optional[int]					):
       SCREAMING_SNAKE_CASE				:					List[Any]		  =			self.model_tester.prepare_config_and_inputs()
       self.model_tester.create_and_check_decoder_model_past(*UpperCAmelCase_					)






     def  _A				( self				:		Optional[int]					):
       return






