

import gc
import unittest

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMParallelScheduler,
    DDPMParallelScheduler,
    StableDiffusionParadigmsPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import slow, torch_device
from diffusers.utils.testing_utils import (
    enable_full_determinism,
    require_torch_gpu,
)

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_IMAGE_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()






class 				A_					(						snake_case__				,      snake_case__				,      unittest.TestCase   ):
     _lowercase     :	Dict        =      StableDiffusionParadigmsPipeline
     _lowercase     :	List[Any]        =      TEXT_TO_IMAGE_PARAMS
     _lowercase     :	str        =      TEXT_TO_IMAGE_BATCH_PARAMS
     _lowercase     :	List[Any]        =      TEXT_TO_IMAGE_IMAGE_PARAMS
     _lowercase     :	Tuple        =      TEXT_TO_IMAGE_IMAGE_PARAMS


     def     UpperCAmelCase    (				self     :						Union[str, Any]   )				->      Any:
      torch.manual_seed(0   )
      __lowerCAmelCase:  Union[str, Any]        = UNetaDConditionModel(
          block_out_channels=(3_2, 6_4)     ,    layers_per_block=2     ,    sample_size=3_2     ,    in_channels=4     ,    out_channels=4     ,    down_block_types=('DownBlock2D', 'CrossAttnDownBlock2D')     ,    up_block_types=('CrossAttnUpBlock2D', 'UpBlock2D')     ,    cross_attention_dim=3_2     ,    attention_head_dim=(2, 4)     ,    use_linear_projection=UpperCAmelCase     ,    )
      __lowerCAmelCase:  str        = DDIMParallelScheduler(
          beta_start=0.00085     ,    beta_end=0.012     ,    beta_schedule='scaled_linear'     ,    clip_sample=UpperCAmelCase     ,    set_alpha_to_one=UpperCAmelCase     ,    )
      torch.manual_seed(0   )
      __lowerCAmelCase:  Optional[Any]        = AutoencoderKL(
          block_out_channels=[3_2, 6_4]     ,    in_channels=3     ,    out_channels=3     ,    down_block_types=['DownEncoderBlock2D', 'DownEncoderBlock2D']     ,    up_block_types=['UpDecoderBlock2D', 'UpDecoderBlock2D']     ,    latent_channels=4     ,    sample_size=1_2_8     ,    )
      torch.manual_seed(0   )
      __lowerCAmelCase:  Optional[int]        = CLIPTextConfig(
          bos_token_id=0     ,    eos_token_id=2     ,    hidden_size=3_2     ,    intermediate_size=3_7     ,    layer_norm_eps=1E-05     ,    num_attention_heads=4     ,    num_hidden_layers=5     ,    pad_token_id=1     ,    vocab_size=1_0_0_0     ,    hidden_act='gelu'     ,    projection_dim=5_1_2     ,    )
      __lowerCAmelCase:  Tuple        = CLIPTextModel(UpperCAmelCase   )
      __lowerCAmelCase:  Tuple        = CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'   )

      __lowerCAmelCase:  List[str]        = {
          'unet': unet,
          'scheduler': scheduler,
          'vae': vae,
          'text_encoder': text_encoder,
          'tokenizer': tokenizer,
          'safety_checker': None,
          'feature_extractor': None,
      }
      return components


     def     UpperCAmelCase    (				self     :						Optional[Any]     ,    UpperCAmelCase     :						str     ,    UpperCAmelCase     :						str=0   )				->      Dict:
      if str(UpperCAmelCase   ).startswith('mps'   ):
       __lowerCAmelCase:  Tuple        = torch.manual_seed(UpperCAmelCase   )
      else:
       __lowerCAmelCase:  List[str]        = torch.Generator(device=UpperCAmelCase   ).manual_seed(UpperCAmelCase   )
      __lowerCAmelCase:  List[Any]        = {
          'prompt': 'a photograph of an astronaut riding a horse',
          'generator': generator,
          'num_inference_steps': 1_0,
          'guidance_scale': 6.0,
          'output_type': 'numpy',
          'parallel': 3,
          'debug': True,
      }
      return inputs


     def     UpperCAmelCase    (				self     :						List[Any]   )				->      int:
      __lowerCAmelCase:  List[Any]        = 'cpu'  # ensure determinism for the device-dependent torch.Generator
      __lowerCAmelCase:  Optional[int]        = self.get_dummy_components()
      __lowerCAmelCase:  Dict        = StableDiffusionParadigmsPipeline(**UpperCAmelCase   )
      __lowerCAmelCase:  str        = sd_pipe.to(UpperCAmelCase   )
      sd_pipe.set_progress_bar_config(disable=UpperCAmelCase   )

      __lowerCAmelCase:  List[Any]        = self.get_dummy_inputs(UpperCAmelCase   )
      __lowerCAmelCase:  Any        = sd_pipe(**UpperCAmelCase   ).images
      __lowerCAmelCase:  Union[str, Any]        = image[0, -3:, -3:, -1]
      assert image.shape == (1, 6_4, 6_4, 3)

      __lowerCAmelCase:  List[str]        = np.array([0.4773, 0.5417, 0.4723, 0.4925, 0.5631, 0.4752, 0.5240, 0.4935, 0.5023]   )

      assert np.abs(image_slice.flatten() - expected_slice   ).max() < 1E-2


     def     UpperCAmelCase    (				self     :						str   )				->      Optional[int]:
      __lowerCAmelCase:  Any        = 'cpu'  # ensure determinism for the device-dependent torch.Generator
      __lowerCAmelCase:  int        = self.get_dummy_components()
      torch.manual_seed(0   )
      __lowerCAmelCase:  Union[str, Any]        = DDPMParallelScheduler()
      torch.manual_seed(0   )
      __lowerCAmelCase:  Optional[int]        = StableDiffusionParadigmsPipeline(**UpperCAmelCase   )
      __lowerCAmelCase:  str        = sd_pipe.to(UpperCAmelCase   )
      sd_pipe.set_progress_bar_config(disable=UpperCAmelCase   )

      __lowerCAmelCase:  Optional[Any]        = self.get_dummy_inputs(UpperCAmelCase   )
      __lowerCAmelCase:  Dict        = sd_pipe(**UpperCAmelCase   ).images
      __lowerCAmelCase:  Tuple        = image[0, -3:, -3:, -1]
      assert image.shape == (1, 6_4, 6_4, 3)

      __lowerCAmelCase:  Any        = np.array([0.3573, 0.4420, 0.4960, 0.4799, 0.3796, 0.3879, 0.4819, 0.4365, 0.4468]   )

      assert np.abs(image_slice.flatten() - expected_slice   ).max() < 1E-2


     def     UpperCAmelCase    (				self     :						List[Any]   )				->      List[Any]:
      super().test_inference_batch_consistent(batch_sizes=[1, 2]   )


     def     UpperCAmelCase    (				self     :						Union[str, Any]   )				->      int:
      super().test_inference_batch_single_identical(batch_size=2     ,    expected_max_diff=3E-3   )


     def     UpperCAmelCase    (				self     :						Tuple   )				->      Any:
      __lowerCAmelCase:  Tuple        = 'cpu'  # ensure determinism for the device-dependent torch.Generator
      __lowerCAmelCase:  Any        = self.get_dummy_components()
      __lowerCAmelCase:  str        = StableDiffusionParadigmsPipeline(**UpperCAmelCase   )
      __lowerCAmelCase:  Tuple        = sd_pipe.to(UpperCAmelCase   )
      sd_pipe.set_progress_bar_config(disable=UpperCAmelCase   )

      __lowerCAmelCase:  Optional[Any]        = self.get_dummy_inputs(UpperCAmelCase   )
      __lowerCAmelCase:  Any        = 'french fries'
      __lowerCAmelCase:  Optional[int]        = sd_pipe(**UpperCAmelCase     ,    negative_prompt=UpperCAmelCase   )
      __lowerCAmelCase:  str        = output.images
      __lowerCAmelCase:  str        = image[0, -3:, -3:, -1]

      assert image.shape == (1, 6_4, 6_4, 3)

      __lowerCAmelCase:  str        = np.array([0.4771, 0.5420, 0.4683, 0.4918, 0.5636, 0.4725, 0.5230, 0.4923, 0.5015]   )

      assert np.abs(image_slice.flatten() - expected_slice   ).max() < 1E-2







@slow
@require_torch_gpu
class 				A_					(						unittest.TestCase   ):


     def     UpperCAmelCase    (				self     :						Dict   )				->      int:
      super().tearDown()
      gc.collect()
      torch.cuda.empty_cache()


     def     UpperCAmelCase    (				self     :						Optional[int]     ,    UpperCAmelCase     :						int=0   )				->      int:
      __lowerCAmelCase:  Optional[int]        = torch.Generator(device=UpperCAmelCase   ).manual_seed(UpperCAmelCase   )
      __lowerCAmelCase:  Any        = {
          'prompt': 'a photograph of an astronaut riding a horse',
          'generator': generator,
          'num_inference_steps': 1_0,
          'guidance_scale': 7.5,
          'output_type': 'numpy',
          'parallel': 3,
          'debug': True,
      }
      return inputs


     def     UpperCAmelCase    (				self     :						Optional[Any]   )				->      Union[str, Any]:
      __lowerCAmelCase:  Union[str, Any]        = 'stabilityai/stable-diffusion-2-base'
      __lowerCAmelCase:  Union[str, Any]        = DDIMParallelScheduler.from_pretrained(UpperCAmelCase     ,    subfolder='scheduler'   )
      __lowerCAmelCase:  str        = StableDiffusionParadigmsPipeline.from_pretrained(UpperCAmelCase     ,    scheduler=UpperCAmelCase     ,    safety_checker=UpperCAmelCase   )
      pipe.to(UpperCAmelCase   )
      pipe.set_progress_bar_config(disable=UpperCAmelCase   )
      pipe.enable_attention_slicing()

      __lowerCAmelCase:  List[str]        = self.get_inputs()
      __lowerCAmelCase:  Optional[int]        = pipe(**UpperCAmelCase   ).images
      __lowerCAmelCase:  Optional[Any]        = image[0, -3:, -3:, -1].flatten()

      assert image.shape == (1, 5_1_2, 5_1_2, 3)

      __lowerCAmelCase:  Optional[Any]        = np.array([0.9622, 0.9602, 0.9748, 0.9591, 0.9630, 0.9691, 0.9661, 0.9631, 0.9741]   )

      assert np.abs(expected_slice - image_slice   ).max() < 1E-2








from collections import OrderedDict
from typing import Any, Mapping, Optional

from ... import PreTrainedTokenizer
from ...configuration_utils import PretrainedConfig
from ...onnx import OnnxConfig, OnnxSeqaSeqConfigWithPast
from ...onnx.utils import compute_effective_axis_dimension
from ...utils import TensorType, is_torch_available, logging


_a									=  logging.get_logger(__name__)

_a									=  {
    '''facebook/m2m100_418M''': '''https://huggingface.co/facebook/m2m100_418M/resolve/main/config.json''',
    # See all M2M100 models at https://huggingface.co/models?filter=m2m_100
}






class 				A_					(						snake_case__   ):
     _lowercase     :	List[Any]        =      'm2m_100'
     _lowercase     :	str        =      ['past_key_values']
     _lowercase     :	List[Any]        =      {'num_attention_heads': 'encoder_attention_heads', 'hidden_size': 'd_model'}


     def __init__(				self     :						int     ,    UpperCAmelCase     :						Any=1_2_8_1_1_2     ,    UpperCAmelCase     :						Optional[int]=1_0_2_4     ,    UpperCAmelCase     :						Tuple=1_2     ,    UpperCAmelCase     :						List[str]=4_0_9_6     ,    UpperCAmelCase     :						Dict=1_6     ,    UpperCAmelCase     :						int=1_2     ,    UpperCAmelCase     :						int=4_0_9_6     ,    UpperCAmelCase     :						List[str]=1_6     ,    UpperCAmelCase     :						Optional[Any]=0.05     ,    UpperCAmelCase     :						Dict=0.05     ,    UpperCAmelCase     :						Union[str, Any]=True     ,    UpperCAmelCase     :						Dict=True     ,    UpperCAmelCase     :						List[str]="relu"     ,    UpperCAmelCase     :						Dict=1_0_2_4     ,    UpperCAmelCase     :						str=0.1     ,    UpperCAmelCase     :						Dict=0.1     ,    UpperCAmelCase     :						List[str]=0.0     ,    UpperCAmelCase     :						Union[str, Any]=0.02     ,    UpperCAmelCase     :						Optional[Any]=2     ,    UpperCAmelCase     :						str=True     ,    UpperCAmelCase     :						Optional[int]=1     ,    UpperCAmelCase     :						List[Any]=0     ,    UpperCAmelCase     :						Union[str, Any]=2     ,    **UpperCAmelCase     :						Union[str, Any]     ,    )				->      Tuple:
      __lowerCAmelCase:  Optional[Any]        = vocab_size
      __lowerCAmelCase:  Dict        = max_position_embeddings
      __lowerCAmelCase:  Tuple        = d_model
      __lowerCAmelCase:  List[str]        = encoder_ffn_dim
      __lowerCAmelCase:  Any        = encoder_layers
      __lowerCAmelCase:  int        = encoder_attention_heads
      __lowerCAmelCase:  str        = decoder_ffn_dim
      __lowerCAmelCase:  Tuple        = decoder_layers
      __lowerCAmelCase:  Tuple        = decoder_attention_heads
      __lowerCAmelCase:  Any        = dropout
      __lowerCAmelCase:  Optional[int]        = attention_dropout
      __lowerCAmelCase:  Tuple        = activation_dropout
      __lowerCAmelCase:  Optional[Any]        = activation_function
      __lowerCAmelCase:  Union[str, Any]        = init_std
      __lowerCAmelCase:  List[str]        = encoder_layerdrop
      __lowerCAmelCase:  Optional[Any]        = decoder_layerdrop
      __lowerCAmelCase:  Dict        = use_cache
      __lowerCAmelCase:  Any        = encoder_layers
      __lowerCAmelCase:  Tuple        = scale_embedding  # scale factor will be sqrt(d_model) if True

      super().__init__(
          pad_token_id=UpperCAmelCase     ,    bos_token_id=UpperCAmelCase     ,    eos_token_id=UpperCAmelCase     ,    is_encoder_decoder=UpperCAmelCase     ,    decoder_start_token_id=UpperCAmelCase     ,    **UpperCAmelCase     ,    )







class 				A_					(						snake_case__   ):


     @property
     def     UpperCAmelCase    (				self     :						List[Any]   )				->      Mapping[str, Mapping[int, str]]:
      __lowerCAmelCase:  Union[str, Any]        = OrderedDict(
          [
              ('input_ids', {0: 'batch', 1: 'encoder_sequence'}),
              ('attention_mask', {0: 'batch', 1: 'encoder_sequence'}),
          ]   )

      if self.use_past:
       __lowerCAmelCase:  str        = {0: 'batch'}
       __lowerCAmelCase:  Union[str, Any]        = {0: 'batch', 1: 'past_decoder_sequence + sequence'}
      else:
       __lowerCAmelCase:  Any        = {0: 'batch', 1: 'decoder_sequence'}
       __lowerCAmelCase:  Dict        = {0: 'batch', 1: 'decoder_sequence'}

      if self.use_past:
       self.fill_with_past_key_values_(UpperCAmelCase     ,    direction='inputs'   )
      return common_inputs


     def     UpperCAmelCase    (				self     :						Tuple     ,    UpperCAmelCase     :						PreTrainedTokenizer     ,    UpperCAmelCase     :						int = -1     ,    UpperCAmelCase     :						int = -1     ,    UpperCAmelCase     :						bool = False     ,    UpperCAmelCase     :						Optional[TensorType] = None     ,    )				->      Mapping[str, Any]:
      # Copied from OnnxConfig.generate_dummy_inputs
      # Did not use super(OnnxConfigWithPast, self).generate_dummy_inputs for code clarity.
      # If dynamic axis (-1) we forward with a fixed dimension of 2 samples to avoid optimizations made by ONNX
      __lowerCAmelCase:  List[str]        = compute_effective_axis_dimension(
          UpperCAmelCase     ,    fixed_dimension=OnnxConfig.default_fixed_batch     ,    num_token_to_add=0   )

      # If dynamic axis (-1) we forward with a fixed dimension of 8 tokens to avoid optimizations made by ONNX
      __lowerCAmelCase:  Tuple        = tokenizer.num_special_tokens_to_add(UpperCAmelCase   )
      __lowerCAmelCase:  List[str]        = compute_effective_axis_dimension(
          UpperCAmelCase     ,    fixed_dimension=OnnxConfig.default_fixed_sequence     ,    num_token_to_add=UpperCAmelCase   )

      # Generate dummy inputs according to compute batch and sequence
      __lowerCAmelCase:  int        = [' '.join([tokenizer.unk_token]   ) * seq_length] * batch_size
      __lowerCAmelCase:  Optional[int]        = dict(tokenizer(UpperCAmelCase     ,    return_tensors=UpperCAmelCase   )   )
      return common_inputs


     def     UpperCAmelCase    (				self     :						str     ,    UpperCAmelCase     :						PreTrainedTokenizer     ,    UpperCAmelCase     :						int = -1     ,    UpperCAmelCase     :						int = -1     ,    UpperCAmelCase     :						bool = False     ,    UpperCAmelCase     :						Optional[TensorType] = None     ,    )				->      Mapping[str, Any]:
      __lowerCAmelCase:  Optional[int]        = self._generate_dummy_inputs_for_sequence_classification_and_question_answering(
          UpperCAmelCase     ,    UpperCAmelCase     ,    UpperCAmelCase     ,    UpperCAmelCase     ,    UpperCAmelCase   )

      # Generate decoder inputs
      __lowerCAmelCase:  Any        = seq_length if not self.use_past else 1
      __lowerCAmelCase:  Union[str, Any]        = self._generate_dummy_inputs_for_sequence_classification_and_question_answering(
          UpperCAmelCase     ,    UpperCAmelCase     ,    UpperCAmelCase     ,    UpperCAmelCase     ,    UpperCAmelCase   )
      __lowerCAmelCase:  List[Any]        = {F'''decoder_{name}''': tensor for name, tensor in decoder_inputs.items()}
      __lowerCAmelCase:  List[str]        = dict(**UpperCAmelCase     ,    **UpperCAmelCase   )

      if self.use_past:
       if not is_torch_available():
        raise ValueError('Cannot generate dummy past_keys inputs without PyTorch installed.'   )
       else:
        import torch
       __lowerCAmelCase      ,				__lowerCAmelCase:  List[Any]        = common_inputs['input_ids'].shape
       __lowerCAmelCase:  int        = common_inputs['decoder_input_ids'].shape[1]
       __lowerCAmelCase      ,				__lowerCAmelCase:  str        = self.num_attention_heads
       __lowerCAmelCase:  str        = (
           batch,
           num_encoder_attention_heads,
           encoder_seq_length,
           self._config.hidden_size // num_encoder_attention_heads,
       )
       __lowerCAmelCase:  Optional[int]        = decoder_seq_length + 3
       __lowerCAmelCase:  List[str]        = (
           batch,
           num_decoder_attention_heads,
           decoder_past_length,
           self._config.hidden_size // num_decoder_attention_heads,
       )

       __lowerCAmelCase:  Dict        = torch.cat(
           [common_inputs['decoder_attention_mask'], torch.ones(UpperCAmelCase     ,    UpperCAmelCase   )]     ,    dim=1   )

       __lowerCAmelCase:  str        = []
       # If the number of encoder and decoder layers are present in the model configuration, both are considered
       __lowerCAmelCase      ,				__lowerCAmelCase:  str        = self.num_layers
       __lowerCAmelCase:  Tuple        = min(UpperCAmelCase     ,    UpperCAmelCase   )
       __lowerCAmelCase:  List[Any]        = max(UpperCAmelCase     ,    UpperCAmelCase   ) - min_num_layers
       __lowerCAmelCase:  Any        = 'encoder' if num_encoder_layers > num_decoder_layers else 'decoder'

       for _ in range(UpperCAmelCase   ):
        common_inputs["past_key_values"].append(
            (
                torch.zeros(UpperCAmelCase   ),
                torch.zeros(UpperCAmelCase   ),
                torch.zeros(UpperCAmelCase   ),
                torch.zeros(UpperCAmelCase   ),
            )   )
       # TODO: test this.
       __lowerCAmelCase:  List[Any]        = encoder_shape if remaining_side_name == 'encoder' else decoder_shape
       for _ in range(UpperCAmelCase     ,    UpperCAmelCase   ):
        common_inputs["past_key_values"].append((torch.zeros(UpperCAmelCase   ), torch.zeros(UpperCAmelCase   ))   )
      return common_inputs


     _lowercase     :	int        =      _generate_dummy_inputs_for_default_and_seqaseq_lm





