code: 
  
  
  
  
  
  
  """simple docstring"""
  
  import itertools
  import os
  import subprocess
  from os.path import dirname
  
  from parameterized import parameterized
  
  from tests.trainer.test_trainer import TrainerIntegrationCommon  # noqa
  from transformers import is_torch_available
  from transformers.testing_utils import (
      TestCasePlus,
      execute_subprocess_async,
      get_gpu_count,
      get_tests_dir,
      require_deepspeed,
      require_torch_gpu,
      slow,
  )
  from transformers.trainer_utils import set_seed
  
  
  if is_torch_available():
  					from tests.trainer.test_trainer import (  # noqa
  					    RegressionModelConfig,
  					    RegressionPreTrainedModel,
  					    get_regression_trainer,
  					)
  
  
  set_seed(42)
  
  lowercase__      =				get_tests_dir("""fixtures""")
  lowercase__      =				os.path.join(dirname(get_tests_dir()))
  lowercase__      =				dirname(os.path.abspath(__file__))
  
  # default torch.distributed port
  lowercase__      =				"""10999"""
  
  lowercase__      =				"""t5-small"""
  
  # *** Working Models ***
  lowercase__      =				"""hf-internal-testing/tiny-albert"""
  lowercase__      =				"""sshleifer/bart-tiny-random"""
  lowercase__      =				"""hf-internal-testing/tiny-bert"""
  lowercase__      =				"""hf-internal-testing/tiny-random-bigbird_pegasus"""
  lowercase__      =				"""hf-internal-testing/tiny-random-big_bird"""
  lowercase__      =				"""hf-internal-testing/tiny-random-blenderbot"""
  lowercase__      =				"""bigscience/bigscience-small-testing"""
  lowercase__      =				"""hf-internal-testing/tiny-random-deberta"""
  lowercase__      =				"""hf-internal-testing/tiny-random-deberta-v2"""
  lowercase__      =				"""sshleifer/tiny-distilbert-base-cased"""
  lowercase__      =				"""hf-internal-testing/tiny-electra"""
  lowercase__      =				"""hf-internal-testing/tiny-random-flaubert"""
  lowercase__      =				"""stas/tiny-wmt19-en-de"""
  lowercase__      =				"""hf-internal-testing/tiny-random-funnel"""
  lowercase__      =				"""sshleifer/tiny-gpt2"""
  lowercase__      =				"""hf-internal-testing/tiny-random-gptj"""
  lowercase__      =				"""hf-internal-testing/tiny-random-gpt_neo"""
  lowercase__      =				"""hf-internal-testing/tiny-layoutlm"""
  lowercase__      =				"""hf-internal-testing/tiny-random-led"""
  lowercase__      =				"""hf-internal-testing/tiny-random-longformer"""
  lowercase__      =				"""stas/tiny-m2m_100"""  # hf tiny model is unsuitable
  lowercase__      =				"""sshleifer/tiny-marian-en-de"""
  lowercase__      =				"""sshleifer/tiny-mbart"""
  lowercase__      =				"""hf-internal-testing/tiny-random-mobilebert"""
  lowercase__      =				"""hf-internal-testing/tiny-random-mpnet"""
  lowercase__      =				"""stas/pegasus-cnn_dailymail-tiny-random"""
  lowercase__      =				"""hf-internal-testing/tiny-random-prophetnet"""
  lowercase__      =				"""sshleifer/tiny-distilroberta-base"""
  lowercase__      =				"""hf-internal-testing/tiny-random-squeezebert"""
  lowercase__      =				"""patrickvonplaten/t5-tiny-random"""
  lowercase__      =				"""hf-internal-testing/tiny-random-t5-v1.1"""
  lowercase__      =				"""hf-internal-testing/tiny-random-vit"""
  lowercase__      =				"""hf-internal-testing/tiny-xlm-roberta"""
  lowercase__      =				"""sshleifer/tiny-xlnet-base-cased"""
  
  
  # *** To Fix ***
  
  
  # *** tiny model issues ***
  # missing model files:
  lowercase__      =				"""hf-internal-testing/tiny-random-mt5"""
  lowercase__      =				"""hf-internal-testing/tiny-random-camembert"""
  lowercase__      =				"""hf-internal-testing/tiny-random-openai-gpt"""
  
  # missing tokenizer files
  lowercase__      =				"""hf-internal-testing/tiny-random-convbert"""
  lowercase__      =				"""hf-internal-testing/tiny-random-layoutlmv2"""
  lowercase__      =				"""hf-internal-testing/tiny-random-hubert"""
  
  # issues with tokenizer
  lowercase__      =				"""hf-internal-testing/tiny-random-ctrl"""
  lowercase__      =				"""hf-internal-testing/tiny-random-transfo-xl"""  # same as ctrl
  
  # other issues with tiny models
  lowercase__      =				"""hf-internal-testing/tiny-random-ibert"""  # multiple issues with either mlm/qa/clas
  lowercase__      =				"""hf-internal-testing/tiny-random-reformer"""  # multiple issues with either mlm/qa/clas
  
  # *** Lacking official examples to test with ***
  # or not working with examples
  lowercase__      =				"""hf-internal-testing/tiny-random-dpr"""
  # - "dpr"  examples/research_projects/rag-end2end-retriever/
  lowercase__      =				"""hf-internal-testing/tiny-random-rag"""
  # - "rag" research_projects
  lowercase__      =				""""""
  # - "luke" Entities classes - no plan to make such example
  lowercase__      =				"""hf-internal-testing/tiny-random-lxmert"""
  # - "lxmert" doesn't work with run_qa.py
  lowercase__      =				"""hf-internal-testing/tiny-random-clip"""
  # - "clip" nothing under pytorch examples - XXX: Suraj is working on adding some - check by end of Sep
  lowercase__      =				"""hf-internal-testing/tiny-random-speech_to_text"""
  # - "speech_to_text", nothing under pytorch examples
  
  
  # *** Reactive mode ***
  # models with low usage, unstable API, things about to change - do nothing about the following until someone runs into a problem
  lowercase__      =				"""hf-internal-testing/tiny-random-tapas"""
  
  def 	__lowerCamelCase			(						__UpperCamelCase=False )		->					str:
  
  		"""simple docstring"""
  
  
  
  
  
  
  
  		lowerCAmelCase_     :							List[str]             =    min(2   , get_gpu_count() ) if distributed else 1
  		lowerCAmelCase_     :							Tuple             =    os.environ.get("DS_TEST_PORT"   , __UpperCamelCase )
  		return f'''deepspeed --num_nodes 1 --num_gpus {num_gpus} --master_port {master_port}'''.split()
  
  def 	__lowerCamelCase			(						)		->					Union[str, Any]:
  
  		"""simple docstring"""
  
  
  
  
  
  
  
  		lowerCAmelCase_     :							Tuple             =    f'''{FIXTURE_DIRECTORY}/tests_samples'''
  		lowerCAmelCase_     :							List[Any]             =    f'''{data_dir_samples}/wmt_en_ro'''
  		lowerCAmelCase_     :							Optional[int]             =    f'''{data_dir_samples}/xsum'''
  		lowerCAmelCase_     :							Any             =    "\n        --do_train\n        --max_train_samples 4\n        --per_device_train_batch_size 2\n        --num_train_epochs 1\n        --fp16\n        --report_to none\n        --overwrite_output_dir\n        ".split()
  
  		# try to cover as many models as possible once (it's enough to run on one task per model)
  		# but need a tiny model for each
  		#
  		# should have "{model_type.upper()}_TINY" corresponding vars defined, e.g., T5_TINY, etc.
  		lowerCAmelCase_     :							Any             =    {
  		    "trans": [
  		        "bart",
  		        "fsmt",
  		        "m2m_100",
  		        "marian",
  		        "mbart",
  		        "t5",
  		        "t5_v1",
  		        # "mt5", missing model files
  		    ],
  		    "sum": [
  		        "pegasus",
  		    ],
  		    "clm": [
  		        "big_bird",
  		        "bigbird_pegasus",
  		        "blenderbot",
  		        "bloom",
  		        "gpt2",
  		        "gpt_neo",
  		        "gptj",
  		        "xlm-roberta",
  		        "prophetnet",
  		        # "camembert", missing model files
  		    ],
  		    "mlm": [
  		        "albert",
  		        "deberta",
  		        "deberta-v2",
  		        "distilbert",
  		        "electra",
  		        "flaubert",
  		        "funnel",
  		        "layoutlm",
  		        # "reformer", # multiple issues with either mlm/qa/clas
  		    ],
  		    "qa": [
  		        "led",
  		        "longformer",
  		        "mobilebert",
  		        "mpnet",
  		        "roberta",
  		        "squeezebert",
  		        # "convbert", # missing tokenizer files
  		        # "layoutlmv2", missing model files
  		    ],
  		    "clas": [
  		        "bert",
  		        "xlnet",
  		        # "hubert", # missing tokenizer files
  		        # "ibert", # multiple issues with either mlm/qa/clas
  		        # "transfo-xl", # tokenizer issues as ctrl
  		        # "ctrl", # tokenizer issues
  		        # "openai-gpt", missing model files
  		        # "tapas", multiple issues
  		    ],
  		    "img_clas": [
  		        "vit",
  		    ],
  		}
  
  		lowerCAmelCase_     :							Optional[int]             =    f'''{ROOT_DIRECTORY}/examples/pytorch'''
  
  		lowerCAmelCase_     :							List[Any]             =    {
  		    "trans": f'''
          {scripts_dir}/translation/run_translation.py
          --train_file {data_dir_wmt}/train.json
          --source_lang en
          --target_lang ro
          ''',
  		    "sum": f'''
          {scripts_dir}/summarization/run_summarization.py
          --train_file {data_dir_xsum}/sample.json
          --max_source_length 12
          --max_target_length 12
          --lang en
          ''',
  		    "clm": f'''
          {scripts_dir}/language-modeling/run_clm.py
          --train_file {FIXTURE_DIRECTORY}/sample_text.txt
          --block_size 8
          ''',
  		    "mlm": f'''
          {scripts_dir}/language-modeling/run_mlm.py
          --train_file {FIXTURE_DIRECTORY}/sample_text.txt
          ''',
  		    "qa": f'''
          {scripts_dir}/question-answering/run_qa.py
          --train_file {data_dir_samples}/SQUAD/sample.json
          ''',
  		    "clas": f'''
          {scripts_dir}/text-classification/run_glue.py
          --train_file {data_dir_samples}/MRPC/train.csv
          --max_seq_length 12
          --task_name MRPC
          ''',
  		    "img_clas": f'''
          {scripts_dir}/image-classification/run_image_classification.py
              --dataset_name hf-internal-testing/cats_vs_dogs_sample
              --remove_unused_columns False
              --max_steps 10
              --image_processor_name {DS_TESTS_DIRECTORY}/vit_feature_extractor.json
          ''',
  		}
  
  		lowerCAmelCase_     :							List[Any]             =    get_launcher(distributed=__UpperCamelCase )
  
  		lowerCAmelCase_     :							Dict             =    {}
  		for task, args in tasks.items():
  				lowerCAmelCase_     :							Any             =    args.split()
  				for model in tasksamodels[task]:
  						lowerCAmelCase_     :							List[str]             =    globals()[f'''{model.upper().replace("-"   , "_" )}_TINY''']
  						lowerCAmelCase_     :							Any             =    f'''--model_name_or_path {model_name}'''.split()
  						lowerCAmelCase_     :							Tuple             =    launcher + args + args_model + args_main
  
  						# # generation special case
  						# if task == "gen":
  						#     launcher = f"deepspeed --num_nodes 1 --num_gpus 1".split()
  						#     args_model += f"--model_type {model}".split()
  						#     cmds[f"{task}_{model}"] = launcher + args + args_model
  						# else:
  
  		return cmds
  
  
  lowercase__      =				make_task_cmds()
  
  lowercase__      =				"""zero2"""
  lowercase__      =				"""zero3"""
  
  lowercase__      =				[ZEROa, ZEROa]
  
  def 	__lowerCamelCase			(						__UpperCamelCase   , __UpperCamelCase   , __UpperCamelCase )		->					List[str]:
  
  		"""simple docstring"""
  
  
  
  
  
  
  
  		lowerCAmelCase_     :							Optional[Any]             =    parameterized.to_safe_name("_".join(str(__UpperCamelCase ) for x in param.args ) )
  		return f'''{func.__name__}_{param_based_name}'''
  
  
  # Cartesian-product of zero stages with models to test
  lowercase__      =				list(itertools.product(stages, task_cmds.keys()))
  
  
  
  @slow
  @require_deepspeed
  @require_torch_gpu
  class 	__lowerCamelCase       (				A__    ):
  
  
  
  
  
  		'''simple docstring'''
  
  
  
  
  
  
  		def      lowerCamelCase (     self							:			Optional[Any]		,    a_							:			List[Any]		,    a_							:			Tuple			):
  				# return a ready to run train cmd
  				if task not in task_cmds:
  						raise ValueError(f'''don\'t know of task {task}, have {task_cmds.keys()}'''			)
  
  				lowerCAmelCase_     :							Any             =    task_cmds[task]
  				lowerCAmelCase_     :							Optional[Any]             =    f'''--deepspeed {self.test_file_dir_str}/ds_config_{stage}.json'''.split()
  
  				lowerCAmelCase_     :							Dict             =    self.get_auto_remove_tmp_dir()
  				lowerCAmelCase_     :							str             =    f'''--output_dir {output_dir}'''.split()
  
  				cmd += args_ds + args_out
  
  				return cmd, output_dir
  
  
  
  
  
  		@parameterized.expand(a_		,    name_func=a_			)
  		def      lowerCamelCase (     self							:			int		,    a_							:			str		,    a_							:			Tuple			):
  				# testing the ability to do a run followed by recovery of full fp32 weights
  
  				lowerCAmelCase_       ,			lowerCAmelCase_     :							Union[str, Any]             =    self.get_task_cmd(a_		,    a_			)
  
  				# 1. generate the checkpoint
  				cmd += "--save_steps 1".split()
  				# keep for quick debug
  				# print(" ".join([f"\nPYTHONPATH={self.src_dir_str}"] + cmd)); die
  				execute_subprocess_async(a_		,    env=self.get_env()			)
  
  				# 2. test that the fp32 weights get reconsolidated
  				lowerCAmelCase_     :							List[str]             =    f'''{output_dir}/checkpoint-1'''
  				lowerCAmelCase_     :							Optional[int]             =    f'''{chkpt_dir}/out.bin'''
  				lowerCAmelCase_     :							Dict             =    f'''{chkpt_dir}/zero_to_fp32.py {chkpt_dir} {recovered_model_path}'''
  				# keep for quick debug
  				# print(" ".join([f"\nPYTHONPATH={self.src_dir_str}"] +cmd)); die
  				subprocess.check_call(a_		,    shell=a_			)
  				assert os.path.exists(a_			), f'''{recovered_model_path} was not found'''
  
  				# possibly could also test that the resulting saved model is usable but given that we use
  				# random models we won't know if it's any good
  
style_context: 
  
  
  
  
  
  
  """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 	__lowerCamelCase       (				A__				,							A__				,							A__				,							unittest.TestCase    ):
  
  
  
  
  
  		'''simple docstring'''
  
  
  
  
  
  
  		a_			:       Union[str, Any]      =							StableDiffusionImageVariationPipeline
  		a_			:       Tuple      =							IMAGE_VARIATION_PARAMS
  		a_			:       Optional[Any]      =							IMAGE_VARIATION_BATCH_PARAMS
  		a_			:       int      =							frozenset([]    )
  		# TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
  		a_			:       Optional[int]      =							frozenset([]    )
  
  
  
  
  
  		def      lowerCamelCase (     self							:			Tuple			):
  				torch.manual_seed(0			)
  				lowerCAmelCase_     :							List[Any]             =    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_     :							int             =    PNDMScheduler(skip_prk_steps=a_			)
  				torch.manual_seed(0			)
  				lowerCAmelCase_     :							List[Any]             =    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_     :							Dict             =    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_     :							List[Any]             =    CLIPVisionModelWithProjection(a_			)
  				lowerCAmelCase_     :							Union[str, Any]             =    CLIPImageProcessor(crop_size=32		,    size=32			)
  
  				lowerCAmelCase_     :							List[Any]             =    {
  				    "unet": unet,
  				    "scheduler": scheduler,
  				    "vae": vae,
  				    "image_encoder": image_encoder,
  				    "feature_extractor": feature_extractor,
  				    "safety_checker": None,
  				}
  				return components
  
  
  
  
  
  		def      lowerCamelCase (     self							:			Optional[int]		,    a_							:			Tuple		,    a_							:			str=0			):
  				lowerCAmelCase_     :							List[Any]             =    floats_tensor((1, 3, 32, 32)		,    rng=random.Random(a_			)			)
  				lowerCAmelCase_     :							int             =    image.cpu().permute(0		,    2		,    3		,    1			)[0]
  				lowerCAmelCase_     :							List[str]             =    Image.fromarray(np.uinta(a_			)			).convert("RGB"			).resize((32, 32)			)
  				if str(a_			).startswith("mps"			):
  						lowerCAmelCase_     :							Optional[int]             =    torch.manual_seed(a_			)
  				else:
  						lowerCAmelCase_     :							Optional[Any]             =    torch.Generator(device=a_			).manual_seed(a_			)
  				lowerCAmelCase_     :							str             =    {
  				    "image": image,
  				    "generator": generator,
  				    "num_inference_steps": 2,
  				    "guidance_scale": 6.0,
  				    "output_type": "numpy",
  				}
  				return inputs
  
  
  
  
  
  		def      lowerCamelCase (     self							:			Any			):
  				lowerCAmelCase_     :							int             =    "cpu"  # ensure determinism for the device-dependent torch.Generator
  				lowerCAmelCase_     :							Dict             =    self.get_dummy_components()
  				lowerCAmelCase_     :							Union[str, Any]             =    StableDiffusionImageVariationPipeline(**a_			)
  				lowerCAmelCase_     :							int             =    sd_pipe.to(a_			)
  				sd_pipe.set_progress_bar_config(disable=a_			)
  
  				lowerCAmelCase_     :							Tuple             =    self.get_dummy_inputs(a_			)
  				lowerCAmelCase_     :							List[str]             =    sd_pipe(**a_			).images
  				lowerCAmelCase_     :							List[Any]             =    image[0, -3:, -3:, -1]
  
  				assert image.shape == (1, 64, 64, 3)
  				lowerCAmelCase_     :							List[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      lowerCamelCase (     self							:			Union[str, Any]			):
  				lowerCAmelCase_     :							int             =    "cpu"  # ensure determinism for the device-dependent torch.Generator
  				lowerCAmelCase_     :							Optional[Any]             =    self.get_dummy_components()
  				lowerCAmelCase_     :							Tuple             =    StableDiffusionImageVariationPipeline(**a_			)
  				lowerCAmelCase_     :							List[str]             =    sd_pipe.to(a_			)
  				sd_pipe.set_progress_bar_config(disable=a_			)
  
  				lowerCAmelCase_     :							List[str]             =    self.get_dummy_inputs(a_			)
  				lowerCAmelCase_     :							Optional[Any]             =    2 * [inputs["image"]]
  				lowerCAmelCase_     :							Optional[Any]             =    sd_pipe(**a_			)
  
  				lowerCAmelCase_     :							Union[str, Any]             =    output.images
  
  				lowerCAmelCase_     :							Optional[Any]             =    image[-1, -3:, -3:, -1]
  
  				assert image.shape == (2, 64, 64, 3)
  				lowerCAmelCase_     :							int             =    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      lowerCamelCase (     self							:			str			):
  				super().test_inference_batch_single_identical(expected_max_diff=3e-3			)
  
  
  
  
  
  
  
  @slow
  @require_torch_gpu
  class 	__lowerCamelCase       (				unittest.TestCase    ):
  
  
  
  
  
  		'''simple docstring'''
  
  
  
  
  
  
  		def      lowerCamelCase (     self							:			int			):
  				super().tearDown()
  				gc.collect()
  				torch.cuda.empty_cache()
  
  
  
  
  
  		def      lowerCamelCase (     self							:			Optional[Any]		,    a_							:			Optional[Any]		,    a_							:			List[str]="cpu"		,    a_							:			Optional[Any]=torch.floataa		,    a_							:			Optional[int]=0			):
  				lowerCAmelCase_     :							str             =    torch.Generator(device=a_			).manual_seed(a_			)
  				lowerCAmelCase_     :							str             =    load_image(
  				    "https://huggingface.co/datasets/diffusers/test-arrays/resolve/main"
  				    "/stable_diffusion_imgvar/input_image_vermeer.png"			)
  				lowerCAmelCase_     :							Optional[Any]             =    np.random.RandomState(a_			).standard_normal((1, 4, 64, 64)			)
  				lowerCAmelCase_     :							List[Any]             =    torch.from_numpy(a_			).to(device=a_		,    dtype=a_			)
  				lowerCAmelCase_     :							Any             =    {
  				    "image": init_image,
  				    "latents": latents,
  				    "generator": generator,
  				    "num_inference_steps": 3,
  				    "guidance_scale": 7.5,
  				    "output_type": "numpy",
  				}
  				return inputs
  
  
  
  
  
  		def      lowerCamelCase (     self							:			Any			):
  				lowerCAmelCase_     :							Tuple             =    StableDiffusionImageVariationPipeline.from_pretrained(
  				    "lambdalabs/sd-image-variations-diffusers"		,    safety_checker=a_			)
  				lowerCAmelCase_     :							int             =    sd_pipe.to(a_			)
  				sd_pipe.set_progress_bar_config(disable=a_			)
  
  				lowerCAmelCase_     :							Union[str, Any]             =    self.get_inputs(a_			)
  				lowerCAmelCase_     :							Union[str, Any]             =    sd_pipe(**a_			).images
  				lowerCAmelCase_     :							Optional[int]             =    image[0, -3:, -3:, -1].flatten()
  
  				assert image.shape == (1, 5_12, 5_12, 3)
  				lowerCAmelCase_     :							Optional[Any]             =    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      lowerCamelCase (     self							:			Dict			):
  				lowerCAmelCase_     :							List[str]             =    0
  
  				def callback_fn(a_							:			int		,    a_							:			int		,    a_							:			torch.FloatTensor			) -> None:
  						lowerCAmelCase_     :							str             =    True
  						nonlocal number_of_steps
  						number_of_steps += 1
  						if step == 1:
  								lowerCAmelCase_     :							str             =    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_     :							Dict             =    latents.detach().cpu().numpy()
  								assert latents.shape == (1, 4, 64, 64)
  								lowerCAmelCase_     :							Tuple             =    latents[0, -3:, -3:, -1]
  								lowerCAmelCase_     :							Dict             =    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_     :							Union[str, Any]             =    False
  
  				lowerCAmelCase_     :							str             =    StableDiffusionImageVariationPipeline.from_pretrained(
  				    "fusing/sd-image-variations-diffusers"		,    safety_checker=a_		,    torch_dtype=torch.floataa		,    )
  				pipe.to(a_			)
  				pipe.set_progress_bar_config(disable=a_			)
  				pipe.enable_attention_slicing()
  
  				lowerCAmelCase_     :							Tuple             =    self.get_inputs(a_		,    dtype=torch.floataa			)
  				pipe(**a_		,    callback=a_		,    callback_steps=1			)
  				assert callback_fn.has_been_called
  				assert number_of_steps == inputs["num_inference_steps"]
  
  
  
  
  
  		def      lowerCamelCase (     self							:			Optional[Any]			):
  				torch.cuda.empty_cache()
  				torch.cuda.reset_max_memory_allocated()
  				torch.cuda.reset_peak_memory_stats()
  
  				lowerCAmelCase_     :							Tuple             =    "fusing/sd-image-variations-diffusers"
  				lowerCAmelCase_     :							str             =    StableDiffusionImageVariationPipeline.from_pretrained(
  				    a_		,    safety_checker=a_		,    torch_dtype=torch.floataa			)
  				lowerCAmelCase_     :							int             =    pipe.to(a_			)
  				pipe.set_progress_bar_config(disable=a_			)
  				pipe.enable_attention_slicing(1			)
  				pipe.enable_sequential_cpu_offload()
  
  				lowerCAmelCase_     :							List[str]             =    self.get_inputs(a_		,    dtype=torch.floataa			)
  				lowerCAmelCase_     :							Any             =    pipe(**a_			)
  
  				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 	__lowerCamelCase       (				unittest.TestCase    ):
  
  
  
  
  
  		'''simple docstring'''
  
  
  
  
  
  
  		def      lowerCamelCase (     self							:			Tuple			):
  				super().tearDown()
  				gc.collect()
  				torch.cuda.empty_cache()
  
  
  
  
  
  		def      lowerCamelCase (     self							:			Optional[int]		,    a_							:			int		,    a_							:			int="cpu"		,    a_							:			str=torch.floataa		,    a_							:			Optional[Any]=0			):
  				lowerCAmelCase_     :							str             =    torch.Generator(device=a_			).manual_seed(a_			)
  				lowerCAmelCase_     :							int             =    load_image(
  				    "https://huggingface.co/datasets/diffusers/test-arrays/resolve/main"
  				    "/stable_diffusion_imgvar/input_image_vermeer.png"			)
  				lowerCAmelCase_     :							Optional[Any]             =    np.random.RandomState(a_			).standard_normal((1, 4, 64, 64)			)
  				lowerCAmelCase_     :							Optional[int]             =    torch.from_numpy(a_			).to(device=a_		,    dtype=a_			)
  				lowerCAmelCase_     :							Union[str, Any]             =    {
  				    "image": init_image,
  				    "latents": latents,
  				    "generator": generator,
  				    "num_inference_steps": 50,
  				    "guidance_scale": 7.5,
  				    "output_type": "numpy",
  				}
  				return inputs
  
  
  
  
  
  		def      lowerCamelCase (     self							:			Optional[int]			):
  				lowerCAmelCase_     :							Any             =    StableDiffusionImageVariationPipeline.from_pretrained("fusing/sd-image-variations-diffusers"			)
  				sd_pipe.to(a_			)
  				sd_pipe.set_progress_bar_config(disable=a_			)
  
  				lowerCAmelCase_     :							List[str]             =    self.get_inputs(a_			)
  				lowerCAmelCase_     :							int             =    sd_pipe(**a_			).images[0]
  
  				lowerCAmelCase_     :							Optional[Any]             =    load_numpy(
  				    "https://huggingface.co/datasets/diffusers/test-arrays/resolve/main"
  				    "/stable_diffusion_imgvar/lambdalabs_variations_pndm.npy"			)
  				lowerCAmelCase_     :							List[Any]             =    np.abs(expected_image - image			).max()
  				assert max_diff < 1e-3
  
  
  
  
  
  		def      lowerCamelCase (     self							:			Any			):
  				lowerCAmelCase_     :							List[Any]             =    StableDiffusionImageVariationPipeline.from_pretrained("fusing/sd-image-variations-diffusers"			)
  				lowerCAmelCase_     :							Tuple             =    DPMSolverMultistepScheduler.from_config(sd_pipe.scheduler.config			)
  				sd_pipe.to(a_			)
  				sd_pipe.set_progress_bar_config(disable=a_			)
  
  				lowerCAmelCase_     :							Optional[int]             =    self.get_inputs(a_			)
  				lowerCAmelCase_     :							Tuple             =    25
  				lowerCAmelCase_     :							int             =    sd_pipe(**a_			).images[0]
  
  				lowerCAmelCase_     :							Union[str, Any]             =    load_numpy(
  				    "https://huggingface.co/datasets/diffusers/test-arrays/resolve/main"
  				    "/stable_diffusion_imgvar/lambdalabs_variations_dpm_multi.npy"			)
  				lowerCAmelCase_     :							Optional[Any]             =    np.abs(expected_image - image			).max()
  				assert max_diff < 1e-3
  