import collections
import inspect
import unittest

from transformers import NatConfig
from transformers.testing_utils import require_natten, require_torch, require_vision, slow, torch_device
from transformers.utils import cached_property, is_torch_available, is_vision_available

from ...test_backbone_common import BackboneTesterMixin
from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, _config_zero_init, floats_tensor, ids_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
    import torch
    from torch import nn

    from transformers import NatBackbone, NatForImageClassification, NatModel
    from transformers.models.nat.modeling_nat import NAT_PRETRAINED_MODEL_ARCHIVE_LIST

if is_vision_available():
    from PIL import Image

    from transformers import AutoImageProcessor
class 				a_ :






 '''simple docstring'''






 def __init__(			self					:  Optional[Any]			,    lowercase__					:  str			,    lowercase__					:  Tuple=13			,    lowercase__					:  Tuple=64			,    lowercase__					:  Tuple=4			,    lowercase__					:  List[str]=3			,    lowercase__					:  Any=16			,    lowercase__					:  Optional[Any]=[1, 2, 1]			,    lowercase__					:  Tuple=[2, 4, 8]			,    lowercase__					:  int=3			,    lowercase__					:  Optional[Any]=2.0			,    lowercase__					:  Optional[Any]=True			,    lowercase__					:  Any=0.0			,    lowercase__					:  List[Any]=0.0			,    lowercase__					:  List[str]=0.1			,    lowercase__					:  Optional[int]="gelu"			,    lowercase__					:  Optional[int]=True			,    lowercase__					:  str=0.02			,    lowercase__					:  Union[str, Any]=1e-5			,    lowercase__					:  Optional[Any]=True			,    lowercase__					:  Dict=None			,    lowercase__					:  str=True			,    lowercase__					:  List[str]=10			,    lowercase__					:  str=["stage1", "stage2"]			,    lowercase__					:  Optional[Any]=[1, 2]			,    ):







        '''simple docstring'''



        lowerCAmelCase__     	=					parent
        lowerCAmelCase__     	=					batch_size
        lowerCAmelCase__     	=					image_size
        lowerCAmelCase__     	=					patch_size
        lowerCAmelCase__     	=					num_channels
        lowerCAmelCase__     	=					embed_dim
        lowerCAmelCase__     	=					depths
        lowerCAmelCase__     	=					num_heads
        lowerCAmelCase__     	=					kernel_size
        lowerCAmelCase__     	=					mlp_ratio
        lowerCAmelCase__     	=					qkv_bias
        lowerCAmelCase__     	=					hidden_dropout_prob
        lowerCAmelCase__     	=					attention_probs_dropout_prob
        lowerCAmelCase__     	=					drop_path_rate
        lowerCAmelCase__     	=					hidden_act
        lowerCAmelCase__     	=					patch_norm
        lowerCAmelCase__     	=					layer_norm_eps
        lowerCAmelCase__     	=					initializer_range
        lowerCAmelCase__     	=					is_training
        lowerCAmelCase__     	=					scope
        lowerCAmelCase__     	=					use_labels
        lowerCAmelCase__     	=					num_labels
        lowerCAmelCase__     	=					out_features
        lowerCAmelCase__     	=					out_indices




 def 		__snake_case						(			self					:  str):







        '''simple docstring'''



        lowerCAmelCase__     	=					floats_tensor([self.batch_size, self.num_channels, self.image_size, self.image_size])

        lowerCAmelCase__     	=					None
        if self.use_labels:
               lowerCAmelCase__     	=					ids_tensor([self.batch_size]			,    self.num_labels)

        lowerCAmelCase__     	=					self.get_config()

        return config, pixel_values, labels




 def 		__snake_case						(			self					:  str):







        '''simple docstring'''



        return NatConfig(
            num_labels=self.num_labels			,    image_size=self.image_size			,    patch_size=self.patch_size			,    num_channels=self.num_channels			,    embed_dim=self.embed_dim			,    depths=self.depths			,    num_heads=self.num_heads			,    kernel_size=self.kernel_size			,    mlp_ratio=self.mlp_ratio			,    qkv_bias=self.qkv_bias			,    hidden_dropout_prob=self.hidden_dropout_prob			,    attention_probs_dropout_prob=self.attention_probs_dropout_prob			,    drop_path_rate=self.drop_path_rate			,    hidden_act=self.hidden_act			,    patch_norm=self.patch_norm			,    layer_norm_eps=self.layer_norm_eps			,    initializer_range=self.initializer_range			,    out_features=self.out_features			,    out_indices=self.out_indices			,    )




 def 		__snake_case						(			self					:  List[str]			,    lowercase__					:  Dict			,    lowercase__					:  Dict			,    lowercase__					:  Any):







        '''simple docstring'''



        lowerCAmelCase__     	=					NatModel(config=lowercase__)
        model.to(lowercase__)
        model.eval()
        lowerCAmelCase__     	=					model(lowercase__)

        lowerCAmelCase__     	=					lowerCAmelCase__     	=					(config.image_size // config.patch_size) // (2 ** (len(config.depths) - 1))
        lowerCAmelCase__     	=					int(config.embed_dim * 2 ** (len(config.depths) - 1))

        self.parent.assertEqual(
            result.last_hidden_state.shape			,    (self.batch_size, expected_height, expected_width, expected_dim))




 def 		__snake_case						(			self					:  str			,    lowercase__					:  List[str]			,    lowercase__					:  List[Any]			,    lowercase__					:  Optional[Any]):







        '''simple docstring'''



        lowerCAmelCase__     	=					NatForImageClassification(lowercase__)
        model.to(lowercase__)
        model.eval()
        lowerCAmelCase__     	=					model(lowercase__			,    labels=lowercase__)
        self.parent.assertEqual(result.logits.shape			,    (self.batch_size, self.num_labels))

        # test greyscale images
        lowerCAmelCase__     	=					1
        lowerCAmelCase__     	=					NatForImageClassification(lowercase__)
        model.to(lowercase__)
        model.eval()

        lowerCAmelCase__     	=					floats_tensor([self.batch_size, 1, self.image_size, self.image_size])
        lowerCAmelCase__     	=					model(lowercase__)
        self.parent.assertEqual(result.logits.shape			,    (self.batch_size, self.num_labels))




 def 		__snake_case						(			self					:  Dict			,    lowercase__					:  Optional[Any]			,    lowercase__					:  Optional[int]			,    lowercase__					:  List[str]):







        '''simple docstring'''



        lowerCAmelCase__     	=					NatBackbone(config=lowercase__)
        model.to(lowercase__)
        model.eval()
        lowerCAmelCase__     	=					model(lowercase__)

        # verify hidden states
        self.parent.assertEqual(len(result.feature_maps)			,    len(config.out_features))
        self.parent.assertListEqual(list(result.feature_maps[0].shape)			,    [self.batch_size, model.channels[0], 16, 16])

        # verify channels
        self.parent.assertEqual(len(model.channels)			,    len(config.out_features))

        # verify backbone works with out_features=None
        lowerCAmelCase__     	=					None
        lowerCAmelCase__     	=					NatBackbone(config=lowercase__)
        model.to(lowercase__)
        model.eval()
        lowerCAmelCase__     	=					model(lowercase__)

        # verify feature maps
        self.parent.assertEqual(len(result.feature_maps)			,    1)
        self.parent.assertListEqual(list(result.feature_maps[0].shape)			,    [self.batch_size, model.channels[-1], 4, 4])

        # verify channels
        self.parent.assertEqual(len(model.channels)			,    1)




 def 		__snake_case						(			self					:  Optional[Any]):







        '''simple docstring'''



        lowerCAmelCase__     	=					self.prepare_config_and_inputs()
        lowerCAmelCase__	,					lowerCAmelCase__	,					lowerCAmelCase__     	=					config_and_inputs
        lowerCAmelCase__     	=					{'pixel_values': pixel_values}
        return config, inputs_dict






@require_natten
@require_torch
class 				a_ (						SCREAMING_SNAKE_CASE       ,     SCREAMING_SNAKE_CASE       ,     unittest.TestCase						):






 '''simple docstring'''






 UpperCAmelCase_	 =					(
     (
         NatModel,
         NatForImageClassification,
         NatBackbone,
     )
     if is_torch_available()
     else ()
 )
 UpperCAmelCase_	 =					(
     {'feature-extraction': NatModel, 'image-classification': NatForImageClassification}
     if is_torch_available()
     else {}
 )
 UpperCAmelCase_	 =					False

 UpperCAmelCase_	 =					False
 UpperCAmelCase_	 =					False
 UpperCAmelCase_	 =					False
 UpperCAmelCase_	 =					False




 def 		__snake_case						(			self					:  Any):







        '''simple docstring'''



        lowerCAmelCase__     	=					NatModelTester(self)
        lowerCAmelCase__     	=					ConfigTester(self			,    config_class=lowercase__			,    embed_dim=37)




 def 		__snake_case						(			self					:  Any):







        '''simple docstring'''



        self.create_and_test_config_common_properties()
        self.config_tester.create_and_test_config_to_json_string()
        self.config_tester.create_and_test_config_to_json_file()
        self.config_tester.create_and_test_config_from_and_save_pretrained()
        self.config_tester.create_and_test_config_with_num_labels()
        self.config_tester.check_config_can_be_init_without_params()
        self.config_tester.check_config_arguments_init()




 def 		__snake_case						(			self					:  Optional[Any]):







        '''simple docstring'''



        return




 def 		__snake_case						(			self					:  Optional[Any]):







        '''simple docstring'''



        lowerCAmelCase__     	=					self.model_tester.prepare_config_and_inputs()
        self.model_tester.create_and_check_model(*lowercase__)




 def 		__snake_case						(			self					:  str):







        '''simple docstring'''



        lowerCAmelCase__     	=					self.model_tester.prepare_config_and_inputs()
        self.model_tester.create_and_check_for_image_classification(*lowercase__)




 def 		__snake_case						(			self					:  Union[str, Any]):







        '''simple docstring'''



        lowerCAmelCase__     	=					self.model_tester.prepare_config_and_inputs()
        self.model_tester.create_and_check_backbone(*lowercase__)




 @unittest.skip(reason='Nat does not use inputs_embeds')
 def 		__snake_case						(			self					:  Optional[Any]):







        '''simple docstring'''



        pass




 @unittest.skip(reason='Nat does not use feedforward chunking')
 def 		__snake_case						(			self					:  Tuple):







        '''simple docstring'''



        pass




 def 		__snake_case						(			self					:  Optional[int]):







        '''simple docstring'''



        lowerCAmelCase__	,					lowerCAmelCase__     	=					self.model_tester.prepare_config_and_inputs_for_common()

        for model_class in self.all_model_classes:
               lowerCAmelCase__     	=					model_class(lowercase__)
               self.assertIsInstance(model.get_input_embeddings()			,    (nn.Module))
               lowerCAmelCase__     	=					model.get_output_embeddings()
               self.assertTrue(x is None or isinstance(lowercase__			,    nn.Linear))




 def 		__snake_case						(			self					:  List[str]):







        '''simple docstring'''



        lowerCAmelCase__	,					lowerCAmelCase__     	=					self.model_tester.prepare_config_and_inputs_for_common()

        for model_class in self.all_model_classes:
               lowerCAmelCase__     	=					model_class(lowercase__)
               lowerCAmelCase__     	=					inspect.signature(model.forward)
               # signature.parameters is an OrderedDict => so arg_names order is deterministic
               lowerCAmelCase__     	=					[*signature.parameters.keys()]

               lowerCAmelCase__     	=					['pixel_values']
               self.assertListEqual(arg_names[:1]			,    lowercase__)




 def 		__snake_case						(			self					:  Dict):







        '''simple docstring'''



        self.skipTest('Nat\'s attention operation is handled entirely by NATTEN.')




 def 		__snake_case						(			self					:  Any			,    lowercase__					:  Tuple			,    lowercase__					:  Tuple			,    lowercase__					:  Tuple			,    lowercase__					:  int):







        '''simple docstring'''



        lowerCAmelCase__     	=					model_class(lowercase__)
        model.to(lowercase__)
        model.eval()

        with torch.no_grad():
               lowerCAmelCase__     	=					model(**self._prepare_for_class(lowercase__			,    lowercase__))

        lowerCAmelCase__     	=					outputs.hidden_states

        lowerCAmelCase__     	=					getattr(
            self.model_tester			,    'expected_num_hidden_layers'			,    len(self.model_tester.depths) + 1)
        self.assertEqual(len(lowercase__)			,    lowercase__)

        # Nat has a different seq_length
        lowerCAmelCase__     	=					(
            config.patch_size
            if isinstance(config.patch_size			,    collections.abc.Iterable)
            else (config.patch_size, config.patch_size)
        )

        lowerCAmelCase__     	=					image_size[0] // patch_size[0]
        lowerCAmelCase__     	=					image_size[1] // patch_size[1]

        self.assertListEqual(
            list(hidden_states[0].shape[-3:])			,    [height, width, self.model_tester.embed_dim]			,    )

        if model_class.__name__ != "NatBackbone":
               lowerCAmelCase__     	=					outputs.reshaped_hidden_states
               self.assertEqual(len(lowercase__)			,    lowercase__)

               lowerCAmelCase__	,					lowerCAmelCase__	,					lowerCAmelCase__	,					lowerCAmelCase__     	=					reshaped_hidden_states[0].shape
               lowerCAmelCase__     	=					(
                   reshaped_hidden_states[0].view(lowercase__			,    lowercase__			,    lowercase__			,    lowercase__).permute(0			,    2			,    3			,    1)
               )
               self.assertListEqual(
                   list(reshaped_hidden_states.shape[-3:])			,    [height, width, self.model_tester.embed_dim]			,    )




 def 		__snake_case						(			self					:  Any):







        '''simple docstring'''



        lowerCAmelCase__	,					lowerCAmelCase__     	=					self.model_tester.prepare_config_and_inputs_for_common()

        lowerCAmelCase__     	=					(
            self.model_tester.image_size
            if isinstance(self.model_tester.image_size			,    collections.abc.Iterable)
            else (self.model_tester.image_size, self.model_tester.image_size)
        )

        for model_class in self.all_model_classes:
               lowerCAmelCase__     	=					True
               self.check_hidden_states_output(lowercase__			,    lowercase__			,    lowercase__			,    lowercase__)

               # check that output_hidden_states also work using config
               del inputs_dict["output_hidden_states"]
               lowerCAmelCase__     	=					True

               self.check_hidden_states_output(lowercase__			,    lowercase__			,    lowercase__			,    lowercase__)




 @slow
 def 		__snake_case						(			self					:  List[Any]):







        '''simple docstring'''



        for model_name in NAT_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
               lowerCAmelCase__     	=					NatModel.from_pretrained(lowercase__)
               self.assertIsNotNone(lowercase__)




 def 		__snake_case						(			self					:  Dict):







        '''simple docstring'''



        lowerCAmelCase__	,					lowerCAmelCase__     	=					self.model_tester.prepare_config_and_inputs_for_common()

        lowerCAmelCase__     	=					_config_zero_init(lowercase__)
        for model_class in self.all_model_classes:
               lowerCAmelCase__     	=					model_class(config=lowercase__)
               for name, param in model.named_parameters():
                      if "embeddings" not in name and param.requires_grad:
                             self.assertIn(
                                 ((param.data.mean() * 1e9).round() / 1e9).item()			,    [0.0, 1.0]			,    msg=F"""Parameter {name} of model {model_class} seems not properly initialized"""			,    )






@require_natten
@require_vision
@require_torch
class 				a_ (						unittest.TestCase						):






 '''simple docstring'''






 @cached_property
 def 		__snake_case						(			self					:  str):







        '''simple docstring'''



        return AutoImageProcessor.from_pretrained('shi-labs/nat-mini-in1k-224') if is_vision_available() else None




 @slow
 def 		__snake_case						(			self					:  int):







        '''simple docstring'''



        lowerCAmelCase__     	=					NatForImageClassification.from_pretrained('shi-labs/nat-mini-in1k-224').to(lowercase__)
        lowerCAmelCase__     	=					self.default_image_processor

        lowerCAmelCase__     	=					Image.open('./tests/fixtures/tests_samples/COCO/000000039769.png')
        lowerCAmelCase__     	=					image_processor(images=lowercase__			,    return_tensors='pt').to(lowercase__)

        # forward pass
        with torch.no_grad():
               lowerCAmelCase__     	=					model(**lowercase__)

        # verify the logits
        lowerCAmelCase__     	=					torch.Size((1, 1_000))
        self.assertEqual(outputs.logits.shape			,    lowercase__)
        lowerCAmelCase__     	=					torch.tensor([0.3_805, -0.8_676, -0.3_912]).to(lowercase__)
        self.assertTrue(torch.allclose(outputs.logits[0, :3]			,    lowercase__			,    atol=1e-4))






@require_torch
@require_natten
class 				a_ (						unittest.TestCase       ,     SCREAMING_SNAKE_CASE						):






 '''simple docstring'''






 UpperCAmelCase_	 =					(NatBackbone,) if is_torch_available() else ()
 UpperCAmelCase_	 =					NatConfig




 def 		__snake_case						(			self					:  List[str]):







        '''simple docstring'''



        lowerCAmelCase__     	=					NatModelTester(self)

import gc
import random
import unittest

import numpy as np
import torch
from PIL import Image
from transformers import CLIPImageProcessor, CLIPVisionConfig, CLIPVisionModelWithProjection

from diffusers import (
    AutoencoderKL,
    DPMSolverMultistepScheduler,
    PNDMScheduler,
    StableDiffusionImageVariationPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import floats_tensor, load_image, load_numpy, nightly, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..pipeline_params import IMAGE_VARIATION_BATCH_PARAMS, IMAGE_VARIATION_PARAMS
from ..test_pipelines_common import PipelineKarrasSchedulerTesterMixin, PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()
class 				a_ (						SCREAMING_SNAKE_CASE       ,     SCREAMING_SNAKE_CASE       ,     SCREAMING_SNAKE_CASE       ,     unittest.TestCase						):






 '''simple docstring'''






 UpperCAmelCase_	 =					StableDiffusionImageVariationPipeline
 UpperCAmelCase_	 =					IMAGE_VARIATION_PARAMS
 UpperCAmelCase_	 =					IMAGE_VARIATION_BATCH_PARAMS
 UpperCAmelCase_	 =					frozenset([]						)
 # TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
 UpperCAmelCase_	 =					frozenset([]						)




 def 		__snake_case						(			self					:  Optional[Any]):







        '''simple docstring'''



        torch.manual_seed(0)
        lowerCAmelCase__     	=					UNetaDConditionModel(
            block_out_channels=(32, 64)			,    layers_per_block=2			,    sample_size=32			,    in_channels=4			,    out_channels=4			,    down_block_types=('DownBlock2D', 'CrossAttnDownBlock2D')			,    up_block_types=('CrossAttnUpBlock2D', 'UpBlock2D')			,    cross_attention_dim=32			,    )
        lowerCAmelCase__     	=					PNDMScheduler(skip_prk_steps=lowercase__)
        torch.manual_seed(0)
        lowerCAmelCase__     	=					AutoencoderKL(
            block_out_channels=[32, 64]			,    in_channels=3			,    out_channels=3			,    down_block_types=['DownEncoderBlock2D', 'DownEncoderBlock2D']			,    up_block_types=['UpDecoderBlock2D', 'UpDecoderBlock2D']			,    latent_channels=4			,    )
        torch.manual_seed(0)
        lowerCAmelCase__     	=					CLIPVisionConfig(
            hidden_size=32			,    projection_dim=32			,    intermediate_size=37			,    layer_norm_eps=1e-05			,    num_attention_heads=4			,    num_hidden_layers=5			,    image_size=32			,    patch_size=4			,    )
        lowerCAmelCase__     	=					CLIPVisionModelWithProjection(lowercase__)
        lowerCAmelCase__     	=					CLIPImageProcessor(crop_size=32			,    size=32)

        lowerCAmelCase__     	=					{
            'unet': unet,
            'scheduler': scheduler,
            'vae': vae,
            'image_encoder': image_encoder,
            'feature_extractor': feature_extractor,
            'safety_checker': None,
        }
        return components




 def 		__snake_case						(			self					:  Optional[int]			,    lowercase__					:  Dict			,    lowercase__					:  Tuple=0):







        '''simple docstring'''



        lowerCAmelCase__     	=					floats_tensor((1, 3, 32, 32)			,    rng=random.Random(lowercase__))
        lowerCAmelCase__     	=					image.cpu().permute(0			,    2			,    3			,    1)[0]
        lowerCAmelCase__     	=					Image.fromarray(np.uinta(lowercase__)).convert('RGB').resize((32, 32))
        if str(lowercase__).startswith('mps'):
               lowerCAmelCase__     	=					torch.manual_seed(lowercase__)
        else:
               lowerCAmelCase__     	=					torch.Generator(device=lowercase__).manual_seed(lowercase__)
        lowerCAmelCase__     	=					{
            'image': image,
            'generator': generator,
            'num_inference_steps': 2,
            'guidance_scale': 6.0,
            'output_type': 'numpy',
        }
        return inputs




 def 		__snake_case						(			self					:  Union[str, Any]):







        '''simple docstring'''



        lowerCAmelCase__     	=					'cpu'  # ensure determinism for the device-dependent torch.Generator
        lowerCAmelCase__     	=					self.get_dummy_components()
        lowerCAmelCase__     	=					StableDiffusionImageVariationPipeline(**lowercase__)
        lowerCAmelCase__     	=					sd_pipe.to(lowercase__)
        sd_pipe.set_progress_bar_config(disable=lowercase__)

        lowerCAmelCase__     	=					self.get_dummy_inputs(lowercase__)
        lowerCAmelCase__     	=					sd_pipe(**lowercase__).images
        lowerCAmelCase__     	=					image[0, -3:, -3:, -1]

        assert image.shape == (1, 64, 64, 3)
        lowerCAmelCase__     	=					np.array([0.5_239, 0.5_723, 0.4_796, 0.5_049, 0.5_550, 0.4_685, 0.5_329, 0.4_891, 0.4_921])

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




 def 		__snake_case						(			self					:  Any):







        '''simple docstring'''



        lowerCAmelCase__     	=					'cpu'  # ensure determinism for the device-dependent torch.Generator
        lowerCAmelCase__     	=					self.get_dummy_components()
        lowerCAmelCase__     	=					StableDiffusionImageVariationPipeline(**lowercase__)
        lowerCAmelCase__     	=					sd_pipe.to(lowercase__)
        sd_pipe.set_progress_bar_config(disable=lowercase__)

        lowerCAmelCase__     	=					self.get_dummy_inputs(lowercase__)
        lowerCAmelCase__     	=					2 * [inputs['image']]
        lowerCAmelCase__     	=					sd_pipe(**lowercase__)

        lowerCAmelCase__     	=					output.images

        lowerCAmelCase__     	=					image[-1, -3:, -3:, -1]

        assert image.shape == (2, 64, 64, 3)
        lowerCAmelCase__     	=					np.array([0.6_892, 0.5_637, 0.5_836, 0.5_771, 0.6_254, 0.6_409, 0.5_580, 0.5_569, 0.5_289])

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




 def 		__snake_case						(			self					:  Optional[int]):







        '''simple docstring'''



        super().test_inference_batch_single_identical(expected_max_diff=3e-3)






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






 '''simple docstring'''






 def 		__snake_case						(			self					:  List[Any]):







        '''simple docstring'''



        super().tearDown()
        gc.collect()
        torch.cuda.empty_cache()




 def 		__snake_case						(			self					:  List[Any]			,    lowercase__					:  List[str]			,    lowercase__					:  Optional[int]="cpu"			,    lowercase__					:  List[str]=torch.floataa			,    lowercase__					:  Dict=0):







        '''simple docstring'''



        lowerCAmelCase__     	=					torch.Generator(device=lowercase__).manual_seed(lowercase__)
        lowerCAmelCase__     	=					load_image(
            'https://huggingface.co/datasets/diffusers/test-arrays/resolve/main'
            '/stable_diffusion_imgvar/input_image_vermeer.png')
        lowerCAmelCase__     	=					np.random.RandomState(lowercase__).standard_normal((1, 4, 64, 64))
        lowerCAmelCase__     	=					torch.from_numpy(lowercase__).to(device=lowercase__			,    dtype=lowercase__)
        lowerCAmelCase__     	=					{
            'image': init_image,
            'latents': latents,
            'generator': generator,
            'num_inference_steps': 3,
            'guidance_scale': 7.5,
            'output_type': 'numpy',
        }
        return inputs




 def 		__snake_case						(			self					:  List[str]):







        '''simple docstring'''



        lowerCAmelCase__     	=					StableDiffusionImageVariationPipeline.from_pretrained(
            'lambdalabs/sd-image-variations-diffusers'			,    safety_checker=lowercase__)
        lowerCAmelCase__     	=					sd_pipe.to(lowercase__)
        sd_pipe.set_progress_bar_config(disable=lowercase__)

        lowerCAmelCase__     	=					self.get_inputs(lowercase__)
        lowerCAmelCase__     	=					sd_pipe(**lowercase__).images
        lowerCAmelCase__     	=					image[0, -3:, -3:, -1].flatten()

        assert image.shape == (1, 512, 512, 3)
        lowerCAmelCase__     	=					np.array([0.84_491, 0.90_789, 0.75_708, 0.78_734, 0.83_485, 0.70_099, 0.66_938, 0.68_727, 0.61_379])
        assert np.abs(image_slice - expected_slice).max() < 6e-3




 def 		__snake_case						(			self					:  Optional[Any]):







        '''simple docstring'''



        lowerCAmelCase__     	=					0

        def callback_fn(lowercase__					:  int			,    lowercase__					:  int			,    lowercase__					:  torch.FloatTensor) -> None:
               lowerCAmelCase__     	=					True
               nonlocal number_of_steps
               number_of_steps += 1
               if step == 1:
                      lowerCAmelCase__     	=					latents.detach().cpu().numpy()
                      assert latents.shape == (1, 4, 64, 64)
                      lowerCAmelCase__     	=					latents[0, -3:, -3:, -1]
                      lowerCAmelCase__     	=					np.array(
                          [-0.1_621, 0.2_837, -0.7_979, -0.1_221, -1.3_057, 0.7_681, -2.1_191, 0.0_464, 1.6_309])

                      assert np.abs(latents_slice.flatten() - expected_slice).max() < 5e-2
               elif step == 2:
                      lowerCAmelCase__     	=					latents.detach().cpu().numpy()
                      assert latents.shape == (1, 4, 64, 64)
                      lowerCAmelCase__     	=					latents[0, -3:, -3:, -1]
                      lowerCAmelCase__     	=					np.array([0.6_299, 1.7_500, 1.1_992, -2.1_582, -1.8_994, 0.7_334, -0.7_090, 1.0_137, 1.5_273])

                      assert np.abs(latents_slice.flatten() - expected_slice).max() < 5e-2

        lowerCAmelCase__     	=					False

        lowerCAmelCase__     	=					StableDiffusionImageVariationPipeline.from_pretrained(
            'fusing/sd-image-variations-diffusers'			,    safety_checker=lowercase__			,    torch_dtype=torch.floataa			,    )
        pipe.to(lowercase__)
        pipe.set_progress_bar_config(disable=lowercase__)
        pipe.enable_attention_slicing()

        lowerCAmelCase__     	=					self.get_inputs(lowercase__			,    dtype=torch.floataa)
        pipe(**lowercase__			,    callback=lowercase__			,    callback_steps=1)
        assert callback_fn.has_been_called
        assert number_of_steps == inputs["num_inference_steps"]




 def 		__snake_case						(			self					:  str):







        '''simple docstring'''



        torch.cuda.empty_cache()
        torch.cuda.reset_max_memory_allocated()
        torch.cuda.reset_peak_memory_stats()

        lowerCAmelCase__     	=					'fusing/sd-image-variations-diffusers'
        lowerCAmelCase__     	=					StableDiffusionImageVariationPipeline.from_pretrained(
            lowercase__			,    safety_checker=lowercase__			,    torch_dtype=torch.floataa)
        lowerCAmelCase__     	=					pipe.to(lowercase__)
        pipe.set_progress_bar_config(disable=lowercase__)
        pipe.enable_attention_slicing(1)
        pipe.enable_sequential_cpu_offload()

        lowerCAmelCase__     	=					self.get_inputs(lowercase__			,    dtype=torch.floataa)
        lowerCAmelCase__     	=					pipe(**lowercase__)

        lowerCAmelCase__     	=					torch.cuda.max_memory_allocated()
        # make sure that less than 2.6 GB is allocated
        assert mem_bytes < 2.6 * 10**9






@nightly
@require_torch_gpu
class 				a_ (						unittest.TestCase						):






 '''simple docstring'''






 def 		__snake_case						(			self					:  str):







        '''simple docstring'''



        super().tearDown()
        gc.collect()
        torch.cuda.empty_cache()




 def 		__snake_case						(			self					:  int			,    lowercase__					:  List[str]			,    lowercase__					:  Dict="cpu"			,    lowercase__					:  Optional[int]=torch.floataa			,    lowercase__					:  Optional[Any]=0):







        '''simple docstring'''



        lowerCAmelCase__     	=					torch.Generator(device=lowercase__).manual_seed(lowercase__)
        lowerCAmelCase__     	=					load_image(
            'https://huggingface.co/datasets/diffusers/test-arrays/resolve/main'
            '/stable_diffusion_imgvar/input_image_vermeer.png')
        lowerCAmelCase__     	=					np.random.RandomState(lowercase__).standard_normal((1, 4, 64, 64))
        lowerCAmelCase__     	=					torch.from_numpy(lowercase__).to(device=lowercase__			,    dtype=lowercase__)
        lowerCAmelCase__     	=					{
            'image': init_image,
            'latents': latents,
            'generator': generator,
            'num_inference_steps': 50,
            'guidance_scale': 7.5,
            'output_type': 'numpy',
        }
        return inputs




 def 		__snake_case						(			self					:  Optional[Any]):







        '''simple docstring'''



        lowerCAmelCase__     	=					StableDiffusionImageVariationPipeline.from_pretrained('fusing/sd-image-variations-diffusers')
        sd_pipe.to(lowercase__)
        sd_pipe.set_progress_bar_config(disable=lowercase__)

        lowerCAmelCase__     	=					self.get_inputs(lowercase__)
        lowerCAmelCase__     	=					sd_pipe(**lowercase__).images[0]

        lowerCAmelCase__     	=					load_numpy(
            'https://huggingface.co/datasets/diffusers/test-arrays/resolve/main'
            '/stable_diffusion_imgvar/lambdalabs_variations_pndm.npy')
        lowerCAmelCase__     	=					np.abs(expected_image - image).max()
        assert max_diff < 1e-3




 def 		__snake_case						(			self					:  Tuple):







        '''simple docstring'''



        lowerCAmelCase__     	=					StableDiffusionImageVariationPipeline.from_pretrained('fusing/sd-image-variations-diffusers')
        lowerCAmelCase__     	=					DPMSolverMultistepScheduler.from_config(sd_pipe.scheduler.config)
        sd_pipe.to(lowercase__)
        sd_pipe.set_progress_bar_config(disable=lowercase__)

        lowerCAmelCase__     	=					self.get_inputs(lowercase__)
        lowerCAmelCase__     	=					25
        lowerCAmelCase__     	=					sd_pipe(**lowercase__).images[0]

        lowerCAmelCase__     	=					load_numpy(
            'https://huggingface.co/datasets/diffusers/test-arrays/resolve/main'
            '/stable_diffusion_imgvar/lambdalabs_variations_dpm_multi.npy')
        lowerCAmelCase__     	=					np.abs(expected_image - image).max()
        assert max_diff < 1e-3
