
"""simple docstring"""




import logging
import math
from functools import partial
from typing import Any, Callable, Dict, Iterable, List, Optional, Sequence, Tuple, Union

import torch

from .tensor_utils import tensor_tree_map, tree_map

def lowercase_  (   _snake_case						):
							SCREAMING_SNAKE_CASE__						:       str           =   []
							if isinstance(_snake_case   ,_snake_case						):
														for v in tree.values():
																					shapes.extend(_fetch_dims(_snake_case						)						)
							elif isinstance(_snake_case   ,(list, tuple)						):
														for t in tree:
																					shapes.extend(_fetch_dims(_snake_case						)						)
							elif isinstance(_snake_case   ,torch.Tensor						):
														shapes.append(tree.shape						)
							else:
														raise ValueError("""Not supported"""						)

							return shapes

@torch.jit.ignore
def lowercase_  (   _snake_case   ,_snake_case						):
							SCREAMING_SNAKE_CASE__						:       Dict           =   []
							for d in reversed(_snake_case						):
														idx.append(flat_idx % d						)
														SCREAMING_SNAKE_CASE__						:       List[str]           =   flat_idx // d

							return tuple(reversed(_snake_case						)						)

@torch.jit.ignore
def lowercase_  (   _snake_case   ,_snake_case   ,_snake_case   ,_snake_case = None   ,_snake_case = None   ,):

							# start_edges and end_edges both indicate whether, starting from any given
							# dimension, the start/end index is at the top/bottom edge of the
							# corresponding tensor, modeled as a tree
							def reduce_edge_list(_snake_case						) -> None:
														SCREAMING_SNAKE_CASE__						:       List[str]           =   True
														for i in range(len(_snake_case						)						):
																					SCREAMING_SNAKE_CASE__						:       Dict           =   -1 * (i + 1)
																					l[reversed_idx] &= tally
																					SCREAMING_SNAKE_CASE__						:       Optional[Any]           =   l[reversed_idx]

							if start_edges is None:
														SCREAMING_SNAKE_CASE__						:       List[str]           =   [s == 0 for s in start]
														reduce_edge_list(_snake_case						)
							if end_edges is None:
														SCREAMING_SNAKE_CASE__						:       List[str]           =   [e == (d - 1) for e, d in zip(_snake_case   ,_snake_case						)]
														reduce_edge_list(_snake_case						)

							# Base cases. Either start/end are empty and we're done, or the final,
							# one-dimensional tensor can be simply sliced
							if len(_snake_case						) == 0:
														return [()]
							elif len(_snake_case						) == 1:
														return [(slice(start[0]   ,end[0] + 1						),)]

							SCREAMING_SNAKE_CASE__						:       List[Tuple[slice, ...]]           =   []
							SCREAMING_SNAKE_CASE__						:       List[slice]           =   []

							# Dimensions common to start and end can be selected directly
							for s, e in zip(_snake_case   ,_snake_case						):
														if s == e:
																					path_list.append(slice(_snake_case   ,s + 1						)						)
														else:
																					break

							SCREAMING_SNAKE_CASE__						:       Tuple[slice, ...]           =   tuple(_snake_case						)
							SCREAMING_SNAKE_CASE__						:       List[Any]           =   len(_snake_case						)

							# start == end, and we're done
							if divergence_idx == len(_snake_case						):
														return [path]

							def upper() -> Tuple[Tuple[slice, ...], ...]:
														assert start_edges is not None
														assert end_edges is not None

														SCREAMING_SNAKE_CASE__						:       Optional[Any]           =   start[divergence_idx]
														return tuple(
														    path + (slice(_snake_case   ,sdi + 1						),) + s
														    for s in _get_minimal_slice_set(
														        start[divergence_idx + 1 :]   ,[d - 1 for d in dims[divergence_idx + 1 :]]   ,dims[divergence_idx + 1 :]   ,start_edges=start_edges[divergence_idx + 1 :]   ,end_edges=[True for _ in end_edges[divergence_idx + 1 :]]   ,)						)

							def lower() -> Tuple[Tuple[slice, ...], ...]:
														assert start_edges is not None
														assert end_edges is not None

														SCREAMING_SNAKE_CASE__						:       List[Any]           =   end[divergence_idx]
														return tuple(
														    path + (slice(_snake_case   ,edi + 1						),) + s
														    for s in _get_minimal_slice_set(
														        [0 for _ in start[divergence_idx + 1 :]]   ,end[divergence_idx + 1 :]   ,dims[divergence_idx + 1 :]   ,start_edges=[True for _ in start_edges[divergence_idx + 1 :]]   ,end_edges=end_edges[divergence_idx + 1 :]   ,)						)

							# If both start and end are at the edges of the subtree rooted at
							# divergence_idx, we can just select the whole subtree at once
							if start_edges[divergence_idx] and end_edges[divergence_idx]:
														slices.append(path + (slice(start[divergence_idx]   ,end[divergence_idx] + 1						),)						)
							# If just start is at the edge, we can grab almost all of the subtree,
							# treating only the ragged bottom edge as an edge case
							elif start_edges[divergence_idx]:
														slices.append(path + (slice(start[divergence_idx]   ,end[divergence_idx]						),)						)
														slices.extend(lower()						)
							# Analogous to the previous case, but the top is ragged this time
							elif end_edges[divergence_idx]:
														slices.extend(upper()						)
														slices.append(path + (slice(start[divergence_idx] + 1   ,end[divergence_idx] + 1						),)						)
							# If both sides of the range are ragged, we need to handle both sides
							# separately. If there's contiguous meat in between them, we can index it
							# in one big chunk
							else:
														slices.extend(upper()						)
														SCREAMING_SNAKE_CASE__						:       Dict           =   end[divergence_idx] - start[divergence_idx]
														if middle_ground > 1:
																					slices.append(path + (slice(start[divergence_idx] + 1   ,end[divergence_idx]						),)						)
														slices.extend(lower()						)

							return slices

@torch.jit.ignore
def lowercase_  (   _snake_case   ,_snake_case   ,_snake_case   ,_snake_case						):
							SCREAMING_SNAKE_CASE__						:       Tuple           =   t.shape[:no_batch_dims]
							SCREAMING_SNAKE_CASE__						:       Dict           =   list(_flat_idx_to_idx(_snake_case   ,_snake_case						)						)
							# _get_minimal_slice_set is inclusive
							SCREAMING_SNAKE_CASE__						:       Any           =   list(_flat_idx_to_idx(flat_end - 1   ,_snake_case						)						)

							# Get an ordered list of slices to perform
							SCREAMING_SNAKE_CASE__						:       List[str]           =   _get_minimal_slice_set(
							    _snake_case   ,_snake_case   ,_snake_case   ,)

							SCREAMING_SNAKE_CASE__						:       Any           =   [t[s] for s in slices]

							return torch.cat([s.view((-1,) + t.shape[no_batch_dims:]						) for s in sliced_tensors]						)

def lowercase_  (   _snake_case   ,_snake_case   ,_snake_case   ,_snake_case   ,_snake_case = False   ,_snake_case = None   ,_snake_case = False   ,):
							if not (len(_snake_case						) > 0):
														raise ValueError("""Must provide at least one input"""						)

							SCREAMING_SNAKE_CASE__						:       str           =   [shape[:no_batch_dims] for shape in _fetch_dims(_snake_case						)]
							SCREAMING_SNAKE_CASE__						:       Dict           =   tuple([max(_snake_case						) for s in zip(*_snake_case						)]						)

							def _prep_inputs(_snake_case						) -> torch.Tensor:
														if not low_mem:
																					if not sum(t.shape[:no_batch_dims]						) == no_batch_dims:
																												SCREAMING_SNAKE_CASE__						:       Any           =   t.expand(orig_batch_dims + t.shape[no_batch_dims:]						)
																					SCREAMING_SNAKE_CASE__						:       Optional[int]           =   t.reshape(-1   ,*t.shape[no_batch_dims:]						)
														else:
																					SCREAMING_SNAKE_CASE__						:       Optional[Any]           =   t.expand(orig_batch_dims + t.shape[no_batch_dims:]						)
														return t

							SCREAMING_SNAKE_CASE__						:       Dict[str, Any]           =   tensor_tree_map(_prep_inputs   ,_snake_case						)
							SCREAMING_SNAKE_CASE__						:       int           =   None
							if _out is not None:
														SCREAMING_SNAKE_CASE__						:       Dict           =   tensor_tree_map(lambda _snake_case						: t.view([-1] + list(t.shape[no_batch_dims:]						)						)   ,_out						)

							SCREAMING_SNAKE_CASE__						:       Optional[Any]           =   1
							for d in orig_batch_dims:
														flat_batch_dim *= d

							SCREAMING_SNAKE_CASE__						:       Optional[Any]           =   flat_batch_dim // chunk_size + (flat_batch_dim % chunk_size != 0)

							def _select_chunk(_snake_case						) -> torch.Tensor:
														return t[i : i + chunk_size] if t.shape[0] != 1 else t

							SCREAMING_SNAKE_CASE__						:       int           =   0
							SCREAMING_SNAKE_CASE__						:       Dict           =   prepped_outputs
							for _ in range(_snake_case						):
														# Chunk the input
														if not low_mem:
																					SCREAMING_SNAKE_CASE__						:       List[str]           =   _select_chunk
														else:
																					SCREAMING_SNAKE_CASE__						:       int           =   partial(
																					    _chunk_slice   ,flat_start=_snake_case   ,flat_end=min(_snake_case   ,i + chunk_size						)   ,no_batch_dims=len(_snake_case						)   ,)

														SCREAMING_SNAKE_CASE__						:       Dict[str, Any]           =   tensor_tree_map(_snake_case   ,_snake_case						)

														# Run the layer on the chunk
														SCREAMING_SNAKE_CASE__						:       str           =   layer(**_snake_case						)

														# Allocate space for the output
														if out is None:
																					SCREAMING_SNAKE_CASE__						:       Dict           =   tensor_tree_map(lambda _snake_case						: t.new_zeros((flat_batch_dim,) + t.shape[1:]						)   ,_snake_case						)

														# Put the chunk in its pre-allocated space
														if isinstance(_snake_case   ,_snake_case						):

																					def assign(_snake_case   ,_snake_case						) -> None:
																												for k, v in da.items():
																																			if isinstance(_snake_case   ,_snake_case						):
																																										assign(_snake_case   ,da[k]						)
																																			else:
																																										if _add_into_out:
																																																	v[i : i + chunk_size] += da[k]
																																										else:
																																																	SCREAMING_SNAKE_CASE__						:       Union[str, Any]           =   da[k]

																					assign(_snake_case   ,_snake_case						)
														elif isinstance(_snake_case   ,_snake_case						):
																					for xa, xa in zip(_snake_case   ,_snake_case						):
																												if _add_into_out:
																																			xa[i : i + chunk_size] += xa
																												else:
																																			SCREAMING_SNAKE_CASE__						:       List[str]           =   xa
														elif isinstance(_snake_case   ,torch.Tensor						):
																					if _add_into_out:
																												out[i : i + chunk_size] += output_chunk
																					else:
																												SCREAMING_SNAKE_CASE__						:       Dict           =   output_chunk
														else:
																					raise ValueError("""Not supported"""						)

														i += chunk_size

							SCREAMING_SNAKE_CASE__						:       Dict           =   tensor_tree_map(lambda _snake_case						: t.view(orig_batch_dims + t.shape[1:]						)   ,_snake_case						)

							return out



class 	lowerCAmelCase_						:


				"""simple docstring"""



				def __init__(self   ,       SCREAMING_SNAKE_CASE__ = 5_12   ,       )					->		Dict:



											"""simple docstring"""




											SCREAMING_SNAKE_CASE__						:       int           =   max_chunk_size
											SCREAMING_SNAKE_CASE__						:       Optional[int]           =   None
											SCREAMING_SNAKE_CASE__						:       Optional[tuple]           =   None



				def 	__magic_name__    (self   ,       SCREAMING_SNAKE_CASE__   ,       SCREAMING_SNAKE_CASE__   ,       SCREAMING_SNAKE_CASE__    )					->		int:



											"""simple docstring"""




											logging.info("""Tuning chunk size..."""    )

											if min_chunk_size >= self.max_chunk_size:
																		return min_chunk_size

											SCREAMING_SNAKE_CASE__						:       List[int]           =   [2**l for l in range(int(math.log(self.max_chunk_size   ,       2    )    ) + 1    )]
											SCREAMING_SNAKE_CASE__						:       Any           =   [c for c in candidates if c > min_chunk_size]
											SCREAMING_SNAKE_CASE__						:       Dict           =   [min_chunk_size] + candidates
											candidates[-1] += 4

											def test_chunk_size(SCREAMING_SNAKE_CASE__    ) -> bool:
																		try:
																									with torch.no_grad():
																																fn(*SCREAMING_SNAKE_CASE__   ,       chunk_size=SCREAMING_SNAKE_CASE__    )
																									return True
																		except RuntimeError:
																									return False

											SCREAMING_SNAKE_CASE__						:       Any           =   0
											SCREAMING_SNAKE_CASE__						:       Dict           =   len(SCREAMING_SNAKE_CASE__    ) - 1
											while i > min_viable_chunk_size_index:
																		SCREAMING_SNAKE_CASE__						:       int           =   test_chunk_size(candidates[i]    )
																		if not viable:
																									SCREAMING_SNAKE_CASE__						:       Dict           =   (min_viable_chunk_size_index + i) // 2
																		else:
																									SCREAMING_SNAKE_CASE__						:       Any           =   i
																									SCREAMING_SNAKE_CASE__						:       Dict           =   (i + len(SCREAMING_SNAKE_CASE__    ) - 1) // 2

											return candidates[min_viable_chunk_size_index]



				def 	__magic_name__    (self   ,       SCREAMING_SNAKE_CASE__   ,       SCREAMING_SNAKE_CASE__    )					->		bool:



											"""simple docstring"""




											SCREAMING_SNAKE_CASE__						:       List[Any]           =   True
											for aa, aa in zip(SCREAMING_SNAKE_CASE__   ,       SCREAMING_SNAKE_CASE__    ):
																		assert type(SCREAMING_SNAKE_CASE__    ) == type(SCREAMING_SNAKE_CASE__    )
																		if isinstance(SCREAMING_SNAKE_CASE__   ,       (list, tuple)    ):
																									consistent &= self._compare_arg_caches(SCREAMING_SNAKE_CASE__   ,       SCREAMING_SNAKE_CASE__    )
																		elif isinstance(SCREAMING_SNAKE_CASE__   ,       SCREAMING_SNAKE_CASE__    ):
																									SCREAMING_SNAKE_CASE__						:       str           =   [v for _, v in sorted(aa.items()   ,       key=lambda SCREAMING_SNAKE_CASE__    : x[0]    )]
																									SCREAMING_SNAKE_CASE__						:       Tuple           =   [v for _, v in sorted(aa.items()   ,       key=lambda SCREAMING_SNAKE_CASE__    : x[0]    )]
																									consistent &= self._compare_arg_caches(SCREAMING_SNAKE_CASE__   ,       SCREAMING_SNAKE_CASE__    )
																		else:
																									consistent &= aa == aa

											return consistent



				def 	__magic_name__    (self   ,       SCREAMING_SNAKE_CASE__   ,       SCREAMING_SNAKE_CASE__   ,       SCREAMING_SNAKE_CASE__   ,       )					->		int:



											"""simple docstring"""




											SCREAMING_SNAKE_CASE__						:       Union[str, Any]           =   True
											SCREAMING_SNAKE_CASE__						:       tuple           =   tree_map(lambda SCREAMING_SNAKE_CASE__    : a.shape if isinstance(SCREAMING_SNAKE_CASE__   ,       torch.Tensor    ) else a   ,       SCREAMING_SNAKE_CASE__   ,       SCREAMING_SNAKE_CASE__    )
											if self.cached_arg_data is not None:
																		# If args have changed shape/value, we need to re-tune
																		assert len(self.cached_arg_data    ) == len(SCREAMING_SNAKE_CASE__    )
																		SCREAMING_SNAKE_CASE__						:       Any           =   self._compare_arg_caches(self.cached_arg_data   ,       SCREAMING_SNAKE_CASE__    )
											else:
																		# Otherwise, we can reuse the precomputed value
																		SCREAMING_SNAKE_CASE__						:       Dict           =   False

											if not consistent:
																		SCREAMING_SNAKE_CASE__						:       Dict           =   self._determine_favorable_chunk_size(
																		    SCREAMING_SNAKE_CASE__   ,       SCREAMING_SNAKE_CASE__   ,       SCREAMING_SNAKE_CASE__   ,       )
																		SCREAMING_SNAKE_CASE__						:       Any           =   arg_data

											assert self.cached_chunk_size is not None

											return self.cached_chunk_size






"""simple docstring"""




import gc
import random
import unittest

import torch

from diffusers import (
    IFImgaImgPipeline,
    IFImgaImgSuperResolutionPipeline,
    IFInpaintingPipeline,
    IFInpaintingSuperResolutionPipeline,
    IFPipeline,
    IFSuperResolutionPipeline,
)
from diffusers.models.attention_processor import AttnAddedKVProcessor
from diffusers.utils.import_utils import is_xformers_available
from diffusers.utils.testing_utils import floats_tensor, load_numpy, require_torch_gpu, skip_mps, slow, torch_device

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineTesterMixin, assert_mean_pixel_difference
from . import IFPipelineTesterMixin



@skip_mps
class 	lowerCAmelCase_						(a__ ,					a__ ,					unittest.TestCase  ):


				"""simple docstring"""



				__UpperCamelCase      :						Optional[Any]					       =   IFPipeline
				__UpperCamelCase      :						Dict					       =   TEXT_TO_IMAGE_PARAMS - {'''width''', '''height''', '''latents'''}
				__UpperCamelCase      :						Any					       =   TEXT_TO_IMAGE_BATCH_PARAMS
				__UpperCamelCase      :						Union[str, Any]					       =   PipelineTesterMixin.required_optional_params - {'''latents'''}



				def 	__magic_name__    (self    )					->		Union[str, Any]:



											"""simple docstring"""




											return self._get_dummy_components()



				def 	__magic_name__    (self   ,       SCREAMING_SNAKE_CASE__   ,       SCREAMING_SNAKE_CASE__=0    )					->		List[Any]:



											"""simple docstring"""




											if str(SCREAMING_SNAKE_CASE__    ).startswith("""mps"""    ):
																		SCREAMING_SNAKE_CASE__						:       Dict           =   torch.manual_seed(SCREAMING_SNAKE_CASE__    )
											else:
																		SCREAMING_SNAKE_CASE__						:       Any           =   torch.Generator(device=SCREAMING_SNAKE_CASE__    ).manual_seed(SCREAMING_SNAKE_CASE__    )

											SCREAMING_SNAKE_CASE__						:       int           =   {
											    """prompt""": """A painting of a squirrel eating a burger""",
											    """generator""": generator,
											    """num_inference_steps""": 2,
											    """output_type""": """numpy""",
											}

											return inputs



				def 	__magic_name__    (self    )					->		Tuple:



											"""simple docstring"""




											self._test_save_load_optional_components()



				@unittest.skipIf(torch_device != """cuda"""   ,       reason="""float16 requires CUDA"""    )
				def 	__magic_name__    (self    )					->		List[str]:



											"""simple docstring"""




											super().test_save_load_floataa(expected_max_diff=1E-1    )



				def 	__magic_name__    (self    )					->		List[Any]:



											"""simple docstring"""




											self._test_attention_slicing_forward_pass(expected_max_diff=1E-2    )



				def 	__magic_name__    (self    )					->		Tuple:



											"""simple docstring"""




											self._test_save_load_local()



				def 	__magic_name__    (self    )					->		Dict:



											"""simple docstring"""




											self._test_inference_batch_single_identical(
											    expected_max_diff=1E-2   ,       )



				@unittest.skipIf(
				    torch_device != """cuda""" or not is_xformers_available()   ,       reason="""XFormers attention is only available with CUDA and `xformers` installed"""   ,       )
				def 	__magic_name__    (self    )					->		Optional[int]:



											"""simple docstring"""




											self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=1E-3    )






@slow
@require_torch_gpu
class 	lowerCAmelCase_						(unittest.TestCase  ):


				"""simple docstring"""



				def 	__magic_name__    (self    )					->		Dict:



											"""simple docstring"""




											super().tearDown()
											gc.collect()
											torch.cuda.empty_cache()



				def 	__magic_name__    (self    )					->		Optional[int]:



											"""simple docstring"""




											SCREAMING_SNAKE_CASE__						:       Optional[int]           =   IFPipeline.from_pretrained("""DeepFloyd/IF-I-XL-v1.0"""   ,       variant="""fp16"""   ,       torch_dtype=torch.floataa    )

											SCREAMING_SNAKE_CASE__						:       Dict           =   IFSuperResolutionPipeline.from_pretrained(
											    """DeepFloyd/IF-II-L-v1.0"""   ,       variant="""fp16"""   ,       torch_dtype=torch.floataa   ,       text_encoder=SCREAMING_SNAKE_CASE__   ,       tokenizer=SCREAMING_SNAKE_CASE__    )

											# pre compute text embeddings and remove T5 to save memory

											pipe_a.text_encoder.to("""cuda"""    )

											SCREAMING_SNAKE_CASE__      ,      SCREAMING_SNAKE_CASE__						:       Union[str, Any]           =   pipe_a.encode_prompt("""anime turtle"""   ,       device="""cuda"""    )

											del pipe_a.tokenizer
											del pipe_a.text_encoder
											gc.collect()

											SCREAMING_SNAKE_CASE__						:       List[str]           =   None
											SCREAMING_SNAKE_CASE__						:       Union[str, Any]           =   None

											pipe_a.enable_model_cpu_offload()
											pipe_a.enable_model_cpu_offload()

											pipe_a.unet.set_attn_processor(AttnAddedKVProcessor()    )
											pipe_a.unet.set_attn_processor(AttnAddedKVProcessor()    )

											self._test_if(SCREAMING_SNAKE_CASE__   ,       SCREAMING_SNAKE_CASE__   ,       SCREAMING_SNAKE_CASE__   ,       SCREAMING_SNAKE_CASE__    )

											pipe_a.remove_all_hooks()
											pipe_a.remove_all_hooks()

											# img2img

											SCREAMING_SNAKE_CASE__						:       Union[str, Any]           =   IFImgaImgPipeline(**pipe_a.components    )
											SCREAMING_SNAKE_CASE__						:       Optional[Any]           =   IFImgaImgSuperResolutionPipeline(**pipe_a.components    )

											pipe_a.enable_model_cpu_offload()
											pipe_a.enable_model_cpu_offload()

											pipe_a.unet.set_attn_processor(AttnAddedKVProcessor()    )
											pipe_a.unet.set_attn_processor(AttnAddedKVProcessor()    )

											self._test_if_imgaimg(SCREAMING_SNAKE_CASE__   ,       SCREAMING_SNAKE_CASE__   ,       SCREAMING_SNAKE_CASE__   ,       SCREAMING_SNAKE_CASE__    )

											pipe_a.remove_all_hooks()
											pipe_a.remove_all_hooks()

											# inpainting

											SCREAMING_SNAKE_CASE__						:       Optional[Any]           =   IFInpaintingPipeline(**pipe_a.components    )
											SCREAMING_SNAKE_CASE__						:       int           =   IFInpaintingSuperResolutionPipeline(**pipe_a.components    )

											pipe_a.enable_model_cpu_offload()
											pipe_a.enable_model_cpu_offload()

											pipe_a.unet.set_attn_processor(AttnAddedKVProcessor()    )
											pipe_a.unet.set_attn_processor(AttnAddedKVProcessor()    )

											self._test_if_inpainting(SCREAMING_SNAKE_CASE__   ,       SCREAMING_SNAKE_CASE__   ,       SCREAMING_SNAKE_CASE__   ,       SCREAMING_SNAKE_CASE__    )



				def 	__magic_name__    (self   ,       SCREAMING_SNAKE_CASE__   ,       SCREAMING_SNAKE_CASE__   ,       SCREAMING_SNAKE_CASE__   ,       SCREAMING_SNAKE_CASE__    )					->		Optional[int]:



											"""simple docstring"""




											_start_torch_memory_measurement()

											SCREAMING_SNAKE_CASE__						:       Tuple           =   torch.Generator(device="""cpu"""    ).manual_seed(0    )
											SCREAMING_SNAKE_CASE__						:       Dict           =   pipe_a(
											    prompt_embeds=SCREAMING_SNAKE_CASE__   ,       negative_prompt_embeds=SCREAMING_SNAKE_CASE__   ,       num_inference_steps=2   ,       generator=SCREAMING_SNAKE_CASE__   ,       output_type="""np"""   ,       )

											SCREAMING_SNAKE_CASE__						:       int           =   output.images[0]

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

											SCREAMING_SNAKE_CASE__						:       Any           =   torch.cuda.max_memory_allocated()
											assert mem_bytes < 13 * 10**9

											SCREAMING_SNAKE_CASE__						:       Optional[int]           =   load_numpy(
											    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/if/test_if.npy"""    )
											assert_mean_pixel_difference(SCREAMING_SNAKE_CASE__   ,       SCREAMING_SNAKE_CASE__    )

											# pipeline 2

											_start_torch_memory_measurement()

											SCREAMING_SNAKE_CASE__						:       int           =   torch.Generator(device="""cpu"""    ).manual_seed(0    )

											SCREAMING_SNAKE_CASE__						:       Optional[int]           =   floats_tensor((1, 3, 64, 64)   ,       rng=random.Random(0    )    ).to(SCREAMING_SNAKE_CASE__    )

											SCREAMING_SNAKE_CASE__						:       Optional[Any]           =   pipe_a(
											    prompt_embeds=SCREAMING_SNAKE_CASE__   ,       negative_prompt_embeds=SCREAMING_SNAKE_CASE__   ,       image=SCREAMING_SNAKE_CASE__   ,       generator=SCREAMING_SNAKE_CASE__   ,       num_inference_steps=2   ,       output_type="""np"""   ,       )

											SCREAMING_SNAKE_CASE__						:       List[str]           =   output.images[0]

											assert image.shape == (2_56, 2_56, 3)

											SCREAMING_SNAKE_CASE__						:       List[Any]           =   torch.cuda.max_memory_allocated()
											assert mem_bytes < 4 * 10**9

											SCREAMING_SNAKE_CASE__						:       int           =   load_numpy(
											    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/if/test_if_superresolution_stage_II.npy"""    )
											assert_mean_pixel_difference(SCREAMING_SNAKE_CASE__   ,       SCREAMING_SNAKE_CASE__    )



				def 	__magic_name__    (self   ,       SCREAMING_SNAKE_CASE__   ,       SCREAMING_SNAKE_CASE__   ,       SCREAMING_SNAKE_CASE__   ,       SCREAMING_SNAKE_CASE__    )					->		str:



											"""simple docstring"""




											_start_torch_memory_measurement()

											SCREAMING_SNAKE_CASE__						:       int           =   floats_tensor((1, 3, 64, 64)   ,       rng=random.Random(0    )    ).to(SCREAMING_SNAKE_CASE__    )

											SCREAMING_SNAKE_CASE__						:       Tuple           =   torch.Generator(device="""cpu"""    ).manual_seed(0    )

											SCREAMING_SNAKE_CASE__						:       Any           =   pipe_a(
											    prompt_embeds=SCREAMING_SNAKE_CASE__   ,       negative_prompt_embeds=SCREAMING_SNAKE_CASE__   ,       image=SCREAMING_SNAKE_CASE__   ,       num_inference_steps=2   ,       generator=SCREAMING_SNAKE_CASE__   ,       output_type="""np"""   ,       )

											SCREAMING_SNAKE_CASE__						:       Dict           =   output.images[0]

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

											SCREAMING_SNAKE_CASE__						:       Optional[Any]           =   torch.cuda.max_memory_allocated()
											assert mem_bytes < 10 * 10**9

											SCREAMING_SNAKE_CASE__						:       str           =   load_numpy(
											    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/if/test_if_img2img.npy"""    )
											assert_mean_pixel_difference(SCREAMING_SNAKE_CASE__   ,       SCREAMING_SNAKE_CASE__    )

											# pipeline 2

											_start_torch_memory_measurement()

											SCREAMING_SNAKE_CASE__						:       Union[str, Any]           =   torch.Generator(device="""cpu"""    ).manual_seed(0    )

											SCREAMING_SNAKE_CASE__						:       List[Any]           =   floats_tensor((1, 3, 2_56, 2_56)   ,       rng=random.Random(0    )    ).to(SCREAMING_SNAKE_CASE__    )
											SCREAMING_SNAKE_CASE__						:       Optional[Any]           =   floats_tensor((1, 3, 64, 64)   ,       rng=random.Random(0    )    ).to(SCREAMING_SNAKE_CASE__    )

											SCREAMING_SNAKE_CASE__						:       Optional[Any]           =   pipe_a(
											    prompt_embeds=SCREAMING_SNAKE_CASE__   ,       negative_prompt_embeds=SCREAMING_SNAKE_CASE__   ,       image=SCREAMING_SNAKE_CASE__   ,       original_image=SCREAMING_SNAKE_CASE__   ,       generator=SCREAMING_SNAKE_CASE__   ,       num_inference_steps=2   ,       output_type="""np"""   ,       )

											SCREAMING_SNAKE_CASE__						:       List[Any]           =   output.images[0]

											assert image.shape == (2_56, 2_56, 3)

											SCREAMING_SNAKE_CASE__						:       Optional[Any]           =   torch.cuda.max_memory_allocated()
											assert mem_bytes < 4 * 10**9

											SCREAMING_SNAKE_CASE__						:       str           =   load_numpy(
											    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/if/test_if_img2img_superresolution_stage_II.npy"""    )
											assert_mean_pixel_difference(SCREAMING_SNAKE_CASE__   ,       SCREAMING_SNAKE_CASE__    )



				def 	__magic_name__    (self   ,       SCREAMING_SNAKE_CASE__   ,       SCREAMING_SNAKE_CASE__   ,       SCREAMING_SNAKE_CASE__   ,       SCREAMING_SNAKE_CASE__    )					->		Union[str, Any]:



											"""simple docstring"""




											_start_torch_memory_measurement()

											SCREAMING_SNAKE_CASE__						:       Dict           =   floats_tensor((1, 3, 64, 64)   ,       rng=random.Random(0    )    ).to(SCREAMING_SNAKE_CASE__    )
											SCREAMING_SNAKE_CASE__						:       Dict           =   floats_tensor((1, 3, 64, 64)   ,       rng=random.Random(1    )    ).to(SCREAMING_SNAKE_CASE__    )

											SCREAMING_SNAKE_CASE__						:       Dict           =   torch.Generator(device="""cpu"""    ).manual_seed(0    )
											SCREAMING_SNAKE_CASE__						:       List[str]           =   pipe_a(
											    prompt_embeds=SCREAMING_SNAKE_CASE__   ,       negative_prompt_embeds=SCREAMING_SNAKE_CASE__   ,       image=SCREAMING_SNAKE_CASE__   ,       mask_image=SCREAMING_SNAKE_CASE__   ,       num_inference_steps=2   ,       generator=SCREAMING_SNAKE_CASE__   ,       output_type="""np"""   ,       )

											SCREAMING_SNAKE_CASE__						:       List[Any]           =   output.images[0]

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

											SCREAMING_SNAKE_CASE__						:       Tuple           =   torch.cuda.max_memory_allocated()
											assert mem_bytes < 10 * 10**9

											SCREAMING_SNAKE_CASE__						:       Optional[Any]           =   load_numpy(
											    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/if/test_if_inpainting.npy"""    )
											assert_mean_pixel_difference(SCREAMING_SNAKE_CASE__   ,       SCREAMING_SNAKE_CASE__    )

											# pipeline 2

											_start_torch_memory_measurement()

											SCREAMING_SNAKE_CASE__						:       int           =   torch.Generator(device="""cpu"""    ).manual_seed(0    )

											SCREAMING_SNAKE_CASE__						:       Dict           =   floats_tensor((1, 3, 64, 64)   ,       rng=random.Random(0    )    ).to(SCREAMING_SNAKE_CASE__    )
											SCREAMING_SNAKE_CASE__						:       int           =   floats_tensor((1, 3, 2_56, 2_56)   ,       rng=random.Random(0    )    ).to(SCREAMING_SNAKE_CASE__    )
											SCREAMING_SNAKE_CASE__						:       int           =   floats_tensor((1, 3, 2_56, 2_56)   ,       rng=random.Random(1    )    ).to(SCREAMING_SNAKE_CASE__    )

											SCREAMING_SNAKE_CASE__						:       int           =   pipe_a(
											    prompt_embeds=SCREAMING_SNAKE_CASE__   ,       negative_prompt_embeds=SCREAMING_SNAKE_CASE__   ,       image=SCREAMING_SNAKE_CASE__   ,       mask_image=SCREAMING_SNAKE_CASE__   ,       original_image=SCREAMING_SNAKE_CASE__   ,       generator=SCREAMING_SNAKE_CASE__   ,       num_inference_steps=2   ,       output_type="""np"""   ,       )

											SCREAMING_SNAKE_CASE__						:       Dict           =   output.images[0]

											assert image.shape == (2_56, 2_56, 3)

											SCREAMING_SNAKE_CASE__						:       List[str]           =   torch.cuda.max_memory_allocated()
											assert mem_bytes < 4 * 10**9

											SCREAMING_SNAKE_CASE__						:       Any           =   load_numpy(
											    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/if/test_if_inpainting_superresolution_stage_II.npy"""    )
											assert_mean_pixel_difference(SCREAMING_SNAKE_CASE__   ,       SCREAMING_SNAKE_CASE__    )






def lowercase_  (   ):
							torch.cuda.empty_cache()
							torch.cuda.reset_max_memory_allocated()
							torch.cuda.reset_peak_memory_stats()




