


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


lowerCamelCase__							:						List[Any]							     =   False
class 		__magic_name__						(unittest.TestCase				):





     '''simple docstring'''



     pass




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





     '''simple docstring'''



     def 	SCREAMING_SNAKE_CASE__						(					self:Optional[int]   ):
       super().tearDown()
       gc.collect()
       torch.cuda.empty_cache()

     def 	SCREAMING_SNAKE_CASE__						(					self:List[str]   ):
       snake_case__						= VersatileDiffusionPipeline.from_pretrained('''shi-labs/versatile-diffusion'''     ,		torch_dtype=torch.floataa   )
       pipe.to(_a   )
       pipe.set_progress_bar_config(disable=_a   )

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

       snake_case__						= torch.manual_seed(0   )
       snake_case__						= pipe.dual_guided(
           prompt='''first prompt'''     ,		image=_a     ,		text_to_image_strength=0.75     ,		generator=_a     ,		guidance_scale=7.5     ,		num_inference_steps=2     ,		output_type='''numpy'''     ,		).images

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

       snake_case__						= generator.manual_seed(0   )
       snake_case__						= pipe.dual_guided(
           prompt='''first prompt'''     ,		image=_a     ,		text_to_image_strength=0.75     ,		generator=_a     ,		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 	SCREAMING_SNAKE_CASE__						(					self:List[str]   ):
       snake_case__						= VersatileDiffusionPipeline.from_pretrained('''shi-labs/versatile-diffusion'''     ,		torch_dtype=torch.floataa   )
       pipe.to(_a   )
       pipe.set_progress_bar_config(disable=_a   )

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

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

       assert image.shape == (1, 5_12, 5_12, 3)
       snake_case__						= np.array([0.1448, 0.1619, 0.1741, 0.1086, 0.1147, 0.1128, 0.1199, 0.1165, 0.1001]   )

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

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

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

       assert image.shape == (1, 5_12, 5_12, 3)
       snake_case__						= np.array([0.3367, 0.3169, 0.2656, 0.3870, 0.4790, 0.3796, 0.4009, 0.4878, 0.4778]   )

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

       snake_case__						= pipe.image_variation(_a     ,		generator=_a     ,		output_type='''numpy'''   ).images

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

       assert image.shape == (1, 5_12, 5_12, 3)
       snake_case__						= np.array([0.3076, 0.3123, 0.3284, 0.3782, 0.3770, 0.3894, 0.4297, 0.4331, 0.4456]   )

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







'''simple docstring'''



from typing import List, Optional, TypeVar

from .arrow_dataset import Dataset, _concatenate_map_style_datasets, _interleave_map_style_datasets
from .dataset_dict import DatasetDict, IterableDatasetDict
from .info import DatasetInfo
from .iterable_dataset import IterableDataset, _concatenate_iterable_datasets, _interleave_iterable_datasets
from .splits import NamedSplit
from .utils import logging
from .utils.py_utils import Literal


UpperCamelCase_    	=				logging.get_logger(__name__)


UpperCamelCase_    	=				TypeVar("DatasetType", Dataset, IterableDataset)




def   lowercase__(			__UpperCamelCase:					List[DatasetType]      ,__UpperCamelCase:					Optional[List[float]] = None      ,__UpperCamelCase:					Optional[int] = None      ,__UpperCamelCase:					Optional[DatasetInfo] = None      ,__UpperCamelCase:					Optional[NamedSplit] = None      ,__UpperCamelCase:					Literal["first_exhausted", "all_exhausted"] = "first_exhausted"      ,):

      """simple docstring"""






      from .arrow_dataset import Dataset
      from .iterable_dataset import IterableDataset

      if not datasets:
            raise ValueError('Unable to interleave an empty list of datasets.'    )
      for i, dataset in enumerate(__UpperCamelCase    ):
            if not isinstance(__UpperCamelCase      ,(Dataset, IterableDataset)    ):
                  if isinstance(__UpperCamelCase      ,(DatasetDict, IterableDatasetDict)    ):
                        if not dataset:
                              raise ValueError(
                                  f"Expected a list of Dataset objects or a list of IterableDataset objects, but element at position {i} "
                                  'is an empty dataset dictionary.'    )
                        raise ValueError(
                            f"Dataset at position {i} has at least one split: {list(__UpperCamelCase    )}\n"
                            f"Please pick one to interleave with the other datasets, for example: dataset['{next(iter(__UpperCamelCase    )    )}']"    )
                  raise ValueError(
                      f"Expected a list of Dataset objects or a list of IterableDataset objects, but element at position {i} is a {type(__UpperCamelCase    ).__name__}."    )
            if i == 0:
                  SCREAMING_SNAKE_CASE					,    SCREAMING_SNAKE_CASE   :      Optional[Any]    					=						(
                      (Dataset, IterableDataset) if isinstance(__UpperCamelCase      ,__UpperCamelCase    ) else (IterableDataset, Dataset)
                  )
            elif not isinstance(__UpperCamelCase      ,__UpperCamelCase    ):
                  raise ValueError(
                      f"Unable to interleave a {dataset_type.__name__} (at position 0) with a {other_type.__name__} (at position {i}). Expected a list of Dataset objects or a list of IterableDataset objects."    )
      if stopping_strategy not in ["first_exhausted", "all_exhausted"]:
            raise ValueError(f"{stopping_strategy} is not supported. Please enter a valid stopping_strategy."    )
      if dataset_type is Dataset:
            return _interleave_map_style_datasets(
                __UpperCamelCase      ,__UpperCamelCase      ,__UpperCamelCase      ,info=__UpperCamelCase      ,split=__UpperCamelCase      ,stopping_strategy=__UpperCamelCase    )
      else:
            return _interleave_iterable_datasets(
                __UpperCamelCase      ,__UpperCamelCase      ,__UpperCamelCase      ,info=__UpperCamelCase      ,split=__UpperCamelCase      ,stopping_strategy=__UpperCamelCase    )





def   lowercase__(			__UpperCamelCase:					List[DatasetType]      ,__UpperCamelCase:					Optional[DatasetInfo] = None      ,__UpperCamelCase:					Optional[NamedSplit] = None      ,__UpperCamelCase:					int = 0      ,):

      """simple docstring"""






      if not dsets:
            raise ValueError('Unable to concatenate an empty list of datasets.'    )
      for i, dataset in enumerate(__UpperCamelCase    ):
            if not isinstance(__UpperCamelCase      ,(Dataset, IterableDataset)    ):
                  if isinstance(__UpperCamelCase      ,(DatasetDict, IterableDatasetDict)    ):
                        if not dataset:
                              raise ValueError(
                                  f"Expected a list of Dataset objects or a list of IterableDataset objects, but element at position {i} "
                                  'is an empty dataset dictionary.'    )
                        raise ValueError(
                            f"Dataset at position {i} has at least one split: {list(__UpperCamelCase    )}\n"
                            f"Please pick one to interleave with the other datasets, for example: dataset['{next(iter(__UpperCamelCase    )    )}']"    )
                  raise ValueError(
                      f"Expected a list of Dataset objects or a list of IterableDataset objects, but element at position {i} is a {type(__UpperCamelCase    ).__name__}."    )
            if i == 0:
                  SCREAMING_SNAKE_CASE					,    SCREAMING_SNAKE_CASE   :      int    					=						(
                      (Dataset, IterableDataset) if isinstance(__UpperCamelCase      ,__UpperCamelCase    ) else (IterableDataset, Dataset)
                  )
            elif not isinstance(__UpperCamelCase      ,__UpperCamelCase    ):
                  raise ValueError(
                      f"Unable to interleave a {dataset_type.__name__} (at position 0) with a {other_type.__name__} (at position {i}). Expected a list of Dataset objects or a list of IterableDataset objects."    )
      if dataset_type is Dataset:
            return _concatenate_map_style_datasets(__UpperCamelCase      ,info=__UpperCamelCase      ,split=__UpperCamelCase      ,axis=__UpperCamelCase    )
      else:
            return _concatenate_iterable_datasets(__UpperCamelCase      ,info=__UpperCamelCase      ,split=__UpperCamelCase      ,axis=__UpperCamelCase    )


