



'''simple docstring'''





import random
import unittest

from torch.utils.data import BatchSampler, DataLoader, IterableDataset

from accelerate import Accelerator
from accelerate.data_loader import (
    BatchSamplerShard,
    DataLoaderDispatcher,
    DataLoaderShard,
    IterableDatasetShard,
    SkipBatchSampler,
    SkipDataLoader,
    skip_first_batches,
)






class 					lowerCamelCase_    (				SCREAMING_SNAKE_CASE_     ):
  """simple docstring"""




  def __init__( self      :	Union[str, Any]			,	_a      :	Union[str, Any]=0.01			,	_a      :	Optional[Any]=1000   )					->					int:
    __lowerCamelCase       :       Any           = p_stop
    __lowerCamelCase       :       List[Any]           = max_length




  def __iter__( self      :	int   )					->					List[str]:
    __lowerCamelCase       :       int           = 0
    __lowerCamelCase       :       Optional[int]           = False
    while not stop and count < self.max_length:
      yield count
      count += 1
      __lowerCamelCase       :       Union[str, Any]           = random.random() < self.p_stop






class 					lowerCamelCase_    (				unittest.TestCase     ):
  """simple docstring"""




  def 						_lowercase     ( self      :	List[str]			,	_a      :	Union[str, Any]			,	_a      :	str			,	_a      :	int=False			,	_a      :	List[Any]=True   )					->					int:
    __lowerCamelCase       :       Dict           = [
        BatchSamplerShard(_a			,	2			,	_a			,	split_batches=_a			,	even_batches=_a   )
        for i in range(2   )
    ]
    __lowerCamelCase       :       Union[str, Any]           = [list(_a   ) for batch_sampler_shard in batch_sampler_shards]
    if not split_batches:
      self.assertListEqual([len(_a   ) for shard in batch_sampler_shards]			,	[len(_a   ) for e in expected]   )
    self.assertListEqual(_a			,	_a   )




  def 						_lowercase     ( self      :	Optional[int]   )					->					Optional[int]:
    # Check the shards when the dataset is a round multiple of total batch size.
    __lowerCamelCase       :       Union[str, Any]           = BatchSampler(range(24   )			,	batch_size=3			,	drop_last=_a   )
    __lowerCamelCase       :       str           = [
        [[0, 1, 2], [6, 7, 8], [12, 13, 14], [18, 19, 20]],
        [[3, 4, 5], [9, 10, 11], [15, 16, 17], [21, 22, 23]],
    ]
    self.check_batch_sampler_shards(_a			,	_a   )

    __lowerCamelCase       :       Any           = BatchSampler(range(24   )			,	batch_size=3			,	drop_last=_a   )
    # Expected shouldn't change
    self.check_batch_sampler_shards(_a			,	_a   )

    # Check the shards when the dataset is a round multiple of batch size but not total batch size.
    __lowerCamelCase       :       Optional[Any]           = BatchSampler(range(21   )			,	batch_size=3			,	drop_last=_a   )
    __lowerCamelCase       :       Optional[Any]           = [
        [[0, 1, 2], [6, 7, 8], [12, 13, 14], [18, 19, 20]],
        [[3, 4, 5], [9, 10, 11], [15, 16, 17], [0, 1, 2]],
    ]
    self.check_batch_sampler_shards(_a			,	_a   )

    __lowerCamelCase       :       Union[str, Any]           = BatchSampler(range(21   )			,	batch_size=3			,	drop_last=_a   )
    __lowerCamelCase       :       Optional[int]           = [
        [[0, 1, 2], [6, 7, 8], [12, 13, 14]],
        [[3, 4, 5], [9, 10, 11], [15, 16, 17]],
    ]
    self.check_batch_sampler_shards(_a			,	_a   )

    # Check the shards when the dataset is not a round multiple of batch size but has a multiple of
    # num_processes batch.
    __lowerCamelCase       :       Optional[Any]           = BatchSampler(range(22   )			,	batch_size=3			,	drop_last=_a   )
    __lowerCamelCase       :       List[Any]           = [
        [[0, 1, 2], [6, 7, 8], [12, 13, 14], [18, 19, 20]],
        [[3, 4, 5], [9, 10, 11], [15, 16, 17], [21, 0, 1]],
    ]
    self.check_batch_sampler_shards(_a			,	_a   )

    __lowerCamelCase       :       Optional[int]           = BatchSampler(range(22   )			,	batch_size=3			,	drop_last=_a   )
    __lowerCamelCase       :       Tuple           = [
        [[0, 1, 2], [6, 7, 8], [12, 13, 14]],
        [[3, 4, 5], [9, 10, 11], [15, 16, 17]],
    ]
    self.check_batch_sampler_shards(_a			,	_a   )

    # Check the shards when the dataset is not a round multiple of batch size but and has not a multiple of
    # num_processes batch.
    __lowerCamelCase       :       Any           = BatchSampler(range(20   )			,	batch_size=3			,	drop_last=_a   )
    __lowerCamelCase       :       Optional[Any]           = [
        [[0, 1, 2], [6, 7, 8], [12, 13, 14], [18, 19, 0]],
        [[3, 4, 5], [9, 10, 11], [15, 16, 17], [1, 2, 3]],
    ]
    self.check_batch_sampler_shards(_a			,	_a   )

    __lowerCamelCase       :       List[Any]           = BatchSampler(range(20   )			,	batch_size=3			,	drop_last=_a   )
    __lowerCamelCase       :       int           = [
        [[0, 1, 2], [6, 7, 8], [12, 13, 14]],
        [[3, 4, 5], [9, 10, 11], [15, 16, 17]],
    ]
    self.check_batch_sampler_shards(_a			,	_a   )

    # Check the shards when the dataset is very small.
    __lowerCamelCase       :       List[Any]           = BatchSampler(range(2   )			,	batch_size=3			,	drop_last=_a   )
    __lowerCamelCase       :       Union[str, Any]           = [[[0, 1, 0]], [[1, 0, 1]]]
    self.check_batch_sampler_shards(_a			,	_a   )

    __lowerCamelCase       :       Any           = BatchSampler(range(2   )			,	batch_size=3			,	drop_last=_a   )
    __lowerCamelCase       :       List[str]           = [[], []]
    self.check_batch_sampler_shards(_a			,	_a   )




  def 						_lowercase     ( self      :	Dict   )					->					str:
    # Check the shards when the dataset is a round multiple of batch size.
    __lowerCamelCase       :       Dict           = BatchSampler(range(24   )			,	batch_size=4			,	drop_last=_a   )
    __lowerCamelCase       :       List[str]           = [
        [[0, 1], [4, 5], [8, 9], [12, 13], [16, 17], [20, 21]],
        [[2, 3], [6, 7], [10, 11], [14, 15], [18, 19], [22, 23]],
    ]
    self.check_batch_sampler_shards(_a			,	_a			,	split_batches=_a   )

    __lowerCamelCase       :       Optional[Any]           = BatchSampler(range(24   )			,	batch_size=4			,	drop_last=_a   )
    # Expected shouldn't change
    self.check_batch_sampler_shards(_a			,	_a			,	split_batches=_a   )

    # Check the shards when the dataset is not a round multiple of batch size.
    __lowerCamelCase       :       int           = BatchSampler(range(22   )			,	batch_size=4			,	drop_last=_a   )
    __lowerCamelCase       :       Dict           = [
        [[0, 1], [4, 5], [8, 9], [12, 13], [16, 17], [20, 21]],
        [[2, 3], [6, 7], [10, 11], [14, 15], [18, 19], [0, 1]],
    ]
    self.check_batch_sampler_shards(_a			,	_a			,	split_batches=_a   )

    __lowerCamelCase       :       Tuple           = BatchSampler(range(22   )			,	batch_size=4			,	drop_last=_a   )
    __lowerCamelCase       :       Any           = [
        [[0, 1], [4, 5], [8, 9], [12, 13], [16, 17]],
        [[2, 3], [6, 7], [10, 11], [14, 15], [18, 19]],
    ]
    self.check_batch_sampler_shards(_a			,	_a			,	split_batches=_a   )

    # Check the shards when the dataset is not a round multiple of batch size or num_processes.
    __lowerCamelCase       :       List[Any]           = BatchSampler(range(21   )			,	batch_size=4			,	drop_last=_a   )
    __lowerCamelCase       :       List[Any]           = [
        [[0, 1], [4, 5], [8, 9], [12, 13], [16, 17], [20, 0]],
        [[2, 3], [6, 7], [10, 11], [14, 15], [18, 19], [1, 2]],
    ]
    self.check_batch_sampler_shards(_a			,	_a			,	split_batches=_a   )

    __lowerCamelCase       :       List[str]           = BatchSampler(range(21   )			,	batch_size=4			,	drop_last=_a   )
    __lowerCamelCase       :       Union[str, Any]           = [
        [[0, 1], [4, 5], [8, 9], [12, 13], [16, 17]],
        [[2, 3], [6, 7], [10, 11], [14, 15], [18, 19]],
    ]
    self.check_batch_sampler_shards(_a			,	_a			,	split_batches=_a   )

    # Check the shards when the dataset is very small.
    __lowerCamelCase       :       Optional[int]           = BatchSampler(range(2   )			,	batch_size=4			,	drop_last=_a   )
    __lowerCamelCase       :       List[str]           = [[[0, 1]], [[0, 1]]]
    self.check_batch_sampler_shards(_a			,	_a			,	split_batches=_a   )

    __lowerCamelCase       :       int           = BatchSampler(range(2   )			,	batch_size=4			,	drop_last=_a   )
    __lowerCamelCase       :       Optional[int]           = [[], []]
    self.check_batch_sampler_shards(_a			,	_a			,	split_batches=_a   )




  def 						_lowercase     ( self      :	Tuple   )					->					List[str]:
    # Check the shards when the dataset is a round multiple of total batch size.
    __lowerCamelCase       :       Optional[int]           = BatchSampler(range(24   )			,	batch_size=3			,	drop_last=_a   )
    __lowerCamelCase       :       str           = [
        [[0, 1, 2], [6, 7, 8], [12, 13, 14], [18, 19, 20]],
        [[3, 4, 5], [9, 10, 11], [15, 16, 17], [21, 22, 23]],
    ]
    self.check_batch_sampler_shards(_a			,	_a			,	even_batches=_a   )

    __lowerCamelCase       :       Tuple           = BatchSampler(range(24   )			,	batch_size=3			,	drop_last=_a   )
    # Expected shouldn't change
    self.check_batch_sampler_shards(_a			,	_a			,	even_batches=_a   )

    # Check the shards when the dataset is a round multiple of batch size but not total batch size.
    __lowerCamelCase       :       int           = BatchSampler(range(21   )			,	batch_size=3			,	drop_last=_a   )
    __lowerCamelCase       :       int           = [
        [[0, 1, 2], [6, 7, 8], [12, 13, 14], [18, 19, 20]],
        [[3, 4, 5], [9, 10, 11], [15, 16, 17]],
    ]
    self.check_batch_sampler_shards(_a			,	_a			,	even_batches=_a   )

    __lowerCamelCase       :       Optional[Any]           = BatchSampler(range(21   )			,	batch_size=3			,	drop_last=_a   )
    __lowerCamelCase       :       str           = [
        [[0, 1, 2], [6, 7, 8], [12, 13, 14]],
        [[3, 4, 5], [9, 10, 11], [15, 16, 17]],
    ]
    self.check_batch_sampler_shards(_a			,	_a			,	even_batches=_a   )

    # Check the shards when the dataset is not a round multiple of batch size but has a multiple of
    # num_processes batch.
    __lowerCamelCase       :       Dict           = BatchSampler(range(22   )			,	batch_size=3			,	drop_last=_a   )
    __lowerCamelCase       :       Union[str, Any]           = [
        [[0, 1, 2], [6, 7, 8], [12, 13, 14], [18, 19, 20]],
        [[3, 4, 5], [9, 10, 11], [15, 16, 17], [21]],
    ]
    self.check_batch_sampler_shards(_a			,	_a			,	even_batches=_a   )

    __lowerCamelCase       :       Any           = BatchSampler(range(22   )			,	batch_size=3			,	drop_last=_a   )
    __lowerCamelCase       :       Dict           = [
        [[0, 1, 2], [6, 7, 8], [12, 13, 14]],
        [[3, 4, 5], [9, 10, 11], [15, 16, 17]],
    ]
    self.check_batch_sampler_shards(_a			,	_a			,	even_batches=_a   )

    # Check the shards when the dataset is not a round multiple of batch size but and has not a multiple of
    # num_processes batch.
    __lowerCamelCase       :       Tuple           = BatchSampler(range(20   )			,	batch_size=3			,	drop_last=_a   )
    __lowerCamelCase       :       Optional[Any]           = [
        [[0, 1, 2], [6, 7, 8], [12, 13, 14], [18, 19]],
        [[3, 4, 5], [9, 10, 11], [15, 16, 17]],
    ]
    self.check_batch_sampler_shards(_a			,	_a			,	even_batches=_a   )

    __lowerCamelCase       :       List[Any]           = BatchSampler(range(20   )			,	batch_size=3			,	drop_last=_a   )
    __lowerCamelCase       :       str           = [
        [[0, 1, 2], [6, 7, 8], [12, 13, 14]],
        [[3, 4, 5], [9, 10, 11], [15, 16, 17]],
    ]
    self.check_batch_sampler_shards(_a			,	_a			,	even_batches=_a   )

    # Check the shards when the dataset is very small.
    __lowerCamelCase       :       Any           = BatchSampler(range(2   )			,	batch_size=3			,	drop_last=_a   )
    __lowerCamelCase       :       Union[str, Any]           = [[[0, 1]], []]
    self.check_batch_sampler_shards(_a			,	_a			,	even_batches=_a   )

    __lowerCamelCase       :       List[str]           = BatchSampler(range(2   )			,	batch_size=3			,	drop_last=_a   )
    __lowerCamelCase       :       Optional[int]           = [[], []]
    self.check_batch_sampler_shards(_a			,	_a			,	even_batches=_a   )




  def 						_lowercase     ( self      :	Any   )					->					Any:
    # Check the shards when the dataset is a round multiple of batch size.
    __lowerCamelCase       :       int           = BatchSampler(range(24   )			,	batch_size=4			,	drop_last=_a   )
    __lowerCamelCase       :       List[str]           = [
        [[0, 1], [4, 5], [8, 9], [12, 13], [16, 17], [20, 21]],
        [[2, 3], [6, 7], [10, 11], [14, 15], [18, 19], [22, 23]],
    ]
    self.check_batch_sampler_shards(_a			,	_a			,	split_batches=_a			,	even_batches=_a   )

    __lowerCamelCase       :       Any           = BatchSampler(range(24   )			,	batch_size=4			,	drop_last=_a   )
    # Expected shouldn't change
    self.check_batch_sampler_shards(_a			,	_a			,	split_batches=_a			,	even_batches=_a   )

    # Check the shards when the dataset is not a round multiple of batch size.
    __lowerCamelCase       :       Dict           = BatchSampler(range(22   )			,	batch_size=4			,	drop_last=_a   )
    __lowerCamelCase       :       List[Any]           = [
        [[0, 1], [4, 5], [8, 9], [12, 13], [16, 17], [20, 21]],
        [[2, 3], [6, 7], [10, 11], [14, 15], [18, 19]],
    ]
    self.check_batch_sampler_shards(_a			,	_a			,	split_batches=_a			,	even_batches=_a   )

    __lowerCamelCase       :       Any           = BatchSampler(range(22   )			,	batch_size=4			,	drop_last=_a   )
    __lowerCamelCase       :       Tuple           = [
        [[0, 1], [4, 5], [8, 9], [12, 13], [16, 17]],
        [[2, 3], [6, 7], [10, 11], [14, 15], [18, 19]],
    ]
    self.check_batch_sampler_shards(_a			,	_a			,	split_batches=_a			,	even_batches=_a   )

    # Check the shards when the dataset is not a round multiple of batch size or num_processes.
    __lowerCamelCase       :       List[str]           = BatchSampler(range(21   )			,	batch_size=4			,	drop_last=_a   )
    __lowerCamelCase       :       Optional[Any]           = [
        [[0, 1], [4, 5], [8, 9], [12, 13], [16, 17], [20]],
        [[2, 3], [6, 7], [10, 11], [14, 15], [18, 19]],
    ]
    self.check_batch_sampler_shards(_a			,	_a			,	split_batches=_a			,	even_batches=_a   )

    __lowerCamelCase       :       Optional[int]           = BatchSampler(range(21   )			,	batch_size=4			,	drop_last=_a   )
    __lowerCamelCase       :       Optional[Any]           = [
        [[0, 1], [4, 5], [8, 9], [12, 13], [16, 17]],
        [[2, 3], [6, 7], [10, 11], [14, 15], [18, 19]],
    ]
    self.check_batch_sampler_shards(_a			,	_a			,	split_batches=_a			,	even_batches=_a   )

    # Check the shards when the dataset is very small.
    __lowerCamelCase       :       Tuple           = BatchSampler(range(2   )			,	batch_size=4			,	drop_last=_a   )
    __lowerCamelCase       :       Union[str, Any]           = [[[0, 1]], []]
    self.check_batch_sampler_shards(_a			,	_a			,	split_batches=_a			,	even_batches=_a   )

    __lowerCamelCase       :       List[str]           = BatchSampler(range(2   )			,	batch_size=4			,	drop_last=_a   )
    __lowerCamelCase       :       int           = [[], []]
    self.check_batch_sampler_shards(_a			,	_a			,	split_batches=_a			,	even_batches=_a   )




  def 						_lowercase     ( self      :	Optional[Any]   )					->					Optional[int]:
    __lowerCamelCase       :       Union[str, Any]           = [[0, 1, 2], [3, 4], [5, 6, 7, 8], [9, 10, 11], [12, 13]]
    __lowerCamelCase       :       List[Any]           = [BatchSamplerShard(_a			,	2			,	_a			,	even_batches=_a   ) for i in range(2   )]

    self.assertEqual(len(batch_sampler_shards[0]   )			,	3   )
    self.assertEqual(len(batch_sampler_shards[1]   )			,	2   )

    self.assertListEqual(list(batch_sampler_shards[0]   )			,	[[0, 1, 2], [5, 6, 7, 8], [12, 13]]   )
    self.assertListEqual(list(batch_sampler_shards[1]   )			,	[[3, 4], [9, 10, 11]]   )




  def 						_lowercase     ( self      :	Optional[int]			,	_a      :	Tuple			,	_a      :	int			,	_a      :	Union[str, Any]			,	_a      :	str=False			,	_a      :	str=2			,	_a      :	Optional[Any]=False   )					->					str:
    random.seed(_a   )
    __lowerCamelCase       :       Tuple           = list(_a   )

    __lowerCamelCase       :       Any           = [
        IterableDatasetShard(
            _a			,	batch_size=_a			,	drop_last=_a			,	num_processes=_a			,	process_index=_a			,	split_batches=_a			,	)
        for i in range(_a   )
    ]
    __lowerCamelCase       :       List[str]           = []
    for iterable_dataset_shard in iterable_dataset_shards:
      # Since our random iterable dataset will be... random... we need to use a seed to get reproducible results.
      random.seed(_a   )
      iterable_dataset_lists.append(list(_a   )   )

    __lowerCamelCase       :       str           = batch_size // num_processes if split_batches else batch_size
    # All iterable dataset shard should have the same length, a round multiple of shard_batch_size
    __lowerCamelCase       :       Dict           = iterable_dataset_lists[0]
    for l in iterable_dataset_lists[1:]:
      self.assertEqual(len(_a   )			,	len(_a   )   )
      self.assertTrue(len(_a   ) % shard_batch_size == 0   )

    __lowerCamelCase       :       Optional[int]           = []
    for idx in range(0			,	len(_a   )			,	_a   ):
      for l in iterable_dataset_lists:
        observed += l[idx : idx + shard_batch_size]

    if not drop_last:
      while len(_a   ) < len(_a   ):
        reference += reference
    self.assertListEqual(_a			,	reference[: len(_a   )]   )




  def 						_lowercase     ( self      :	str   )					->					Optional[int]:
    __lowerCamelCase       :       Any           = 42
    __lowerCamelCase       :       str           = RandomIterableDataset()

    self.check_iterable_dataset_shards(_a			,	_a			,	batch_size=4			,	drop_last=_a			,	split_batches=_a   )
    self.check_iterable_dataset_shards(_a			,	_a			,	batch_size=4			,	drop_last=_a			,	split_batches=_a   )
    self.check_iterable_dataset_shards(_a			,	_a			,	batch_size=4			,	drop_last=_a			,	split_batches=_a   )
    self.check_iterable_dataset_shards(_a			,	_a			,	batch_size=4			,	drop_last=_a			,	split_batches=_a   )

    # Edge case with a very small dataset
    __lowerCamelCase       :       str           = RandomIterableDataset(max_length=2   )

    self.check_iterable_dataset_shards(_a			,	_a			,	batch_size=4			,	drop_last=_a			,	split_batches=_a   )
    self.check_iterable_dataset_shards(_a			,	_a			,	batch_size=4			,	drop_last=_a			,	split_batches=_a   )
    self.check_iterable_dataset_shards(_a			,	_a			,	batch_size=4			,	drop_last=_a			,	split_batches=_a   )
    self.check_iterable_dataset_shards(_a			,	_a			,	batch_size=4			,	drop_last=_a			,	split_batches=_a   )




  def 						_lowercase     ( self      :	Union[str, Any]   )					->					Optional[Any]:
    __lowerCamelCase       :       int           = BatchSampler(range(16   )			,	batch_size=4			,	drop_last=_a   )
    __lowerCamelCase       :       Union[str, Any]           = SkipBatchSampler(_a			,	2   )
    self.assertListEqual(list(_a   )			,	[[8, 9, 10, 11], [12, 13, 14, 15]]   )




  def 						_lowercase     ( self      :	str   )					->					List[str]:
    __lowerCamelCase       :       int           = SkipDataLoader(list(range(16   )   )			,	batch_size=4			,	skip_batches=2   )
    self.assertListEqual([t.tolist() for t in dataloader]			,	[[8, 9, 10, 11], [12, 13, 14, 15]]   )




  def 						_lowercase     ( self      :	str   )					->					str:
    __lowerCamelCase       :       str           = DataLoader(list(range(16   )   )			,	batch_size=4   )
    __lowerCamelCase       :       List[Any]           = skip_first_batches(_a			,	num_batches=2   )
    self.assertListEqual([t.tolist() for t in new_dataloader]			,	[[8, 9, 10, 11], [12, 13, 14, 15]]   )




  def 						_lowercase     ( self      :	Optional[int]   )					->					Dict:
    __lowerCamelCase       :       Optional[int]           = DataLoaderShard(list(range(16   )   )			,	batch_size=4   )
    for idx, _ in enumerate(_a   ):
      self.assertEqual(dataloader.end_of_dataloader			,	idx == 3   )

    # Test it also works on the second iteration
    for idx, _ in enumerate(_a   ):
      self.assertEqual(dataloader.end_of_dataloader			,	idx == 3   )




  def 						_lowercase     ( self      :	Any   )					->					List[str]:
    Accelerator()
    __lowerCamelCase       :       Union[str, Any]           = DataLoaderDispatcher(range(16   )			,	batch_size=4   )
    for idx, _ in enumerate(_a   ):
      self.assertEqual(dataloader.end_of_dataloader			,	idx == 3   )

    # Test it also works on the second iteration
    for idx, _ in enumerate(_a   ):
      self.assertEqual(dataloader.end_of_dataloader			,	idx == 3   )











'''simple docstring'''





# This model implementation is heavily inspired by https://github.com/haofanwang/ControlNet-for-Diffusers/

import gc
import random
import tempfile
import unittest

import numpy as np
import torch
from PIL import Image
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    ControlNetModel,
    DDIMScheduler,
    StableDiffusionControlNetImgaImgPipeline,
    UNetaDConditionModel,
)
from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion_controlnet import MultiControlNetModel
from diffusers.utils import floats_tensor, load_image, load_numpy, randn_tensor, slow, torch_device
from diffusers.utils.import_utils import is_xformers_available
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..pipeline_params import (
    IMAGE_TO_IMAGE_IMAGE_PARAMS,
    TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS,
    TEXT_GUIDED_IMAGE_VARIATION_PARAMS,
)
from ..test_pipelines_common import (
    PipelineKarrasSchedulerTesterMixin,
    PipelineLatentTesterMixin,
    PipelineTesterMixin,
)


enable_full_determinism()






class 					lowerCamelCase_    (				SCREAMING_SNAKE_CASE_   ,   SCREAMING_SNAKE_CASE_   ,   SCREAMING_SNAKE_CASE_   ,   unittest.TestCase     ):
  """simple docstring"""




  a_												=StableDiffusionControlNetImgaImgPipeline
  a_												=TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {"""height""", """width"""}
  a_												=TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
  a_												=IMAGE_TO_IMAGE_IMAGE_PARAMS.union({"""control_image"""}     )
  a_												=IMAGE_TO_IMAGE_IMAGE_PARAMS




  def 						_lowercase     ( self      :	Union[str, Any]   )					->					List[str]:
    torch.manual_seed(0   )
    __lowerCamelCase       :       Tuple           = 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			,	)
    torch.manual_seed(0   )
    __lowerCamelCase       :       Any           = ControlNetModel(
        block_out_channels=(32, 64)			,	layers_per_block=2			,	in_channels=4			,	down_block_types=('DownBlock2D', 'CrossAttnDownBlock2D')			,	cross_attention_dim=32			,	conditioning_embedding_out_channels=(16, 32)			,	)
    torch.manual_seed(0   )
    __lowerCamelCase       :       Any           = DDIMScheduler(
        beta_start=0.00085			,	beta_end=0.012			,	beta_schedule='scaled_linear'			,	clip_sample=_a			,	set_alpha_to_one=_a			,	)
    torch.manual_seed(0   )
    __lowerCamelCase       :       Dict           = 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           = CLIPTextConfig(
        bos_token_id=0			,	eos_token_id=2			,	hidden_size=32			,	intermediate_size=37			,	layer_norm_eps=1e-05			,	num_attention_heads=4			,	num_hidden_layers=5			,	pad_token_id=1			,	vocab_size=1000			,	)
    __lowerCamelCase       :       Union[str, Any]           = CLIPTextModel(_a   )
    __lowerCamelCase       :       Any           = CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'   )

    __lowerCamelCase       :       int           = {
        'unet': unet,
        'controlnet': controlnet,
        'scheduler': scheduler,
        'vae': vae,
        'text_encoder': text_encoder,
        'tokenizer': tokenizer,
        'safety_checker': None,
        'feature_extractor': None,
    }
    return components




  def 						_lowercase     ( self      :	Union[str, Any]			,	_a      :	Any			,	_a      :	List[str]=0   )					->					List[str]:
    if str(_a   ).startswith('mps'   ):
      __lowerCamelCase       :       Any           = torch.manual_seed(_a   )
    else:
      __lowerCamelCase       :       Any           = torch.Generator(device=_a   ).manual_seed(_a   )

    __lowerCamelCase       :       Union[str, Any]           = 2
    __lowerCamelCase       :       Any           = randn_tensor(
        (1, 3, 32 * controlnet_embedder_scale_factor, 32 * controlnet_embedder_scale_factor)			,	generator=_a			,	device=torch.device(_a   )			,	)
    __lowerCamelCase       :       Optional[Any]           = floats_tensor(control_image.shape			,	rng=random.Random(_a   )   ).to(_a   )
    __lowerCamelCase       :       List[Any]           = image.cpu().permute(0			,	2			,	3			,	1   )[0]
    __lowerCamelCase       :       int           = Image.fromarray(np.uinta(_a   )   ).convert('RGB'   ).resize((64, 64)   )
    __lowerCamelCase       :       int           = {
        'prompt': 'A painting of a squirrel eating a burger',
        'generator': generator,
        'num_inference_steps': 2,
        'guidance_scale': 6.0,
        'output_type': 'numpy',
        'image': image,
        'control_image': control_image,
    }

    return inputs




  def 						_lowercase     ( self      :	List[str]   )					->					Optional[Any]:
    return self._test_attention_slicing_forward_pass(expected_max_diff=2e-3   )




  @unittest.skipIf(
      torch_device != 'cuda' or not is_xformers_available()			,	reason='XFormers attention is only available with CUDA and `xformers` installed'			,	)
  def 						_lowercase     ( self      :	str   )					->					List[str]:
    self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=2e-3   )




  def 						_lowercase     ( self      :	Any   )					->					List[str]:
    self._test_inference_batch_single_identical(expected_max_diff=2e-3   )






class 					lowerCamelCase_    (				SCREAMING_SNAKE_CASE_   ,   SCREAMING_SNAKE_CASE_   ,   unittest.TestCase     ):
  """simple docstring"""




  a_												=StableDiffusionControlNetImgaImgPipeline
  a_												=TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {"""height""", """width"""}
  a_												=TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
  a_												=frozenset([]     )  # TO_DO: add image_params once refactored VaeImageProcessor.preprocess




  def 						_lowercase     ( self      :	List[Any]   )					->					List[str]:
    torch.manual_seed(0   )
    __lowerCamelCase       :       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			,	)
    torch.manual_seed(0   )

    def init_weights(_a      :	str   ):
      if isinstance(_a			,	torch.nn.Convad   ):
        torch.nn.init.normal(m.weight   )
        m.bias.data.fill_(1.0   )

    __lowerCamelCase       :       Any           = ControlNetModel(
        block_out_channels=(32, 64)			,	layers_per_block=2			,	in_channels=4			,	down_block_types=('DownBlock2D', 'CrossAttnDownBlock2D')			,	cross_attention_dim=32			,	conditioning_embedding_out_channels=(16, 32)			,	)
    controlneta.controlnet_down_blocks.apply(_a   )

    torch.manual_seed(0   )
    __lowerCamelCase       :       Any           = ControlNetModel(
        block_out_channels=(32, 64)			,	layers_per_block=2			,	in_channels=4			,	down_block_types=('DownBlock2D', 'CrossAttnDownBlock2D')			,	cross_attention_dim=32			,	conditioning_embedding_out_channels=(16, 32)			,	)
    controlneta.controlnet_down_blocks.apply(_a   )

    torch.manual_seed(0   )
    __lowerCamelCase       :       List[Any]           = DDIMScheduler(
        beta_start=0.00085			,	beta_end=0.012			,	beta_schedule='scaled_linear'			,	clip_sample=_a			,	set_alpha_to_one=_a			,	)
    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       :       Optional[int]           = CLIPTextConfig(
        bos_token_id=0			,	eos_token_id=2			,	hidden_size=32			,	intermediate_size=37			,	layer_norm_eps=1e-05			,	num_attention_heads=4			,	num_hidden_layers=5			,	pad_token_id=1			,	vocab_size=1000			,	)
    __lowerCamelCase       :       Dict           = CLIPTextModel(_a   )
    __lowerCamelCase       :       Optional[Any]           = CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'   )

    __lowerCamelCase       :       List[str]           = MultiControlNetModel([controlneta, controlneta]   )

    __lowerCamelCase       :       str           = {
        'unet': unet,
        'controlnet': controlnet,
        'scheduler': scheduler,
        'vae': vae,
        'text_encoder': text_encoder,
        'tokenizer': tokenizer,
        'safety_checker': None,
        'feature_extractor': None,
    }
    return components




  def 						_lowercase     ( self      :	Dict			,	_a      :	int			,	_a      :	Union[str, Any]=0   )					->					Union[str, Any]:
    if str(_a   ).startswith('mps'   ):
      __lowerCamelCase       :       Optional[Any]           = torch.manual_seed(_a   )
    else:
      __lowerCamelCase       :       Union[str, Any]           = torch.Generator(device=_a   ).manual_seed(_a   )

    __lowerCamelCase       :       Any           = 2

    __lowerCamelCase       :       Any           = [
        randn_tensor(
            (1, 3, 32 * controlnet_embedder_scale_factor, 32 * controlnet_embedder_scale_factor)			,	generator=_a			,	device=torch.device(_a   )			,	),
        randn_tensor(
            (1, 3, 32 * controlnet_embedder_scale_factor, 32 * controlnet_embedder_scale_factor)			,	generator=_a			,	device=torch.device(_a   )			,	),
    ]

    __lowerCamelCase       :       List[Any]           = floats_tensor(control_image[0].shape			,	rng=random.Random(_a   )   ).to(_a   )
    __lowerCamelCase       :       Union[str, Any]           = image.cpu().permute(0			,	2			,	3			,	1   )[0]
    __lowerCamelCase       :       Dict           = Image.fromarray(np.uinta(_a   )   ).convert('RGB'   ).resize((64, 64)   )
    __lowerCamelCase       :       Dict           = {
        'prompt': 'A painting of a squirrel eating a burger',
        'generator': generator,
        'num_inference_steps': 2,
        'guidance_scale': 6.0,
        'output_type': 'numpy',
        'image': image,
        'control_image': control_image,
    }

    return inputs




  def 						_lowercase     ( self      :	Tuple   )					->					Union[str, Any]:
    __lowerCamelCase       :       Any           = self.get_dummy_components()
    __lowerCamelCase       :       Any           = self.pipeline_class(**_a   )
    pipe.to(_a   )

    __lowerCamelCase       :       Optional[int]           = 10.0
    __lowerCamelCase       :       Tuple           = 4

    __lowerCamelCase       :       str           = self.get_dummy_inputs(_a   )
    __lowerCamelCase       :       int           = steps
    __lowerCamelCase       :       List[Any]           = scale
    __lowerCamelCase       :       Optional[Any]           = pipe(**_a   )[0]

    __lowerCamelCase       :       Dict           = self.get_dummy_inputs(_a   )
    __lowerCamelCase       :       List[str]           = steps
    __lowerCamelCase       :       List[Any]           = scale
    __lowerCamelCase       :       Dict           = pipe(**_a			,	control_guidance_start=0.1			,	control_guidance_end=0.2   )[0]

    __lowerCamelCase       :       Any           = self.get_dummy_inputs(_a   )
    __lowerCamelCase       :       Union[str, Any]           = steps
    __lowerCamelCase       :       Any           = scale
    __lowerCamelCase       :       Optional[int]           = pipe(**_a			,	control_guidance_start=[0.1, 0.3]			,	control_guidance_end=[0.2, 0.7]   )[0]

    __lowerCamelCase       :       Any           = self.get_dummy_inputs(_a   )
    __lowerCamelCase       :       Tuple           = steps
    __lowerCamelCase       :       List[Any]           = scale
    __lowerCamelCase       :       List[str]           = pipe(**_a			,	control_guidance_start=0.4			,	control_guidance_end=[0.5, 0.8]   )[0]

    # make sure that all outputs are different
    assert np.sum(np.abs(output_a - output_a   )   ) > 1e-3
    assert np.sum(np.abs(output_a - output_a   )   ) > 1e-3
    assert np.sum(np.abs(output_a - output_a   )   ) > 1e-3




  def 						_lowercase     ( self      :	Optional[Any]   )					->					List[str]:
    return self._test_attention_slicing_forward_pass(expected_max_diff=2e-3   )




  @unittest.skipIf(
      torch_device != 'cuda' or not is_xformers_available()			,	reason='XFormers attention is only available with CUDA and `xformers` installed'			,	)
  def 						_lowercase     ( self      :	Dict   )					->					Tuple:
    self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=2e-3   )




  def 						_lowercase     ( self      :	str   )					->					int:
    self._test_inference_batch_single_identical(expected_max_diff=2e-3   )




  def 						_lowercase     ( self      :	List[Any]   )					->					str:
    __lowerCamelCase       :       int           = self.get_dummy_components()
    __lowerCamelCase       :       List[Any]           = self.pipeline_class(**_a   )
    pipe.to(_a   )
    pipe.set_progress_bar_config(disable=_a   )
    with tempfile.TemporaryDirectory() as tmpdir:
      try:
        # save_pretrained is not implemented for Multi-ControlNet
        pipe.save_pretrained(_a   )
      except NotImplementedError:
        pass






@slow
@require_torch_gpu
class 					lowerCamelCase_    (				unittest.TestCase     ):
  """simple docstring"""




  def 						_lowercase     ( self      :	List[str]   )					->					List[str]:
    super().tearDown()
    gc.collect()
    torch.cuda.empty_cache()




  def 						_lowercase     ( self      :	Optional[Any]   )					->					Union[str, Any]:
    __lowerCamelCase       :       Optional[Any]           = ControlNetModel.from_pretrained('lllyasviel/sd-controlnet-canny'   )

    __lowerCamelCase       :       List[str]           = StableDiffusionControlNetImgaImgPipeline.from_pretrained(
        'runwayml/stable-diffusion-v1-5'			,	safety_checker=_a			,	controlnet=_a   )
    pipe.enable_model_cpu_offload()
    pipe.set_progress_bar_config(disable=_a   )

    __lowerCamelCase       :       Optional[int]           = torch.Generator(device='cpu'   ).manual_seed(0   )
    __lowerCamelCase       :       List[Any]           = 'evil space-punk bird'
    __lowerCamelCase       :       List[str]           = load_image(
        'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/sd_controlnet/bird_canny.png'   ).resize((512, 512)   )
    __lowerCamelCase       :       Any           = load_image(
        'https://huggingface.co/lllyasviel/sd-controlnet-canny/resolve/main/images/bird.png'   ).resize((512, 512)   )

    __lowerCamelCase       :       Dict           = pipe(
        _a			,	_a			,	control_image=_a			,	generator=_a			,	output_type='np'			,	num_inference_steps=50			,	strength=0.6			,	)

    __lowerCamelCase       :       Optional[Any]           = output.images[0]

    assert image.shape == (512, 512, 3)

    __lowerCamelCase       :       List[str]           = load_numpy(
        'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/sd_controlnet/img2img.npy'   )

    assert np.abs(expected_image - image   ).max() < 9e-2






