






from abc import ABC, abstractmethod
from typing import List, Optional







class 	_UpperCAmelCase  (  _a		):




      """simple docstring"""



      def __init__(   self		):



          '''simple docstring'''
          # test for the above condition
          self.test()
      def 			snake_case_      (   self		):



          '''simple docstring'''
          lowerCAmelCase__   :str      							=				0
          lowerCAmelCase__   :Dict      							=				False
          while not completed:
              if counter == 1:
                  self.reset()
              lowerCAmelCase__   :Optional[int]      							=				self.advance()
              if not self.does_advance(snake_case_		):
                  raise Exception(
                      "Custom Constraint is not defined correctly. self.does_advance(self.advance()) must be true."		)

              lowerCAmelCase__							,lowerCAmelCase__							,lowerCAmelCase__   :Optional[int]      							=				self.update(snake_case_		)
              counter += 1

              if counter > 10_000:
                  raise Exception("update() does not fulfill the constraint."		)

          if self.remaining() != 0:
              raise Exception("Custom Constraint is not defined correctly."		)
      @abstractmethod
      def 			snake_case_      (   self		):



          '''simple docstring'''
          raise NotImplementedError(
              f'''{self.__class__} is an abstract class. Only classes inheriting this class can be called.'''		)
      @abstractmethod
      def 			snake_case_      (   self    ,	_lowerCAmelCase		):



          '''simple docstring'''
          raise NotImplementedError(
              f'''{self.__class__} is an abstract class. Only classes inheriting this class can be called.'''		)
      @abstractmethod
      def 			snake_case_      (   self    ,	_lowerCAmelCase		):



          '''simple docstring'''
          raise NotImplementedError(
              f'''{self.__class__} is an abstract class. Only classes inheriting this class can be called.'''		)
      @abstractmethod
      def 			snake_case_      (   self		):



          '''simple docstring'''
          raise NotImplementedError(
              f'''{self.__class__} is an abstract class. Only classes inheriting this class can be called.'''		)
      @abstractmethod
      def 			snake_case_      (   self		):



          '''simple docstring'''
          raise NotImplementedError(
              f'''{self.__class__} is an abstract class. Only classes inheriting this class can be called.'''		)



      @abstractmethod
      def 			snake_case_      (   self    ,	_lowerCAmelCase=False		):



          '''simple docstring'''
          raise NotImplementedError(
              f'''{self.__class__} is an abstract class. Only classes inheriting this class can be called.'''		)







class 	_UpperCAmelCase  (  _a		):




      """simple docstring"""



      def __init__(   self    ,	_lowerCAmelCase		):



          '''simple docstring'''
          super(snake_case_    ,	self		).__init__()

          if not isinstance(snake_case_    ,	snake_case_		) or len(snake_case_		) == 0:
              raise ValueError(f'''`token_ids` has to be a non-empty list, but is {token_ids}.'''		)
          if any((not isinstance(snake_case_    ,	snake_case_		) or token_id < 0) for token_id in token_ids		):
              raise ValueError(f'''Each list in `token_ids` has to be a list of positive integers, but is {token_ids}.'''		)

          lowerCAmelCase__   :Tuple      							=				token_ids

          lowerCAmelCase__   :List[Any]      							=				len(self.token_ids		)
          lowerCAmelCase__   :int      							=				-1  # the index of the currently fulfilled step
          lowerCAmelCase__   :List[Any]      							=				False
      def 			snake_case_      (   self		):



          '''simple docstring'''
          if self.completed:
              return None
          return self.token_ids[self.fulfilled_idx + 1]
      def 			snake_case_      (   self    ,	_lowerCAmelCase		):



          '''simple docstring'''
          if not isinstance(snake_case_    ,	snake_case_		):
              raise ValueError(f'''`token_id` has to be an `int`, but is {token_id} of type {type(snake_case_		)}'''		)

          if self.completed:
              return False

          return token_id == self.token_ids[self.fulfilled_idx + 1]
      def 			snake_case_      (   self    ,	_lowerCAmelCase		):



          '''simple docstring'''
          if not isinstance(snake_case_    ,	snake_case_		):
              raise ValueError(f'''`token_id` has to be an `int`, but is {token_id} of type {type(snake_case_		)}'''		)

          lowerCAmelCase__   :Optional[int]      							=				False
          lowerCAmelCase__   :Optional[int]      							=				False
          lowerCAmelCase__   :Tuple      							=				False

          if self.does_advance(snake_case_		):
              self.fulfilled_idx += 1
              lowerCAmelCase__   :Tuple      							=				True
              if self.fulfilled_idx == (self.seqlen - 1):
                  lowerCAmelCase__   :int      							=				True
              lowerCAmelCase__   :List[str]      							=				completed
          else:
              # failed to make progress.
              lowerCAmelCase__   :Optional[int]      							=				True
              self.reset()
          return stepped, completed, reset
      def 			snake_case_      (   self		):



          '''simple docstring'''
          lowerCAmelCase__   :Optional[int]      							=				False
          lowerCAmelCase__   :List[str]      							=				0
      def 			snake_case_      (   self		):



          '''simple docstring'''
          return self.seqlen - (self.fulfilled_idx + 1)



      def 			snake_case_      (   self    ,	_lowerCAmelCase=False		):



          '''simple docstring'''
          lowerCAmelCase__   :Union[str, Any]      							=				PhrasalConstraint(self.token_ids		)

          if stateful:
              lowerCAmelCase__   :int      							=				self.seqlen
              lowerCAmelCase__   :Any      							=				self.fulfilled_idx
              lowerCAmelCase__   :List[str]      							=				self.completed

          return new_constraint







class 	_UpperCAmelCase  :




      """simple docstring"""



      def __init__(   self    ,	_lowerCAmelCase    ,	_lowerCAmelCase=True		):



          '''simple docstring'''
          lowerCAmelCase__   :str      							=				max([len(snake_case_		) for one in nested_token_ids]		)

          lowerCAmelCase__   :List[str]      							=				{}
          for token_ids in nested_token_ids:
              lowerCAmelCase__   :List[Any]      							=				root
              for tidx, token_id in enumerate(snake_case_		):
                  if token_id not in level:
                      lowerCAmelCase__   :Any      							=				{}

                  lowerCAmelCase__   :str      							=				level[token_id]

          if no_subsets and self.has_subsets(snake_case_    ,	snake_case_		):
              raise ValueError(
                  "Each list in `nested_token_ids` can\'t be a complete subset of another list, but is"
                  f''' {nested_token_ids}.'''		)

          lowerCAmelCase__   :Dict      							=				root
      def 			snake_case_      (   self    ,	_lowerCAmelCase		):



          '''simple docstring'''
          lowerCAmelCase__   :str      							=				self.trie

          for current_token in current_seq:
              lowerCAmelCase__   :Dict      							=				start[current_token]

          lowerCAmelCase__   :Dict      							=				list(start.keys()		)

          return next_tokens
      def 			snake_case_      (   self    ,	_lowerCAmelCase		):



          '''simple docstring'''
          lowerCAmelCase__   :Dict      							=				self.next_tokens(snake_case_		)

          return len(snake_case_		) == 0
      def 			snake_case_      (   self    ,	_lowerCAmelCase		):



          '''simple docstring'''
          lowerCAmelCase__   :Dict      							=				list(root.values()		)
          if len(snake_case_		) == 0:
              return 1
          else:
              return sum([self.count_leaves(snake_case_		) for nn in next_nodes]		)



      def 			snake_case_      (   self    ,	_lowerCAmelCase    ,	_lowerCAmelCase		):



          '''simple docstring'''
          lowerCAmelCase__   :Optional[int]      							=				self.count_leaves(snake_case_		)
          return len(snake_case_		) != leaf_count







class 	_UpperCAmelCase  (  _a		):




      """simple docstring"""



      def __init__(   self    ,	_lowerCAmelCase		):



          '''simple docstring'''
          super(snake_case_    ,	self		).__init__()

          if not isinstance(snake_case_    ,	snake_case_		) or len(snake_case_		) == 0:
              raise ValueError(f'''`nested_token_ids` has to be a non-empty list, but is {nested_token_ids}.'''		)
          if any(not isinstance(snake_case_    ,	snake_case_		) for token_ids in nested_token_ids		):
              raise ValueError(f'''`nested_token_ids` has to be a list of lists, but is {nested_token_ids}.'''		)
          if any(
              any((not isinstance(snake_case_    ,	snake_case_		) or token_id < 0) for token_id in token_ids		)
              for token_ids in nested_token_ids		):
              raise ValueError(
                  f'''Each list in `nested_token_ids` has to be a list of positive integers, but is {nested_token_ids}.'''		)

          lowerCAmelCase__   :Union[str, Any]      							=				DisjunctiveTrie(snake_case_		)
          lowerCAmelCase__   :Optional[Any]      							=				nested_token_ids

          lowerCAmelCase__   :Tuple      							=				self.trie.max_height
          lowerCAmelCase__   :Union[str, Any]      							=				[]
          lowerCAmelCase__   :int      							=				False
      def 			snake_case_      (   self		):



          '''simple docstring'''
          lowerCAmelCase__   :Any      							=				self.trie.next_tokens(self.current_seq		)

          if len(snake_case_		) == 0:
              return None
          else:
              return token_list
      def 			snake_case_      (   self    ,	_lowerCAmelCase		):



          '''simple docstring'''
          if not isinstance(snake_case_    ,	snake_case_		):
              raise ValueError(f'''`token_id` is supposed to be type `int`, but is {token_id} of type {type(snake_case_		)}'''		)

          lowerCAmelCase__   :Optional[Any]      							=				self.trie.next_tokens(self.current_seq		)

          return token_id in next_tokens
      def 			snake_case_      (   self    ,	_lowerCAmelCase		):



          '''simple docstring'''
          if not isinstance(snake_case_    ,	snake_case_		):
              raise ValueError(f'''`token_id` is supposed to be type `int`, but is {token_id} of type {type(snake_case_		)}'''		)

          lowerCAmelCase__   :Any      							=				False
          lowerCAmelCase__   :Union[str, Any]      							=				False
          lowerCAmelCase__   :Dict      							=				False

          if self.does_advance(snake_case_		):
              self.current_seq.append(snake_case_		)
              lowerCAmelCase__   :List[str]      							=				True
          else:
              lowerCAmelCase__   :Tuple      							=				True
              self.reset()

          lowerCAmelCase__   :int      							=				self.trie.reached_leaf(self.current_seq		)
          lowerCAmelCase__   :Dict      							=				completed

          return stepped, completed, reset
      def 			snake_case_      (   self		):



          '''simple docstring'''
          lowerCAmelCase__   :Tuple      							=				False
          lowerCAmelCase__   :str      							=				[]
      def 			snake_case_      (   self		):



          '''simple docstring'''
          if self.completed:
              # since this can be completed without reaching max height
              return 0
          else:
              return self.seqlen - len(self.current_seq		)



      def 			snake_case_      (   self    ,	_lowerCAmelCase=False		):



          '''simple docstring'''
          lowerCAmelCase__   :Any      							=				DisjunctiveConstraint(self.token_ids		)

          if stateful:
              lowerCAmelCase__   :List[str]      							=				self.seqlen
              lowerCAmelCase__   :str      							=				self.current_seq
              lowerCAmelCase__   :Union[str, Any]      							=				self.completed

          return new_constraint







class 	_UpperCAmelCase  :




      """simple docstring"""



      def __init__(   self    ,	_lowerCAmelCase		):



          '''simple docstring'''
          lowerCAmelCase__   :Optional[int]      							=				constraints

          # max # of steps required to fulfill a given constraint
          lowerCAmelCase__   :Dict      							=				max([c.seqlen for c in constraints]		)
          lowerCAmelCase__   :List[Any]      							=				len(snake_case_		)
          lowerCAmelCase__   :int      							=				False

          self.init_state()
      def 			snake_case_      (   self		):



          '''simple docstring'''
          lowerCAmelCase__   :Tuple      							=				[]
          lowerCAmelCase__   :Any      							=				None
          lowerCAmelCase__   :Optional[Any]      							=				[constraint.copy(stateful=snake_case_		) for constraint in self.constraints]
      def 			snake_case_      (   self		):



          '''simple docstring'''
          lowerCAmelCase__   :List[Any]      							=				0
          if self.inprogress_constraint:
              # extra points for having a constraint mid-fulfilled
              add += self.max_seqlen - self.inprogress_constraint.remaining()

          return (len(self.complete_constraints		) * self.max_seqlen) + add
      def 			snake_case_      (   self		):



          '''simple docstring'''
          lowerCAmelCase__   :Optional[int]      							=				[]
          if self.inprogress_constraint is None:
              for constraint in self.pending_constraints:  # "pending" == "unfulfilled yet"
                  lowerCAmelCase__   :Optional[Any]      							=				constraint.advance()
                  if isinstance(snake_case_    ,	snake_case_		):
                      token_list.append(snake_case_		)
                  elif isinstance(snake_case_    ,	snake_case_		):
                      token_list.extend(snake_case_		)
          else:
              lowerCAmelCase__   :Dict      							=				self.inprogress_constraint.advance()
              if isinstance(snake_case_    ,	snake_case_		):
                  token_list.append(snake_case_		)
              elif isinstance(snake_case_    ,	snake_case_		):
                  token_list.extend(snake_case_		)

          if len(snake_case_		) == 0:
              return None
          else:
              return token_list
      def 			snake_case_      (   self    ,	_lowerCAmelCase		):



          '''simple docstring'''
          self.init_state()

          if token_ids is not None:
              for token in token_ids:
                  # completes or steps **one** constraint
                  lowerCAmelCase__							,lowerCAmelCase__   :List[str]      							=				self.add(snake_case_		)

                  # the entire list of constraints are fulfilled
                  if self.completed:
                      break
      def 			snake_case_      (   self    ,	_lowerCAmelCase		):



          '''simple docstring'''
          if not isinstance(snake_case_    ,	snake_case_		):
              raise ValueError(f'''`token_id` should be an `int`, but is `{token_id}`.'''		)

          lowerCAmelCase__							,lowerCAmelCase__   :int      							=				False, False

          if self.completed:
              lowerCAmelCase__   :List[Any]      							=				True
              lowerCAmelCase__   :List[Any]      							=				False
              return complete, stepped

          if self.inprogress_constraint is not None:
              # In the middle of fulfilling a constraint. If the `token_id` *does* makes an incremental progress to current
              # job, simply update the state

              lowerCAmelCase__							,lowerCAmelCase__							,lowerCAmelCase__   :Optional[int]      							=				self.inprogress_constraint.update(snake_case_		)
              if reset:
                  # 1. If the next token breaks the progress, then we must restart.
                  #     e.g. constraint = "I love pies" and sequence so far is "I love" but `token_id` == "books".

                  #     But that doesn't mean we self.init_state(), since we only reset the state for this particular
                  #     constraint, not the full list of constraints.

                  self.pending_constraints.append(self.inprogress_constraint.copy(stateful=snake_case_		)		)
                  lowerCAmelCase__   :List[Any]      							=				None

              if complete:
                  # 2. If the next token completes the constraint, move it to completed list, set
                  #     inprogress to None. If there are no pending constraints either, then this full list of constraints
                  #     is complete.

                  self.complete_constraints.append(self.inprogress_constraint		)
                  lowerCAmelCase__   :List[str]      							=				None

                  if len(self.pending_constraints		) == 0:
                      # we're done!
                      lowerCAmelCase__   :Any      							=				True

          else:
              # Not in the middle of fulfilling a constraint. So does this `token_id` helps us step towards any of our list
              # of constraints?

              for cidx, pending_constraint in enumerate(self.pending_constraints		):
                  if pending_constraint.does_advance(snake_case_		):
                      lowerCAmelCase__							,lowerCAmelCase__							,lowerCAmelCase__   :Dict      							=				pending_constraint.update(snake_case_		)

                      if not stepped:
                          raise Exception(
                              "`constraint.update(token_id)` is not yielding incremental progress, "
                              "even though `constraint.does_advance(token_id)` is true."		)

                      if complete:
                          self.complete_constraints.append(snake_case_		)
                          lowerCAmelCase__   :List[Any]      							=				None

                      if not complete and stepped:
                          lowerCAmelCase__   :List[Any]      							=				pending_constraint

                      if complete or stepped:
                          # If we made any progress at all, then it's at least not a "pending constraint".

                          lowerCAmelCase__   :str      							=				(
                              self.pending_constraints[:cidx] + self.pending_constraints[cidx + 1 :]
                          )

                          if len(self.pending_constraints		) == 0 and self.inprogress_constraint is None:
                              # If there's no longer any pending after this and no inprogress either, then we must be
                              # complete.

                              lowerCAmelCase__   :int      							=				True

                          break  # prevent accidentally stepping through multiple constraints with just one token.

          return complete, stepped



      def 			snake_case_      (   self    ,	_lowerCAmelCase=True		):



          '''simple docstring'''
          lowerCAmelCase__   :List[Any]      							=				ConstraintListState(self.constraints		)  # we actually never though self.constraints objects
          # throughout this process. So it's at initialization state.

          if stateful:
              lowerCAmelCase__   :int      							=				[
                  constraint.copy(stateful=snake_case_		) for constraint in self.complete_constraints
              ]
              if self.inprogress_constraint is not None:
                  lowerCAmelCase__   :int      							=				self.inprogress_constraint.copy(stateful=snake_case_		)
              lowerCAmelCase__   :Optional[Any]      							=				[constraint.copy() for constraint in self.pending_constraints]

          return new_state











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		)



