



import itertools
import random
import unittest

import numpy as np

from transformers import EncodecFeatureExtractor
from transformers.testing_utils import require_torch
from transformers.utils.import_utils import is_torch_available

from ...test_sequence_feature_extraction_common import SequenceFeatureExtractionTestMixin


if is_torch_available():
      import torch


UpperCamelCase_									=   random.Random()







def       lowerCamelCase_  (    _a       :						Any   ,		_a       :						List[Any]=1.0   ,		_a       :						Tuple=None   ,		_a       :						Optional[int]=None    ):



    '''simple docstring'''




    if rng is None:
        UpperCAmelCase_     :       int  		=					global_rng

    UpperCAmelCase_     :       Tuple  		=					[]
    for batch_idx in range(shape[0]    ):
        values.append([]    )
        for _ in range(shape[1]    ):
            values[-1].append(rng.random() * scale    )

    return values







@require_torch
class       _snake_case				(     unittest.TestCase ):






   '''simple docstring'''






   def __init__(    self:     Optional[Any]  ,lowerCamelCase_:     str  ,lowerCamelCase_:     List[str]=7  ,lowerCamelCase_:     int=400  ,lowerCamelCase_:     Tuple=2000  ,lowerCamelCase_:     Optional[Any]=1  ,lowerCamelCase_:     Optional[int]=0.0  ,lowerCamelCase_:     Dict=24000  ,lowerCamelCase_:     Dict=True  ,)	->      Any:
       UpperCAmelCase_     :       List[str]  		=					parent
       UpperCAmelCase_     :       List[str]  		=					batch_size
       UpperCAmelCase_     :       List[str]  		=					min_seq_length
       UpperCAmelCase_     :       Optional[int]  		=					max_seq_length
       UpperCAmelCase_     :       int  		=					(self.max_seq_length - self.min_seq_length) // (self.batch_size - 1)
       UpperCAmelCase_     :       int  		=					feature_size
       UpperCAmelCase_     :       List[str]  		=					padding_value
       UpperCAmelCase_     :       Optional[Any]  		=					sampling_rate
       UpperCAmelCase_     :       Optional[int]  		=					return_attention_mask


   def       A__    (    self:     Optional[int]							)	->      Optional[Any]:
       return {
           "feature_size": self.feature_size,
           "padding_value": self.padding_value,
           "sampling_rate": self.sampling_rate,
           "return_attention_mask": self.return_attention_mask,
       }






   def       A__    (    self:     List[Any]  ,lowerCamelCase_:     List[Any]=False  ,lowerCamelCase_:     Dict=False							)	->      List[str]:
       def _flatten(lowerCamelCase_:     Optional[int]							):
           return list(itertools.chain(*lowerCamelCase_							)							)

       if equal_length:
           UpperCAmelCase_     :       Optional[int]  		=					floats_list((self.batch_size, self.max_seq_length)							)
       else:
           # make sure that inputs increase in size
           UpperCAmelCase_     :       List[Any]  		=					[
               _flatten(floats_list((x, self.feature_size)							)							)
               for x in range(self.min_seq_length  ,self.max_seq_length  ,self.seq_length_diff							)
           ]

       if numpify:
           UpperCAmelCase_     :       str  		=					[np.asarray(lowerCamelCase_							) for x in audio_inputs]

       return audio_inputs







@require_torch
class       _snake_case				(     __snake_case ,							unittest.TestCase ):






   '''simple docstring'''






   A__      : List[Any]       =						EncodecFeatureExtractor


   def       A__    (    self:     Union[str, Any]							)	->      Union[str, Any]:
       UpperCAmelCase_     :       List[str]  		=					EnCodecFeatureExtractionTester(self							)


   def       A__    (    self:     List[Any]							)	->      Dict:
       # Tests that all call wrap to encode_plus and batch_encode_plus
       UpperCAmelCase_     :       List[Any]  		=					self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()							)
       # create three inputs of length 800, 1000, and 1200
       UpperCAmelCase_     :       str  		=					[floats_list((1, x)							)[0] for x in range(800  ,1400  ,200							)]
       UpperCAmelCase_     :       int  		=					[np.asarray(lowerCamelCase_							) for audio_input in audio_inputs]

       # Test not batched input
       UpperCAmelCase_     :       Union[str, Any]  		=					feat_extract(audio_inputs[0]  ,return_tensors="""np"""							).input_values
       UpperCAmelCase_     :       Dict  		=					feat_extract(np_audio_inputs[0]  ,return_tensors="""np"""							).input_values
       self.assertTrue(np.allclose(lowerCamelCase_  ,lowerCamelCase_  ,atol=1e-3							)							)

       # Test batched
       UpperCAmelCase_     :       Tuple  		=					feat_extract(lowerCamelCase_  ,padding=lowerCamelCase_  ,return_tensors="""np"""							).input_values
       UpperCAmelCase_     :       Any  		=					feat_extract(lowerCamelCase_  ,padding=lowerCamelCase_  ,return_tensors="""np"""							).input_values
       for enc_seq_a, enc_seq_a in zip(lowerCamelCase_  ,lowerCamelCase_							):
           self.assertTrue(np.allclose(lowerCamelCase_  ,lowerCamelCase_  ,atol=1e-3							)							)


   def       A__    (    self:     str							)	->      List[str]:
       UpperCAmelCase_     :       Union[str, Any]  		=					self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()							)
       UpperCAmelCase_     :       Union[str, Any]  		=					np.random.rand(100							).astype(np.floataa							)
       UpperCAmelCase_     :       Union[str, Any]  		=					np_audio_inputs.tolist()

       for inputs in [py_audio_inputs, np_audio_inputs]:
           UpperCAmelCase_     :       Dict  		=					feature_extractor.pad([{"""input_values""": inputs}]  ,return_tensors="""np"""							)
           self.assertTrue(np_processed.input_values.dtype == np.floataa							)
           UpperCAmelCase_     :       Dict  		=					feature_extractor.pad([{"""input_values""": inputs}]  ,return_tensors="""pt"""							)
           self.assertTrue(pt_processed.input_values.dtype == torch.floataa							)


   def       A__    (    self:     List[str]  ,lowerCamelCase_:     Union[str, Any]							)	->      Optional[Any]:
       from datasets import load_dataset

       UpperCAmelCase_     :       int  		=					load_dataset("""hf-internal-testing/librispeech_asr_dummy"""  ,"""clean"""  ,split="""validation"""							)
       # automatic decoding with librispeech
       UpperCAmelCase_     :       int  		=					ds.sort("""id"""							).select(range(lowerCamelCase_							)							)[:num_samples]["""audio"""]

       return [x["array"] for x in audio_samples]


   def       A__    (    self:     Optional[int]							)	->      Union[str, Any]:
       # fmt: off
       UpperCAmelCase_     :       Dict  		=					torch.tensor(
           [2.3804e-03, 2.0752e-03, 1.9836e-03, 2.1057e-03, 1.6174e-03,
            3.0518e-04, 9.1553e-05, 3.3569e-04, 9.7656e-04, 1.8311e-03,
            2.0142e-03, 2.1057e-03, 1.7395e-03, 4.5776e-04, -3.9673e-04,
            4.5776e-04, 1.0071e-03, 9.1553e-05, 4.8828e-04, 1.1597e-03,
            7.3242e-04, 9.4604e-04, 1.8005e-03, 1.8311e-03, 8.8501e-04,
            4.2725e-04, 4.8828e-04, 7.3242e-04, 1.0986e-03, 2.1057e-03]							)
       # fmt: on
       UpperCAmelCase_     :       Optional[Any]  		=					self._load_datasamples(1							)
       UpperCAmelCase_     :       Union[str, Any]  		=					EncodecFeatureExtractor()
       UpperCAmelCase_     :       List[Any]  		=					feature_extractor(lowerCamelCase_  ,return_tensors="""pt"""							).input_values
       self.assertEquals(input_values.shape  ,(1, 1, 93680)							)
       self.assertTrue(torch.allclose(input_values[0, 0, :30]  ,lowerCamelCase_  ,atol=1e-6							)							)


   def       A__    (    self:     List[str]							)	->      Dict:
       # fmt: off
       UpperCAmelCase_     :       Optional[int]  		=					torch.tensor(
           [2.3804e-03, 2.0752e-03, 1.9836e-03, 2.1057e-03, 1.6174e-03,
            3.0518e-04, 9.1553e-05, 3.3569e-04, 9.7656e-04, 1.8311e-03,
            2.0142e-03, 2.1057e-03, 1.7395e-03, 4.5776e-04, -3.9673e-04,
            4.5776e-04, 1.0071e-03, 9.1553e-05, 4.8828e-04, 1.1597e-03,
            7.3242e-04, 9.4604e-04, 1.8005e-03, 1.8311e-03, 8.8501e-04,
            4.2725e-04, 4.8828e-04, 7.3242e-04, 1.0986e-03, 2.1057e-03]							)
       # fmt: on
       UpperCAmelCase_     :       Tuple  		=					self._load_datasamples(1							)
       UpperCAmelCase_     :       List[Any]  		=					[np.tile(input_audio[0][None]  ,reps=(2, 1)							)]
       input_audio[0][1] *= 0.5
       UpperCAmelCase_     :       int  		=					EncodecFeatureExtractor(feature_size=2							)
       UpperCAmelCase_     :       int  		=					feature_extractor(lowerCamelCase_  ,return_tensors="""pt"""							).input_values
       self.assertEquals(input_values.shape  ,(1, 2, 93680)							)
       self.assertTrue(torch.allclose(input_values[0, 0, :30]  ,lowerCamelCase_  ,atol=1e-6							)							)
       self.assertTrue(torch.allclose(input_values[0, 1, :30]  ,EXPECTED_INPUT_VALUES * 0.5  ,atol=1e-6							)							)






   def       A__    (    self:     Optional[int]							)	->      List[str]:
       UpperCAmelCase_     :       Dict  		=					self._load_datasamples(2							)
       # would be easier if the stride was like
       UpperCAmelCase_     :       Any  		=					EncodecFeatureExtractor(feature_size=1  ,chunk_length_s=1  ,overlap=0.0_1							)

       # pad and trunc raise an error ?
       with self.assertRaisesRegex(
           lowerCamelCase_  ,"""^Both padding and truncation were set. Make sure you only set one.$"""  ,):
           UpperCAmelCase_     :       Optional[int]  		=					feature_extractor(
               lowerCamelCase_  ,padding="""max_length"""  ,truncation=lowerCamelCase_  ,return_tensors="""pt"""							).input_values

       # truncate to chunk
       UpperCAmelCase_     :       str  		=					feature_extractor(lowerCamelCase_  ,truncation=lowerCamelCase_  ,return_tensors="""pt"""							).input_values
       self.assertEquals(truncated_outputs.shape  ,(2, 1, 71520)							)  # 2 chunks

       # force truncate to max_length
       UpperCAmelCase_     :       Optional[Any]  		=					feature_extractor(
           lowerCamelCase_  ,truncation=lowerCamelCase_  ,max_length=48000  ,return_tensors="""pt"""							).input_values
       self.assertEquals(truncated_outputs.shape  ,(2, 1, 48000)							)

       # pad to chunk
       UpperCAmelCase_     :       Dict  		=					feature_extractor(lowerCamelCase_  ,padding=lowerCamelCase_  ,return_tensors="""pt"""							).input_values
       self.assertEquals(padded_outputs.shape  ,(2, 1, 95280)							)

       # pad to chunk
       UpperCAmelCase_     :       Optional[Any]  		=					feature_extractor(lowerCamelCase_  ,return_tensors="""pt"""							).input_values
       self.assertEquals(truncated_outputs.shape  ,(2, 1, 95280)							)

       # force pad to max length
       UpperCAmelCase_     :       str  		=					feature_extractor(
           lowerCamelCase_  ,padding="""max_length"""  ,max_length=100000  ,return_tensors="""pt"""							).input_values
       self.assertEquals(truncated_outputs.shape  ,(2, 1, 100000)							)

       # force no pad
       with self.assertRaisesRegex(
           lowerCamelCase_  ,"""^Unable to create tensor, you should probably activate padding with 'padding=True' to have batched tensors with the same length.$"""  ,):
           UpperCAmelCase_     :       Dict  		=					feature_extractor(lowerCamelCase_  ,padding=lowerCamelCase_  ,return_tensors="""pt"""							).input_values

       UpperCAmelCase_     :       Any  		=					feature_extractor(input_audio[0]  ,padding=lowerCamelCase_  ,return_tensors="""pt"""							).input_values
       self.assertEquals(truncated_outputs.shape  ,(1, 1, 93680)							)

       # no pad if no chunk_length_s
       UpperCAmelCase_     :       str  		=					None
       with self.assertRaisesRegex(
           lowerCamelCase_  ,"""^Unable to create tensor, you should probably activate padding with 'padding=True' to have batched tensors with the same length.$"""  ,):
           UpperCAmelCase_     :       Dict  		=					feature_extractor(lowerCamelCase_  ,padding=lowerCamelCase_  ,return_tensors="""pt"""							).input_values

       UpperCAmelCase_     :       Any  		=					feature_extractor(input_audio[0]  ,padding=lowerCamelCase_  ,return_tensors="""pt"""							).input_values
       self.assertEquals(truncated_outputs.shape  ,(1, 1, 93680)							)

       # no pad if no overlap
       UpperCAmelCase_     :       int  		=					2
       UpperCAmelCase_     :       Optional[Any]  		=					None
       with self.assertRaisesRegex(
           lowerCamelCase_  ,"""^Unable to create tensor, you should probably activate padding with 'padding=True' to have batched tensors with the same length.$"""  ,):
           UpperCAmelCase_     :       int  		=					feature_extractor(lowerCamelCase_  ,padding=lowerCamelCase_  ,return_tensors="""pt"""							).input_values

       UpperCAmelCase_     :       Optional[Any]  		=					feature_extractor(input_audio[0]  ,padding=lowerCamelCase_  ,return_tensors="""pt"""							).input_values
       self.assertEquals(truncated_outputs.shape  ,(1, 1, 93680)							)


"""simple docstring"""

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 			_lowercase  (							__UpperCAmelCase							,			__UpperCAmelCase							,			__UpperCAmelCase							,			unittest.TestCase						):
	lowercase_								=      StableDiffusionImageVariationPipeline
	lowercase_								=      IMAGE_VARIATION_PARAMS
	lowercase_								=      IMAGE_VARIATION_BATCH_PARAMS
	lowercase_								=      frozenset([]						)
	# TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
	lowercase_								=      frozenset([]						)




	def      _UpperCamelCase       (					self   )						->				int:
			torch.manual_seed(0   )
			lowerCamelCase	:     List[str]            =       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	:     List[Any]            =       PNDMScheduler(skip_prk_steps=UpperCAmelCase_   )
			torch.manual_seed(0   )
			lowerCamelCase	:     str            =       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	:     int            =       CLIPVisionConfig(
			    hidden_size=32						,					projection_dim=32						,					intermediate_size=37						,					layer_norm_eps=1E-0_5						,					num_attention_heads=4						,					num_hidden_layers=5						,					image_size=32						,					patch_size=4						,					)
			lowerCamelCase	:     Optional[int]            =       CLIPVisionModelWithProjection(UpperCAmelCase_   )
			lowerCamelCase	:     int            =       CLIPImageProcessor(crop_size=32						,					size=32   )

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




	def      _UpperCamelCase       (					self						,					UpperCAmelCase_						,					UpperCAmelCase_=0   )						->				Optional[int]:
			lowerCamelCase	:     int            =       floats_tensor((1, 3, 32, 32)						,					rng=random.Random(UpperCAmelCase_   )   )
			lowerCamelCase	:     List[str]            =       image.cpu().permute(0						,					2						,					3						,					1   )[0]
			lowerCamelCase	:     str            =       Image.fromarray(np.uinta(UpperCAmelCase_   )   ).convert('RGB'   ).resize((32, 32)   )
			if str(UpperCAmelCase_   ).startswith('mps'   ):
					lowerCamelCase	:     Tuple            =       torch.manual_seed(UpperCAmelCase_   )
			else:
					lowerCamelCase	:     int            =       torch.Generator(device=UpperCAmelCase_   ).manual_seed(UpperCAmelCase_   )
			lowerCamelCase	:     Optional[Any]            =       {
			    'image': image,
			    'generator': generator,
			    'num_inference_steps': 2,
			    'guidance_scale': 6.0,
			    'output_type': 'numpy',
			}
			return inputs




	def      _UpperCamelCase       (					self   )						->				str:
			lowerCamelCase	:     Tuple            =       'cpu'  # ensure determinism for the device-dependent torch.Generator
			lowerCamelCase	:     List[str]            =       self.get_dummy_components()
			lowerCamelCase	:     Tuple            =       StableDiffusionImageVariationPipeline(**UpperCAmelCase_   )
			lowerCamelCase	:     Optional[Any]            =       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	:     Dict            =       image[0, -3:, -3:, -1]

			assert image.shape == (1, 64, 64, 3)
			lowerCamelCase	:     str            =       np.array([0.5239, 0.5723, 0.4796, 0.5049, 0.5550, 0.4685, 0.5329, 0.4891, 0.4921]   )

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




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

			lowerCamelCase	:     Any            =       self.get_dummy_inputs(UpperCAmelCase_   )
			lowerCamelCase	:     Any            =       2 * [inputs['image']]
			lowerCamelCase	:     Dict            =       sd_pipe(**UpperCAmelCase_   )

			lowerCamelCase	:     List[str]            =       output.images

			lowerCamelCase	:     Dict            =       image[-1, -3:, -3:, -1]

			assert image.shape == (2, 64, 64, 3)
			lowerCamelCase	:     Dict            =       np.array([0.6892, 0.5637, 0.5836, 0.5771, 0.6254, 0.6409, 0.5580, 0.5569, 0.5289]   )

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




	def      _UpperCamelCase       (					self   )						->				List[str]:
			super().test_inference_batch_single_identical(expected_max_diff=3E-3   )






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




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




	def      _UpperCamelCase       (					self						,					UpperCAmelCase_						,					UpperCAmelCase_="cpu"						,					UpperCAmelCase_=torch.floataa						,					UpperCAmelCase_=0   )						->				Optional[int]:
			lowerCamelCase	:     List[Any]            =       torch.Generator(device=UpperCAmelCase_   ).manual_seed(UpperCAmelCase_   )
			lowerCamelCase	:     List[str]            =       load_image(
			    'https://huggingface.co/datasets/diffusers/test-arrays/resolve/main'
			    '/stable_diffusion_imgvar/input_image_vermeer.png'   )
			lowerCamelCase	:     Union[str, Any]            =       np.random.RandomState(UpperCAmelCase_   ).standard_normal((1, 4, 64, 64)   )
			lowerCamelCase	:     int            =       torch.from_numpy(UpperCAmelCase_   ).to(device=UpperCAmelCase_						,					dtype=UpperCAmelCase_   )
			lowerCamelCase	:     Optional[Any]            =       {
			    'image': init_image,
			    'latents': latents,
			    'generator': generator,
			    'num_inference_steps': 3,
			    'guidance_scale': 7.5,
			    'output_type': 'numpy',
			}
			return inputs




	def      _UpperCamelCase       (					self   )						->				Optional[int]:
			lowerCamelCase	:     List[Any]            =       StableDiffusionImageVariationPipeline.from_pretrained(
			    'lambdalabs/sd-image-variations-diffusers'						,					safety_checker=UpperCAmelCase_   )
			lowerCamelCase	:     Dict            =       sd_pipe.to(UpperCAmelCase_   )
			sd_pipe.set_progress_bar_config(disable=UpperCAmelCase_   )

			lowerCamelCase	:     Optional[int]            =       self.get_inputs(UpperCAmelCase_   )
			lowerCamelCase	:     Any            =       sd_pipe(**UpperCAmelCase_   ).images
			lowerCamelCase	:     Union[str, Any]            =       image[0, -3:, -3:, -1].flatten()

			assert image.shape == (1, 512, 512, 3)
			lowerCamelCase	:     List[str]            =       np.array([0.84491, 0.90789, 0.75708, 0.78734, 0.83485, 0.70099, 0.66938, 0.68727, 0.61379]   )
			assert np.abs(image_slice - expected_slice   ).max() < 6E-3




	def      _UpperCamelCase       (					self   )						->				List[Any]:
			lowerCamelCase	:     Optional[Any]            =       0

			def callback_fn(UpperCAmelCase_						,					UpperCAmelCase_						,					UpperCAmelCase_   ) -> None:
					lowerCamelCase	:     int            =       True
					nonlocal number_of_steps
					number_of_steps += 1
					if step == 1:
							lowerCamelCase	:     List[Any]            =       latents.detach().cpu().numpy()
							assert latents.shape == (1, 4, 64, 64)
							lowerCamelCase	:     Dict            =       latents[0, -3:, -3:, -1]
							lowerCamelCase	:     List[Any]            =       np.array(
							    [-0.1621, 0.2837, -0.7979, -0.1221, -1.3057, 0.7681, -2.1191, 0.0464, 1.6309]   )

							assert np.abs(latents_slice.flatten() - expected_slice   ).max() < 5E-2
					elif step == 2:
							lowerCamelCase	:     Union[str, Any]            =       latents.detach().cpu().numpy()
							assert latents.shape == (1, 4, 64, 64)
							lowerCamelCase	:     Optional[int]            =       latents[0, -3:, -3:, -1]
							lowerCamelCase	:     Optional[Any]            =       np.array([0.6299, 1.7500, 1.1992, -2.1582, -1.8994, 0.7334, -0.7090, 1.0137, 1.5273]   )

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

			lowerCamelCase	:     List[str]            =       False

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

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




	def      _UpperCamelCase       (					self   )						->				str:
			torch.cuda.empty_cache()
			torch.cuda.reset_max_memory_allocated()
			torch.cuda.reset_peak_memory_stats()

			lowerCamelCase	:     Any            =       'fusing/sd-image-variations-diffusers'
			lowerCamelCase	:     Optional[int]            =       StableDiffusionImageVariationPipeline.from_pretrained(
			    UpperCAmelCase_						,					safety_checker=UpperCAmelCase_						,					torch_dtype=torch.floataa   )
			lowerCamelCase	:     Any            =       pipe.to(UpperCAmelCase_   )
			pipe.set_progress_bar_config(disable=UpperCAmelCase_   )
			pipe.enable_attention_slicing(1   )
			pipe.enable_sequential_cpu_offload()

			lowerCamelCase	:     Optional[int]            =       self.get_inputs(UpperCAmelCase_						,					dtype=torch.floataa   )
			lowerCamelCase	:     int            =       pipe(**UpperCAmelCase_   )

			lowerCamelCase	:     List[Any]            =       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 			_lowercase  (							unittest.TestCase						):




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




	def      _UpperCamelCase       (					self						,					UpperCAmelCase_						,					UpperCAmelCase_="cpu"						,					UpperCAmelCase_=torch.floataa						,					UpperCAmelCase_=0   )						->				Any:
			lowerCamelCase	:     List[str]            =       torch.Generator(device=UpperCAmelCase_   ).manual_seed(UpperCAmelCase_   )
			lowerCamelCase	:     Optional[Any]            =       load_image(
			    'https://huggingface.co/datasets/diffusers/test-arrays/resolve/main'
			    '/stable_diffusion_imgvar/input_image_vermeer.png'   )
			lowerCamelCase	:     Union[str, Any]            =       np.random.RandomState(UpperCAmelCase_   ).standard_normal((1, 4, 64, 64)   )
			lowerCamelCase	:     List[Any]            =       torch.from_numpy(UpperCAmelCase_   ).to(device=UpperCAmelCase_						,					dtype=UpperCAmelCase_   )
			lowerCamelCase	:     List[Any]            =       {
			    'image': init_image,
			    'latents': latents,
			    'generator': generator,
			    'num_inference_steps': 50,
			    'guidance_scale': 7.5,
			    'output_type': 'numpy',
			}
			return inputs




	def      _UpperCamelCase       (					self   )						->				Any:
			lowerCamelCase	:     int            =       StableDiffusionImageVariationPipeline.from_pretrained('fusing/sd-image-variations-diffusers'   )
			sd_pipe.to(UpperCAmelCase_   )
			sd_pipe.set_progress_bar_config(disable=UpperCAmelCase_   )

			lowerCamelCase	:     int            =       self.get_inputs(UpperCAmelCase_   )
			lowerCamelCase	:     int            =       sd_pipe(**UpperCAmelCase_   ).images[0]

			lowerCamelCase	:     List[Any]            =       load_numpy(
			    'https://huggingface.co/datasets/diffusers/test-arrays/resolve/main'
			    '/stable_diffusion_imgvar/lambdalabs_variations_pndm.npy'   )
			lowerCamelCase	:     str            =       np.abs(expected_image - image   ).max()
			assert max_diff < 1E-3




	def      _UpperCamelCase       (					self   )						->				List[str]:
			lowerCamelCase	:     List[Any]            =       StableDiffusionImageVariationPipeline.from_pretrained('fusing/sd-image-variations-diffusers'   )
			lowerCamelCase	:     Optional[int]            =       DPMSolverMultistepScheduler.from_config(sd_pipe.scheduler.config   )
			sd_pipe.to(UpperCAmelCase_   )
			sd_pipe.set_progress_bar_config(disable=UpperCAmelCase_   )

			lowerCamelCase	:     Union[str, Any]            =       self.get_inputs(UpperCAmelCase_   )
			lowerCamelCase	:     Optional[Any]            =       25
			lowerCamelCase	:     List[str]            =       sd_pipe(**UpperCAmelCase_   ).images[0]

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






