import unittest

from transformers import is_torch_available
from transformers.testing_utils import require_sentencepiece, require_tokenizers, require_torch, slow, torch_device


if is_torch_available():
 import torch

 from transformers import CamembertModel




@require_torch
@require_sentencepiece
@require_tokenizers
class       __lowerCAmelCase     (				unittest.TestCase  ):



 '''simple docstring'''

 @slow
 def     __snake_case    (	self				:						Any     )       ->							Any:
        _UpperCamelCase   :				int       =     CamembertModel.from_pretrained('camembert-base'     )
        model.to(lowerCamelCase     )

        _UpperCamelCase   :				Dict       =     torch.tensor(
            [[5, 1_2_1, 1_1, 6_6_0, 1_6, 7_3_0, 2_5_5_4_3, 1_1_0, 8_3, 6]]				,	device=lowerCamelCase				,	dtype=torch.long				,	)  # J'aime le camembert !
        with torch.no_grad():
               _UpperCamelCase   :				Any       =     model(lowerCamelCase     )['last_hidden_state']
        _UpperCamelCase   :				str       =     torch.Size((1, 1_0, 7_6_8)     )
        self.assertEqual(output.shape				,	lowerCamelCase     )
        # compare the actual values for a slice.
        _UpperCamelCase   :				Union[str, Any]       =     torch.tensor(
            [[[-0.0_2_5_4, 0.0_2_3_5, 0.1_0_2_7], [0.0_6_0_6, -0.1_8_1_1, -0.0_4_1_8], [-0.1_5_6_1, -0.1_1_2_7, 0.2_6_8_7]]]				,	device=lowerCamelCase				,	dtype=torch.float				,	)
        # camembert = torch.hub.load('pytorch/fairseq', 'camembert.v0')
        # camembert.eval()
        # expected_slice = roberta.model.forward(input_ids)[0][:, :3, :3].detach()

        self.assertTrue(torch.allclose(output[:, :3, :3]				,	lowerCamelCase				,	atol=1e-4     )     )




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       __lowerCAmelCase     (				a_,		a_,		unittest.TestCase  ):



 '''simple docstring'''
 _lowerCamelCase  :		Any 			=       StableDiffusionParadigmsPipeline
 _lowerCamelCase  :		Any 			=       TEXT_TO_IMAGE_PARAMS
 _lowerCamelCase  :		str 			=       TEXT_TO_IMAGE_BATCH_PARAMS
 _lowerCamelCase  :		int 			=       TEXT_TO_IMAGE_IMAGE_PARAMS
 _lowerCamelCase  :		List[Any] 			=       TEXT_TO_IMAGE_IMAGE_PARAMS

 def     __snake_case    (	self				:						int     )       ->							List[str]:
        torch.manual_seed(0     )
        _UpperCamelCase   :				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=lowerCamelCase				,	)
        _UpperCamelCase   :				str       =     DDIMParallelScheduler(
            beta_start=0.0_0_0_8_5				,	beta_end=0.0_1_2				,	beta_schedule='scaled_linear'				,	clip_sample=lowerCamelCase				,	set_alpha_to_one=lowerCamelCase				,	)
        torch.manual_seed(0     )
        _UpperCamelCase   :				str       =     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     )
        _UpperCamelCase   :				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				,	)
        _UpperCamelCase   :				Any       =     CLIPTextModel(lowerCamelCase     )
        _UpperCamelCase   :				Optional[Any]       =     CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'     )

        _UpperCamelCase   :				Tuple       =     {
            'unet': unet,
            'scheduler': scheduler,
            'vae': vae,
            'text_encoder': text_encoder,
            'tokenizer': tokenizer,
            'safety_checker': None,
            'feature_extractor': None,
        }
        return components

 def     __snake_case    (	self				:						List[Any]				,	lowerCamelCase				:						Any				,	lowerCamelCase				:						str=0     )       ->							Dict:
        if str(lowerCamelCase     ).startswith('mps'     ):
               _UpperCamelCase   :				List[Any]       =     torch.manual_seed(lowerCamelCase     )
        else:
               _UpperCamelCase   :				Union[str, Any]       =     torch.Generator(device=lowerCamelCase     ).manual_seed(lowerCamelCase     )
        _UpperCamelCase   :				Optional[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     __snake_case    (	self				:						List[Any]     )       ->							str:
        _UpperCamelCase   :				List[str]       =     'cpu'  # ensure determinism for the device-dependent torch.Generator
        _UpperCamelCase   :				List[Any]       =     self.get_dummy_components()
        _UpperCamelCase   :				Tuple       =     StableDiffusionParadigmsPipeline(**lowerCamelCase     )
        _UpperCamelCase   :				str       =     sd_pipe.to(lowerCamelCase     )
        sd_pipe.set_progress_bar_config(disable=lowerCamelCase     )

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

        _UpperCamelCase   :				Any       =     np.array([0.4_7_7_3, 0.5_4_1_7, 0.4_7_2_3, 0.4_9_2_5, 0.5_6_3_1, 0.4_7_5_2, 0.5_2_4_0, 0.4_9_3_5, 0.5_0_2_3]     )

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

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

        _UpperCamelCase   :				int       =     self.get_dummy_inputs(lowerCamelCase     )
        _UpperCamelCase   :				Optional[int]       =     sd_pipe(**lowerCamelCase     ).images
        _UpperCamelCase   :				Dict       =     image[0, -3:, -3:, -1]
        assert image.shape == (1, 6_4, 6_4, 3)

        _UpperCamelCase   :				List[Any]       =     np.array([0.3_5_7_3, 0.4_4_2_0, 0.4_9_6_0, 0.4_7_9_9, 0.3_7_9_6, 0.3_8_7_9, 0.4_8_1_9, 0.4_3_6_5, 0.4_4_6_8]     )

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

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

 def     __snake_case    (	self				:						Union[str, Any]     )       ->							Tuple:
        super().test_inference_batch_single_identical(batch_size=2				,	expected_max_diff=3e-3     )

 def     __snake_case    (	self				:						List[str]     )       ->							List[str]:
        _UpperCamelCase   :				List[str]       =     'cpu'  # ensure determinism for the device-dependent torch.Generator
        _UpperCamelCase   :				Tuple       =     self.get_dummy_components()
        _UpperCamelCase   :				Any       =     StableDiffusionParadigmsPipeline(**lowerCamelCase     )
        _UpperCamelCase   :				List[Any]       =     sd_pipe.to(lowerCamelCase     )
        sd_pipe.set_progress_bar_config(disable=lowerCamelCase     )

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

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

        _UpperCamelCase   :				Dict       =     np.array([0.4_7_7_1, 0.5_4_2_0, 0.4_6_8_3, 0.4_9_1_8, 0.5_6_3_6, 0.4_7_2_5, 0.5_2_3_0, 0.4_9_2_3, 0.5_0_1_5]     )

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







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



 '''simple docstring'''

 def     __snake_case    (	self				:						Optional[Any]     )       ->							Optional[int]:
        super().tearDown()
        gc.collect()
        torch.cuda.empty_cache()

 def     __snake_case    (	self				:						Dict				,	lowerCamelCase				:						Any=0     )       ->							Dict:
        _UpperCamelCase   :				List[str]       =     torch.Generator(device=lowerCamelCase     ).manual_seed(lowerCamelCase     )
        _UpperCamelCase   :				Tuple       =     {
            '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     __snake_case    (	self				:						List[Any]     )       ->							str:
        _UpperCamelCase   :				List[str]       =     'stabilityai/stable-diffusion-2-base'
        _UpperCamelCase   :				int       =     DDIMParallelScheduler.from_pretrained(lowerCamelCase				,	subfolder='scheduler'     )
        _UpperCamelCase   :				List[Any]       =     StableDiffusionParadigmsPipeline.from_pretrained(lowerCamelCase				,	scheduler=lowerCamelCase				,	safety_checker=lowerCamelCase     )
        pipe.to(lowerCamelCase     )
        pipe.set_progress_bar_config(disable=lowerCamelCase     )
        pipe.enable_attention_slicing()

        _UpperCamelCase   :				Optional[Any]       =     self.get_inputs()
        _UpperCamelCase   :				Optional[int]       =     pipe(**lowerCamelCase     ).images
        _UpperCamelCase   :				Optional[int]       =     image[0, -3:, -3:, -1].flatten()

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

        _UpperCamelCase   :				int       =     np.array([0.9_6_2_2, 0.9_6_0_2, 0.9_7_4_8, 0.9_5_9_1, 0.9_6_3_0, 0.9_6_9_1, 0.9_6_6_1, 0.9_6_3_1, 0.9_7_4_1]     )

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



