






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 				snake_case__							(							UpperCAmelCase							:  Union[dict, list, tuple, torch.Tensor]       ):
    lowerCAmelCase__   :Optional[int]      							=				[]
    if isinstance(UpperCamelCase__    ,			UpperCamelCase__       ):
        for v in tree.values():
            shapes.extend(_fetch_dims(UpperCamelCase__       )       )
    elif isinstance(UpperCamelCase__    ,			(list, tuple)       ):
        for t in tree:
            shapes.extend(_fetch_dims(UpperCamelCase__       )       )
    elif isinstance(UpperCamelCase__    ,			torch.Tensor       ):
        shapes.append(tree.shape       )
    else:
        raise ValueError("Not supported"       )

    return shapes


@torch.jit.ignore
def 				snake_case__							(							UpperCAmelCase							:  int    ,			UpperCAmelCase							:  Tuple[int, ...]       ):
    lowerCAmelCase__   :str      							=				[]
    for d in reversed(UpperCamelCase__       ):
        idx.append(flat_idx % d       )
        lowerCAmelCase__   :Any      							=				flat_idx // d

    return tuple(reversed(UpperCamelCase__       )       )


@torch.jit.ignore
def 				snake_case__							(							UpperCAmelCase							:  Sequence[int]    ,			UpperCAmelCase							:  Sequence[int]    ,			UpperCAmelCase							:  Sequence[int]    ,			UpperCAmelCase							:  Optional[Sequence[bool]] = None    ,			UpperCAmelCase							:  Optional[Sequence[bool]] = 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(UpperCAmelCase							:  List[bool]       ) -> None:
        lowerCAmelCase__   :str      							=				True
        for i in range(len(UpperCamelCase__       )       ):
            lowerCAmelCase__   :int      							=				-1 * (i + 1)
            l[reversed_idx] &= tally
            lowerCAmelCase__   :Dict      							=				l[reversed_idx]

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

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

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

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

    lowerCAmelCase__   :Tuple[slice, ...]      							=				tuple(UpperCamelCase__       )
    lowerCAmelCase__   :List[str]      							=				len(UpperCamelCase__       )

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

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

        lowerCAmelCase__   :Optional[Any]      							=				start[divergence_idx]
        return tuple(
            path + (slice(UpperCamelCase__    ,			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

        lowerCAmelCase__   :str      							=				end[divergence_idx]
        return tuple(
            path + (slice(UpperCamelCase__    ,			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()       )
        lowerCAmelCase__   :Optional[Any]      							=				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 				snake_case__							(							UpperCAmelCase							:  torch.Tensor    ,			UpperCAmelCase							:  int    ,			UpperCAmelCase							:  int    ,			UpperCAmelCase							:  int       ):
    lowerCAmelCase__   :str      							=				t.shape[:no_batch_dims]
    lowerCAmelCase__   :Dict      							=				list(_flat_idx_to_idx(UpperCamelCase__    ,			UpperCamelCase__       )       )
    # _get_minimal_slice_set is inclusive
    lowerCAmelCase__   :List[str]      							=				list(_flat_idx_to_idx(flat_end - 1    ,			UpperCamelCase__       )       )

    # Get an ordered list of slices to perform
    lowerCAmelCase__   :Any      							=				_get_minimal_slice_set(
        UpperCamelCase__    ,			UpperCamelCase__    ,			UpperCamelCase__    ,			)

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

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


def 				snake_case__							(							UpperCAmelCase							:  Callable    ,			UpperCAmelCase							:  Dict[str, Any]    ,			UpperCAmelCase							:  int    ,			UpperCAmelCase							:  int    ,			UpperCAmelCase							:  bool = False    ,			UpperCAmelCase							:  Any = None    ,			UpperCAmelCase							:  bool = False    ,			):
    if not (len(UpperCamelCase__       ) > 0):
        raise ValueError("Must provide at least one input"       )

    lowerCAmelCase__   :str      							=				[shape[:no_batch_dims] for shape in _fetch_dims(UpperCamelCase__       )]
    lowerCAmelCase__   :Any      							=				tuple([max(UpperCamelCase__       ) for s in zip(*UpperCamelCase__       )]       )

    def _prep_inputs(UpperCAmelCase							:  torch.Tensor       ) -> torch.Tensor:
        if not low_mem:
            if not sum(t.shape[:no_batch_dims]       ) == no_batch_dims:
                lowerCAmelCase__   :List[Any]      							=				t.expand(orig_batch_dims + t.shape[no_batch_dims:]       )
            lowerCAmelCase__   :Union[str, Any]      							=				t.reshape(-1    ,			*t.shape[no_batch_dims:]       )
        else:
            lowerCAmelCase__   :Optional[Any]      							=				t.expand(orig_batch_dims + t.shape[no_batch_dims:]       )
        return t

    lowerCAmelCase__   :Dict[str, Any]      							=				tensor_tree_map(_prep_inputs    ,			UpperCamelCase__       )
    lowerCAmelCase__   :str      							=				None
    if _out is not None:
        lowerCAmelCase__   :Union[str, Any]      							=				tensor_tree_map(lambda UpperCAmelCase       : t.view([-1] + list(t.shape[no_batch_dims:]       )       )    ,			_out       )

    lowerCAmelCase__   :int      							=				1
    for d in orig_batch_dims:
        flat_batch_dim *= d

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

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

    lowerCAmelCase__   :int      							=				0
    lowerCAmelCase__   :Optional[int]      							=				prepped_outputs
    for _ in range(UpperCamelCase__       ):
        # Chunk the input
        if not low_mem:
            lowerCAmelCase__   :str      							=				_select_chunk
        else:
            lowerCAmelCase__   :Union[str, Any]      							=				partial(
                _chunk_slice    ,			flat_start=UpperCamelCase__    ,			flat_end=min(UpperCamelCase__    ,			i + chunk_size       )    ,			no_batch_dims=len(UpperCamelCase__       )    ,			)

        lowerCAmelCase__   :Dict[str, Any]      							=				tensor_tree_map(UpperCamelCase__    ,			UpperCamelCase__       )

        # Run the layer on the chunk
        lowerCAmelCase__   :Any      							=				layer(**UpperCamelCase__       )

        # Allocate space for the output
        if out is None:
            lowerCAmelCase__   :List[str]      							=				tensor_tree_map(lambda UpperCAmelCase       : t.new_zeros((flat_batch_dim,) + t.shape[1:]       )    ,			UpperCamelCase__       )

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

            def assign(UpperCAmelCase							:  dict    ,			UpperCAmelCase							:  dict       ) -> None:
                for k, v in da.items():
                    if isinstance(UpperCamelCase__    ,			UpperCamelCase__       ):
                        assign(UpperCamelCase__    ,			da[k]       )
                    else:
                        if _add_into_out:
                            v[i : i + chunk_size] += da[k]
                        else:
                            lowerCAmelCase__   :List[Any]      							=				da[k]

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

        i += chunk_size

    lowerCAmelCase__   :Optional[int]      							=				tensor_tree_map(lambda UpperCAmelCase       : t.view(orig_batch_dims + t.shape[1:]       )    ,			UpperCamelCase__       )

    return out







class 	_UpperCAmelCase  :




      """simple docstring"""



      def __init__(   self    ,	_lowerCAmelCase = 512    ,	):



          '''simple docstring'''
          lowerCAmelCase__   :List[Any]      							=				max_chunk_size
          lowerCAmelCase__   :Optional[int]      							=				None
          lowerCAmelCase__   :Optional[tuple]      							=				None
      def 			snake_case_      (   self    ,	_lowerCAmelCase    ,	_lowerCAmelCase    ,	_lowerCAmelCase		):



          '''simple docstring'''
          logging.info("Tuning chunk size..."		)

          if min_chunk_size >= self.max_chunk_size:
              return min_chunk_size

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

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

          lowerCAmelCase__   :Optional[Any]      							=				0
          lowerCAmelCase__   :Optional[int]      							=				len(__a		) - 1
          while i > min_viable_chunk_size_index:
              lowerCAmelCase__   :List[str]      							=				test_chunk_size(candidates[i]		)
              if not viable:
                  lowerCAmelCase__   :int      							=				(min_viable_chunk_size_index + i) // 2
              else:
                  lowerCAmelCase__   :Union[str, Any]      							=				i
                  lowerCAmelCase__   :Dict      							=				(i + len(__a		) - 1) // 2

          return candidates[min_viable_chunk_size_index]
      def 			snake_case_      (   self    ,	_lowerCAmelCase    ,	_lowerCAmelCase		):



          '''simple docstring'''
          lowerCAmelCase__   :Dict      							=				True
          for aa, aa in zip(__a    ,	__a		):
              assert type(__a		) == type(__a		)
              if isinstance(__a    ,	(list, tuple)		):
                  consistent &= self._compare_arg_caches(__a    ,	__a		)
              elif isinstance(__a    ,	__a		):
                  lowerCAmelCase__   :str      							=				[v for _, v in sorted(aa.items()    ,	key=lambda _lowerCAmelCase		: x[0]		)]
                  lowerCAmelCase__   :List[str]      							=				[v for _, v in sorted(aa.items()    ,	key=lambda _lowerCAmelCase		: x[0]		)]
                  consistent &= self._compare_arg_caches(__a    ,	__a		)
              else:
                  consistent &= aa == aa

          return consistent



      def 			snake_case_      (   self    ,	_lowerCAmelCase    ,	_lowerCAmelCase    ,	_lowerCAmelCase    ,	):



          '''simple docstring'''
          lowerCAmelCase__   :List[str]      							=				True
          lowerCAmelCase__   :tuple      							=				tree_map(lambda _lowerCAmelCase		: a.shape if isinstance(__a    ,	torch.Tensor		) else a    ,	__a    ,	__a		)
          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(__a		)
              lowerCAmelCase__   :Any      							=				self._compare_arg_caches(self.cached_arg_data    ,	__a		)
          else:
              # Otherwise, we can reuse the precomputed value
              lowerCAmelCase__   :int      							=				False

          if not consistent:
              lowerCAmelCase__   :Optional[Any]      							=				self._determine_favorable_chunk_size(
                  __a    ,	__a    ,	__a    ,	)
              lowerCAmelCase__   :List[Any]      							=				arg_data

          assert self.cached_chunk_size is not None

          return self.cached_chunk_size











import tempfile

import numpy as np
import torch
from transformers import AutoTokenizer, TaEncoderModel

from diffusers import DDPMScheduler, UNetaDConditionModel
from diffusers.models.attention_processor import AttnAddedKVProcessor
from diffusers.pipelines.deepfloyd_if import IFWatermarker
from diffusers.utils.testing_utils import torch_device

from ..test_pipelines_common import to_np







class 	_UpperCAmelCase  :




      """simple docstring"""



      def 			snake_case_      (   self		):



          '''simple docstring'''
          torch.manual_seed(0		)
          lowerCAmelCase__   :List[str]      							=				TaEncoderModel.from_pretrained("hf-internal-testing/tiny-random-t5"		)

          torch.manual_seed(0		)
          lowerCAmelCase__   :List[Any]      							=				AutoTokenizer.from_pretrained("hf-internal-testing/tiny-random-t5"		)

          torch.manual_seed(0		)
          lowerCAmelCase__   :Tuple      							=				UNetaDConditionModel(
              sample_size=32    ,	layers_per_block=1    ,	block_out_channels=[32, 64]    ,	down_block_types=[
                  "ResnetDownsampleBlock2D",
                  "SimpleCrossAttnDownBlock2D",
              ]    ,	mid_block_type="UNetMidBlock2DSimpleCrossAttn"    ,	up_block_types=["SimpleCrossAttnUpBlock2D", "ResnetUpsampleBlock2D"]    ,	in_channels=3    ,	out_channels=6    ,	cross_attention_dim=32    ,	encoder_hid_dim=32    ,	attention_head_dim=8    ,	addition_embed_type="text"    ,	addition_embed_type_num_heads=2    ,	cross_attention_norm="group_norm"    ,	resnet_time_scale_shift="scale_shift"    ,	act_fn="gelu"    ,	)
          unet.set_attn_processor(AttnAddedKVProcessor()		)  # For reproducibility tests

          torch.manual_seed(0		)
          lowerCAmelCase__   :Dict      							=				DDPMScheduler(
              num_train_timesteps=1_000    ,	beta_schedule="squaredcos_cap_v2"    ,	beta_start=0.0001    ,	beta_end=0.02    ,	thresholding=_lowerCAmelCase    ,	dynamic_thresholding_ratio=0.95    ,	sample_max_value=1.0    ,	prediction_type="epsilon"    ,	variance_type="learned_range"    ,	)

          torch.manual_seed(0		)
          lowerCAmelCase__   :List[str]      							=				IFWatermarker()

          return {
              "text_encoder": text_encoder,
              "tokenizer": tokenizer,
              "unet": unet,
              "scheduler": scheduler,
              "watermarker": watermarker,
              "safety_checker": None,
              "feature_extractor": None,
          }
      def 			snake_case_      (   self		):



          '''simple docstring'''
          torch.manual_seed(0		)
          lowerCAmelCase__   :int      							=				TaEncoderModel.from_pretrained("hf-internal-testing/tiny-random-t5"		)

          torch.manual_seed(0		)
          lowerCAmelCase__   :Dict      							=				AutoTokenizer.from_pretrained("hf-internal-testing/tiny-random-t5"		)

          torch.manual_seed(0		)
          lowerCAmelCase__   :Union[str, Any]      							=				UNetaDConditionModel(
              sample_size=32    ,	layers_per_block=[1, 2]    ,	block_out_channels=[32, 64]    ,	down_block_types=[
                  "ResnetDownsampleBlock2D",
                  "SimpleCrossAttnDownBlock2D",
              ]    ,	mid_block_type="UNetMidBlock2DSimpleCrossAttn"    ,	up_block_types=["SimpleCrossAttnUpBlock2D", "ResnetUpsampleBlock2D"]    ,	in_channels=6    ,	out_channels=6    ,	cross_attention_dim=32    ,	encoder_hid_dim=32    ,	attention_head_dim=8    ,	addition_embed_type="text"    ,	addition_embed_type_num_heads=2    ,	cross_attention_norm="group_norm"    ,	resnet_time_scale_shift="scale_shift"    ,	act_fn="gelu"    ,	class_embed_type="timestep"    ,	mid_block_scale_factor=1.414    ,	time_embedding_act_fn="gelu"    ,	time_embedding_dim=32    ,	)
          unet.set_attn_processor(AttnAddedKVProcessor()		)  # For reproducibility tests

          torch.manual_seed(0		)
          lowerCAmelCase__   :Tuple      							=				DDPMScheduler(
              num_train_timesteps=1_000    ,	beta_schedule="squaredcos_cap_v2"    ,	beta_start=0.0001    ,	beta_end=0.02    ,	thresholding=_lowerCAmelCase    ,	dynamic_thresholding_ratio=0.95    ,	sample_max_value=1.0    ,	prediction_type="epsilon"    ,	variance_type="learned_range"    ,	)

          torch.manual_seed(0		)
          lowerCAmelCase__   :str      							=				DDPMScheduler(
              num_train_timesteps=1_000    ,	beta_schedule="squaredcos_cap_v2"    ,	beta_start=0.0001    ,	beta_end=0.02    ,	)

          torch.manual_seed(0		)
          lowerCAmelCase__   :List[str]      							=				IFWatermarker()

          return {
              "text_encoder": text_encoder,
              "tokenizer": tokenizer,
              "unet": unet,
              "scheduler": scheduler,
              "image_noising_scheduler": image_noising_scheduler,
              "watermarker": watermarker,
              "safety_checker": None,
              "feature_extractor": None,
          }
      def 			snake_case_      (   self		):



          '''simple docstring'''
          lowerCAmelCase__   :Optional[Any]      							=				self.get_dummy_components()
          lowerCAmelCase__   :Any      							=				self.pipeline_class(**_lowerCAmelCase		)
          pipe.to(_lowerCAmelCase		)
          pipe.set_progress_bar_config(disable=_lowerCAmelCase		)

          lowerCAmelCase__   :int      							=				self.get_dummy_inputs(_lowerCAmelCase		)

          lowerCAmelCase__   :Optional[int]      							=				inputs["prompt"]
          lowerCAmelCase__   :Tuple      							=				inputs["generator"]
          lowerCAmelCase__   :List[Any]      							=				inputs["num_inference_steps"]
          lowerCAmelCase__   :Tuple      							=				inputs["output_type"]

          if "image" in inputs:
              lowerCAmelCase__   :Dict      							=				inputs["image"]
          else:
              lowerCAmelCase__   :Union[str, Any]      							=				None

          if "mask_image" in inputs:
              lowerCAmelCase__   :int      							=				inputs["mask_image"]
          else:
              lowerCAmelCase__   :str      							=				None

          if "original_image" in inputs:
              lowerCAmelCase__   :List[Any]      							=				inputs["original_image"]
          else:
              lowerCAmelCase__   :Dict      							=				None

          lowerCAmelCase__							,lowerCAmelCase__   :int      							=				pipe.encode_prompt(_lowerCAmelCase		)

          # inputs with prompt converted to embeddings
          lowerCAmelCase__   :Optional[Any]      							=				{
              "prompt_embeds": prompt_embeds,
              "negative_prompt_embeds": negative_prompt_embeds,
              "generator": generator,
              "num_inference_steps": num_inference_steps,
              "output_type": output_type,
          }

          if image is not None:
              lowerCAmelCase__   :List[Any]      							=				image

          if mask_image is not None:
              lowerCAmelCase__   :Any      							=				mask_image

          if original_image is not None:
              lowerCAmelCase__   :Tuple      							=				original_image

          # set all optional components to None
          for optional_component in pipe._optional_components:
              setattr(_lowerCAmelCase    ,	_lowerCAmelCase    ,	_lowerCAmelCase		)

          lowerCAmelCase__   :Optional[int]      							=				pipe(**_lowerCAmelCase		)[0]

          with tempfile.TemporaryDirectory() as tmpdir:
              pipe.save_pretrained(_lowerCAmelCase		)
              lowerCAmelCase__   :str      							=				self.pipeline_class.from_pretrained(_lowerCAmelCase		)
              pipe_loaded.to(_lowerCAmelCase		)
              pipe_loaded.set_progress_bar_config(disable=_lowerCAmelCase		)

          pipe_loaded.unet.set_attn_processor(AttnAddedKVProcessor()		)  # For reproducibility tests

          for optional_component in pipe._optional_components:
              self.assertTrue(
                  getattr(_lowerCAmelCase    ,	_lowerCAmelCase		) is None    ,	f'''`{optional_component}` did not stay set to None after loading.'''    ,	)

          lowerCAmelCase__   :Any      							=				self.get_dummy_inputs(_lowerCAmelCase		)

          lowerCAmelCase__   :Tuple      							=				inputs["generator"]
          lowerCAmelCase__   :Union[str, Any]      							=				inputs["num_inference_steps"]
          lowerCAmelCase__   :str      							=				inputs["output_type"]

          # inputs with prompt converted to embeddings
          lowerCAmelCase__   :Union[str, Any]      							=				{
              "prompt_embeds": prompt_embeds,
              "negative_prompt_embeds": negative_prompt_embeds,
              "generator": generator,
              "num_inference_steps": num_inference_steps,
              "output_type": output_type,
          }

          if image is not None:
              lowerCAmelCase__   :Tuple      							=				image

          if mask_image is not None:
              lowerCAmelCase__   :str      							=				mask_image

          if original_image is not None:
              lowerCAmelCase__   :Union[str, Any]      							=				original_image

          lowerCAmelCase__   :List[Any]      							=				pipe_loaded(**_lowerCAmelCase		)[0]

          lowerCAmelCase__   :Dict      							=				np.abs(to_np(_lowerCAmelCase		) - to_np(_lowerCAmelCase		)		).max()
          self.assertLess(_lowerCAmelCase    ,	1e-4		)



      def 			snake_case_      (   self		):



          '''simple docstring'''
          lowerCAmelCase__   :Tuple      							=				self.get_dummy_components()
          lowerCAmelCase__   :Optional[int]      							=				self.pipeline_class(**_lowerCAmelCase		)
          pipe.to(_lowerCAmelCase		)
          pipe.set_progress_bar_config(disable=_lowerCAmelCase		)

          lowerCAmelCase__   :Optional[int]      							=				self.get_dummy_inputs(_lowerCAmelCase		)
          lowerCAmelCase__   :List[str]      							=				pipe(**_lowerCAmelCase		)[0]

          with tempfile.TemporaryDirectory() as tmpdir:
              pipe.save_pretrained(_lowerCAmelCase		)
              lowerCAmelCase__   :Dict      							=				self.pipeline_class.from_pretrained(_lowerCAmelCase		)
              pipe_loaded.to(_lowerCAmelCase		)
              pipe_loaded.set_progress_bar_config(disable=_lowerCAmelCase		)

          pipe_loaded.unet.set_attn_processor(AttnAddedKVProcessor()		)  # For reproducibility tests

          lowerCAmelCase__   :Optional[int]      							=				self.get_dummy_inputs(_lowerCAmelCase		)
          lowerCAmelCase__   :List[Any]      							=				pipe_loaded(**_lowerCAmelCase		)[0]

          lowerCAmelCase__   :str      							=				np.abs(to_np(_lowerCAmelCase		) - to_np(_lowerCAmelCase		)		).max()
          self.assertLess(_lowerCAmelCase    ,	1e-4		)



