





'''simple docstring'''






import gc
import tempfile
import unittest

import numpy as np
import torch

from diffusers import VersatileDiffusionPipeline
from diffusers.utils.testing_utils import load_image, nightly, require_torch_gpu, torch_device


A__    :       Optional[Any]              =False




class 			UpperCAmelCase					(      unittest.TestCase       ):
      pass




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




      def     lowercase__   (      self      :			Dict  )    ->				Optional[Any]:
            # clean up the VRAM after each test
            super().tearDown()
            gc.collect()
            torch.cuda.empty_cache()




      def     lowercase__   (      self      :			str  )    ->				Optional[int]:
            _lowerCAmelCase												=							VersatileDiffusionPipeline.from_pretrained("""shi-labs/versatile-diffusion"""      ,      torch_dtype=torch.floataa  )
            pipe.to(__snake_case  )
            pipe.set_progress_bar_config(disable=__snake_case  )

            _lowerCAmelCase												=							load_image(
                """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/versatile_diffusion/benz.jpg"""  )

            _lowerCAmelCase												=							torch.manual_seed(0  )
            _lowerCAmelCase												=							pipe.dual_guided(
                prompt="""first prompt"""      ,      image=__snake_case      ,      text_to_image_strength=0.75      ,      generator=__snake_case      ,      guidance_scale=7.5      ,      num_inference_steps=2      ,      output_type="""numpy"""      ,      ).images

            with tempfile.TemporaryDirectory() as tmpdirname:
                  pipe.save_pretrained(__snake_case  )
                  _lowerCAmelCase												=							VersatileDiffusionPipeline.from_pretrained(__snake_case      ,      torch_dtype=torch.floataa  )
            pipe.to(__snake_case  )
            pipe.set_progress_bar_config(disable=__snake_case  )

            _lowerCAmelCase												=							generator.manual_seed(0  )
            _lowerCAmelCase												=							pipe.dual_guided(
                prompt="""first prompt"""      ,      image=__snake_case      ,      text_to_image_strength=0.75      ,      generator=__snake_case      ,      guidance_scale=7.5      ,      num_inference_steps=2      ,      output_type="""numpy"""      ,      ).images

            assert np.abs(image - new_image  ).sum() < 1E-5, "Models don't have the same forward pass"




      def     lowercase__   (      self      :			Optional[int]  )    ->				List[str]:
            _lowerCAmelCase												=							VersatileDiffusionPipeline.from_pretrained("""shi-labs/versatile-diffusion"""      ,      torch_dtype=torch.floataa  )
            pipe.to(__snake_case  )
            pipe.set_progress_bar_config(disable=__snake_case  )

            _lowerCAmelCase												=							"""cyberpunk 2077"""
            _lowerCAmelCase												=							load_image(
                """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/versatile_diffusion/benz.jpg"""  )
            _lowerCAmelCase												=							torch.manual_seed(0  )
            _lowerCAmelCase												=							pipe.dual_guided(
                prompt=__snake_case      ,      image=__snake_case      ,      text_to_image_strength=0.75      ,      generator=__snake_case      ,      guidance_scale=7.5      ,      num_inference_steps=50      ,      output_type="""numpy"""      ,      ).images

            _lowerCAmelCase												=							image[0, 2_53:2_56, 2_53:2_56, -1]

            assert image.shape == (1, 5_12, 5_12, 3)
            _lowerCAmelCase												=							np.array([0.14_48, 0.16_19, 0.17_41, 0.10_86, 0.11_47, 0.11_28, 0.11_99, 0.11_65, 0.10_01]  )

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

            _lowerCAmelCase												=							"""A painting of a squirrel eating a burger """
            _lowerCAmelCase												=							torch.manual_seed(0  )
            _lowerCAmelCase												=							pipe.text_to_image(
                prompt=__snake_case      ,      generator=__snake_case      ,      guidance_scale=7.5      ,      num_inference_steps=50      ,      output_type="""numpy"""  ).images

            _lowerCAmelCase												=							image[0, 2_53:2_56, 2_53:2_56, -1]

            assert image.shape == (1, 5_12, 5_12, 3)
            _lowerCAmelCase												=							np.array([0.33_67, 0.31_69, 0.26_56, 0.38_70, 0.47_90, 0.37_96, 0.40_09, 0.48_78, 0.47_78]  )

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

            _lowerCAmelCase												=							pipe.image_variation(__snake_case      ,      generator=__snake_case      ,      output_type="""numpy"""  ).images

            _lowerCAmelCase												=							image[0, 2_53:2_56, 2_53:2_56, -1]

            assert image.shape == (1, 5_12, 5_12, 3)
            _lowerCAmelCase												=							np.array([0.30_76, 0.31_23, 0.32_84, 0.37_82, 0.37_70, 0.38_94, 0.42_97, 0.43_31, 0.44_56]  )

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












'''simple docstring'''






import json
import os
import tempfile

import datasets
from utils import generate_example_dataset, get_duration


A__    :       str              =5_00_00
A__    :       Optional[int]              =50_00

A__       ,						A__    :       Optional[int]              =os.path.split(__file__)
A__    :       Tuple              =os.path.join(RESULTS_BASEPATH, '''results''', RESULTS_FILENAME.replace('''.py''', '''.json'''))
@get_duration
def        UpperCamelCase__				(		lowerCAmelCase							,     lowerCAmelCase  ):





      """simple docstring"""


      for i in range(lowerCAmelCase  ):
            _lowerCAmelCase												=							dataset[i]
@get_duration
def        UpperCamelCase__				(		lowerCAmelCase							,     lowerCAmelCase							,     lowerCAmelCase  ):





      """simple docstring"""


      for i in range(0							,     len(lowerCAmelCase  )							,     lowerCAmelCase  ):
            _lowerCAmelCase												=							dataset[i : i + batch_size]
@get_duration
def        UpperCamelCase__				(		lowerCAmelCase							,     lowerCAmelCase							,     lowerCAmelCase  ):





      """simple docstring"""


      with dataset.formatted_as(type=lowerCAmelCase  ):
            for i in range(lowerCAmelCase  ):
                  _lowerCAmelCase												=							dataset[i]
@get_duration
def        UpperCamelCase__				(		lowerCAmelCase							,     lowerCAmelCase							,     lowerCAmelCase							,     lowerCAmelCase  ):





      """simple docstring"""


      with dataset.formatted_as(type=lowerCAmelCase  ):
            for i in range(0							,     lowerCAmelCase							,     lowerCAmelCase  ):
                  _lowerCAmelCase												=							dataset[i : i + batch_size]
def        UpperCamelCase__				(		):





      """simple docstring"""


      _lowerCAmelCase												=							{"""num examples""": SPEED_TEST_N_EXAMPLES}
      _lowerCAmelCase												=							[
          (read, {"""length""": SMALL_TEST}),
          (read, {"""length""": SPEED_TEST_N_EXAMPLES}),
          (read_batch, {"""length""": SPEED_TEST_N_EXAMPLES, """batch_size""": 10}),
          (read_batch, {"""length""": SPEED_TEST_N_EXAMPLES, """batch_size""": 1_00}),
          (read_batch, {"""length""": SPEED_TEST_N_EXAMPLES, """batch_size""": 10_00}),
          (read_formatted, {"""type""": """numpy""", """length""": SMALL_TEST}),
          (read_formatted, {"""type""": """pandas""", """length""": SMALL_TEST}),
          (read_formatted, {"""type""": """torch""", """length""": SMALL_TEST}),
          (read_formatted, {"""type""": """tensorflow""", """length""": SMALL_TEST}),
          (read_formatted_batch, {"""type""": """numpy""", """length""": SMALL_TEST, """batch_size""": 10}),
          (read_formatted_batch, {"""type""": """numpy""", """length""": SMALL_TEST, """batch_size""": 10_00}),
      ]

      _lowerCAmelCase												=							[
          (read, {"""length""": SMALL_TEST}),
          (read, {"""length""": SPEED_TEST_N_EXAMPLES}),
          (read_batch, {"""length""": SPEED_TEST_N_EXAMPLES, """batch_size""": 10}),
          (read_batch, {"""length""": SPEED_TEST_N_EXAMPLES, """batch_size""": 1_00}),
          (read_batch, {"""length""": SPEED_TEST_N_EXAMPLES, """batch_size""": 10_00}),
          (read_formatted, {"""type""": """numpy""", """length""": SMALL_TEST}),
          (read_formatted_batch, {"""type""": """numpy""", """length""": SMALL_TEST, """batch_size""": 10}),
          (read_formatted_batch, {"""type""": """numpy""", """length""": SMALL_TEST, """batch_size""": 10_00}),
      ]
      with tempfile.TemporaryDirectory() as tmp_dir:
            print("""generating dataset"""  )
            _lowerCAmelCase												=							datasets.Features(
                {"""list""": datasets.Sequence(datasets.Value("""float32"""  )  ), """numbers""": datasets.Value("""float32"""  )}  )
            _lowerCAmelCase												=							generate_example_dataset(
                os.path.join(lowerCAmelCase							,     """dataset.arrow"""  )							,     lowerCAmelCase							,     num_examples=lowerCAmelCase							,     seq_shapes={"""list""": (1_00,)}							,     )
            print("""first set of iterations"""  )
            for func, kwargs in functions:
                  print(func.__name__							,     str(lowerCAmelCase  )  )
                  _lowerCAmelCase												=							func(lowerCAmelCase							,     **lowerCAmelCase  )

            print("""shuffling dataset"""  )
            _lowerCAmelCase												=							dataset.shuffle()
            print("""Second set of iterations (after shuffling"""  )
            for func, kwargs in functions_shuffled:
                  print("""shuffled """							,     func.__name__							,     str(lowerCAmelCase  )  )
                  _lowerCAmelCase												=							func(
                      lowerCAmelCase							,     **lowerCAmelCase  )

      with open(lowerCAmelCase							,     """wb"""  ) as f:
            f.write(json.dumps(lowerCAmelCase  ).encode("""utf-8"""  )  )


if __name__ == "__main__":  # useful to run the profiler
      benchmark_iterating()






