code: 
  
  
  
  
  
  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    (__UpperCAmelCase       ,__UpperCAmelCase       ,__UpperCAmelCase       ,unittest.TestCase				):
        _SCREAMING_SNAKE_CASE					=							StableDiffusionImageVariationPipeline
        _SCREAMING_SNAKE_CASE					=							IMAGE_VARIATION_PARAMS
        _SCREAMING_SNAKE_CASE					=							IMAGE_VARIATION_BATCH_PARAMS
        _SCREAMING_SNAKE_CASE					=							frozenset([]				)
        # TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
        _SCREAMING_SNAKE_CASE					=							frozenset([]				)
  
  
  
        def    __a (					self					)      ->   Any:
  
  
  
               '''simple docstring'''
  
               torch.manual_seed(0					)
               _snake_case      :							Optional[int]     				=						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     ,  )
               _snake_case      :							Dict     				=						PNDMScheduler(skip_prk_steps=lowercase_					)
               torch.manual_seed(0					)
               _snake_case      :							Optional[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					)
               _snake_case      :							Tuple     				=						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     ,  )
               _snake_case      :							List[Any]     				=						CLIPVisionModelWithProjection(lowercase_					)
               _snake_case      :							str     				=						CLIPImageProcessor(crop_size=32     ,  size=32					)
  
               _snake_case      :							List[str]     				=						{
                   '''unet''': unet,
                   '''scheduler''': scheduler,
                   '''vae''': vae,
                   '''image_encoder''': image_encoder,
                   '''feature_extractor''': feature_extractor,
                   '''safety_checker''': None,
               }
               return components
  
  
  
        def    __a (					self     ,  lowercase_     ,  lowercase_=0					)      ->   Optional[int]:
  
  
  
               '''simple docstring'''
  
               _snake_case      :							List[Any]     				=						floats_tensor((1, 3, 32, 32)     ,  rng=random.Random(lowercase_					)					)
               _snake_case      :							Tuple     				=						image.cpu().permute(0     ,  2     ,  3     ,  1					)[0]
               _snake_case      :							List[Any]     				=						Image.fromarray(np.uinta(lowercase_					)					).convert('''RGB'''					).resize((32, 32)					)
               if str(lowercase_					).startswith('''mps'''					):
                      _snake_case      :							Tuple     				=						torch.manual_seed(lowercase_					)
               else:
                      _snake_case      :							Union[str, Any]     				=						torch.Generator(device=lowercase_					).manual_seed(lowercase_					)
               _snake_case      :							List[str]     				=						{
                   '''image''': image,
                   '''generator''': generator,
                   '''num_inference_steps''': 2,
                   '''guidance_scale''': 6.0,
                   '''output_type''': '''numpy''',
               }
               return inputs
  
  
  
        def    __a (					self					)      ->   int:
  
  
  
               '''simple docstring'''
  
               _snake_case      :							Optional[int]     				=						'''cpu'''  # ensure determinism for the device-dependent torch.Generator
               _snake_case      :							Optional[Any]     				=						self.get_dummy_components()
               _snake_case      :							Any     				=						StableDiffusionImageVariationPipeline(**lowercase_					)
               _snake_case      :							Optional[Any]     				=						sd_pipe.to(lowercase_					)
               sd_pipe.set_progress_bar_config(disable=lowercase_					)
  
               _snake_case      :							List[str]     				=						self.get_dummy_inputs(lowercase_					)
               _snake_case      :							Dict     				=						sd_pipe(**lowercase_					).images
               _snake_case      :							Tuple     				=						image[0, -3:, -3:, -1]
  
               assert image.shape == (1, 64, 64, 3)
               _snake_case      :							int     				=						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    __a (					self					)      ->   Optional[int]:
  
  
  
               '''simple docstring'''
  
               _snake_case      :							str     				=						'''cpu'''  # ensure determinism for the device-dependent torch.Generator
               _snake_case      :							Dict     				=						self.get_dummy_components()
               _snake_case      :							Dict     				=						StableDiffusionImageVariationPipeline(**lowercase_					)
               _snake_case      :							Optional[int]     				=						sd_pipe.to(lowercase_					)
               sd_pipe.set_progress_bar_config(disable=lowercase_					)
  
               _snake_case      :							List[str]     				=						self.get_dummy_inputs(lowercase_					)
               _snake_case      :							Any     				=						2 * [inputs['''image''']]
               _snake_case      :							int     				=						sd_pipe(**lowercase_					)
  
               _snake_case      :							Optional[Any]     				=						output.images
  
               _snake_case      :							Optional[Any]     				=						image[-1, -3:, -3:, -1]
  
               assert image.shape == (2, 64, 64, 3)
               _snake_case      :							List[Any]     				=						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    __a (					self					)      ->   Dict:
  
  
  
               '''simple docstring'''
  
               super().test_inference_batch_single_identical(expected_max_diff=3E-3					)
  
  
  @slow
  @require_torch_gpu
  class 				A    (unittest.TestCase				):
  
  
  
        def    __a (					self					)      ->   Union[str, Any]:
  
  
  
               '''simple docstring'''
  
               super().tearDown()
               gc.collect()
               torch.cuda.empty_cache()
  
  
  
        def    __a (					self     ,  lowercase_     ,  lowercase_="cpu"     ,  lowercase_=torch.floataa     ,  lowercase_=0					)      ->   str:
  
  
  
               '''simple docstring'''
  
               _snake_case      :							Dict     				=						torch.Generator(device=lowercase_					).manual_seed(lowercase_					)
               _snake_case      :							Tuple     				=						load_image(
                   '''https://huggingface.co/datasets/diffusers/test-arrays/resolve/main'''
                   '''/stable_diffusion_imgvar/input_image_vermeer.png'''					)
               _snake_case      :							Union[str, Any]     				=						np.random.RandomState(lowercase_					).standard_normal((1, 4, 64, 64)					)
               _snake_case      :							Dict     				=						torch.from_numpy(lowercase_					).to(device=lowercase_     ,  dtype=lowercase_					)
               _snake_case      :							List[Any]     				=						{
                   '''image''': init_image,
                   '''latents''': latents,
                   '''generator''': generator,
                   '''num_inference_steps''': 3,
                   '''guidance_scale''': 7.5,
                   '''output_type''': '''numpy''',
               }
               return inputs
  
  
  
        def    __a (					self					)      ->   Any:
  
  
  
               '''simple docstring'''
  
               _snake_case      :							List[str]     				=						StableDiffusionImageVariationPipeline.from_pretrained(
                   '''lambdalabs/sd-image-variations-diffusers'''     ,  safety_checker=lowercase_					)
               _snake_case      :							str     				=						sd_pipe.to(lowercase_					)
               sd_pipe.set_progress_bar_config(disable=lowercase_					)
  
               _snake_case      :							str     				=						self.get_inputs(lowercase_					)
               _snake_case      :							Any     				=						sd_pipe(**lowercase_					).images
               _snake_case      :							Any     				=						image[0, -3:, -3:, -1].flatten()
  
               assert image.shape == (1, 512, 512, 3)
               _snake_case      :							List[Any]     				=						np.array([0.8_4491, 0.9_0789, 0.7_5708, 0.7_8734, 0.8_3485, 0.7_0099, 0.6_6938, 0.6_8727, 0.6_1379]					)
               assert np.abs(image_slice - expected_slice					).max() < 6E-3
  
  
  
        def    __a (					self					)      ->   List[Any]:
  
  
  
               '''simple docstring'''
  
               _snake_case      :							Union[str, Any]     				=						0
  
               def callback_fn(lowercase_     ,  lowercase_     ,  lowercase_					) -> None:
                      _snake_case      :							Tuple     				=						True
                      nonlocal number_of_steps
                      number_of_steps += 1
                      if step == 1:
                             _snake_case      :							List[Any]     				=						latents.detach().cpu().numpy()
                             assert latents.shape == (1, 4, 64, 64)
                             _snake_case      :							Union[str, Any]     				=						latents[0, -3:, -3:, -1]
                             _snake_case      :							List[str]     				=						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:
                             _snake_case      :							Optional[Any]     				=						latents.detach().cpu().numpy()
                             assert latents.shape == (1, 4, 64, 64)
                             _snake_case      :							Dict     				=						latents[0, -3:, -3:, -1]
                             _snake_case      :							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
  
               _snake_case      :							str     				=						False
  
               _snake_case      :							Optional[Any]     				=						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()
  
               _snake_case      :							Any     				=						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    __a (					self					)      ->   Tuple:
  
  
  
               '''simple docstring'''
  
               torch.cuda.empty_cache()
               torch.cuda.reset_max_memory_allocated()
               torch.cuda.reset_peak_memory_stats()
  
               _snake_case      :							Optional[int]     				=						'''fusing/sd-image-variations-diffusers'''
               _snake_case      :							Union[str, Any]     				=						StableDiffusionImageVariationPipeline.from_pretrained(
                   lowercase_     ,  safety_checker=lowercase_     ,  torch_dtype=torch.floataa					)
               _snake_case      :							Optional[int]     				=						pipe.to(lowercase_					)
               pipe.set_progress_bar_config(disable=lowercase_					)
               pipe.enable_attention_slicing(1					)
               pipe.enable_sequential_cpu_offload()
  
               _snake_case      :							List[Any]     				=						self.get_inputs(lowercase_     ,  dtype=torch.floataa					)
               _snake_case      :							Dict     				=						pipe(**lowercase_					)
  
               _snake_case      :							Tuple     				=						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				):
  
  
  
        def    __a (					self					)      ->   List[Any]:
  
  
  
               '''simple docstring'''
  
               super().tearDown()
               gc.collect()
               torch.cuda.empty_cache()
  
  
  
        def    __a (					self     ,  lowercase_     ,  lowercase_="cpu"     ,  lowercase_=torch.floataa     ,  lowercase_=0					)      ->   Optional[Any]:
  
  
  
               '''simple docstring'''
  
               _snake_case      :							List[Any]     				=						torch.Generator(device=lowercase_					).manual_seed(lowercase_					)
               _snake_case      :							Optional[Any]     				=						load_image(
                   '''https://huggingface.co/datasets/diffusers/test-arrays/resolve/main'''
                   '''/stable_diffusion_imgvar/input_image_vermeer.png'''					)
               _snake_case      :							Optional[int]     				=						np.random.RandomState(lowercase_					).standard_normal((1, 4, 64, 64)					)
               _snake_case      :							Any     				=						torch.from_numpy(lowercase_					).to(device=lowercase_     ,  dtype=lowercase_					)
               _snake_case      :							List[Any]     				=						{
                   '''image''': init_image,
                   '''latents''': latents,
                   '''generator''': generator,
                   '''num_inference_steps''': 50,
                   '''guidance_scale''': 7.5,
                   '''output_type''': '''numpy''',
               }
               return inputs
  
  
  
        def    __a (					self					)      ->   Union[str, Any]:
  
  
  
               '''simple docstring'''
  
               _snake_case      :							Optional[int]     				=						StableDiffusionImageVariationPipeline.from_pretrained('''fusing/sd-image-variations-diffusers'''					)
               sd_pipe.to(lowercase_					)
               sd_pipe.set_progress_bar_config(disable=lowercase_					)
  
               _snake_case      :							Tuple     				=						self.get_inputs(lowercase_					)
               _snake_case      :							Optional[Any]     				=						sd_pipe(**lowercase_					).images[0]
  
               _snake_case      :							List[Any]     				=						load_numpy(
                   '''https://huggingface.co/datasets/diffusers/test-arrays/resolve/main'''
                   '''/stable_diffusion_imgvar/lambdalabs_variations_pndm.npy'''					)
               _snake_case      :							List[Any]     				=						np.abs(expected_image - image					).max()
               assert max_diff < 1E-3
  
  
  
        def    __a (					self					)      ->   Dict:
  
  
  
               '''simple docstring'''
  
               _snake_case      :							List[Any]     				=						StableDiffusionImageVariationPipeline.from_pretrained('''fusing/sd-image-variations-diffusers'''					)
               _snake_case      :							Dict     				=						DPMSolverMultistepScheduler.from_config(sd_pipe.scheduler.config					)
               sd_pipe.to(lowercase_					)
               sd_pipe.set_progress_bar_config(disable=lowercase_					)
  
               _snake_case      :							List[Any]     				=						self.get_inputs(lowercase_					)
               _snake_case      :							List[Any]     				=						25
               _snake_case      :							Tuple     				=						sd_pipe(**lowercase_					).images[0]
  
               _snake_case      :							List[Any]     				=						load_numpy(
                   '''https://huggingface.co/datasets/diffusers/test-arrays/resolve/main'''
                   '''/stable_diffusion_imgvar/lambdalabs_variations_dpm_multi.npy'''					)
               _snake_case      :							Any     				=						np.abs(expected_image - image					).max()
               assert max_diff < 1E-3
  
  
  
  
  
  
  
style_context: 
  
  
  
  
  
  import unittest
  
  from transformers import load_tool
  from transformers.tools.agent_types import AGENT_TYPE_MAPPING
  
  from .test_tools_common import ToolTesterMixin, output_types
  
  class 				A    (unittest.TestCase       ,__UpperCAmelCase				):
  
  
  
        def    __a (					self					)      ->   str:
  
  
  
               '''simple docstring'''
  
               _snake_case      :							List[Any]     				=						load_tool('''translation'''					)
               self.tool.setup()
               _snake_case      :							Any     				=						load_tool('''translation'''     ,  remote=lowercase_					)
  
  
  
        def    __a (					self					)      ->   Optional[int]:
  
  
  
               '''simple docstring'''
  
               _snake_case      :							Tuple     				=						self.tool('''Hey, what\'s up?'''     ,  src_lang='''English'''     ,  tgt_lang='''French'''					)
               self.assertEqual(lowercase_     ,  '''- Hé, comment ça va?'''					)
  
  
  
        def    __a (					self					)      ->   Union[str, Any]:
  
  
  
               '''simple docstring'''
  
               _snake_case      :							Optional[int]     				=						self.remote_tool('''Hey, what\'s up?'''     ,  src_lang='''English'''     ,  tgt_lang='''French'''					)
               self.assertEqual(lowercase_     ,  '''- Hé, comment ça va?'''					)
  
  
  
        def    __a (					self					)      ->   List[Any]:
  
  
  
               '''simple docstring'''
  
               _snake_case      :							Dict     				=						self.tool(text='''Hey, what\'s up?'''     ,  src_lang='''English'''     ,  tgt_lang='''French'''					)
               self.assertEqual(lowercase_     ,  '''- Hé, comment ça va?'''					)
  
  
  
        def    __a (					self					)      ->   Dict:
  
  
  
               '''simple docstring'''
  
               _snake_case      :							List[Any]     				=						self.remote_tool(text='''Hey, what\'s up?'''     ,  src_lang='''English'''     ,  tgt_lang='''French'''					)
               self.assertEqual(lowercase_     ,  '''- Hé, comment ça va?'''					)
  
  
  
        def    __a (					self					)      ->   str:
  
  
  
               '''simple docstring'''
  
               _snake_case      :							Optional[int]     				=						['''Hey, what\'s up?''', '''English''', '''Spanish''']
               _snake_case      :							Dict     				=						self.tool(*lowercase_					)
  
               # There is a single output
               if len(self.tool.outputs					) == 1:
                      _snake_case      :							Optional[Any]     				=						[outputs]
  
               self.assertListEqual(output_types(lowercase_					)     ,  self.tool.outputs					)
  
  
  
        def    __a (					self					)      ->   Tuple:
  
  
  
               '''simple docstring'''
  
               _snake_case      :							Union[str, Any]     				=						['''Hey, what\'s up?''', '''English''', '''Spanish''']
               _snake_case      :							Dict     				=						self.tool(*lowercase_					)
  
               if not isinstance(lowercase_     ,  lowercase_					):
                      _snake_case      :							Optional[Any]     				=						[outputs]
  
               self.assertEqual(len(lowercase_					)     ,  len(self.tool.outputs					)					)
  
               for output, output_type in zip(lowercase_     ,  self.tool.outputs					):
                      _snake_case      :							List[str]     				=						AGENT_TYPE_MAPPING[output_type]
                      self.assertTrue(isinstance(lowercase_     ,  lowercase_					)					)
  
  
  
        def    __a (					self					)      ->   List[Any]:
  
  
  
               '''simple docstring'''
  
               _snake_case      :							Dict     				=						['''Hey, what\'s up?''', '''English''', '''Spanish''']
  
               _snake_case      :							Any     				=						[]
  
               for _input, input_type in zip(lowercase_     ,  self.tool.inputs					):
                      if isinstance(lowercase_     ,  lowercase_					):
                             _inputs.append([AGENT_TYPE_MAPPING[_input_type](_input					) for _input_type in input_type]					)
                      else:
                             _inputs.append(AGENT_TYPE_MAPPING[input_type](_input					)					)
  
          # Should not raise an error
               _snake_case      :							List[Any]     				=						self.tool(*lowercase_					)
  
               if not isinstance(lowercase_     ,  lowercase_					):
                      _snake_case      :							Any     				=						[outputs]
  
               self.assertEqual(len(lowercase_					)     ,  len(self.tool.outputs					)					)
  
  
  
  
  
  
  